perf: add rpc exec worker pool

Author: terry-xiaoyu

.github/workflows/run_test_case.yaml

@@ -19,7 +19,7 @@ jobs:
           run: |
             make test
 
-        - uses: actions/upload-artifact@v1
+        - uses: actions/upload-artifact@v4
           if: always()
           with:
             name: logs

src/gen_rpc_acceptor.erl

@@ -340,13 +340,8 @@ handle_cast(M, F, A, Ordered, #state{socket=Socket, driver=Driver, peer=Peer, co
             ?log(debug, "event=request_not_allowed driver=~s socket=\"~s\" control=~s method=cast module=~s",[Driver, gen_rpc_helper:socket_to_string(Socket), Control, RealM])
     end.
 
-exec_cast(M, F, A, _PreserveOrder = true) ->
-    {Pid, MRef} = erlang:spawn_monitor(M, F, A),
-    receive
-        {'DOWN', MRef, process, Pid, _} -> ok
-    end;
-exec_cast(M, F, A, _PreserveOrder = false) ->
-    _ = erlang:spawn(M, F, A),
+exec_cast(M, F, A, PreserveOrder) ->
+    ok = gen_rpc_exec_worker_sup:submit(M, F, A, PreserveOrder),
     ok.
 
 reply_immediately(Payload, #state{driver_mod = DriverMod, driver = Driver, socket = Socket}) ->

src/gen_rpc_exec_worker.erl

@@ -0,0 +1,65 @@
+%%% -*-mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*-
+%%% ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et:
+
+-module(gen_rpc_exec_worker).
+
+%%% Behaviour
+-behaviour(gen_server).
+
+-include("logger.hrl").
+
+%%% API
+-export([start_link/1, worker_name/1]).
+
+%%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+%%% ===================================================
+%%% API
+%%% ===================================================
+-spec start_link(pos_integer()) -> gen_server:startlink_ret().
+start_link(Index) when is_integer(Index), Index > 0 ->
+    gen_server:start_link({via, gen_rpc_registry, worker_name(Index)}, ?MODULE, Index, []).
+
+-spec worker_name(pos_integer()) -> {exec_cast_worker, pos_integer()}.
+worker_name(Index) when is_integer(Index), Index > 0 ->
+    {exec_cast_worker, Index}.
+
+%%% ===================================================
+%%% gen_server callbacks
+%%% ===================================================
+init(Index) ->
+    {ok, Index}.
+
+handle_call(_Request, _From, State) ->
+    {reply, {error, unsupported_call}, State}.
+
+handle_cast({exec_cast, M, F, A}, State) ->
+    execute_cast(M, F, A),
+    {noreply, State};
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+terminate(_Reason, _State) ->
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%% ===================================================
+%%% Internal functions
+%%% ===================================================
+execute_cast(M, F, A) ->
+    try erlang:apply(M, F, A) of
+        _Result ->
+            ok
+    catch
+        Class:Reason:Stacktrace ->
+            ?log(error,
+                 "event=exec_cast_failed module=~p function=~p args=\"~0p\" class=~p reason=\"~0p\" stacktrace=\"~0p\"",
+                 [M, F, A, Class, Reason, Stacktrace]),
+            ok
+    end.

src/supervisor/gen_rpc_exec_worker_sup.erl

@@ -0,0 +1,77 @@
+%%% -*-mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*-
+%%% ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et:
+
+-module(gen_rpc_exec_worker_sup).
+
+%%% Behaviour
+-behaviour(supervisor).
+
+-include("logger.hrl").
+
+%%% API
+-export([start_link/0, submit/4]).
+
+%%% Supervisor callbacks
+-export([init/1]).
+
+-define(FIRST_WORKER_INDEX, 1).
+
+%%% ===================================================
+%%% API
+%%% ===================================================
+-spec start_link() -> supervisor:startlink_ret().
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+-spec submit(module(), atom(), list(), boolean()) -> ok.
+submit(M, F, A, PreserveOrder) ->
+    Name = gen_rpc_exec_worker:worker_name(worker_index(PreserveOrder)),
+    case gen_rpc_registry:whereis_name(Name) of
+        undefined ->
+            ?log(notice, "event=exec_cast_worker_unavailable, name=~p", [Name]),
+            exec_cast(M, F, A, PreserveOrder);
+        Pid ->
+            gen_server:cast(Pid, {exec_cast, M, F, A}),
+            ok
+    end.
+
+%%% ===================================================
+%%% Supervisor callbacks
+%%% ===================================================
+init([]) ->
+    Children = [worker_spec(Index) || Index <- lists:seq(1, worker_count())],
+    {ok, {{one_for_one, 15, 1}, Children}}.
+
+%%% ===================================================
+%%% Internal functions
+%%% ===================================================
+%%% The `exec_cast/4` function is a fallback that executes the cast directly if the worker is unavailable.
+%%% Usually happens when relup failed to start the worker pool.
+exec_cast(M, F, A, _PreserveOrder = true) ->
+    {Pid, MRef} = erlang:spawn_monitor(M, F, A),
+    receive
+        {'DOWN', MRef, process, Pid, _} -> ok
+    end;
+exec_cast(M, F, A, _PreserveOrder = false) ->
+    _ = erlang:spawn(M, F, A),
+    ok.
+
+worker_spec(Index) ->
+    Name = gen_rpc_exec_worker:worker_name(Index),
+    {Name, {gen_rpc_exec_worker, start_link, [Index]}, permanent, 5000, worker, [gen_rpc_exec_worker]}.
+
+worker_count() ->
+    erlang:system_info(schedulers).
+
+worker_index(true) ->
+    ?FIRST_WORKER_INDEX;
+worker_index(false) ->
+    random_worker_index().
+
+random_worker_index() ->
+    case worker_count() of
+        1 ->
+            1;
+        Count ->
+            erlang:phash2(make_ref(), Count) + 1
+    end.

src/supervisor/gen_rpc_sup.erl

@@ -29,6 +29,7 @@ start_link() ->
 init([]) ->
     {ok, {{one_for_one, 100, 1}, [
         {gen_rpc_registry, {gen_rpc_registry, start_link, []}, permanent, 5000, worker, [gen_rpc_registry]},
+        {gen_rpc_exec_worker_sup, {gen_rpc_exec_worker_sup, start_link, []}, permanent, 5000, supervisor, [gen_rpc_exec_worker_sup]},
         {gen_rpc_server_tcp, {gen_rpc_server,start_link,[tcp]}, permanent, 5000, worker, [gen_rpc_server]},
         {gen_rpc_server_ssl, {gen_rpc_server,start_link,[ssl]}, permanent, 5000, worker, [gen_rpc_server]},
         {gen_rpc_acceptor_sup, {gen_rpc_acceptor_sup,start_link, []}, permanent, 5000, supervisor, [gen_rpc_acceptor_sup]},