From 4e835325cb01ff56aaec4916fcbc3a2356bbb490 Mon Sep 17 00:00:00 2001 From: Maxim Fedorov Date: Sat, 11 Nov 2023 22:12:20 -0800 Subject: [PATCH] [spg] Support for OTP 26 Also includes minor bugfixes which were fixed in upstream. --- .github/workflows/erlang.yml | 2 +- README.md | 8 ++++++ src/spg.app.src | 2 +- src/spg.erl | 48 ++++++++++++++++++------------------ 4 files changed, 34 insertions(+), 26 deletions(-) diff --git a/.github/workflows/erlang.yml b/.github/workflows/erlang.yml index 603beb4..eb00e61 100644 --- a/.github/workflows/erlang.yml +++ b/.github/workflows/erlang.yml @@ -14,7 +14,7 @@ jobs: strategy: matrix: - otp_version: ['21.3.8.9', '22.3.4.9', '23.3.4.9', '24.3.2', '25.0'] + otp_version: ['21.3.8.9', '22.3.4.9', '23.3.4.9', '24.3.2', '25.0', '26.1.2'] os: [ubuntu-latest] container: diff --git a/README.md b/README.md index be0a564..2fedf29 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,10 @@ spg: Scalable Process Groups ===== +> **Warning** +> This project is no longer maintained. OTP 26 is the last supported version. +> Consider switching to `pg`, which is an official part of OTP distribution. + Replacement for Erlang/OTP pg2 implementation. Implements Strong Eventual Consistency (SEC), and an overlay network over Erlang Distribution Cluster. @@ -144,6 +148,10 @@ Feature requests that are being considered (feel free to submit PR implementing ## Changelog +Version 1.2.1: + - tested to OTP 26 support + - minor bugfixes + Version 1.2: - updated for OTP 25 support (property-based test requires OTP 25 now) - removed benchmarking suites, generic cleanup diff --git a/src/spg.app.src b/src/spg.app.src index 7730111..036764a 100644 --- a/src/spg.app.src +++ b/src/spg.app.src @@ -1,6 +1,6 @@ {application, spg, [{description, "Scalable Process Groups"}, - {vsn, "1.2"}, + {vsn, "1.2.1"}, {registered, [spg_sup, spg]}, {mod, {spg_app, []}}, {applications, diff --git a/src/spg.erl b/src/spg.erl index e15330b..38432de 100644 --- a/src/spg.erl +++ b/src/spg.erl @@ -290,13 +290,13 @@ handle_call({leave_local, Group, PidOrPids}, _From, #state{scope = Scope, local handle_call(monitor, {Pid, _Tag}, #state{scope = Scope, scope_monitors = ScopeMon} = State) -> %% next line could also be done with iterating over process state, but it appears to be slower Local = maps:from_list([{G,P} || [G,P] <- ets:match(Scope, {'$1', '$2', '_'})]), - MRef = erlang:monitor(process, Pid), %% monitor the monitor, to discard it upon termination, and generate MRef + MRef = erlang:monitor(process, Pid, [{tag, {'DOWN', scope_monitors}}]), %% monitor the monitor, to discard it upon termination, and generate MRef {reply, {MRef, Local}, State#state{scope_monitors = ScopeMon#{MRef => Pid}}}; handle_call({monitor, Group}, {Pid, _Tag}, #state{scope = Scope, group_monitors = GM, monitored_groups = MG} = State) -> %% ETS cache is writable only from this process - so get_members is safe to use Members = get_members(Scope, Group), - MRef = erlang:monitor(process, Pid), + MRef = erlang:monitor(process, Pid, [{tag, {'DOWN', group_monitors}}]), NewMG = maps:update_with(Group, fun (Ex) -> [{Pid, MRef} | Ex] end, [{Pid, MRef}], MG), {reply, {MRef, Members}, State#state{group_monitors = GM#{MRef => {Pid, Group}}, monitored_groups = NewMG}}; @@ -385,12 +385,14 @@ handle_info({discover, Peer}, State) -> handle_info({discover, Peer, _ProtocolVersion}, State) -> handle_discover(Peer, State); -%% handle local process exit, or a local monitor exit + +%% handle local process exit handle_info({'DOWN', MRef, process, Pid, _Info}, #state{scope = Scope, local = Local, remote = Remote, scope_monitors = ScopeMon, monitored_groups = MG} = State) when node(Pid) =:= node() -> case maps:take(Pid, Local) of error -> - {noreply, maybe_drop_monitor(MRef, State)}; + %% ignore late monitor: this can only happen when leave request and 'DOWN' are in pg queue + {noreply, State}; {{MRef, Groups}, NewLocal} -> [leave_local_update_ets(Scope, ScopeMon, MG, Group, Pid) || Group <- Groups], %% send update to all remote peers @@ -398,16 +400,31 @@ handle_info({'DOWN', MRef, process, Pid, _Info}, #state{scope = Scope, local = L {noreply, State#state{local = NewLocal}} end; -%% handle remote node down or scope leaving overlay network, or a monitor from the remote node went down +%% handle remote node down or scope leaving overlay network handle_info({'DOWN', MRef, process, Pid, _Info}, #state{scope = Scope, remote = Remote, scope_monitors = ScopeMon, monitored_groups = MG} = State) -> case maps:take(Pid, Remote) of {{MRef, RemoteMap}, NewRemote} -> - foreach(fun (Group, Pids) -> + maps:foreach(fun (Group, Pids) -> leave_remote_update_ets(Scope, ScopeMon, MG, Pids, [Group]) end, RemoteMap), {noreply, State#state{remote = NewRemote}}; error -> - {noreply, maybe_drop_monitor(MRef, State)} + {noreply, State} + end; + +%% handle scope monitor exiting +handle_info({{'DOWN', scope_monitors}, MRef, process, _Pid, _Info}, #state{scope_monitors = ScopeMon} = State) -> + {noreply, State#state{scope_monitors = maps:remove(MRef, ScopeMon)}}; + +%% handle group monitor exiting +handle_info({{'DOWN', group_monitors}, MRef, process, Pid, _Info}, #state{ + group_monitors = GMs, monitored_groups = MG} = State) -> + case maps:take(MRef, GMs) of + error -> + {noreply, State}; + {{Pid, Group}, NewGM} -> + %% clean up the inverse map + {noreply, State#state{group_monitors = NewGM, monitored_groups = demonitor_group({Pid, MRef}, Group, MG)}} end; %% nodedown: ignore, and wait for 'DOWN' signal for monitored process @@ -654,23 +671,6 @@ broadcast([Dest | Tail], Msg) -> erlang:send(Dest, Msg, [noconnect]), broadcast(Tail, Msg). -%% drops a monitor if DOWN was received -maybe_drop_monitor(MRef, #state{scope_monitors = ScopeMon, group_monitors = GMs, monitored_groups = MG} = State) -> - %% could be a local monitor going DOWN. Since it's a rare event, check should - %% not stay in front of any other, more frequent events - case maps:take(MRef, ScopeMon) of - error -> - case maps:take(MRef, GMs) of - error -> - State; - {{Pid, Group}, NewGM} -> - %% clean up the inverse map - State#state{group_monitors = NewGM, monitored_groups = demonitor_group({Pid, MRef}, Group, MG)} - end; - {_Pid, NewScopeMon} -> - State#state{scope_monitors = NewScopeMon} - end. - demonitor_group(Tag, Group, MG) -> case maps:find(Group, MG) of {ok, [Tag]} ->