couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From vatam...@apache.org
Subject [37/50] couch-replicator commit: updated refs/heads/63012-scheduler to 27a5eae
Date Tue, 14 Mar 2017 19:26:32 GMT
Make sure doc processor workers do not re-add deleted replication jobs

Previously, especially in case of filtered replications, doc processor workers
could inadvertently re-add a replication job after it was deleted.

Workers after finishing fetching filter code and computing the replication
id, would try to add the replication job to the scheduler. They did that without
checking if replication document was already deleted, or another worker
was spawned.

The fix is to create a unique worker reference, pass it to the
worker, then worker confirms they are still current and document was not
deleted before adding the job, otherwise they exit with an `ignore` result.


Project: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/commit/700a9295
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/tree/700a9295
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/diff/700a9295

Branch: refs/heads/63012-scheduler
Commit: 700a9295590d789b33c34999c75a4410e22eb43d
Parents: f2b3ac5
Author: Nick Vatamaniuc <vatamane@apache.org>
Authored: Wed Dec 7 20:51:45 2016 -0500
Committer: Nick Vatamaniuc <vatamane@apache.org>
Committed: Wed Jan 11 15:00:23 2017 -0500

----------------------------------------------------------------------
 src/couch_replicator.hrl                      |  2 +
 src/couch_replicator_doc_processor.erl        | 48 +++++++++--
 src/couch_replicator_doc_processor_worker.erl | 94 +++++++++++++++-------
 3 files changed, 107 insertions(+), 37 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/700a9295/src/couch_replicator.hrl
----------------------------------------------------------------------
diff --git a/src/couch_replicator.hrl b/src/couch_replicator.hrl
index 339e162..b8669e8 100644
--- a/src/couch_replicator.hrl
+++ b/src/couch_replicator.hrl
@@ -30,11 +30,13 @@
 -type seconds() :: non_neg_integer().
 -type rep_start_result() ::
     {ok, rep_id()} |
+    ignore |
     {temporary_error, binary()} |
     {permanent_failure, binary()}.
 
 
 -record(doc_worker_result, {
     id :: db_doc_id(),
+    wref :: reference(),
     result :: rep_start_result()
 }).

http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/700a9295/src/couch_replicator_doc_processor.erl
----------------------------------------------------------------------
diff --git a/src/couch_replicator_doc_processor.erl b/src/couch_replicator_doc_processor.erl
index 402a72f..9c2e2b3 100644
--- a/src/couch_replicator_doc_processor.erl
+++ b/src/couch_replicator_doc_processor.erl
@@ -16,6 +16,7 @@
 -export([start_link/0]).
 -export([docs/1, doc/2]).
 -export([update_docs/0]).
+-export([get_worker_ref/1]).
 
 % multidb changes callback
 -export([db_created/2, db_deleted/2, db_found/2, db_change/3]).
@@ -88,6 +89,18 @@ db_change(DbName, {ChangeProps} = Change, Server) ->
     Server.
 
 
