anchor.diff:              State
       end.
   
  -%% monitor worker and put it into the Working queue
  +%% @doc monitor a working process
   -spec monitor_worker(pid(), comm:my_pid(), state()) -> state().
   monitor_worker(Pid, Source, {Max, Working, Waiting}) ->
       monitor(process, Pid),
       {Max, [{Pid, Source} | Working], Waiting}.
   
  -%% remove worker from Working queue and start a waiting job if present
  +%% @doc cleanup wortker after it is finished.
  +%%      when a worker finishes (either normal or crashed) it is removed from the
  +%%      working queue and the next wainting job is started.
   -spec cleanup_worker(pid(), state()) -> state().
   cleanup_worker(Pid, {Max, Working, Waiting}) ->
       NewWorking = lists:keydelete(Pid, 1, Working),
  
positive.code: work(DHTNodeGroup, {_Round, map, {jsanon, FunBin}, Data}) ->
      %% ?TRACE("worker: should apply ~p to ~p~n", [FunBin, Data]),
      {ok, VM} = js_driver:new(),
      return(DHTNodeGroup, lists:flatten([apply_js(FunBin, X, VM) || X <- Data]));
  work(DHTNodeGroup, {_Round, reduce, {jsanon, FunBin}, Data}) ->
      {ok, VM} = js_driver:new(),
      return(DHTNodeGroup, apply_js(FunBin, Data, VM)).
  
  apply_erl(Fun, Data) ->
      Fun(Data).
  
  apply_js(FunBin, Data, VM) ->
      AutoJS = define_auto_js(FunBin, Data),
      {ok, Result} = js:eval(VM, AutoJS),
      Result.
  
  define_auto_js(FunBin, Args) ->
      EncodedArgs = encode_args(Args, []),
      iolist_to_binary([FunBin, "(", EncodedArgs, ")"]).
  
  encode_args([], Acc) ->
      lists:reverse(Acc);
  encode_args([H | []], Acc) ->
      encode_args([], [value_to_json(H) | Acc]);
  encode_args([H | T], Acc) ->
      encode_args(T, [[value_to_json(H), ","] | Acc]).
  
  value_to_json({Key, Value}) ->
      js_mochijson2:encode({struct, [{Key, value_to_json(Value)}]});
  value_to_json([{_, _} | _] = List) ->
      js_mochijson2:encode({struct, [{K, value_to_json(V)} || {K, V} <- List]});
  value_to_json(List) when is_list(List) ->
      js_mochijson2:encode(List);
  value_to_json(X) ->
      js_mochijson2:encode(X).
  
  %% send results back to wpool
  -spec return(pid_groups:groupname(), any()) -> ok.
  
negative.code: -type(mr_job() :: {Round::pos_integer(),
                     map | reduce,
                     {erlanon | jsanon, binary()},
                     Data::[tuple()]}).
  
  -type(generic_job() :: {erlanon | jsanon, binary(), [tuple()]}).
  