rabbit_process: Move process liveness functions here from rabbit_mnesia

These functions extend the functionality of `erlang:is_process_alive/1`
to take into account the node a process is running on and its cluster
membership.

These functions are moved away from `rabbit_mnesia` because we don't
want `rabbit_mnesia` to be a central piece of RabbitMQ.

Classic-mirrored-queue-related modules continue to use `rabbit_mnesia`
functions, therefore relying on Mnesia, because they depend entirely on
Mnesia anyway. They will go away at the same time as our use of Mnesia.
So by keeping this code untouched, we avoid possible regressions.

Author: dumbbell

deps/rabbit/src/rabbit_amqqueue.erl

@@ -574,7 +574,7 @@ retry_wait(Q, F, E, RetriesLeft) ->
         {stopped, false} ->
             E({absent, Q, stopped});
         _ ->
-            case rabbit_mnesia:is_process_alive(QPid) of
+            case rabbit_process:is_process_alive(QPid) of
                 true ->
                     % rabbitmq-server#1682
                     % The old check would have crashed here,
@@ -1814,7 +1814,7 @@ is_dead_exclusive(Q) when ?amqqueue_exclusive_owner_is(Q, none) ->
     false;
 is_dead_exclusive(Q) when ?amqqueue_exclusive_owner_is_pid(Q) ->
     Pid = amqqueue:get_pid(Q),
-    not rabbit_mnesia:is_process_alive(Pid).
+    not rabbit_process:is_process_alive(Pid).
 
 -spec has_synchronised_mirrors_online(amqqueue:amqqueue()) -> boolean().
 has_synchronised_mirrors_online(Q) ->
@@ -1881,7 +1881,7 @@ on_node_down(Node) ->
 filter_transient_queues_to_delete(Node) ->
     fun(Q) ->
             amqqueue:qnode(Q) == Node andalso
-                not rabbit_mnesia:is_process_alive(amqqueue:get_pid(Q))
+                not rabbit_process:is_process_alive(amqqueue:get_pid(Q))
                 andalso (not amqqueue:is_classic(Q) orelse not amqqueue:is_durable(Q))
                 andalso (not rabbit_amqqueue:is_replicated(Q)
                          orelse rabbit_amqqueue:is_dead_exclusive(Q))

deps/rabbit/src/rabbit_classic_queue.erl

@@ -129,7 +129,7 @@ is_recoverable(Q) when ?is_amqqueue(Q) ->
     %% the policy). Thus, we must check if the local pid is alive
     %% - if the record is present - in order to restart.
     (not rabbit_db_queue:consistent_exists(amqqueue:get_name(Q))
-     orelse not rabbit_mnesia:is_process_alive(amqqueue:get_pid(Q))).
+     orelse not rabbit_process:is_process_alive(amqqueue:get_pid(Q))).
 
 recover(VHost, Queues) ->
     {ok, BQ} = application:get_env(rabbit, backing_queue_module),
@@ -438,11 +438,11 @@ wait_for_promoted_or_stopped(Q0) ->
         {ok, Q} ->
             QPid = amqqueue:get_pid(Q),
             SPids = amqqueue:get_slave_pids(Q),
-            case rabbit_mnesia:is_process_alive(QPid) of
+            case rabbit_process:is_process_alive(QPid) of
                 true  -> {promoted, Q};
                 false ->
                     case lists:any(fun(Pid) ->
-                                       rabbit_mnesia:is_process_alive(Pid)
+                                       rabbit_process:is_process_alive(Pid)
                                    end, SPids) of
                         %% There is a live slave. May be promoted
                         true ->

deps/rabbit/src/rabbit_mnesia.erl

@@ -56,6 +56,13 @@
 %% Used internally in rpc calls
 -export([node_info/0, remove_node_if_mnesia_running/1]).
 
+-deprecated({on_running_node, 1,
+             "Use rabbit_process:on_running_node/1 instead"}).
+-deprecated({is_process_alive, 1,
+             "Use rabbit_process:is_process_alive/1 instead"}).
+-deprecated({is_registered_process_alive, 1,
+             "Use rabbit_process:is_registered_process_alive/1 instead"}).
+
 -ifdef(TEST).
 -compile(export_all).
 -export([init_with_lock/3]).

