anchor.diff:      end.
   
   pull_rep(DbTarget, DbSource, SourceSeqNum) ->
  -    SaveDocsPid = spawn_link(fun() ->
  -            save_docs_loop(DbTarget, 0) end),
  -    OpenDocsPid = spawn_link(fun() ->
  -            open_doc_revs_loop(DbSource, SaveDocsPid, 0) end),
  -    OpenDocsPid ! got_it, % prime queue with got_it
  -    MissingRevsPid = spawn_link(fun() ->
  -            get_missing_revs_loop(DbTarget, OpenDocsPid, 0, 0) end),
  -    MissingRevsPid ! got_it, % prime queue with got_it
  -    self() ! got_it,
  -    {ok, NewSeq} = enum_docs_since(DbSource, SourceSeqNum,
  -        fun(SrcDocInfo, _, _) ->
  -            #doc_info{id=Id,
  -                rev=Rev,
  -                conflict_revs=Conflicts,
  -                deleted_conflict_revs=DelConflicts,
  -                update_seq=Seq} = SrcDocInfo,
  -            SrcRevs = [Rev | Conflicts] ++ DelConflicts,
  -            receive got_it -> ok end,
  -            MissingRevsPid !  {self(), Id, SrcRevs}, % send to the missing revs process
  -            {ok, Seq}
  -        end, SourceSeqNum),
  -    
  -    receive got_it -> ok end,
  -    
  -    MissingRevsPid ! {self(), shutdown},
  -    receive {done, MissingRevsPid, Stats1} -> ok end,
  -    
  -    OpenDocsPid ! {self(), shutdown},
  -    receive {done, OpenDocsPid, Stats2} -> ok end,
  -    
  -    SaveDocsPid ! {self(), shutdown},
  -    receive {done, SaveDocsPid, Stats3} -> ok end,
  -    
  -    {NewSeq, Stats1 ++ Stats2 ++ Stats3}.
  -
  -
  -get_missing_revs_loop(DbTarget, OpenDocsPid, RevsChecked, MissingFound) ->
  -    receive got_it -> ok end,
  -    receive
  -    {Src, Id, Revs} ->
  -        Src ! got_it,
  -
  -        MissingRevs =
  -        case get_missing_revs(DbTarget, [{Id, Revs}]) of
  -        {ok, [{Id, MissingRevs0}]} ->
  -            OpenDocsPid ! {self(), Id, MissingRevs0},
  -            MissingRevs0;
  -        {ok, []} ->
  -            % prime our message queue
  -            self() ! got_it,
  -            []
  -        end,
  -        get_missing_revs_loop(DbTarget, OpenDocsPid,
  -                RevsChecked + length(Revs),
  -                MissingFound + length(MissingRevs));
  -    {Src, shutdown} ->
  -        Src ! {done, self(), [{<<"missing_checked">>, RevsChecked},
  -                                 {<<"missing_found">>, MissingFound}]}
  -    end.
  -    
  -
  -open_doc_revs_loop(DbSource, SaveDocsPid, DocsRead) ->
  -    receive got_it -> ok end,
  -    receive
  -    {Src, Id, MissingRevs} ->
  -        Src ! got_it,
  -        {ok, DocResults} = open_doc_revs(DbSource, Id, MissingRevs, [latest]),
  -        % only save successful reads
  -        Docs = [RevDoc || {ok, RevDoc} <- DocResults],
  -        SaveDocsPid ! {self(), docs, Docs},
  -        open_doc_revs_loop(DbSource, SaveDocsPid, DocsRead + length(Docs));
  -    {Src, shutdown} ->
  -        Src ! {done, self(), [{<<"docs_read">>, DocsRead}]}
  -    end.
  -
  -
  -        
  -save_docs_loop(DbTarget, DocsWritten) ->
  -    receive
  -    {Src, docs, Docs} ->
  -        Src ! got_it,
  -        ok = update_docs(DbTarget, Docs, [], false),
  -        save_docs_loop(DbTarget, DocsWritten + length(Docs));
  -    {Src, shutdown} ->
  -        Src ! {done, self(), [{<<"docs_written">>, DocsWritten}]}
  -    end.
  -
  +    http:set_options([{max_pipeline_length, 101}, {pipeline_timeout, 5000}]),
  +    {ok, {NewSeq, Stats}} = 
  +        enum_docs_since(DbSource, DbTarget, SourceSeqNum, {SourceSeqNum, []}),
  +    http:set_options([{max_pipeline_length, 2}, {pipeline_timeout, 0}]),
  +    {NewSeq, Stats}.
   
   do_http_request(Url, Action, Headers) ->
       do_http_request(Url, Action, Headers, []).
  
