basho/riak_pipe — pull request #59 by beerriot (2012-11-03 14:53:10+00:00)
Title: "Sink type" that can provide backpressure

At the head of a pipe, and between its fittings, passing inputs/outputs is a blocking operation, which allows it to provide backpressure to keep downstream processes from being overwhelmed. Before this commit, there was no such behavior between the tail of the pipe and its sink. For very large Riak KV MapReduce jobs, it was common for the sink to become overwhelmed either with results or with log messages during failures.

This PR introduces blocking sink-delivery. It does this by providing the ability to specify the "type" of the sink, which indicates the method by which messages are delivered.

At this time there are two types:

* `raw`: The behavior that existed before this commit, and the default if none is specified. Messages are delivered by simply sending them to the sink process (`Sink ! Message.` in Erlang).

* `{fsm_sync, Timeout}`: Messages are delivered as synchronous `gen_fsm` events (`gen_fsm:sync_send_event(Sink, Message, Timeout)`).

The type is set in the `Options` parameter of `riak_pipe:exec/2` via the `sink_type` property: `riak_pipe:exec(Spec, [{sink_type, {fsm_sync, infinity}}|OtherOptions])`.

The spec of `riak_sink:*` has changed to facilitate error detection. Instead of returning the message it sent, it now returns the atom `ok` on success (always for `raw` sink types), or `{error, Reason}` for failure. Current `Reason`s are `timeout` if the request times out, and `sink_died` if the sink was gone before the call or vanished before replying.

New tests for this facility are in `riak_pipe:sink_type_test_/0`. A mock sink FSM for testing is included as `test/riak_pipe_test_sink_fsm.erl`.

Notes:

* Yes, synchronous delivery of each output is likely to introduce additional latency. This latency is likely more desirable than the alternative of unbounded memory consumption.

* Yes, synchronous log delivery is a hinderance to logging lots of messages. Again, it's likely more desirable than blowing up a sink with messages from many concurrent log producers. Additionally, the synchronicity is only encountered if the trace is actually delivered, and not if it is filtered out. Finally, the only current use of logging is in Riak KV's MapReduce, where it is used to deliver errors that cause the pipe to teardown immediately. Once one log is delivered there, the rate at which others can be delivered is not relevant.

* Synchronous delivery of the single eoi message seems unnecessary, but it makes the API more predictable.