deps/rabbit/src/rabbit_prequeue.erl

@@ -55,9 +55,9 @@ init(Q0, restart) when ?is_amqqueue(Q0) ->
     QPid = amqqueue:get_pid(Q1),
     SPids = amqqueue:get_slave_pids(Q1),
     LocalOrMasterDown = node(QPid) =:= node()
-        orelse not rabbit_mnesia:on_running_node(QPid),
-    Slaves = [SPid || SPid <- SPids, rabbit_mnesia:is_process_alive(SPid)],
-    case rabbit_mnesia:is_process_alive(QPid) of
+        orelse not rabbit_process:on_running_node(QPid),
+    Slaves = [SPid || SPid <- SPids, rabbit_process:is_process_alive(SPid)],
+    case rabbit_process:is_process_alive(QPid) of
         true  -> false = LocalOrMasterDown, %% assertion
                  rabbit_mirror_queue_slave:go(self(), async),
                  rabbit_mirror_queue_slave:init(Q1); %% [1]

deps/rabbit/src/rabbit_process.erl

@@ -0,0 +1,77 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_process).
+
+-export([on_running_node/1,
+         is_process_alive/1,
+         is_registered_process_alive/1]).
+
+-spec on_running_node(Pid) -> OnRunningNode when
+      Pid :: pid(),
+      OnRunningNode :: boolean().
+%% @doc Indicates if the specified process runs on a member of the cluster.
+%%
+%% @param Pid the PID of the process to check
+%% @returns true if the process runs on one of the cluster members, false
+%% otherwise.
+
+on_running_node(Pid) ->
+    Node = node(Pid),
+    rabbit_nodes:is_running(Node).
+
+%% This requires the process be in the same running cluster as us
+%% (i.e. not partitioned or some random node).
+%%
+%% See also rabbit_misc:is_process_alive/1 which does not.
+
+-spec is_process_alive(Pid) -> IsAlive when
+      Pid :: pid() | {RegisteredName, Node},
+      RegisteredName :: atom(),
+      Node :: node(),
+      IsAlive :: boolean().
+%% @doc Indicates if the specified process is alive.
+%%
+%% Unlike {@link erlang:is_process_alive/1}, this function works with remote
+%% processes and registered processes. However, the process must run on one of
+%% the cluster members.
+%%
+%% @param Pid the PID or name of the process to check
+%% @returns true if the process is alive runs on one of the cluster members,
+%% false otherwise.
+
+is_process_alive(Pid) when is_pid(Pid) ->
+    on_running_node(Pid)
+    andalso
+    rpc:call(node(Pid), erlang, is_process_alive, [Pid]) =:= true;
+is_process_alive({Name, Node}) when is_atom(Name) andalso is_atom(Node) ->
+    case rabbit_nodes:is_running(Node) of
+        true ->
+            try
+                erpc:call(Node, ?MODULE, is_registered_process_alive, [Name])
+            catch
+                error:{exception, undef, [{?MODULE, _, _, _} | _]} ->
+                    rpc:call(
+                      Node,
+                      rabbit_mnesia, is_registered_process_alive, [Name])
+            end;
+        false ->
+            false
+    end.
+
+-spec is_registered_process_alive(RegisteredName) -> IsAlive when
+      RegisteredName :: atom(),
+      IsAlive :: boolean().
+%% @doc Indicates if the specified registered process is alive.
+%%
+%% The process must be local to this node.
+%%
+%% @param RegisteredName the name of the registered process
+%% @returns true if the process is alive, false otherwise.
+
+is_registered_process_alive(Name) ->
+    is_pid(whereis(Name)).

deps/rabbit_common/src/rabbit_misc.erl

@@ -881,7 +881,7 @@ ntoab(IP) ->
 %% loop in rabbit_amqqueue:on_node_down/1 and any delays we incur
 %% would be bad news.
 %%
-%% See also rabbit_mnesia:is_process_alive/1 which also requires the
+%% See also rabbit_process:is_process_alive/1 which also requires the
 %% process be in the same running cluster as us (i.e. not partitioned
 %% or some random node).
 is_process_alive(Pid) when node(Pid) =:= node() ->