anchor.diff:        [],
         fun (H, T) -> [H|T] end,
         fun lists:reverse/1,
  -      fun start_task_proc/2).
  +      fun start_phase_proc/2).
   
   % groupby proc
   groupby_proc() ->
  
positive.code: %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
  % N: the max number of workers
  % Tasks: [Task]
  % Task: {map, Map} or {reduce, Reduce}
  % Map: {K, V} X Emit-> void
  % Emit: {K, V} -> void
  % Reduce: {K, [V]} -> V
  
  mapreduce(_, _, []) -> {error, "The tasks is empty."};
  mapreduce(N, InputList, Tasks) when is_list(Tasks) ->
      process_flag(trap_exit, true),
      WorkerPids = start_workern(N, fun input_proc/0, 1),
      info("spliting input data", []),
      lists:foreach(mk_emit_hash(WorkerPids), InputList),
      info("go~~", []),
      mapreduce_iter(N, WorkerPids, Tasks, 2).
  
  mapreduce_iter(_, WorkerPids, [Task], _) ->
      OutputPid = start_output(),
      start_and_wait(WorkerPids, Task, mk_emit_one(OutputPid)),
      get_output(OutputPid);
  mapreduce_iter(N, WorkerPids, [Task|Tail], RoundNo) ->
      OutputPids = start_workern(N, fun groupby_proc/0, RoundNo),
      start_and_wait(WorkerPids, Task, mk_emit_hash(OutputPids)),
      mapreduce_iter(N, OutputPids, Tail, RoundNo + 1).
  
  start_and_wait(WorkerPids, Task, Emit) ->
      info("start task: ~p", [Task]),
      lists:foreach(
        fun (Pid) -> Pid ! {Task, Emit} end,
        WorkerPids),
      wait_workers(WorkerPids),
      info("task done", []).
  
  % worker
  start_workern(N, F, RoundNo) ->
  
negative.code:               dict:update(K, fun (Vs) -> [V|Vs] end, [V], Bucket)
        end,
        fun dict:to_list/1,
        fun start_phase_proc/2).
  
  % start phase proc
  start_phase_proc({{map, Map}, Emit}, Output) ->
      map_proc(Map, Output, Emit);
  start_phase_proc({{reduce, Reduce}, Emit}, Output) ->
      reduce_proc(Reduce, Output, Emit).
  
  % map proc
  map_proc(Map,InputList, Emit) ->
      phase_proc(fun (KV) -> Map(KV, Emit) end, InputList).
  
  % reduce proc
  reduce_proc(Reduce, InputList, Emit) ->
      phase_proc(fun (KV) -> Emit(Reduce(KV)) end, InputList).
  
  % phase proc
  phase_proc(_, []) -> ok;
  phase_proc(ProcessAndEmit, [KV|Tail]) ->
      ProcessAndEmit(KV),
      phase_proc(ProcessAndEmit, Tail).
  