couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fdman...@apache.org
Subject svn commit: r1149412 - in /couchdb/trunk: src/couchdb/Makefile.am src/couchdb/couch_replicator.erl src/couchdb/couch_replicator_doc_copier.erl src/couchdb/couch_replicator_rev_finder.erl src/couchdb/couch_replicator_worker.erl test/etap/001-load.t
Date Fri, 22 Jul 2011 00:04:32 GMT
Author: fdmanana
Date: Fri Jul 22 00:04:30 2011
New Revision: 1149412

URL: http://svn.apache.org/viewvc?rev=1149412&view=rev
Log:
Refactor/simplify replicator

For each worker (doc copier) process there used to be
a rev finder process. A rev finder dequeues from the
_changes rows queue, finds which document IDs and revisions
are missing in the target and queues the IDs and revisions
of those that are missing into a second queue. Finally
worker processes dequeue up to N elements from this queue.

This model was reduced to have the workers dequeing directly
from the _changes rows queue and find which document IDs and
revisions are missing in the target. This makes the model
simpler without suffering any performance penalty - there's
still enough parallelism to continue to be as efficient as
before.


Added:
    couchdb/trunk/src/couchdb/couch_replicator_worker.erl
Removed:
    couchdb/trunk/src/couchdb/couch_replicator_doc_copier.erl
    couchdb/trunk/src/couchdb/couch_replicator_rev_finder.erl
Modified:
    couchdb/trunk/src/couchdb/Makefile.am
    couchdb/trunk/src/couchdb/couch_replicator.erl
    couchdb/trunk/test/etap/001-load.t

Modified: couchdb/trunk/src/couchdb/Makefile.am
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/Makefile.am?rev=1149412&r1=1149411&r2=1149412&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/Makefile.am (original)
+++ couchdb/trunk/src/couchdb/Makefile.am Fri Jul 22 00:04:30 2011
@@ -73,8 +73,7 @@ source_files = \
     couch_replication_manager.erl \
     couch_replication_notifier.erl \
     couch_replicator.erl \
-    couch_replicator_doc_copier.erl \
-    couch_replicator_rev_finder.erl \
+    couch_replicator_worker.erl \
     couch_replicator_utils.erl \
     couch_secondary_sup.erl \
     couch_server.erl \
@@ -143,8 +142,7 @@ compiled_files = \
     couch_replication_manager.beam \
     couch_replication_notifier.beam \
     couch_replicator.beam \
-    couch_replicator_doc_copier.beam \
-    couch_replicator_rev_finder.beam \
+    couch_replicator_worker.beam \
     couch_replicator_utils.beam \
     couch_secondary_sup.beam \
     couch_server.beam \

Modified: couchdb/trunk/src/couchdb/couch_replicator.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_replicator.erl?rev=1149412&r1=1149411&r2=1149412&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_replicator.erl (original)
+++ couchdb/trunk/src/couchdb/couch_replicator.erl Fri Jul 22 00:04:30 2011
@@ -58,10 +58,9 @@
     src_starttime,
     tgt_starttime,
     timer, % checkpoint timer
-    missing_revs_queue,
     changes_queue,
+    changes_manager,
     changes_reader,
-    missing_rev_finders,
     workers,
     stats = #rep_stats{},
     session_id,
@@ -217,41 +216,28 @@ do_init(#rep{options = Options, id = {Ba
         source_seq = SourceCurSeq
     } = State = init_state(Rep),
 
-    CopiersCount = get_value(worker_processes, Options),
-    RevFindersCount = CopiersCount,
+    NumWorkers = get_value(worker_processes, Options),
     BatchSize = get_value(worker_batch_size, Options),
-    {ok, MissingRevsQueue} = couch_work_queue:new([
-        {multi_workers, true},
-        {max_items, trunc(CopiersCount * 2.0)}
-    ]),
     {ok, ChangesQueue} = couch_work_queue:new([
-        {multi_workers, true},
-        {max_items, trunc(BatchSize * RevFindersCount * 2.0)}
+        {max_items, trunc(BatchSize * NumWorkers * 2.0)}
     ]),
     % This starts the _changes reader process. It adds the changes from
     % the source db to the ChangesQueue.
