anchor.diff:  
   handle_event({register_mapper, Id, MapperPid}, #state{mapper_data=MapperData}=State) ->
       MapperData0 = case lists:keyfind(Id, 1, MapperData) of
  -        {Id, MapperProps} -> lists:keyreplace(Id, 1, MapperData, {Id, MapperProps ++ [{pid, MapperPid}]});
  +        {Id, MapperProps} -> lists:keyreplace(Id, 1, MapperData, {Id, [{pid, MapperPid}|MapperProps]});
           false -> MapperData
       end,
       MapperData1 = MapperData0 ++ [{MapperPid, Id}],
  
positive.code:             end;
          false -> throw(bad_mapper_props_no_id)
      end.
  
  %% log_event(Event, Args) ->
  %%     Template0 = lists:flatten(["~p," || _X <- lists:seq(1, length(Args) + 2)]),
  %%     Template = string:left(Template0, length(Template0) - 1) ++ "\n",
  %%     Msg = io_lib:format(Template, [timestamp(), Event] ++ Args),
  %%     file:write_file("/tmp/riak_kv_map_phase.csv", Msg, [append]).
  
  %% timestamp() ->
  %%     TS = {_,_,Micro} = os:timestamp(),
  %%     {{Year,Month,Day},{Hour,Minute,Second}} = calendar:now_to_universal_time(TS),
  %%     Mstr = element(Month,{"Jan","Feb","Mar","Apr","May","Jun","Jul", "Aug","Sep","Oct","Nov","Dec"}),
  %%     lists:flatten(io_lib:format("~2w ~s ~4w ~2w:~2..0w:~2..0w.~6..0w", [Day,Mstr,Year,Hour,Minute,Second,Micro])).
  
negative.code: 
  start_mappers([], _QTerm, Accum, FsmKeys) ->
      {Accum, FsmKeys};
  start_mappers([{Partition, Inputs}|T], QTerm, Accum, FsmKeys) ->
      case riak_kv_map_master:new_mapper(Partition, QTerm, Inputs, self()) of
          {ok, FSM} ->
              Accum1 = dict:store(FSM, length(Inputs), Accum),
              start_mappers(T, QTerm, Accum1, FsmKeys ++ [{FSM, [{keys, {Partition, Inputs}}]}]);
          Error ->
  