anchor.diff:  %% Mochijson2 conversion will fail on the bare proplist, but will
   %% succeed at encoding this form.
   %%
  -%% The exception to the batching controls is handoff.  Whenever a
  -%% worker receives handoff from another worker, it immediately reduces
  -%% the concatenation of the two inputs.
  -%%
   %% If no inputs are received before eoi, this fitting evaluated the
   %% function once, with an empty list as `Inputs'.
   %%
  
positive.code:     %% just send state of reduce so far
      {ok, Acc ++ lists:reverse(InAcc)}.
  
  %% @doc Handoff simply concatenates the accumulator from the remote
  %% worker with the accumulator from this worker, and immediately
  %% reduces the list.
  -spec handoff(list(), state()) -> {ok, state()}.
  handoff(HandoffAcc, #state{acc=Acc}=State) ->
      %% for each Acc, add to local accs;
      NewAcc = handoff_acc(HandoffAcc, Acc, State),
      {ok, State#state{acc=NewAcc}}.
  
  -spec handoff_acc([term()], [term()], state()) -> [term()].
  handoff_acc(HandoffAcc, LocalAcc, State) ->
      InAcc = LocalAcc++HandoffAcc,
      reduce(InAcc, State, "reducing handoff").
  
  %% @doc Actually evaluate the aggregation function.
  -spec reduce([term()], state(), string()) ->
  
negative.code:          validate_arg/1]).
  -export([reduce_compat/1]).
  -export([no_input_run_reduce_once/0]).
  
  -ifdef(TEST).
  -include_lib("eunit/include/eunit.hrl").
  -endif.
  
  -include_lib("riak_pipe/include/riak_pipe.hrl").
  -include_lib("riak_pipe/include/riak_pipe_log.hrl").
  