+-spec get_worker_ref(db_doc_id()) -> reference() | nil.
+get_worker_ref({DbName, DocId}) when is_binary(DbName), is_binary(DocId) ->
+    case ets:lookup(?MODULE, {DbName, DocId}) of
+        [#rdoc{worker = WRef}] when is_reference(WRef) ->
+            WRef;
+        [#rdoc{worker = nil}] ->
+            nil;
+        [] ->
+            nil
+    end.
+
+
 % Private helpers for multidb changes API, these updates into the doc
 % processor gen_server
 
@@ -203,8 +216,8 @@ handle_cast(Msg, State) ->
     {stop, {error, unexpected_message, Msg}, State}.
 
 
-handle_info({'DOWN', Ref, _, _, #doc_worker_result{id = Id, result = Res}},
-        State) ->
+handle_info({'DOWN', _, _, _, #doc_worker_result{id = Id, wref = Ref,
+        result = Res}}, State) ->
     ok = worker_returned(Ref, Id, Res),
     {noreply, State};
 
@@ -324,6 +337,9 @@ worker_returned(Ref, Id, {ok, RepId}) ->
     end,
     ok;
 
+worker_returned(_Ref, _Id, ignore) ->
+    ok;
+
 worker_returned(Ref, Id, {temporary_error, Reason}) ->
     case ets:lookup(?MODULE, Id) of
     [#rdoc{worker = Ref, errcnt = ErrCnt} = Row] ->
@@ -413,8 +429,9 @@ maybe_start_worker(Id) ->
         ok;
     [#rdoc{rep = Rep} = Doc] ->
         Wait = get_worker_wait(Doc),
-        WRef = couch_replicator_doc_processor_worker:spawn_worker(Id, Rep, Wait),
-        true = ets:insert(?MODULE, Doc#rdoc{worker = WRef}),
+        Ref = make_ref(),
+        true = ets:insert(?MODULE, Doc#rdoc{worker = Ref}),
+        couch_replicator_doc_processor_worker:spawn_worker(Id, Rep, Wait, Ref),
         ok
     end.
 
@@ -773,6 +790,22 @@ normalize_rep_test_() ->
     }.
 
 
+get_worker_ref_test_() ->
+    {
+        setup,
+        fun() -> ets:new(?MODULE, [named_table, public, {keypos, #rdoc.id}]) end,
+        fun(_) -> ets:delete(?MODULE) end,
+        ?_test(begin
+            Id = {<<"db">>, <<"doc">>},
+            ?assertEqual(nil, get_worker_ref(Id)),
+            ets:insert(?MODULE, #rdoc{id = Id, worker = nil}),
+            ?assertEqual(nil, get_worker_ref(Id)),
+            Ref = make_ref(),
+            ets:insert(?MODULE, #rdoc{id = Id, worker = Ref}),
+            ?assertEqual(Ref, get_worker_ref(Id))
+        end)
+    }.
+
 
 % Test helper functions
 
@@ -786,7 +819,7 @@ setup() ->
     meck:expect(config, listen_for_changes, 2, ok),
     meck:expect(couch_replicator_clustering, owner, 2, node()),
     meck:expect(couch_replicator_clustering, link_cluster_event_listener, 1, ok),
-    meck:expect(couch_replicator_doc_processor_worker, spawn_worker, 3, wref),
+    meck:expect(couch_replicator_doc_processor_worker, spawn_worker, 4, pid),
     meck:expect(couch_replicator_scheduler, remove_job, 1, ok),
     meck:expect(couch_replicator_docs, remove_state_fields, 2, ok),
     meck:expect(couch_replicator_docs, update_failed, 4, ok),
@@ -804,9 +837,8 @@ removed_state_fields() ->
     meck:called(couch_replicator_docs, remove_state_fields, [?DB, ?DOC1]).
 
 
-started_worker(Id) ->
-    meck:called(couch_replicator_doc_processor_worker, spawn_worker,
-        [Id, '_', '_']).
+started_worker(_Id) ->
+    1 == meck:num_calls(couch_replicator_doc_processor_worker, spawn_worker, 4).
 
 
 removed_job(Id) ->

http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/700a9295/src/couch_replicator_doc_processor_worker.erl
----------------------------------------------------------------------
diff --git a/src/couch_replicator_doc_processor_worker.erl b/src/couch_replicator_doc_processor_worker.erl
index a6bdeef..30a6988 100644
--- a/src/couch_replicator_doc_processor_worker.erl
+++ b/src/couch_replicator_doc_processor_worker.erl
@@ -12,7 +12,7 @@
 
 -module(couch_replicator_doc_processor_worker).
 
--export([spawn_worker/3]).
+-export([spawn_worker/4]).
 
 -include("couch_replicator.hrl").
 
@@ -27,19 +27,19 @@
 % a worker will then exit with the #doc_worker_result{} record within
 % ?WORKER_TIMEOUT_MSEC timeout period.A timeout is considered a `temporary_error`.
 % Result will be sent as the `Reason` in the {'DOWN',...} message.
--spec spawn_worker(db_doc_id(), #rep{}, seconds()) -> reference().
-spawn_worker(Id, Rep, WaitSec) ->
-    {_Pid, WRef} = spawn_monitor(fun() -> worker_fun(Id, Rep, WaitSec) end),
-    WRef.
+-spec spawn_worker(db_doc_id(), #rep{}, seconds(), reference()) -> pid().
+spawn_worker(Id, Rep, WaitSec, WRef) ->
+    {Pid, _Ref} = spawn_monitor(fun() -> worker_fun(Id, Rep, WaitSec, WRef) end),
+    Pid.
 
 
 % Private functions
 
--spec worker_fun(db_doc_id(), #rep{}, seconds()) -> no_return().
-worker_fun(Id, Rep, WaitSec) ->
+-spec worker_fun(db_doc_id(), #rep{}, seconds(), reference()) -> no_return().
+worker_fun(Id, Rep, WaitSec, WRef) ->
     timer:sleep(WaitSec * 1000),
     Fun = fun() ->
-        try maybe_start_replication(Id, Rep) of
+        try maybe_start_replication(Id, Rep, WRef) of
             Res ->
                 exit(Res)
         catch
@@ -52,7 +52,7 @@ worker_fun(Id, Rep, WaitSec) ->
     {Pid, Ref} = spawn_monitor(Fun),
     receive
         {'DOWN', Ref, _, Pid, Result} ->
-            exit(#doc_worker_result{id = Id, result = Result})
+            exit(#doc_worker_result{id = Id, wref = WRef, result = Result})
     after ?WORKER_TIMEOUT_MSEC ->
         erlang:demonitor(Ref, [flush]),
         exit(Pid, kill),
@@ -61,7 +61,7 @@ worker_fun(Id, Rep, WaitSec) ->
         Msg = io_lib:format("Replication for db ~p doc ~p failed to start due "
             "to timeout after ~B seconds", [DbName, DocId, TimeoutSec]),
         Result = {temporary_error, couch_util:to_binary(Msg)},
-        exit(#doc_worker_result{id = Id, result = Result})
+        exit(#doc_worker_result{id = Id, wref = WRef, result = Result})
     end.
 
 
@@ -69,9 +69,11 @@ worker_fun(Id, Rep, WaitSec) ->
 % rep_start_result(), also throws {filter_fetch_error, Reason} if cannot fetch filter.
 % It can also block for an indeterminate amount of time while fetching the
 % filter.
-maybe_start_replication(Id, RepWithoutId) ->
+maybe_start_replication(Id, RepWithoutId, WRef) ->
     Rep = couch_replicator_docs:update_rep_id(RepWithoutId),
-    case maybe_add_job_to_scheduler(Id, Rep) of
+    case maybe_add_job_to_scheduler(Id, Rep, WRef) of
+    ignore ->
+        ignore;
     {ok, RepId} ->
         {ok, RepId};
     {temporary_error, Reason} ->
@@ -84,18 +86,23 @@ maybe_start_replication(Id, RepWithoutId) ->
     end.
 
 
--spec maybe_add_job_to_scheduler(db_doc_id(), #rep{}) -> rep_start_result().
-maybe_add_job_to_scheduler({_DbName, DocId}, Rep) ->
+-spec maybe_add_job_to_scheduler(db_doc_id(), #rep{}, reference()) ->
+   rep_start_result().
+maybe_add_job_to_scheduler({DbName, DocId}, Rep, WRef) ->
     RepId = Rep#rep.id,
     case couch_replicator_scheduler:rep_state(RepId) of
     nil ->
-        case couch_replicator_scheduler:add_job(Rep) of
-        ok ->
-           ok;
-        {error, already_added} ->
-            couch_log:warning("replicator scheduler: ~p was already added", [Rep])
-        end,
-        {ok, RepId};
+        % Before adding a job check that this worker is still the current
+        % worker. This is to handle a race condition where a worker which was
+        % sleeping and then checking a replication filter may inadvertently re-add
+        % a replication which was already deleted.
+        case couch_replicator_doc_processor:get_worker_ref({DbName, DocId}) of
+        WRef ->
+            ok = couch_replicator_scheduler:add_job(Rep),
+            {ok, RepId};
+        _NilOrOtherWRef ->
+            ignore
+        end;
     #rep{doc_id = DocId} ->
         {ok, RepId};
     #rep{doc_id = null} ->
@@ -130,7 +137,9 @@ doc_processor_worker_test_() ->
             t_already_running_same_docid(),
             t_already_running_transient(),
             t_already_running_other_db_other_doc(),
-            t_spawn_worker()
+            t_spawn_worker(),
+            t_ignore_if_doc_deleted(),
+            t_ignore_if_worker_ref_does_not_match()
         ]
     }.
 
@@ -140,7 +149,7 @@ t_should_add_job() ->
    ?_test(begin
        Id = {?DB, ?DOC1},
        Rep = couch_replicator_docs:parse_rep_doc_without_id(change()),
-       ?assertEqual({ok, ?R1}, maybe_start_replication(Id, Rep)),
+       ?assertEqual({ok, ?R1}, maybe_start_replication(Id, Rep, nil)),
        ?assert(added_job())
    end).
 
@@ -151,7 +160,7 @@ t_already_running_same_docid() ->
        Id = {?DB, ?DOC1},
        mock_already_running(?DB, ?DOC1),
        Rep = couch_replicator_docs:parse_rep_doc_without_id(change()),
-       ?assertEqual({ok, ?R1}, maybe_start_replication(Id, Rep)),
+       ?assertEqual({ok, ?R1}, maybe_start_replication(Id, Rep, nil)),
        ?assert(did_not_add_job())
    end).
 
@@ -162,7 +171,7 @@ t_already_running_transient() ->
        Id = {?DB, ?DOC1},
        mock_already_running(null, null),
        Rep = couch_replicator_docs:parse_rep_doc_without_id(change()),
-       ?assertMatch({temporary_error, _}, maybe_start_replication(Id, Rep)),
+       ?assertMatch({temporary_error, _}, maybe_start_replication(Id, Rep, nil)),
        ?assert(did_not_add_job())
    end).
 
@@ -174,7 +183,7 @@ t_already_running_other_db_other_doc() ->
        Id = {?DB, ?DOC1},
        mock_already_running(<<"otherdb">>, <<"otherdoc">>),
        Rep = couch_replicator_docs:parse_rep_doc_without_id(change()),
-       ?assertMatch({permanent_failure, _}, maybe_start_replication(Id, Rep)),
+       ?assertMatch({permanent_failure, _}, maybe_start_replication(Id, Rep, nil)),
        ?assert(did_not_add_job()),
        1 == meck:num_calls(couch_replicator_docs, update_failed, '_')
    end).
@@ -185,15 +194,41 @@ t_spawn_worker() ->
    ?_test(begin
        Id = {?DB, ?DOC1},
        Rep = couch_replicator_docs:parse_rep_doc_without_id(change()),
-       Ref = spawn_worker(Id, Rep, 0),
-       Res = receive  {'DOWN', Ref, _, _, Reason} -> Reason
+       WRef = make_ref(),
+       meck:expect(couch_replicator_doc_processor, get_worker_ref, 1, WRef),
+       Pid = spawn_worker(Id, Rep, 0, WRef),
+       Res = receive  {'DOWN', _Ref, process, Pid, Reason} -> Reason
            after 1000 -> timeout end,
-       Expect = #doc_worker_result{id = Id, result = {ok, ?R1}},
+       Expect = #doc_worker_result{id = Id, wref = WRef, result = {ok, ?R1}},
        ?assertEqual(Expect, Res),
        ?assert(added_job())
    end).
 
 
+% Should not add job if by the time worker got to fetching the filter
+% and getting a replication id, replication doc was deleted
+t_ignore_if_doc_deleted() ->
+   ?_test(begin
+       Id = {?DB, ?DOC1},
+       Rep = couch_replicator_docs:parse_rep_doc_without_id(change()),
+       meck:expect(couch_replicator_doc_processor, get_worker_ref, 1, nil),
+       ?assertEqual(ignore, maybe_start_replication(Id, Rep, make_ref())),
+       ?assertNot(added_job())
+   end).
+
+
+% Should not add job if by the time worker got to fetchign the filter
+% and building a replication id, another worker was spawned.
+t_ignore_if_worker_ref_does_not_match() ->
+    ?_test(begin
+       Id = {?DB, ?DOC1},
+       Rep = couch_replicator_docs:parse_rep_doc_without_id(change()),
+       meck:expect(couch_replicator_doc_processor, get_worker_ref, 1, make_ref()),
+       ?assertEqual(ignore, maybe_start_replication(Id, Rep, make_ref())),
+       ?assertNot(added_job())
+   end).
+
+
 % Test helper functions
 
 setup() ->
@@ -202,6 +237,7 @@ setup() ->
     meck:expect(couch_server, get_uuid, 0, this_is_snek),
     meck:expect(couch_replicator_docs, update_failed, 4, ok),
     meck:expect(couch_replicator_scheduler, rep_state, 1, nil),
+    meck:expect(couch_replicator_doc_processor, get_worker_ref, 1, nil),
     ok.
 
 


Mime
View raw message