anchor.diff:      dht_node_state:set_mr_state(State, JobId, NewMRState);
   
   on({mr, next_phase, JobId, Round, _DeliveryInterval}, State) ->
  +    %% master started next round.
       ?TRACE("master initiated phase ~p in ~p ~p~n",
                 [Round, JobId, self()]),
       work_on_phase(JobId, State, Round),
       State;
   
   on({mr, terminate_job, JobId, _DeliveryInterval}, State) ->
  +    %% master wants to terminate job.
       MRState = dht_node_state:get_mr_state(State, JobId),
       _ = mr_state:clean_up(MRState),
       dht_node_state:delete_mr_state(State, JobId);
  
positive.code: 
  on({bulk_distribute, _Id, Interval,
      {mr, job, JobId, Master, Client, Job, InitalData}, _Parents}, State) ->
      ?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) ->
      ?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),
  
negative.code:     dht_node_state:set_mr_state(State, JobId, NewMRState);
  
  on({mr, next_phase_data_ack, AckInterval, JobId, DeliveryInterval}, State) ->
      %% ack from other mr nodes.
      %% check if the whole interval waas acked. If so inform master, wait
      %% otherwise.
      NewMRState = mr_state:set_acked(dht_node_state:get_mr_state(State, JobId),
                                      AckInterval),
      case mr_state:is_acked_complete(NewMRState) of
  