anchor.diff:  
   on({bulk_distribute, _Id, Interval,
       {mr, job, JobId, Master, Client, Job, InitalData}, _Parents}, State) ->
  +    %% this message starts the worker supervisor and adds a job specific state
  +    %% to the dht node
       ?TRACE("mr_~s on ~p: received job with initial data: ~p...~n",
              [JobId, self(), InitalData]),
  -    %% @doc
  -    %% this message starts the worker supervisor and adds a job specific state
  -    %% to the dht node
       JobState = mr_state:new(JobId, Client, Master, InitalData, Job, Interval),
       %% send acc to master
       comm:send(Master, {mr, phase_completed, Interval}),
       dht_node_state:set_mr_state(State, JobId, JobState);
   
   on({mr, phase_result, JobId, {work_done, Data}, Range, Round}, State) ->
  +    %% processing of phase results from worker.
  +    %% distribute data and start sync phase
       ?TRACE("mr_~s on ~p: received phase results: ~p...~ndistributing...~n",
              [JobId, self(), ets:tab2list(Data)]),
       MRState = dht_node_state:get_mr_state(State, JobId),
  
positive.code: 
  on({bulk_distribute, _Id, Interval,
     {mr, next_phase_data, JobId, AckRange, Data, Round}, _Parents}, State) ->
      ?TRACE("mr_~s on ~p: received next phase data for ~p: ~p~n",
             [JobId, self(), Interval, Data]),
      NewMRState = mr_state:add_data_to_phase(dht_node_state:get_mr_state(State,
  
negative.code:     dht_node_state:set_mr_state(State, JobId, NewMRState);
  
  on({mr, phase_result, JobId, {worker_died, Reason, _Round}, Range}, State) ->
      %% processing of a failed worker result.
      %% for now abort the job
      ?TRACE("runtime error in phase...terminating job~n", []),
      MRState = dht_node_state:get_mr_state(State, JobId),
      Master = mr_state:get(MRState, master),
  