-    ChangesReader = spawn_changes_reader(
-        StartSeq, Source, ChangesQueue, Options),
-    % This starts the missing rev finders. They check the target for changes
-    % in the ChangesQueue to see if they exist on the target or not. If not,
-    % adds them to MissingRevsQueue.
-    MissingRevFinders = lists:map(
-        fun(_) ->
-            {ok, Pid} = couch_replicator_rev_finder:start_link(
-                self(), Target, ChangesQueue, MissingRevsQueue, BatchSize),
-            Pid
-        end,
-        lists:seq(1, RevFindersCount)),
-    % This starts the doc copy processes. They fetch documents from the
-    % MissingRevsQueue and copy them from the source to the target database.
+    ChangesReader = spawn_changes_reader(StartSeq, Source, ChangesQueue, Options),
+    % Changes manager - responsible for dequeing batches from the changes queue
+    % and deliver them to the worker processes.
+    ChangesManager = spawn_changes_manager(self(), ChangesQueue, BatchSize),
+    % This starts the worker processes. They ask the changes queue manager for a
+    % a batch of _changes rows to process -> check which revs are missing in the
+    % target, and for the missing ones, it copies them from the source to the target.
     MaxConns = get_value(http_connections, Options),
     Workers = lists:map(
         fun(_) ->
-            {ok, Pid} = couch_replicator_doc_copier:start_link(
-                self(), Source, Target, MissingRevsQueue, MaxConns),
+            {ok, Pid} = couch_replicator_worker:start_link(
+                self(), Source, Target, ChangesManager, MaxConns),
             Pid
         end,
-        lists:seq(1, CopiersCount)),
+        lists:seq(1, NumWorkers)),
 
     couch_task_status:add_task(
         "Replication",
@@ -275,7 +261,7 @@ do_init(#rep{options = Options, id = {Ba
         "~c~p HTTP connections~n"
         "~ca connection timeout of ~p milliseconds~n"
         "~csocket options are: ~s~s",
-        [BaseId ++ Ext, $\t, CopiersCount, $\t, BatchSize, $\t,
+        [BaseId ++ Ext, $\t, NumWorkers, $\t, BatchSize, $\t,
             MaxConns, $\t, get_value(connection_timeout, Options),
             $\t, io_lib:format("~p", [get_value(socket_options, Options)]),
             case StartSeq of
@@ -285,16 +271,14 @@ do_init(#rep{options = Options, id = {Ba
                 io_lib:format("~n~csource start sequence ~p", [$\t, StartSeq])
             end]),
 
-    ?LOG_DEBUG("Missing rev finder pids are: ~p", [MissingRevFinders]),
     ?LOG_DEBUG("Worker pids are: ~p", [Workers]),
 
     couch_replication_manager:replication_started(Rep),
 
     {ok, State#rep_state{
-            missing_revs_queue = MissingRevsQueue,
             changes_queue = ChangesQueue,
+            changes_manager = ChangesManager,
             changes_reader = ChangesReader,
-            missing_rev_finders = MissingRevFinders,
             workers = Workers
         }
     }.
@@ -315,12 +299,12 @@ handle_info({'EXIT', Pid, Reason}, #rep_
     ?LOG_ERROR("ChangesReader process died with reason: ~p", [Reason]),
     {stop, changes_reader_died, cancel_timer(State)};
 
-handle_info({'EXIT', Pid, normal}, #rep_state{missing_revs_queue=Pid} = St) ->
-    {noreply, St};
+handle_info({'EXIT', Pid, normal}, #rep_state{changes_manager = Pid} = State) ->
+    {noreply, State};
 
-handle_info({'EXIT', Pid, Reason}, #rep_state{missing_revs_queue=Pid} = St) ->
-    ?LOG_ERROR("MissingRevsQueue process died with reason: ~p", [Reason]),
-    {stop, missing_revs_queue_died, cancel_timer(St)};
+handle_info({'EXIT', Pid, Reason}, #rep_state{changes_manager = Pid} = State) ->
+    ?LOG_ERROR("ChangesManager process died with reason: ~p", [Reason]),
+    {stop, changes_manager_died, cancel_timer(State)};
 
 handle_info({'EXIT', Pid, normal}, #rep_state{changes_queue=Pid} = State) ->
     {noreply, State};
@@ -329,53 +313,26 @@ handle_info({'EXIT', Pid, Reason}, #rep_
     ?LOG_ERROR("ChangesQueue process died with reason: ~p", [Reason]),
     {stop, changes_queue_died, cancel_timer(State)};
 
-handle_info({'EXIT', Pid, normal}, State) ->
-    #rep_state{
-        workers = Workers,
-        missing_rev_finders = RevFinders,
-        missing_revs_queue = RevsQueue
-    } = State,
-    case lists:member(Pid, RevFinders) of
-    false ->
-        case lists:member(Pid, Workers) of
-        false ->
-            {stop, {unknown_process_died, Pid, normal}, State};
-        true ->
-            case Workers -- [Pid] of
-            [] ->
-                do_last_checkpoint(State);
-            Workers2 ->
-                {noreply, State#rep_state{workers = Workers2}}
-            end
-        end;
-    true ->
-        case RevFinders -- [Pid] of
-        [] ->
-            couch_work_queue:close(RevsQueue),
-            {noreply, State#rep_state{missing_rev_finders = []}};
-        RevFinders2 ->
-            {noreply, State#rep_state{missing_rev_finders = RevFinders2}}
-        end
+handle_info({'EXIT', Pid, normal}, #rep_state{workers = Workers} = State) ->
+    case Workers -- [Pid] of
+    Workers ->
+        {stop, {unknown_process_died, Pid, normal}, State};
+    [] ->
+        catch unlink(State#rep_state.changes_manager),
+        catch exit(State#rep_state.changes_manager, kill),
+        do_last_checkpoint(State);
+    Workers2 ->
+        {noreply, State#rep_state{workers = Workers2}}
     end;
 
-handle_info({'EXIT', Pid, Reason}, State) ->
-    #rep_state{
-        workers = Workers,
-        missing_rev_finders = RevFinders
-    } = State,
+handle_info({'EXIT', Pid, Reason}, #rep_state{workers = Workers} = State) ->
     State2 = cancel_timer(State),
     case lists:member(Pid, Workers) of
     false ->
-        case lists:member(Pid, RevFinders) of
-        false ->
-            {stop, {unknown_process_died, Pid, Reason}, State2};
-        true ->
-            ?LOG_ERROR("RevsFinder ~p died with reason: ~p", [Pid, Reason]),
-            {stop, {revs_finder_died, Pid, Reason}, State2}
-        end;
+        {stop, {unknown_process_died, Pid, Reason}, State2};
     true ->
-        ?LOG_ERROR("DocCopier ~p died with reason: ~p", [Pid, Reason]),
-        {stop, {doc_copier_died, Pid, Reason}, State2}
+        ?LOG_ERROR("Worker ~p died with reason: ~p", [Pid, Reason]),
+        {stop, {worker_died, Pid, Reason}, State2}
     end.
 
 
@@ -592,12 +549,10 @@ spawn_changes_reader(StartSeq, #httpdb{}
     spawn_link(fun() ->
         put(last_seq, StartSeq),
         put(retries_left, Db#httpdb.retries),
-        put(row_ts, 1),
         read_changes(StartSeq, Db#httpdb{retries = 0}, ChangesQueue, Options)
     end);
 spawn_changes_reader(StartSeq, Db, ChangesQueue, Options) ->
     spawn_link(fun() ->
-        put(row_ts, 1),
         read_changes(StartSeq, Db, ChangesQueue, Options)
     end).
 
@@ -605,9 +560,7 @@ read_changes(StartSeq, Db, ChangesQueue,
     try
         couch_api_wrap:changes_since(Db, all_docs, StartSeq,
             fun(#doc_info{high_seq = Seq} = DocInfo) ->
-                Ts = get(row_ts),
-                ok = couch_work_queue:queue(ChangesQueue, {Ts, DocInfo}),
-                put(row_ts, Ts + 1),
+                ok = couch_work_queue:queue(ChangesQueue, DocInfo),
                 put(last_seq, Seq)
             end, Options),
         couch_work_queue:close(ChangesQueue)
@@ -635,6 +588,27 @@ read_changes(StartSeq, Db, ChangesQueue,
     end.
 
 
+spawn_changes_manager(Parent, ChangesQueue, BatchSize) ->
+    spawn_link(fun() ->
+        changes_manager_loop_open(Parent, ChangesQueue, BatchSize, 1)
+    end).
+
+changes_manager_loop_open(Parent, ChangesQueue, BatchSize, Ts) ->
+    receive
+    {get_changes, From} ->
+        case couch_work_queue:dequeue(ChangesQueue, BatchSize) of
+        closed ->
+            From ! {closed, self()};
+        {ok, Changes} ->
+            #doc_info{high_seq = Seq} = lists:last(Changes),
+            ReportSeq = {Ts, Seq},
+            ok = gen_server:cast(Parent, {report_seq, ReportSeq}),
+            From ! {changes, self(), Changes, ReportSeq}
+        end,
+        changes_manager_loop_open(Parent, ChangesQueue, BatchSize, Ts + 1)
+    end.
+
+
 checkpoint_interval(_State) ->
     5000.
 

Added: couchdb/trunk/src/couchdb/couch_replicator_worker.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_replicator_worker.erl?rev=1149412&view=auto
==============================================================================
--- couchdb/trunk/src/couchdb/couch_replicator_worker.erl (added)
+++ couchdb/trunk/src/couchdb/couch_replicator_worker.erl Fri Jul 22 00:04:30 2011
@@ -0,0 +1,521 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+%   http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_replicator_worker).
+-behaviour(gen_server).
+
+% public API
+-export([start_link/5]).
+
+% gen_server callbacks
+-export([init/1, terminate/2, code_change/3]).
+-export([handle_call/3, handle_cast/2, handle_info/2]).
+
+-include("couch_db.hrl").
+-include("couch_api_wrap.hrl").
+-include("couch_replicator.hrl").
+
+% TODO: maybe make both buffer max sizes configurable
+-define(DOC_BUFFER_BYTE_SIZE, 512 * 1024).   % for remote targets
+-define(DOC_BUFFER_LEN, 10).                 % for local targets, # of documents
+-define(MAX_BULK_ATT_SIZE, 64 * 1024).
+-define(MAX_BULK_ATTS_PER_DOC, 8).
+
+-import(couch_replicator_utils, [
+    open_db/1,
+    close_db/1,
+    start_db_compaction_notifier/2,
+    stop_db_compaction_notifier/1
+]).
+-import(couch_util, [
+    to_binary/1,
+    get_value/2,
+    get_value/3
+]).
+
+
+-record(batch, {
+    docs = [],
+    size = 0
+}).
+
+-record(state, {
+    loop,
+    max_parallel_conns,
+    source,
+    target,
+    readers = [],
+    writer = nil,
+    pending_fetch = nil,
+    flush_waiter = nil,
+    stats = #rep_stats{},
+    source_db_compaction_notifier = nil,
+    target_db_compaction_notifier = nil,
+    batch = #batch{}
+}).
+
+
+
+start_link(Cp, #db{} = Source, Target, ChangesManager, _MaxConns) ->
+    Pid = spawn_link(fun() ->
+        queue_fetch_loop(Source, Target, Cp, Cp, ChangesManager)
+    end),
+    {ok, Pid};
+
+start_link(Cp, Source, Target, ChangesManager, MaxConns) ->
+    gen_server:start_link(
+        ?MODULE, {Cp, Source, Target, ChangesManager, MaxConns}, []).
+
+
+init({Cp, Source, Target, ChangesManager, MaxConns}) ->
+    process_flag(trap_exit, true),
+    Parent = self(),
+    LoopPid = spawn_link(fun() ->
+        queue_fetch_loop(Source, Target, Parent, Cp, ChangesManager)
+    end),
+    State = #state{
+        max_parallel_conns = MaxConns,
+        loop = LoopPid,
+        source = open_db(Source),
+        target = open_db(Target),
+        source_db_compaction_notifier =
+            start_db_compaction_notifier(Source, self()),
+        target_db_compaction_notifier =
+            start_db_compaction_notifier(Target, self())
+    },
+    {ok, State}.
+
+
+handle_call({fetch_doc, {_Id, Revs, _PAs} = Params}, {Pid, _} = From,
+    #state{loop = Pid, readers = Readers, pending_fetch = nil,
+        stats = Stats, source = Src, target = Tgt,
+        max_parallel_conns = MaxConns} = State) ->
+    Stats2 = Stats#rep_stats{
+        missing_checked = Stats#rep_stats.missing_checked + length(Revs),
+        missing_found = Stats#rep_stats.missing_found + length(Revs)
+    },
+    case length(Readers) of
+    Size when Size < MaxConns ->
+        Reader = spawn_doc_reader(Src, Tgt, Params),
+        NewState = State#state{
+            stats = Stats2,
+            readers = [Reader | Readers]
+        },
+        {reply, ok, NewState};
+    _ ->
+        NewState = State#state{
+            stats = Stats2,
+            pending_fetch = {From, Params}
+        },
+        {noreply, NewState}
+    end;
+
+handle_call({batch_doc, Doc}, From, State) ->
+    gen_server:reply(From, ok),
+    {noreply, maybe_flush_docs(Doc, State)};
+
+handle_call({doc_flushed, true}, _From, #state{stats = Stats} = State) ->
+    NewStats = Stats#rep_stats{
+        docs_read = Stats#rep_stats.docs_read + 1,
+        docs_written = Stats#rep_stats.docs_written + 1
+    },
+    {reply, ok, State#state{stats = NewStats}};
+
+handle_call({doc_flushed, false}, _From, #state{stats = Stats} = State) ->
+    NewStats = Stats#rep_stats{
+        docs_read = Stats#rep_stats.docs_read + 1,
+        doc_write_failures = Stats#rep_stats.doc_write_failures + 1
+    },
+    {reply, ok, State#state{stats = NewStats}};
+
+handle_call({add_write_stats, Written, Failed}, _From,
+    #state{stats = Stats} = State) ->
+    NewStats = Stats#rep_stats{
+        docs_written = Stats#rep_stats.docs_written + Written,
+        doc_write_failures = Stats#rep_stats.doc_write_failures + Failed
+    },
+    {reply, ok, State#state{stats = NewStats}};
+
+handle_call(flush, {Pid, _} = From,
+    #state{loop = Pid, writer = nil, flush_waiter = nil,
+        target = Target, batch = Batch} = State) ->
+    State2 = case State#state.readers of
+    [] ->
+        State#state{writer = spawn_writer(Target, Batch)};
+    _ ->
+        State
+    end,
+    {noreply, State2#state{flush_waiter = From}}.
+
+
+handle_cast({db_compacted, DbName},
+    #state{source = #db{name = DbName} = Source} = State) ->
+    {ok, NewSource} = couch_db:reopen(Source),
+    {noreply, State#state{source = NewSource}};
+
+handle_cast({db_compacted, DbName},
+    #state{target = #db{name = DbName} = Target} = State) ->
+    {ok, NewTarget} = couch_db:reopen(Target),
+    {noreply, State#state{target = NewTarget}};
+
+handle_cast(Msg, State) ->
+    {stop, {unexpected_async_call, Msg}, State}.
+
+
+handle_info({'EXIT', Pid, normal}, #state{loop = Pid} = State) ->
+    #state{
+        batch = #batch{docs = []}, readers = [], writer = nil,
+        pending_fetch = nil, flush_waiter = nil
+    } = State,
+    {stop, normal, State};
+
+handle_info({'EXIT', Pid, normal}, #state{writer = Pid} = State) ->
+    {noreply, after_full_flush(State)};
+
+handle_info({'EXIT', Pid, normal}, #state{writer = nil} = State) ->
+    #state{
+        readers = Readers, writer = Writer, batch = Batch,
+        source = Source, target = Target,
+        pending_fetch = Fetch, flush_waiter = FlushWaiter
+    } = State,
+    case Readers -- [Pid] of
+    Readers ->
+        {noreply, State};
+    Readers2 ->
+        State2 = case Fetch of
+        nil ->
+            case (FlushWaiter =/= nil) andalso (Writer =:= nil) andalso
+                (Readers2 =:= [])  of
+            true ->
+                State#state{
+                    readers = Readers2,
+                    writer = spawn_writer(Target, Batch)
+                };
+            false ->
+                State#state{readers = Readers2}
+            end;
+        {From, FetchParams} ->
+            Reader = spawn_doc_reader(Source, Target, FetchParams),
+            gen_server:reply(From, ok),
+            State#state{
+                readers = [Reader | Readers2],
+                pending_fetch = nil
+            }
+        end,
+        {noreply, State2}
+    end;
+
+handle_info({'EXIT', Pid, Reason}, State) ->
+   {stop, {process_died, Pid, Reason}, State}.
+
+
+terminate(_Reason, State) ->
+    close_db(State#state.source),
+    close_db(State#state.target),
+    stop_db_compaction_notifier(State#state.source_db_compaction_notifier),
+    stop_db_compaction_notifier(State#state.target_db_compaction_notifier).
+
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+
+queue_fetch_loop(Source, Target, Parent, Cp, ChangesManager) ->
+    ChangesManager ! {get_changes, self()},
+    receive
+    {closed, ChangesManager} ->
+        ok;
+    {changes, ChangesManager, Changes, ReportSeq} ->
+        Target2 = open_db(Target),
+        {IdRevs, NotMissingCount} = find_missing(Changes, Target2),
+        ok = gen_server:cast(Cp, {report_seq, ReportSeq}),
+        case Source of
+        #db{} ->
+            Source2 = open_db(Source),
+            Stats = local_process_batch(
+                IdRevs, Source2, Target2, #batch{}, #rep_stats{}),
+            close_db(Source2);
+        #httpdb{} ->
+            remote_process_batch(IdRevs, Parent),
+            {ok, Stats} = gen_server:call(Parent, flush, infinity)
+        end,
+        close_db(Target2),
+        Stats2 = Stats#rep_stats{
+            missing_checked = Stats#rep_stats.missing_checked + NotMissingCount
+        },
+        ok = gen_server:cast(Cp, {report_seq_done, ReportSeq, Stats2}),
+        ?LOG_DEBUG("Worker reported completion of seq ~p", [ReportSeq]),
+        queue_fetch_loop(Source, Target, Parent, Cp, ChangesManager)
+    end.
+
+
+local_process_batch([], _Src, _Tgt, #batch{docs = []}, Stats) ->
+    Stats;
+
+local_process_batch([], _Source, Target, #batch{docs = Docs, size = Size}, Stats) ->
+    case Target of
+    #httpdb{} ->
+        ?LOG_DEBUG("Worker flushing doc batch of size ~p bytes", [Size]);
+    #db{} ->
+        ?LOG_DEBUG("Worker flushing doc batch of ~p docs", [Size])
+    end,
+    {Written, WriteFailures} = flush_docs(Target, Docs),
+    Stats#rep_stats{
+        docs_written = Stats#rep_stats.docs_written + Written,
+        doc_write_failures = Stats#rep_stats.doc_write_failures + WriteFailures
+    };
+
+local_process_batch([IdRevs | Rest], Source, Target, Batch, Stats) ->
+    {_Id, Revs, _PAs} = IdRevs,
+    {ok, {_, DocList, Written0, WriteFailures0}} = fetch_doc(
+        Source, IdRevs, fun local_doc_handler/2, {Target, [], 0, 0}),
+    Read = length(DocList) + Written0 + WriteFailures0,
+    {Batch2, Written, WriteFailures} = lists:foldl(
+        fun(Doc, {Batch0, W0, F0}) ->
+            {Batch1, W, F} = maybe_flush_docs(Target, Batch0, Doc),
+            {Batch1, W0 + W, F0 + F}
+        end,
+        {Batch, Written0, WriteFailures0}, DocList),
+    Stats2 = Stats#rep_stats{
+        missing_checked = Stats#rep_stats.missing_checked + length(Revs),
+        missing_found = Stats#rep_stats.missing_found + length(Revs),
+        docs_read = Stats#rep_stats.docs_read + Read,
+        docs_written = Stats#rep_stats.docs_written + Written,
+        doc_write_failures = Stats#rep_stats.doc_write_failures + WriteFailures
+    },
+    local_process_batch(Rest, Source, Target, Batch2, Stats2).
+
+
+remote_process_batch([], _Parent) ->
+    ok;
+
+remote_process_batch([{Id, Revs, PAs} | Rest], Parent) ->
+    % When the source is a remote database, we fetch a single document revision
+    % per HTTP request. This is mostly to facilitate retrying of HTTP requests
+    % due to network transient failures. It also helps not exceeding the maximum
+    % URL length allowed by proxies and Mochiweb.
+    lists:foreach(
+        fun(Rev) ->
+            ok = gen_server:call(Parent, {fetch_doc, {Id, [Rev], PAs}}, infinity)
+        end,
+        Revs),
+    remote_process_batch(Rest, Parent).
+
+
+spawn_doc_reader(Source, Target, FetchParams) ->
+    Parent = self(),
+    spawn_link(fun() ->
+        Source2 = open_db(Source),
+        fetch_doc(
+            Source2, FetchParams, fun remote_doc_handler/2, {Parent, Target}),
+        close_db(Source2)
+    end).
+
+
+fetch_doc(Source, {Id, Revs, PAs}, DocHandler, Acc) ->
+    try
+        couch_api_wrap:open_doc_revs(
+            Source, Id, Revs, [{atts_since, PAs}], DocHandler, Acc)
+    catch
+    throw:{missing_stub, _} ->
+        ?LOG_ERROR("Retrying fetch and update of document `~p` due to out of "
+            "sync attachment stubs. Missing revisions are: ~s",
+            [Id, couch_doc:revs_to_strs(Revs)]),
+        couch_api_wrap:open_doc_revs(Source, Id, Revs, [], DocHandler, Acc)
+    end.
+
+
+local_doc_handler({ok, Doc}, {Target, DocList, W, F}) ->
+    case batch_doc(Doc) of
+    true ->
+        {ok, {Target, [Doc | DocList], W, F}};
+    false ->
+        ?LOG_DEBUG("Worker flushing doc with attachments", []),
+        Target2 = open_db(Target),
+        Success = (flush_doc(Target2, Doc) =:= ok),
+        close_db(Target2),
+        case Success of
+        true ->
+            {ok, {Target, DocList, W + 1, F}};
+        false ->
+            {ok, {Target, DocList, W, F + 1}}
+        end
+    end;
+local_doc_handler(_, Acc) ->
+    {ok, Acc}.
+
+
+remote_doc_handler({ok, #doc{atts = []} = Doc}, {Parent, _} = Acc) ->
+    ok = gen_server:call(Parent, {batch_doc, Doc}, infinity),
+    {ok, Acc};
+remote_doc_handler({ok, Doc}, {Parent, Target} = Acc) ->
+    % Immediately flush documents with attachments received from a remote
+    % source. The data property of each attachment is a function that starts
+    % streaming the attachment data from the remote source, therefore it's
+    % convenient to call it ASAP to avoid ibrowse inactivity timeouts.
+    ?LOG_DEBUG("Worker flushing doc with attachments", []),
+    Target2 = open_db(Target),
+    Success = (flush_doc(Target2, Doc) =:= ok),
+    ok = gen_server:call(Parent, {doc_flushed, Success}, infinity),
+    close_db(Target2),
+    case Success of
+    true ->
+        {ok, Acc};
+    false ->
+        {skip, Acc}
+    end;
+remote_doc_handler(_, Acc) ->
+    {ok, Acc}.
+
+
+spawn_writer(Target, #batch{docs = DocList, size = Size}) ->
+    case {Target, Size > 0} of
+    {#httpdb{}, true} ->
+        ?LOG_DEBUG("Worker flushing doc batch of size ~p bytes", [Size]);
+    {#db{}, true} ->
+        ?LOG_DEBUG("Worker flushing doc batch of ~p docs", [Size]);
+    _ ->
+        ok
+    end,
+    Parent = self(),
+    spawn_link(
+        fun() ->
+            Target2 = open_db(Target),
+            {Written, Failed} = flush_docs(Target2, DocList),
+            close_db(Target2),
+            ok = gen_server:call(
+                Parent, {add_write_stats, Written, Failed}, infinity)
+        end).
+
+
+after_full_flush(#state{stats = Stats, flush_waiter = Waiter} = State) ->
+    gen_server:reply(Waiter, {ok, Stats}),
+    State#state{
+        stats = #rep_stats{},
+        flush_waiter = nil,
+        writer = nil,
+        batch = #batch{}
+    }.
+
+
+maybe_flush_docs(Doc, #state{target = Target, batch = Batch,
+        stats = Stats} = State) ->
+    {Batch2, W, F} = maybe_flush_docs(Target, Batch, Doc),
+    Stats2 = Stats#rep_stats{
+        docs_read = Stats#rep_stats.docs_read + 1,
+        docs_written = Stats#rep_stats.docs_written + W,
+        doc_write_failures = Stats#rep_stats.doc_write_failures + F
+    },
+    State#state{
+        stats = Stats2,
+        batch = Batch2
+    }.
+
+
+maybe_flush_docs(#httpdb{} = Target,
+    #batch{docs = DocAcc, size = SizeAcc} = Batch, Doc) ->
+    case batch_doc(Doc) of
+    false ->
+        ?LOG_DEBUG("Worker flushing doc with attachments", []),
+        case flush_doc(Target, Doc) of
+        ok ->
+            {Batch, 1, 0};
+        _ ->
+            {Batch, 0, 1}
+        end;
+    true ->
+        JsonDoc = ?JSON_ENCODE(couch_doc:to_json_obj(Doc, [revs, attachments])),
+        case SizeAcc + iolist_size(JsonDoc) of
+        SizeAcc2 when SizeAcc2 > ?DOC_BUFFER_BYTE_SIZE ->
+            ?LOG_DEBUG("Worker flushing doc batch of size ~p bytes", [SizeAcc2]),
+            {Written, Failed} = flush_docs(Target, [JsonDoc | DocAcc]),
+            {#batch{}, Written, Failed};
+        SizeAcc2 ->
+            {#batch{docs = [JsonDoc | DocAcc], size = SizeAcc2}, 0, 0}
+        end
+    end;
+
+maybe_flush_docs(#db{} = Target, #batch{docs = DocAcc, size = SizeAcc}, Doc) ->
+    case SizeAcc + 1 of
+    SizeAcc2 when SizeAcc2 >= ?DOC_BUFFER_LEN ->
+        ?LOG_DEBUG("Worker flushing doc batch of ~p docs", [SizeAcc2]),
+        {Written, Failed} = flush_docs(Target, [Doc | DocAcc]),
+        {#batch{}, Written, Failed};
+    SizeAcc2 ->
+        {#batch{docs = [Doc | DocAcc], size = SizeAcc2}, 0, 0}
+    end.
+
+
+batch_doc(#doc{atts = []}) ->
+    true;
+batch_doc(#doc{atts = Atts}) ->
+    (length(Atts) =< ?MAX_BULK_ATTS_PER_DOC) andalso
+        lists:all(
+            fun(#att{disk_len = L, data = Data}) ->
+                (L =< ?MAX_BULK_ATT_SIZE) andalso (Data =/= stub)
+            end, Atts).
+
+
+flush_docs(_Target, []) ->
+    {0, 0};
+
+flush_docs(Target, DocList) ->
+    {ok, Errors} = couch_api_wrap:update_docs(
+        Target, DocList, [delay_commit], replicated_changes),
+    DbUri = couch_api_wrap:db_uri(Target),
+    lists:foreach(
+        fun({Props}) ->
+            ?LOG_ERROR("Replicator: couldn't write document `~s`, revision `~s`,"
+                " to target database `~s`. Error: `~s`, reason: `~s`.",
+                [get_value(id, Props, ""), get_value(rev, Props, ""), DbUri,
+                    get_value(error, Props, ""), get_value(reason, Props, "")])
+        end, Errors),
+    {length(DocList) - length(Errors), length(Errors)}.
+
+flush_doc(Target, #doc{id = Id, revs = {Pos, [RevId | _]}} = Doc) ->
+    try couch_api_wrap:update_doc(Target, Doc, [], replicated_changes) of
+    {ok, _} ->
+        ok;
+    Error ->
+        ?LOG_ERROR("Replicator: error writing document `~s` to `~s`: ~s",
+            [Id, couch_api_wrap:db_uri(Target), couch_util:to_binary(Error)]),
+        Error
+    catch
+    throw:{missing_stub, _} = MissingStub ->
+        throw(MissingStub);
+    throw:{Error, Reason} ->
+        ?LOG_ERROR("Replicator: couldn't write document `~s`, revision `~s`,"
+            " to target database `~s`. Error: `~s`, reason: `~s`.",
+            [Id, couch_doc:rev_to_str({Pos, RevId}),
+                couch_api_wrap:db_uri(Target), to_binary(Error), to_binary(Reason)]),
+        {error, Error};
+    throw:Err ->
+        ?LOG_ERROR("Replicator: couldn't write document `~s`, revision `~s`,"
+            " to target database `~s`. Error: `~s`.",
+            [Id, couch_doc:rev_to_str({Pos, RevId}),
+                couch_api_wrap:db_uri(Target), to_binary(Err)]),
+        {error, Err}
+    end.
+
+
+find_missing(DocInfos, Target) ->
+    {IdRevs, AllRevsCount} = lists:foldr(
+        fun(#doc_info{id = Id, revs = RevsInfo}, {IdRevAcc, CountAcc}) ->
+            Revs = [Rev || #rev_info{rev = Rev} <- RevsInfo],
+            {[{Id, Revs} | IdRevAcc], CountAcc + length(Revs)}
+        end,
+        {[], 0}, DocInfos),
+    {ok, Missing} = couch_api_wrap:get_missing_revs(Target, IdRevs),
+    MissingRevsCount = lists:foldl(
+        fun({_Id, MissingRevs, _PAs}, Acc) -> Acc + length(MissingRevs) end,
+        0, Missing),
+    {Missing, AllRevsCount - MissingRevsCount}.

Modified: couchdb/trunk/test/etap/001-load.t
URL: http://svn.apache.org/viewvc/couchdb/trunk/test/etap/001-load.t?rev=1149412&r1=1149411&r2=1149412&view=diff
==============================================================================
--- couchdb/trunk/test/etap/001-load.t (original)
+++ couchdb/trunk/test/etap/001-load.t Fri Jul 22 00:04:30 2011
@@ -17,7 +17,7 @@
 
 main(_) ->
     test_util:init_code_path(),
-    etap:plan(51),
+    etap:plan(50),
     Modules = [
         couch_auth_cache,
         couch_api_wrap,
@@ -55,8 +55,7 @@ main(_) ->
         couch_replication_manager,
         couch_replication_notifier,
         couch_replicator,
-        couch_replicator_doc_copier,
-        couch_replicator_rev_finder,
+        couch_replicator_worker,
         couch_replicator_utils,
         couch_rep_sup,
         couch_server,



Mime
View raw message