anchor.diff:      end.
   
   %% actual worker functions
  -
  +-spec init_worker(pid_groups:groupname(), job()) -> {ok, pid()}.
   init_worker(DHTNodeGroup, Workload) ->
       Pid = spawn_link(?MODULE, work, [DHTNodeGroup, Workload]),
       {ok, Pid}.
   
  +%% do the actual work
  +%% TODO join pid_group as worker_RANDOMID so more than one worker can be active
  +-spec work(pid_groups:groupname(), job()) -> ok.
   work(DHTNodeGroup, {_Round, map, {erlanon, FunBin}, _Keep, Data}) ->
       pid_groups:join_as(DHTNodeGroup, worker),
       %% ?TRACE("worker: should apply ~p to ~p~n", [FunBin, Data]),
  
positive.code:             State
      end.
  
  monitor_worker(Pid, Source, {Max, Working, Waiting}) ->
      monitor(process, Pid),
      {Max, [{Pid, Source} | Working], Waiting}.
  
  cleanup_worker(Pid, {Max, Working, Waiting}) ->
      NewWorking = lists:keydelete(Pid, 1, Working),
      case length(Waiting) of
  
negative.code: 
  -type(job() :: mr_job() | generic_job()).
  
  %% running jobs are defined by the pid the worker process has and the source of
  %% the job
  -type(active_jobs() :: [{pid(), comm:mypid()}]).
  
  %% a waiting job is defined by the source and the job spec
  -type(waiting_jobs() :: [{comm:mypid(), job()}]).
  
  -type(state() :: {MaxWorkers::pos_integer(), active_jobs(), waiting_jobs()}).
  
  