positive.code: close_db(Db)->
      couch_db:close(Db).
  
  
  enum_docs_since(#http_db{uri=DbUrl, headers=Headers}=Db, Start, InFun, InAcc)->
      Url = DbUrl ++ "_all_docs_by_seq?count=100&startkey="
              ++ integer_to_list(Start),
      {Results} = do_http_request(Url, get, Headers),
      DocInfoList=
      lists:map(fun({RowInfoList}) ->
          {RowValueProps} = proplists:get_value(<<"value">>, RowInfoList),
          #doc_info{
  
negative.code:         ?JSON_DECODE(ResponseBody)
      end.
  
  save_docs_buffer(DbTarget, DocsBuffer, []) ->
      receive
      {Src, shutdown} ->
          ok = update_docs(DbTarget, lists:reverse(DocsBuffer), [], false),
          Src ! {done, self(), [{<<"docs_written">>, length(DocsBuffer)}]}
      end;
  save_docs_buffer(DbTarget, DocsBuffer, UpdateSequences) ->
      [NextSeq|Rest] = UpdateSequences,
      receive
      {Src, skip, NextSeq} ->
          Src ! got_it,
          save_docs_buffer(DbTarget, DocsBuffer, Rest);
      {Src, docs, {NextSeq, Docs}} ->
          Src ! got_it,
          case couch_util:should_flush() of
              true ->
                  ok = update_docs(DbTarget, lists:reverse(Docs++DocsBuffer), [], 
                      false),
                  save_docs_buffer(DbTarget, [], Rest);
              false ->
                  save_docs_buffer(DbTarget, Docs++DocsBuffer, Rest)
          end;
          {Src, shutdown} ->
          ?LOG_ERROR("received shutdown while waiting for more update_seqs", []),
          ok = update_docs(DbTarget, lists:reverse(DocsBuffer), [], false),
          Src ! {done, self(), [{<<"docs_written">>, length(DocsBuffer)}]}
      end.
  
  pmap(F,List) ->
      [wait_result(Worker) || Worker <- [spawn_worker(self(),F,E) || E <- List]].
  
  spawn_worker(Parent, F, E) ->
      erlang:spawn_monitor(fun() -> Parent ! {self(), F(E)} end).
  
  wait_result({Pid,Ref}) ->
      receive
      {'DOWN', Ref, _, _, normal} -> receive {Pid,Result} -> Result end;
      {'DOWN', Ref, _, _, Reason} -> exit(Reason)
  end.
  
  enum_docs_parallel(DbS, DbT, DocInfoList) ->
      UpdateSeqs = [D#doc_info.update_seq || D <- DocInfoList],
      SaveDocsPid = spawn_link(fun() -> save_docs_buffer(DbT,[],UpdateSeqs) end),
      
      Stats = pmap(fun(SrcDocInfo) ->
          #doc_info{id=Id,
              rev=Rev,
              conflict_revs=Conflicts,
              deleted_conflict_revs=DelConflicts,
              update_seq=Seq} = SrcDocInfo,
          SrcRevs = [Rev | Conflicts] ++ DelConflicts,
          
          case get_missing_revs(DbT, [{Id, SrcRevs}]) of
          {ok, [{Id, MissingRevs}]} ->
              {ok, DocResults} = open_doc_revs(DbS, Id, MissingRevs, [latest]),
              
              % only save successful reads
              Docs = [RevDoc || {ok, RevDoc} <- DocResults],
              
              % include update_seq so we save docs in order
              SaveDocsPid ! {self(), docs, {Seq, Docs}},
              receive got_it -> ok end,
              [{missing_checked, length(SrcRevs)},
               {missing_found, length(MissingRevs)},
               {docs_read, length(Docs)}];
          {ok, []} ->
              SaveDocsPid ! {self(), skip, Seq},
              receive got_it -> ok end,
              [{missing_checked, length(SrcRevs)}]
          end    
      end, DocInfoList),
      
      SaveDocsPid ! {self(), shutdown},
      
      {MissingChecked, MissingFound, DocsRead} = lists:foldl(fun(S, {C, F, R}) ->
          C1 = C + proplists:get_value(missing_checked, S, 0),
          F1 = F + proplists:get_value(missing_found, S, 0),
          R1 = R + proplists:get_value(docs_read, S, 0),
          {C1, F1, R1}
      end, {0, 0, 0}, Stats),
      
      receive
          {done, SaveDocsPid, [{<<"docs_written">>, DocsWritten}]} -> ok
      end,
      
      [ {<<"missing_checked">>, MissingChecked},
        {<<"missing_found">>, MissingFound}, 
        {<<"docs_read">>, DocsRead},
        {<<"docs_written">>, DocsWritten} ].
  
  fix_url(UrlBin) ->
      Url = binary_to_list(UrlBin),
  