anchor.diff:              Master = mr_state:get(MRState, master),
               comm:send(Master, {mr, job_completed, Range}),
               Client = mr_state:get(MRState, client),
  -            %% TODO send back only {K, V} without HK
               comm:send(Client, {mr_results,
                                  ets:select(Data, [{{'_','$1','$2'}, [], [{{'$1', '$2'}}]}]),
                                  %% ets:tab2list(Data),
  
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, 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
  