[spg] Support for OTP 26

.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:

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

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,

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,29 +385,46 @@ 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
             broadcast(maps:keys(Remote), {leave, self(), Pid, Groups}),
             {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]} ->