You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2022/12/01 20:29:36 UTC
[couchdb] branch remove-global created (now b5d96d744)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch remove-global
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at b5d96d744 Remove all usage of global
This branch includes the following new commits:
new b5d96d744 Remove all usage of global
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[couchdb] 01/01: Remove all usage of global
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch remove-global
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit b5d96d744b4c138f08059f36afe32b5c7e52d081
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Thu Dec 1 15:02:24 2022 -0500
Remove all usage of global
Global has a tendency to create deadlocks [1], and when that happens
replication jobs can't start on any of the nodes in the cluster.
We don't really need strict consistent locking for replication jobs. It's
mostly to avoid replication jobs thrashing the same checkpoint docs back and
forth between different session IDs. So, remove global to avoid any issues
around it, and replace it with `pg` -- the new (Erlang 23+) process group
implementation. (Technically `global` is still running in the runtime system as
it's started by the `kernel` app. We just avoid interacting with it and
registering any names to avoid deadlocks).
In `pg` we start a replication `scope`, and then in that scope make every RepId
a process `group`. When replication processes spawn, their Pids becomes
`members` of that group:
```
couch_replicator_pg (scope):
"a12c+create_target" (RepId group):
[Pid1, Pid2] (members)
...
```
As per `pg` implementation, groups are created and remove automatically as
members are added/removed from it, so we don't have to do anything there.
If there are already any running Pids in the same group, we avoid starting the
jobs, and fail like we did before when we used global. In the even more rare
case of a race condition, when 2+ jobs do manage to start, we do a membership
check before each checkpoint. One of the jobs then stops to yield to another.
For simplicity pick the one running on the lowest lexicographically sorted node
name to survive.
[1] https://github.com/erlang/otp/issues/6524
[2] https://www.erlang.org/doc/man/pg.html
---
src/couch_event/src/couch_event_listener.erl | 6 --
src/couch_replicator/src/couch_replicator_pg.erl | 86 ++++++++++++++++++++++
.../src/couch_replicator_scheduler_job.erl | 66 +++++++++++------
src/couch_replicator/src/couch_replicator_sup.erl | 1 +
.../test/eunit/couch_replicator_test_helper.erl | 2 +-
5 files changed, 133 insertions(+), 28 deletions(-)
diff --git a/src/couch_event/src/couch_event_listener.erl b/src/couch_event/src/couch_event_listener.erl
index 40f1a5c65..22bf23327 100644
--- a/src/couch_event/src/couch_event_listener.erl
+++ b/src/couch_event/src/couch_event_listener.erl
@@ -175,14 +175,8 @@ do_terminate(Reason, #st{module = Module, state = State}) ->
end,
erlang:exit(Status).
-where({global, Name}) -> global:whereis_name(Name);
where({local, Name}) -> whereis(Name).
-name_register({global, Name} = GN) ->
- case global:register_name(Name, self()) of
- yes -> true;
- no -> {false, where(GN)}
- end;
name_register({local, Name} = LN) ->
try register(Name, self()) of
true -> true
diff --git a/src/couch_replicator/src/couch_replicator_pg.erl b/src/couch_replicator/src/couch_replicator_pg.erl
new file mode 100644
index 000000000..8d05f2b38
--- /dev/null
+++ b/src/couch_replicator/src/couch_replicator_pg.erl
@@ -0,0 +1,86 @@
+% 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.
+
+% Use pg process groups to reduce the chance of duplicate replication jobs
+% running on the same cluster.
+%
+% A custom replicator pg group is started via start_link/0. Then, replication
+% jobs check if they would be leaders before starting. If, by chance, two jobs
+% with the same RepId start anyway, then replication jobs would do an extra
+% check before each checkpoint. If the they are not leaders any longer, they
+% should stop running. The "leader" is just the first sorted element in the
+% [node(Pid), ...] list.
+
+-module(couch_replicator_pg).
+
+-export([
+ start_link/0,
+ join/2,
+ leave/2,
+ pids/1,
+ should_start/2,
+ should_run/2
+]).
+
+% Start a custom pg group. Should be called from the replication supervisor.
+%
+start_link() ->
+ pg:start_link(?MODULE).
+
+% Join a replication job pid to a RepId group
+%
+join({_, _} = RepId, Pid) when is_pid(Pid) ->
+ pg:join(?MODULE, id(RepId), Pid).
+
+% Leave a replication RepId group. This doesn't have to be called explicitly as
+% the processes are monitored and automatically removed by pg. It may be nice,
+% to call it from terminate/2 to speed things along a bit and clear the group
+% quicker.
+%
+leave({_, _} = RepId, Pid) when is_pid(Pid) ->
+ pg:leave(?MODULE, id(RepId), Pid).
+
+% Determine if a replication job should start on a particular node. If it
+% should, return `yes`, otherwise return `{no, OtherPid}`. `OtherPid` is
+% the pid of the replication job that is already running.
+%
+should_start({_, _} = RepId, Node) when is_atom(Node) ->
+ no_other_nodes(Node, pids(RepId)).
+
+% Determine if the replication job should keep running as the main job for that
+% RepId. If it is, return yes, otherwise return `{no, OtherPid}`. `OtherPid` is
+% the pid of the replication job that should stay running instead of this one.
+%
+should_run({_, _} = RepId, Pid) when is_pid(Pid) ->
+ case pids(RepId) of
+ [OtherPid | _] when OtherPid =/= Pid -> {no, OtherPid};
+ _ -> yes
+ end.
+
+% Sort all the pids by node first to get some deterministic order. For all we
+% know, pids may already sort that way, but we're just making it explicit here
+% in case it somehow changes in the future.
+%
+pids({_, _} = RepId) ->
+ NodePids = [{node(P), P} || P <- pg:get_members(?MODULE, id(RepId))],
+ {_, Pids} = lists:unzip(lists:usort(NodePids)),
+ Pids.
+
+id({Base, Ext}) ->
+ iolist_to_binary([Base, Ext]).
+
+no_other_nodes(_, []) ->
+ yes;
+no_other_nodes(Node, [Pid | _]) when Node =/= node(Pid) ->
+ {no, Pid};
+no_other_nodes(Node, [Pid | Pids]) when Node =:= node(Pid) ->
+ no_other_nodes(Node, Pids).
diff --git a/src/couch_replicator/src/couch_replicator_scheduler_job.erl b/src/couch_replicator/src/couch_replicator_scheduler_job.erl
index 1d3e70c5a..416220efd 100644
--- a/src/couch_replicator/src/couch_replicator_scheduler_job.erl
+++ b/src/couch_replicator/src/couch_replicator_scheduler_job.erl
@@ -81,21 +81,25 @@
view = nil
}).
-start_link(#rep{id = {BaseId, Ext}, source = Src, target = Tgt} = Rep) ->
+start_link(#rep{id = Id = {BaseId, Ext}, source = Src, target = Tgt} = Rep) ->
RepChildId = BaseId ++ Ext,
Source = couch_replicator_api_wrap:db_uri(Src),
Target = couch_replicator_api_wrap:db_uri(Tgt),
- ServerName = {global, {?MODULE, Rep#rep.id}},
-
- case gen_server:start_link(ServerName, ?MODULE, Rep, []) of
- {ok, Pid} ->
- {ok, Pid};
- {error, Reason} ->
- couch_log:warning(
- "failed to start replication `~s` (`~s` -> `~s`)",
- [RepChildId, Source, Target]
- ),
- {error, Reason}
+ case couch_replicator_pg:should_start(Id, node()) of
+ yes ->
+ case gen_server:start_link({local, ?MODULE}, ?MODULE, Rep, []) of
+ {ok, Pid} ->
+ couch_replicator_pg:join(Id, Pid),
+ {ok, Pid};
+ {error, Reason} ->
+ couch_log:warning(
+ "failed to start replication `~s` (`~s` -> `~s`)",
+ [RepChildId, Source, Target]
+ ),
+ {error, Reason}
+ end;
+ {no, OtherPid} ->
+ {error, {already_started, OtherPid}}
end.
init(InitArgs) ->
@@ -234,14 +238,19 @@ handle_call({report_seq_done, Seq, StatsInc}, From, State) ->
handle_cast({sum_stats, Stats}, State) ->
NewStats = couch_replicator_utils:sum_stats(State#rep_state.stats, Stats),
{noreply, State#rep_state{stats = NewStats}};
-handle_cast(checkpoint, State) ->
- case do_checkpoint(State) of
- {ok, NewState} ->
- couch_stats:increment_counter([couch_replicator, checkpoints, success]),
- {noreply, NewState#rep_state{timer = start_timer(State)}};
- Error ->
- couch_stats:increment_counter([couch_replicator, checkpoints, failure]),
- {stop, Error, State}
+handle_cast(checkpoint, #rep_state{rep_details = Rep} = State) ->
+ case couch_replicator_pg:should_run(Rep#rep.id, self()) of
+ yes ->
+ case do_checkpoint(State) of
+ {ok, NewState} ->
+ couch_stats:increment_counter([couch_replicator, checkpoints, success]),
+ {noreply, NewState#rep_state{timer = start_timer(State)}};
+ Error ->
+ couch_stats:increment_counter([couch_replicator, checkpoints, failure]),
+ {stop, Error, State}
+ end;
+ {no, OtherPid} ->
+ {stop, {shutdown, {duplicate_job, OtherPid}}, State}
end;
handle_cast(
{report_seq, Seq},
@@ -367,6 +376,7 @@ terminate(shutdown, #rep_state{rep_details = #rep{id = RepId}} = State) ->
terminate_cleanup(State1);
terminate({shutdown, max_backoff}, {error, InitArgs}) ->
#rep{id = {BaseId, Ext} = RepId} = InitArgs,
+ couch_replicator_pg:leave(RepId, self()),
couch_stats:increment_counter([couch_replicator, failed_starts]),
couch_log:warning("Replication `~s` reached max backoff ", [BaseId ++ Ext]),
couch_replicator_notifier:notify({error, RepId, max_backoff});
@@ -378,6 +388,7 @@ terminate({shutdown, {error, Error}}, {error, Class, Stack, InitArgs}) ->
doc_id = DocId,
db_name = DbName
} = InitArgs,
+ couch_replicator_pg:leave(RepId, self()),
Source = couch_replicator_api_wrap:db_uri(Source0),
Target = couch_replicator_api_wrap:db_uri(Target0),
RepIdStr = BaseId ++ Ext,
@@ -406,6 +417,18 @@ terminate({shutdown, max_backoff}, State) ->
),
terminate_cleanup(State),
couch_replicator_notifier:notify({error, RepId, max_backoff});
+terminate({shutdown, {duplicate_job, OtherPid}}, State) ->
+ #rep_state{
+ source_name = Source,
+ target_name = Target,
+ rep_details = #rep{id = {BaseId, Ext} = RepId}
+ } = State,
+ couch_log:error(
+ "Replication `~s` (`~s` -> `~s`) with pid ~p was usurped by ~p on node ~p",
+ [BaseId ++ Ext, Source, Target, self(), OtherPid, node(OtherPid)]
+ ),
+ terminate_cleanup(State),
+ couch_replicator_notifier:notify({error, RepId, duplicate_job});
terminate({shutdown, Reason}, State) ->
% Unwrap so when reporting we don't have an extra {shutdown, ...} tuple
% wrapped around the message
@@ -423,7 +446,8 @@ terminate(Reason, State) ->
terminate_cleanup(State),
couch_replicator_notifier:notify({error, RepId, Reason}).
-terminate_cleanup(State) ->
+terminate_cleanup(#rep_state{rep_details = #rep{id = RepId}} = State) ->
+ couch_replicator_pg:leave(RepId, self()),
update_task(State),
couch_replicator_api_wrap:db_close(State#rep_state.source),
couch_replicator_api_wrap:db_close(State#rep_state.target).
diff --git a/src/couch_replicator/src/couch_replicator_sup.erl b/src/couch_replicator/src/couch_replicator_sup.erl
index 33eee8659..1cf973f71 100644
--- a/src/couch_replicator/src/couch_replicator_sup.erl
+++ b/src/couch_replicator/src/couch_replicator_sup.erl
@@ -22,6 +22,7 @@ init(_Args) ->
Children = [
{couch_replication_event, {gen_event, start_link, [{local, couch_replication}]}, permanent,
brutal_kill, worker, dynamic},
+ {couch_replicator_pg, {couch_replicator_pg, start_link, []}, permanent, 1000, worker, [pg]},
{couch_replicator_clustering, {couch_replicator_clustering, start_link, []}, permanent,
brutal_kill, worker, [couch_replicator_clustering]},
{couch_replicator_connection, {couch_replicator_connection, start_link, []}, permanent,
diff --git a/src/couch_replicator/test/eunit/couch_replicator_test_helper.erl b/src/couch_replicator/test/eunit/couch_replicator_test_helper.erl
index f862527f4..e28b4a28b 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_test_helper.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_test_helper.erl
@@ -160,7 +160,7 @@ cluster_db_url(Path) ->
<<(cluster_url())/binary, "/", Path/binary>>.
get_pid(RepId) ->
- Pid = global:whereis_name({couch_replicator_scheduler_job, RepId}),
+ [Pid] = couch_replicator_pg:pids(RepId),
?assert(is_pid(Pid)),
Pid.