anchor.diff:      output :: function(),
       write :: function(),
   
  -    sem :: pid(),
  +    sem_map :: pid(),
  +    sem_merge :: pid(),
       mem_threshold = undefined :: number(),
       last_flush = {0,0,0} :: erlang:timestamp(),
   
  
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),
  