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:         write = Write,
          partition = proplists:get_value(partition, JobSpec),
          output = proplists:get_value(output, JobSpec),
          sem = erlduce_utils:sem_new(1),
          dir = proplists:get_value(dir, JobSpec)
      }}.
  
  
negative.code:     {noreply,State};
  
  handle_cast( {merge, BlobID, Items}, State=#state{
          master=Master, write=Write, sem_merge=SemMerge, mem_threshold=MemThreshold, last_flush=LastFlush, dir=Dir, files=Files }) ->
      erlduce_utils:sem_wait(SemMerge),
      {Time, _} = timer:tc(fun()->
          [Write(Item) || Item <- Items], ok
      end),
      erlduce_utils:sem_signal(SemMerge),
      erlduce_job:update_counter(Master, reduce_time, Time),
  
      HighMem = p_check_mem(MemThreshold),
  