From 4fb3c75f3db7b8a4bec28e9f0369e63cce52753b Mon Sep 17 00:00:00 2001 From: Filipe David Borba Manana Date: Wed, 2 Feb 2011 06:24:55 +0000 Subject: Merged revision 1066340 from trunk Replicator DB: make number of replication retry attempts configurable Closes COUCHDB-1051 git-svn-id: https://svn.apache.org/repos/asf/couchdb/branches/1.1.x@1066342 13f79535-47bb-0310-9956-ffa450edef68 --- src/couchdb/couch_rep_db_listener.erl | 55 ++++++++++++++++++++--------------- 1 file changed, 31 insertions(+), 24 deletions(-) (limited to 'src/couchdb/couch_rep_db_listener.erl') diff --git a/src/couchdb/couch_rep_db_listener.erl b/src/couchdb/couch_rep_db_listener.erl index 819f167d..89a18e85 100644 --- a/src/couchdb/couch_rep_db_listener.erl +++ b/src/couchdb/couch_rep_db_listener.erl @@ -20,14 +20,14 @@ -define(DOC_ID_TO_REP_ID, rep_doc_id_to_rep_id). -define(REP_ID_TO_DOC_ID, rep_id_to_rep_doc_id). --define(MAX_RETRIES, 10). -define(INITIAL_WAIT, 5). -record(state, { changes_feed_loop = nil, db_notifier = nil, rep_db_name = nil, - rep_start_pids = [] + rep_start_pids = [], + max_retries }). -import(couch_util, [ @@ -41,40 +41,42 @@ start_link() -> init(_) -> process_flag(trap_exit, true), - ?DOC_ID_TO_REP_ID = ets:new(?DOC_ID_TO_REP_ID, [named_table, set, private]), - ?REP_ID_TO_DOC_ID = ets:new(?REP_ID_TO_DOC_ID, [named_table, set, private]), + _ = ets:new(?DOC_ID_TO_REP_ID, [named_table, set, protected]), + _ = ets:new(?REP_ID_TO_DOC_ID, [named_table, set, private]), Server = self(), ok = couch_config:register( fun("replicator", "db", NewName) -> - ok = gen_server:cast(Server, {rep_db_changed, ?l2b(NewName)}) + ok = gen_server:cast(Server, {rep_db_changed, ?l2b(NewName)}); + ("replicator", "max_replication_retry_count", NewMaxRetries1) -> + NewMaxRetries = list_to_integer(NewMaxRetries1), + ok = gen_server:cast(Server, {set_max_retries, NewMaxRetries}) end ), {Loop, RepDbName} = changes_feed_loop(), {ok, #state{ changes_feed_loop = Loop, rep_db_name = RepDbName, - db_notifier = db_update_notifier()} - }. + db_notifier = db_update_notifier(), + max_retries = list_to_integer( + couch_config:get("replicator", "max_replication_retry_count", "10")) + }}. handle_call({rep_db_update, Change}, _From, State) -> {reply, ok, process_update(State, Change)}; handle_call({triggered, {BaseId, _}}, _From, State) -> - case ets:lookup(?REP_ID_TO_DOC_ID, BaseId) of - [{BaseId, {DocId, true}}] -> - true = ets:insert(?REP_ID_TO_DOC_ID, {BaseId, {DocId, false}}); - _ -> - ok - end, + [{BaseId, {DocId, true}}] = ets:lookup(?REP_ID_TO_DOC_ID, BaseId), + true = ets:insert(?REP_ID_TO_DOC_ID, {BaseId, {DocId, false}}), {reply, ok, State}; handle_call({restart_failure, {Props} = RepDoc, Error}, _From, State) -> DocId = get_value(<<"_id">>, Props), - [{DocId, {BaseId, _} = RepId}] = ets:lookup(?DOC_ID_TO_REP_ID, DocId), + [{DocId, {{BaseId, _} = RepId, MaxRetries}}] = ets:lookup( + ?DOC_ID_TO_REP_ID, DocId), ?LOG_ERROR("Failed to start replication `~s` after ~p attempts using " "the document `~s`. Last error reason was: ~p", - [pp_rep_id(RepId), ?MAX_RETRIES, DocId, Error]), + [pp_rep_id(RepId), MaxRetries, DocId, Error]), couch_rep:update_rep_doc( RepDoc, [{<<"_replication_state">>, <<"error">>}, @@ -103,6 +105,9 @@ handle_cast({rep_db_created, NewName}, handle_cast({rep_db_created, _NewName}, State) -> {noreply, restart(State)}; +handle_cast({set_max_retries, MaxRetries}, State) -> + {noreply, State#state{max_retries = MaxRetries}}; + handle_cast(Msg, State) -> ?LOG_ERROR("Replicator DB listener received unexpected cast ~p", [Msg]), {stop, {error, {unexpected_cast, Msg}}, State}. @@ -259,16 +264,17 @@ rep_user_ctx({RepDoc}) -> end. -maybe_start_replication(State, DocId, JsonRepDoc) -> +maybe_start_replication(#state{max_retries = MaxRetries} = State, + DocId, JsonRepDoc) -> UserCtx = rep_user_ctx(JsonRepDoc), {BaseId, _} = RepId = couch_rep:make_replication_id(JsonRepDoc, UserCtx), case ets:lookup(?REP_ID_TO_DOC_ID, BaseId) of [] -> true = ets:insert(?REP_ID_TO_DOC_ID, {BaseId, {DocId, true}}), - true = ets:insert(?DOC_ID_TO_REP_ID, {DocId, RepId}), + true = ets:insert(?DOC_ID_TO_REP_ID, {DocId, {RepId, MaxRetries}}), Server = self(), Pid = spawn_link(fun() -> - start_replication(Server, JsonRepDoc, RepId, UserCtx) + start_replication(Server, JsonRepDoc, RepId, UserCtx, MaxRetries) end), State#state{rep_start_pids = [Pid | State#state.rep_start_pids]}; [{BaseId, {DocId, _}}] -> @@ -295,7 +301,7 @@ maybe_tag_rep_doc({Props} = JsonRepDoc, RepId) -> end. -start_replication(Server, {RepProps} = RepDoc, RepId, UserCtx) -> +start_replication(Server, {RepProps} = RepDoc, RepId, UserCtx, MaxRetries) -> case (catch couch_rep:start_replication(RepDoc, RepId, UserCtx)) of Pid when is_pid(Pid) -> ?LOG_INFO("Document `~s` triggered replication `~s`", @@ -304,7 +310,7 @@ start_replication(Server, {RepProps} = RepDoc, RepId, UserCtx) -> couch_rep:get_result(Pid, RepId, RepDoc, UserCtx); Error -> keep_retrying( - Server, RepId, RepDoc, UserCtx, Error, ?INITIAL_WAIT, ?MAX_RETRIES) + Server, RepId, RepDoc, UserCtx, Error, ?INITIAL_WAIT, MaxRetries) end. @@ -319,9 +325,10 @@ keep_retrying(Server, RepId, RepDoc, UserCtx, Error, Wait, RetriesLeft) -> Pid when is_pid(Pid) -> ok = gen_server:call(Server, {triggered, RepId}, infinity), {RepProps} = RepDoc, + DocId = get_value(<<"_id">>, RepProps), + [{DocId, {RepId, MaxRetries}}] = ets:lookup(?DOC_ID_TO_REP_ID, DocId), ?LOG_INFO("Document `~s` triggered replication `~s` after ~p attempts", - [get_value(<<"_id">>, RepProps), pp_rep_id(RepId), - ?MAX_RETRIES - RetriesLeft + 1]), + [DocId, pp_rep_id(RepId), MaxRetries - RetriesLeft + 1]), couch_rep:get_result(Pid, RepId, RepDoc, UserCtx); NewError -> keep_retrying( @@ -351,7 +358,7 @@ replication_complete(DocId) -> stop_replication(DocId) -> case ets:lookup(?DOC_ID_TO_REP_ID, DocId) of - [{DocId, {BaseId, _} = RepId}] -> + [{DocId, {{BaseId, _} = RepId, _MaxRetries}}] -> couch_rep:end_replication(RepId), true = ets:delete(?REP_ID_TO_DOC_ID, BaseId), true = ets:delete(?DOC_ID_TO_REP_ID, DocId), @@ -365,7 +372,7 @@ stop_all_replications() -> ?LOG_INFO("Stopping all ongoing replications because the replicator DB " "was deleted or changed", []), ets:foldl( - fun({_, RepId}, _) -> couch_rep:end_replication(RepId) end, + fun({_, {RepId, _}}, _) -> couch_rep:end_replication(RepId) end, ok, ?DOC_ID_TO_REP_ID ), -- cgit v1.2.3