anchor.diff:  
   -record(pipe_ctx, {pipe,     % pipe handling mapred request
                      ref,      % easier-access ref/reqid
  +                   timer,    % ref() for timeout send_after
                      sender}). % {pid(), monitor()} of process sending inputs
   
   -define(PROTO_MAJOR, 1).
  
positive.code:                          req_ctx=#pipe_ctx{ref=ReqId}=PipeCtx}) ->
      case encode_mapred_phase([Res], ContentType) of
          {error, Reason} ->
              riak_pipe:destroy(PipeCtx#pipe_ctx.pipe),
              case PipeCtx#pipe_ctx.sender of
                  {SenderPid, _} ->
  
negative.code: %% PIPE Handle response from mapred_stream
  handle_info(#pipe_eoi{ref=ReqId},
              State=#state{req=#rpbmapredreq{},
                           req_ctx=#pipe_ctx{ref=ReqId,
                                             timer=Timer}}) ->
      NewState = send_msg(#rpbmapredresp{done = 1}, State),
      erlang:cancel_timer(Timer),
      {noreply, NewState#state{req = undefined, req_ctx = undefined}};
  
  handle_info(#pipe_result{ref=ReqId, from=PhaseId, result=Res},
  