anchor.diff:      end.
   
   
  -p_map_worker(Slave, Master, {BlobID, Path, Hosts}, Slaves, Sem, Write, Map, Part) ->
  +p_map_worker(Slave, Master, {BlobID, Path, Hosts}, Slaves, {SemMap,SemMerge}, Write, Map, Part) ->
       case edfs:read({BlobID,Hosts}) of
           {ok, Bytes} ->
  -            erlduce_utils:sem_wait(Sem),
  +            erlduce_utils:sem_wait(SemMap),
  +            erlduce_utils:sem_wait(SemMerge),
               erlduce_job:slave_req_input(Master, Slave),
               {MapTime, Items} = timer:tc(fun()->
                   erlang:erase(),
                   Map(Path,Bytes,Write),
                   erlang:erase()
               end),
  -            erlduce_utils:sem_signal(Sem),
  +            erlduce_utils:sem_signal(SemMerge),
  +            erlduce_utils:sem_signal(SemMap),
               erlduce_job:update_counter(Master,[{input_bytes, byte_size(Bytes)}, {map_time, MapTime} ]),
               case Write of
                   undefined -> erlduce_job:slave_ack_input_all(Master, BlobID);
  
positive.code:     {noreply,State};
  
  handle_cast( {merge, BlobID, Items}, State=#state{
          master=Master, write=Write, sem=Sem, mem_threshold=MemThreshold, last_flush=LastFlush, dir=Dir, files=Files }) ->
      erlduce_utils:sem_wait(Sem),
      {Time, _} = timer:tc(fun()->
          [Write(Item) || Item <- Items], ok
      end),
      erlduce_utils:sem_signal(Sem),
      erlduce_job:update_counter(Master, reduce_time, Time),
  
      HighMem = p_check_mem(MemThreshold),
  
negative.code:     {reply, ignored, State}.
  
  
  handle_cast( {input, eof}, State=#state{ master=Master, dir=Dir, files=Files, sem_map=SemMap}) ->
      State2 = case Files of
          [] -> State;
          _ ->
              erlduce_utils:sem_wait(SemMap),
              {TmpWriteTime,File} = timer:tc(fun()->
                  p_flush_to_disk(Dir, length(Files))
              end),
              erlduce_utils:sem_signal(SemMap),
              erlduce_job:update_counter(Master, reduce_time, TmpWriteTime),
              State#state{ files=[File|Files] }
      end,
      {noreply, State2};
  handle_cast( {input, Blob}, State=#state{ master=Master, slaves=Slaves, sem_map=SemMap, sem_merge=SemMerge, write=Write, map=Map, partition=Part }) ->
      spawn_link(?MODULE, p_map_worker, [self(),Master,Blob,Slaves,{SemMap,SemMerge},Write,Map,Part]),
      {noreply, State};
  
  handle_cast( {merge, BlobID, []}, State=#state{ master=Master }) ->
  