basho/riak_kv — pull request #407 by beerriot (2012-10-19 20:47:59+00:00)
Title: separate the socket handler and sink processes in MR endpoints

I think this is the correct fix for #366.

Prior to this commit, the process holding the HTTP/PB socket was also the process acting as the Pipe sink for MapReduce requests. This led to many problems, including overwhelming mailboxes with error logging, and extra unexpected messages that had to be ignored if the socket was kept open to handle other requests.

This commit creates a new process to act as the sink, and sets up a request-response interface between them. The sink process spools pipe output until the socket-handling process asks for them.

This removes our dependence on the undocumented `mochiweb_request_force_close` feature, which was used to forcibly close an HTTP connection after a MapReduce failure to avoid incorrect 400 errors later (introduced in #224)

I've made this PR against 1.2, in case we want to include it in a point release, but the patch should be easily rebase-able onto master.

Some why's (structure):

* *Why start these sinks under a supervisor?* I agree, it seems like a little bit of overkill. Mostly, we want to make sure they're tied to the app's supervision tree. They could just be linked directly to the processes handling the sockets, but then those processes would have to trap exits in order to be able to provide error messages to their clients. However, the mapred modules are not the only handlers in those processes, so others would have to care about whether exits were trapped as well (or we'd have to be super careful about cleaning up). Putting the sinks under a supervisor also makes it super easy to check for leaks: `supervisor:which_children(riak_kv_mrc_sink_sup)`.

* *What's with the `self()`-messaging `#sink_part{}` in `riak_kv_pb_mapred`?* The module has to operate asynchronously, in order to be able to watch for a timeout message. The design of `riak_api` only allows one message to be sent to the client per erlang message to the process. The `#rbpmapredresp{}` record only allows one phase's outputs per message. The `riak_kv_mrc_sink` fsm might send results for multiple phases in one message. Thus, we must break them down to allow multiple messages to be sent to the client.

* *If the point was to get rid of extra messages, what's with the speculative receives in the cleanup?* It is still possible that a reply from `riak_kv_mrc_sink` might arrive after our timeout message arrives, for example. We can't let it slip. The difference between this speculative receiving and other attempts, is that we know that there is at most one message waiting, and that it will have been delivered before we check here, if it will be delivered at all.

* *Why am I still getting "fitting was gone before startup" messages in the log?* This patch does not target those message. This one is aimed at the "unrecognized message" errors coming out of `riak_api`.

Some what's (testing, mostly):

* *What has been tested already?* I've been running the Java client unit tests via `riak_test` against this branch, as they're good at finding these kinds of errors. There are loadgen tests in issues attached to #366 that should be tried as well.

* *How does this affect performance?* I don't know. It does introduce an extra process for results to hop through, so copying them could make things slower. The `basho_bench` tests for mapred should be run.

* *Are there any compatibility concerns?* Because of the buffering that `riak_kv_mrc_sink` is doing, it's possible that clients will once again see multiple results per streaming message, as they did before the transition to pipe. This *should not* affect compatibility, but it's something to watch for. (It also might improve performance from some perspectives.)