mc_gen_rpc

%%% -*-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:
%%%
%%% Copyright 2015 Panagiotis Papadomitsos. All Rights Reserved.
%%% Copyright 2021 Miniclip. All Rights Reserved.
%%%
%%% Original concept inspired and some code copied from
%%% https://erlangcentral.org/wiki/index.php?title=Building_a_Non-blocking_TCP_server_using_OTP_principles

-module(gen_rpc_client).
-author("Panagiotis Papadomitsos <pj@ezgr.net>").

%%% Behaviour
-behaviour(gen_server).

%%% Include the HUT library
-include_lib("hut/include/hut.hrl").
%%% Include this library's name macro
-include("app.hrl").
%%% Include helpful guard macros
-include("guards.hrl").
%%% Include helpful guard macros
-include("types.hrl").

%%% Local state
-record(state, {socket :: port(),
        driver :: atom(),
        driver_mod :: atom(),
        driver_closed :: atom(),
        driver_error :: atom()}).
-elvis([{elvis_style, state_record_and_type, disable}]).

%%% Supervisor functions
-export([start_link/1, stop/1]).

%%% Server functions
-export([call/3, call/4, call/5, call/6, cast/3, cast/4, cast/5]).

-export([async_call/3, async_call/4, yield/1, nb_yield/1, nb_yield/2]).

-export([eval_everywhere/3, eval_everywhere/4, eval_everywhere/5]).

-export([multicall/3, multicall/4, multicall/5]).

-export([abcast/2, abcast/3, sbcast/2, sbcast/3]).

%%% Behaviour callbacks
-export([init/1, handle_call/3, handle_cast/2,
        handle_info/2, terminate/2, code_change/3]).

%%% Process exports
-export([async_call_worker/5, cast_worker/4]).

-define(IS_TIMEOUT(A), (is_integer(A) andalso A >= 0) orelse A =:= infinity).

-ignore_xref(start_link/1).
-ignore_xref(stop/1).

-elvis([{elvis_style, god_modules, disable}]).

%%% ===================================================
%%% Supervisor functions
%%% ===================================================
-spec start_link(node_or_tuple()) -> {ok, pid()} | ignore | {error, term()}.
start_link(NodeOrTuple) when ?is_node_or_tuple(NodeOrTuple) ->
    PidName = gen_rpc_helper:make_process_name("client", NodeOrTuple),
    gen_server:start_link({local,PidName}, ?MODULE, {NodeOrTuple}, []).

-spec stop(node_or_tuple()) -> ok.
stop(NodeOrTuple) when ?is_node_or_tuple(NodeOrTuple) ->
    PidName = gen_rpc_helper:make_process_name("client", NodeOrTuple),
    gen_server:stop(PidName, normal, infinity).

%%% ===================================================
%%% Server functions
%%% ===================================================
%% Simple server call with no args and default timeout values
-spec call(node_or_tuple(), atom() | tuple(), atom()) -> term() | {badrpc,term()} | {badtcp,term()}.
call(NodeOrTuple, M, F) ->
    call(NodeOrTuple, M, F, [], undefined, undefined).

%% Simple server call with args and default timeout values
-spec call(node_or_tuple(), atom() | tuple(), atom(), list()) -> term() | {badrpc,term()} | {badtcp,term()}.
call(NodeOrTuple, M, F, A) ->
    call(NodeOrTuple, M, F, A, undefined, undefined).

%% Simple server call with custom receive timeout value
-spec call(node_or_tuple(), atom() | tuple(), atom(), list(), timeout() | undefined) -> term() | {badrpc,term()} | {badtcp,term()}.
call(NodeOrTuple, M, F, A, RecvTO) ->
    call(NodeOrTuple, M, F, A, RecvTO, undefined).

%% Simple server call with custom receive and send timeout values
%% This is the function that all of the above call
-spec call(node_or_tuple(), atom() | tuple(), atom(), list(), timeout() | undefined, timeout() | undefined) ->
    term() | {badrpc,term()} | {badtcp,term()}.
call(NodeOrTuple, M, F, A, RecvTO, SendTO) when ?is_node_or_tuple(NodeOrTuple), is_atom(M) orelse is_tuple(M), is_atom(F), is_list(A),
                                         RecvTO =:= undefined orelse ?IS_TIMEOUT(RecvTO),
                                         SendTO =:= undefined orelse ?IS_TIMEOUT(SendTO) ->
    %% Create a unique name for the client because we register as such
    PidName = gen_rpc_helper:make_process_name("client", NodeOrTuple),
    case erlang:whereis(PidName) of
        undefined ->
            ?log(info, "event=client_process_not_found target=\"~p\" action=spawning_client", [NodeOrTuple]),
            case gen_rpc_dispatcher:start_client(NodeOrTuple) of
                {ok, NewPid} ->
                    %% We take care of CALL inside the gen_server
                    %% This is not resilient enough if the caller's mailbox is full
                    %% but it's good enough for now
                    try
                        gen_server:call(NewPid, {{call,M,F,A}, SendTO}, gen_rpc_helper:get_call_receive_timeout(RecvTO))
                    catch
                        exit:{timeout,_Reason} -> {badrpc,timeout};
                        exit:OtherReason -> {badrpc, {unknown_error, OtherReason}}
                    end;
                {error, Reason} ->
                    Reason
            end;
        Pid ->
            ?log(debug, "event=client_process_found pid=\"~p\" target=\"~p\"", [Pid, NodeOrTuple]),
            try
                gen_server:call(Pid, {{call,M,F,A}, SendTO}, gen_rpc_helper:get_call_receive_timeout(RecvTO))
            catch
                exit:{timeout,_Reason} -> {badrpc,timeout};
                exit:OtherReason -> {badrpc, {unknown_error, OtherReason}}
            end
    end.

%% Simple server cast with no args and default timeout values
-spec cast(node_or_tuple(), atom() | tuple(), atom()) -> true.
cast(NodeOrTuple, M, F) ->
    cast(NodeOrTuple, M, F, [], undefined).

%% Simple server cast with args and default timeout values
-spec cast(node_or_tuple(), atom() | tuple(), atom(), list()) -> true.
cast(NodeOrTuple, M, F, A) ->
    cast(NodeOrTuple, M, F, A, undefined).

%% Simple server cast with custom send timeout value
%% This is the function that all of the above casts call
-spec cast(node_or_tuple(), atom() | tuple(), atom(), list(), timeout() | undefined) -> true.
cast(NodeOrTuple, M, F, A, SendTO) when ?is_node_or_tuple(NodeOrTuple), is_atom(M) orelse is_tuple(M), is_atom(F), is_list(A),
                                 SendTO =:= undefined orelse ?IS_TIMEOUT(SendTO) ->
    _WorkerPid = erlang:spawn(?MODULE, cast_worker, [NodeOrTuple, {cast,M,F,A}, undefined, SendTO]),
    true.

%% Evaluate {M, F, A} on connected nodes.
-spec eval_everywhere([node_or_tuple()], atom() | tuple(), atom()) -> abcast.
eval_everywhere(Nodes, M, F) ->
    eval_everywhere(Nodes, M, F, [], undefined).

%% Evaluate {M, F, A} on connected nodes.
-spec eval_everywhere([node_or_tuple()], atom() | tuple(), atom(), list()) -> abcast.
eval_everywhere(Nodes, M, F, A) ->
    eval_everywhere(Nodes, M, F, A, undefined).

%% Evaluate {M, F, A} on connected nodes.
-spec eval_everywhere([node_or_tuple()], atom() | tuple(), atom(), list(), timeout() | undefined) -> abcast.
eval_everywhere(Nodes, M, F, A, SendTO) when is_list(Nodes), is_atom(M) orelse is_tuple(M), is_atom(F), is_list(A),
                                             SendTO =:= undefined orelse ?IS_TIMEOUT(SendTO) ->
    _ = [erlang:spawn(?MODULE, cast_worker, [Node, {cast,M,F,A}, abcast, SendTO]) || Node <- Nodes],
    abcast.

%% Simple server async_call with no args
-spec async_call(node_or_tuple(), atom() | tuple(), atom()) -> {pid(), reference()}.
async_call(NodeOrTuple, M, F) ->
    async_call(NodeOrTuple, M, F, []).

%% Simple server async_call with args
-spec async_call(node_or_tuple(), atom() | tuple(), atom(), list()) -> {pid(), reference()}.
async_call(NodeOrTuple, M, F, A) when ?is_node_or_tuple(NodeOrTuple), is_atom(M) orelse is_tuple(M), is_atom(F), is_list(A) ->
    Ref = erlang:make_ref(),
    Pid = erlang:spawn(?MODULE, async_call_worker, [NodeOrTuple, M, F, A, Ref]),
    {Pid, Ref}.

%% Simple server yield with key. Delegate to nb_yield. Default timeout form configuration.
-spec yield({pid(), reference()}) -> term().
yield(Key) ->
    {value,Result} = nb_yield(Key, infinity),
    Result.

%% Simple server non-blocking yield with key, default timeout value of 0
-spec nb_yield({pid(), reference()}) -> {value,term()} | timeout.
nb_yield(Key)->
    nb_yield(Key, 0).

%% Simple server non-blocking yield with key and custom timeout value
-spec nb_yield({pid(), reference()}, timeout()) -> {value,term()} | timeout.
nb_yield({Pid,Ref}, Timeout) when is_pid(Pid), is_reference(Ref), ?IS_TIMEOUT(Timeout) ->
    Pid ! {self(), Ref, yield},
    receive
        {Pid, Ref, async_call, Result} ->
            {value,Result}
    after
        Timeout ->
            ?log(debug, "event=nb_yield_timeout async_call_pid=\"~p\" async_call_ref=\"~p\"", [Pid, Ref]),
            timeout
    end.

%% "Concurrent" call to a set of servers
-spec multicall(atom() | tuple(), atom(), list()) -> {list(), list()}.
multicall(M, F, A) when is_atom(M) orelse is_tuple(M), is_atom(F), is_list(A) ->
    multicall([node()|gen_rpc:nodes()], M, F, A).

-spec multicall(list() | atom() | tuple(), atom() | tuple(), atom() | list(), list() | timeout()) -> {list(), list()}.
multicall(M, F, A, Timeout) when is_atom(M) orelse is_tuple(M), is_atom(F), is_list(A), ?IS_TIMEOUT(Timeout) ->
    multicall([node()|gen_rpc:nodes()], M, F, A, Timeout);

multicall(Nodes, M, F, A) when is_list(Nodes), is_atom(M) orelse is_tuple(M), is_atom(F), is_list(A) ->
    Keys = [async_call(Node, M, F, A) || Node <- Nodes],
    parse_multicall_results(Keys, Nodes, undefined).

-spec multicall(list(), atom() | tuple(), atom(), list(), timeout()) -> {list(), list()}.
multicall(Nodes, M, F, A, Timeout) when is_list(Nodes), is_atom(M) orelse is_tuple(M), is_atom(F), is_list(A), ?IS_TIMEOUT(Timeout) ->
    Keys = [async_call(Node, M, F, A) || Node <- Nodes],
    parse_multicall_results(Keys, Nodes, Timeout).

-spec abcast(atom(), term()) -> abcast.
abcast(Name, Msg) when is_atom(Name) ->
    abcast([node()|gen_rpc:nodes()], Name, Msg).

-spec abcast(list(), atom(), term()) -> abcast.
abcast(Nodes, Name, Msg) when is_list(Nodes), is_atom(Name) ->
    _ = [erlang:spawn(?MODULE, cast_worker, [Node, {abcast,Name,Msg}, abcast, undefined]) || Node <- Nodes],
    abcast.

-spec sbcast(atom(), term()) -> {list(), list()}.
sbcast(Name, Msg) when is_atom(Name) ->
    sbcast([node()|gen_rpc:nodes()], Name, Msg).

-spec sbcast(list(), atom(), term()) -> {list(), list()}.
sbcast(Nodes, Name, Msg) when is_list(Nodes), is_atom(Name) ->
    Ref = erlang:make_ref(),
    Workers = [{erlang:spawn(?MODULE, cast_worker, [Node, {sbcast, Name, Msg, {self(), Ref, Node}}, undefined, undefined]), Node} || Node <- Nodes],
    parse_sbcast_results(Workers, Ref).

%%% ===================================================
%%% Behaviour callbacks
%%% ===================================================
%% If we're called with a key, remove it, the key is only relevant
%% at the process name level
init({{Node,_Key}}) ->
    init({Node});

init({Node}) ->
    ok = gen_rpc_helper:set_optimal_process_flags(),
    case gen_rpc_helper:get_client_config_per_node(Node) of
        {error, Reason} ->
            ?log(error, "event=external_source_error action=falling_back_to_local reason=\"~p\"", [Reason]),
            {stop, {badrpc, {external_source_error, Reason}}};
        {Driver, Port} ->
            {DriverMod, DriverClosed, DriverError} = gen_rpc_helper:get_client_driver_options(Driver),
            ?log(info, "event=initializing_client driver=~s node=\"~s\" port=~B", [Driver, Node, Port]),
            case DriverMod:connect(Node, Port) of
                {ok, Socket} ->
                    case DriverMod:authenticate_server(Socket) of
                        ok ->
                            {ok, #state{socket=Socket,
                                        driver=Driver,
                                        driver_mod=DriverMod,
                                        driver_closed=DriverClosed,
                                        driver_error=DriverError}, gen_rpc_helper:get_inactivity_timeout(?MODULE)};
                        {error, ReasonTuple} ->
                            ?log(error, "event=client_authentication_failed driver=~s reason=\"~p\"", [Driver, ReasonTuple]),
                            {stop, ReasonTuple}
                    end;
                {error, {_Class,Reason}} ->
                    %% This should be badtcp but to conform with
                    %% the RPC library we return badrpc
                    {stop, {badrpc,Reason}}
            end
    end.

%% This is the actual CALL handler
handle_call({{call,_M,_F,_A} = PacketTuple, SendTO}, Caller, #state{socket=Socket, driver=Driver, driver_mod=DriverMod} = State) ->
    Packet = erlang:term_to_binary({PacketTuple, Caller}),
    ?log(debug, "message=call event=constructing_call_term driver=~s socket=\"~s\" caller=\"~p\"",
         [Driver, gen_rpc_helper:socket_to_string(Socket), Caller]),
    ok = DriverMod:set_send_timeout(Socket, SendTO),
    case DriverMod:send(Socket, Packet) of
        {error, Reason} ->
            ?log(error, "message=call event=transmission_failed driver=~s socket=\"~s\" caller=\"~p\" reason=\"~p\"",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), Caller, Reason]),
            {stop, Reason, Reason, State};
        ok ->
            ?log(debug, "message=call event=transmission_succeeded driver=~s socket=\"~s\" caller=\"~p\"",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), Caller]),
            %% We need to enable the socket and perform the call only if the call succeeds
            ok = DriverMod:activate_socket(Socket),
            {noreply, State, gen_rpc_helper:get_inactivity_timeout(?MODULE)}
    end;

%% Catch-all for calls - die if we get a message we don't expect
handle_call(Msg, _Caller, #state{socket=Socket, driver=Driver} = State) ->
    ?log(error, "event=uknown_call_received driver=~s socket=\"~s\" message=\"~p\" action=stopping",
         [Driver, gen_rpc_helper:socket_to_string(Socket), Msg]),
    {stop, {unknown_call, Msg}, {unknown_call, Msg}, State}.

%% This is the actual CAST handler for CAST
handle_cast({{cast,_M,_F,_A} = PacketTuple, SendTO}, State) ->
    send_cast(PacketTuple, State, SendTO, false);

%% This is the actual CAST handler for ABCAST
handle_cast({{abcast,_Name,_Msg} = PacketTuple, undefined}, State) ->
    send_cast(PacketTuple, State, undefined, false);

%% This is the actual CAST handler for SBCAST
handle_cast({{sbcast,_Name,_Msg,_Caller} = PacketTuple, undefined}, State) ->
    send_cast(PacketTuple, State, undefined, true);

%% This is the actual ASYNC CALL handler
handle_cast({{async_call,_M,_F,_A} = PacketTuple, Caller, Ref}, #state{socket=Socket, driver=Driver, driver_mod=DriverMod} = State) ->
    Packet = erlang:term_to_binary({PacketTuple, {Caller,Ref}}),
    ?log(debug, "message=async_call event=constructing_async_call_term socket=\"~s\" worker_pid=\"~p\" async_call_ref=\"~p\"",
         [gen_rpc_helper:socket_to_string(Socket), Caller, Ref]),
    ok = DriverMod:set_send_timeout(Socket, undefined),
    case DriverMod:send(Socket, Packet) of
        {error, Reason} ->
            ?log(error, "message=async_call event=transmission_failed driver=~s socket=\"~s\" worker_pid=\"~p\" call_ref=\"~p\" reason=\"~p\"",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), Caller, Ref, Reason]),
            {stop, Reason, Reason, State};
        ok ->
            ?log(debug, "message=async_call event=transmission_succeeded driver=~s socket=\"~s\" worker_pid=\"~p\" call_ref=\"~p\"",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), Caller, Ref]),
            %% We need to enable the socket and perform the call only if the call succeeds
            ok = DriverMod:activate_socket(Socket),
            %% Reply will be handled from the worker
            {noreply, State, gen_rpc_helper:get_inactivity_timeout(?MODULE)}
    end;

%% Catch-all for casts - die if we get a message we don't expect
handle_cast(Msg, #state{socket=Socket, driver=Driver} = State) ->
    ?log(error, "event=uknown_cast_received driver=~s socket=\"~s\" message=\"~p\" action=stopping",
         [Driver, gen_rpc_helper:socket_to_string(Socket), Msg]),
    {stop, {unknown_cast, Msg}, State}.

%% Handle any TCP packet coming in
handle_info({Driver,Socket,Data}, #state{socket=Socket, driver=Driver, driver_mod=DriverMod} = State) ->
    _Reply = case erlang:binary_to_term(Data) of
        {call, Caller, Reply} ->
            ?log(debug, "event=call_reply_received driver=~s socket=\"~s\" caller=\"~p\" action=sending_reply",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), Caller]),
            gen_server:reply(Caller, Reply);
        {async_call, {Caller, Ref}, Reply} ->
            ?log(debug, "event=async_call_reply_received driver=~s socket=\"~s\" caller=\"~p\" action=sending_reply",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), Caller]),
            Caller ! {self(), Ref, async_call, Reply};
        {sbcast, {Caller, Ref, Node}, Reply} ->
            ?log(debug, "event=sbcast_reply_received driver=~s socket=\"~s\" caller=\"~p\" reference=\"~p\" action=sending_reply",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), Caller, Ref]),
            Caller ! {Ref, Node, Reply};
        OtherData ->
            ?log(error, "event=erroneous_reply_received driver=~s socket=\"~s\" data=\"~p\" action=ignoring",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), OtherData])
    end,
    ok = DriverMod:activate_socket(Socket),
    {noreply, State, gen_rpc_helper:get_inactivity_timeout(?MODULE)};

handle_info({DriverClosed, Socket}, #state{socket=Socket, driver=Driver, driver_closed=DriverClosed} = State) ->
    ?log(error, "message=channel_closed driver=~s socket=\"~s\" action=stopping", [Driver, gen_rpc_helper:socket_to_string(Socket)]),
    {stop, normal, State};

handle_info({DriverError, Socket, Reason}, #state{socket=Socket, driver=Driver, driver_error=DriverError} = State) ->
    ?log(error, "message=channel_error driver=~s socket=\"~s\" reason=\"~p\" action=stopping",
         [Driver, gen_rpc_helper:socket_to_string(Socket), Reason]),
    {stop, normal, State};

%% Handle the inactivity timeout gracefully
handle_info(timeout, #state{socket=Socket, driver=Driver} = State) ->
    ?log(info, "message=timeout event=client_inactivity_timeout driver=~s socket=\"~s\" action=stopping",
         [Driver, gen_rpc_helper:socket_to_string(Socket)]),
    {stop, normal, State};

%% Catch-all for info - our protocol is strict so die!
handle_info(Msg, #state{socket=Socket, driver=Driver} = State) ->
    ?log(error, "event=uknown_message_received driver=~s socket=\"~s\" message=\"~p\" action=stopping",
         [Driver, gen_rpc_helper:socket_to_string(Socket), Msg]),
    {stop, {unknown_info, Msg}, State}.

%% Stub functions
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

terminate(_Reason, _State) ->
    ok.

%%% ===================================================
%%% Private functions
%%% ===================================================
send_cast(PacketTuple, #state{socket=Socket, driver=Driver, driver_mod=DriverMod} = State, SendTO, Activate) ->
    Packet = erlang:term_to_binary(PacketTuple),
    ?log(debug, "event=constructing_cast_term driver=~s socket=\"~s\" cast=\"~p\"",
         [Driver, gen_rpc_helper:socket_to_string(Socket), PacketTuple]),
    ok = DriverMod:set_send_timeout(Socket, SendTO),
    case DriverMod:send(Socket, Packet) of
        {error, Reason} ->
            ?log(error, "message=cast event=transmission_failed driver=~s socket=\"~s\" reason=\"~p\"",
                 [Driver, gen_rpc_helper:socket_to_string(Socket), Reason]),
            {stop, Reason, State};
        ok ->
            ok = case Activate of
                     true -> DriverMod:activate_socket(Socket);
                     _ -> ok
                 end,
            ?log(debug, "message=cast event=transmission_succeeded driver=~s socket=\"~s\"",
                 [Driver, gen_rpc_helper:socket_to_string(Socket)]),
            {noreply, State, gen_rpc_helper:get_inactivity_timeout(?MODULE)}
    end.

cast_worker(NodeOrTuple, Cast, Ret, SendTO) ->
    %% Create a unique name for the client because we register as such
    PidName = gen_rpc_helper:make_process_name("client", NodeOrTuple),
    case erlang:whereis(PidName) of
        undefined ->
            ?log(info, "event=client_process_not_found target=\"~p\" action=spawning_client", [NodeOrTuple]),
            case gen_rpc_dispatcher:start_client(NodeOrTuple) of
                {ok, NewPid} ->
                    %% We take care of CALL inside the gen_server
                    %% This is not resilient enough if the caller's mailbox is full
                    %% but it's good enough for now
                    ok = gen_server:cast(NewPid, {Cast,SendTO}),
                    Ret;
                {error, _Reason} ->
                    Ret
            end;
        Pid ->
            ?log(debug, "event=client_process_found pid=\"~p\" target=\"~p\"", [Pid, NodeOrTuple]),
            ok = gen_server:cast(Pid, {Cast,SendTO}),
            Ret
    end.

async_call_worker(NodeOrTuple, M, F, A, Ref) ->
    TTL = gen_rpc_helper:get_async_call_inactivity_timeout(),
    PidName = gen_rpc_helper:make_process_name("client", NodeOrTuple),
    SrvPid = case erlang:whereis(PidName) of
        undefined ->
            ?log(info, "event=client_process_not_found target=\"~p\" action=spawning_client", [NodeOrTuple]),
            case gen_rpc_dispatcher:start_client(NodeOrTuple) of
                {ok, NewPid} ->
                    ok = gen_server:cast(NewPid, {{async_call,M,F,A}, self(), Ref}),
                    NewPid;
                {error, {badrpc,_} = RpcError} ->
                    RpcError
            end;
        Pid ->
            ?log(debug, "event=client_process_found pid=\"~p\" target=\"~p\"", [Pid, NodeOrTuple]),
            ok = gen_server:cast(Pid, {{async_call,M,F,A}, self(), Ref}),
            Pid
    end,
    case SrvPid of
        SrvPid when is_pid(SrvPid) ->
            receive
                %% Wait for the reply from the node's gen_rpc client process
                {SrvPid,Ref,async_call,Reply} ->
                    %% Wait for a yield request from the caller
                    receive
                        {YieldPid,Ref,yield} ->
                            YieldPid ! {self(), Ref, async_call, Reply}
                    after
                        TTL ->
                            exit({error, async_call_cleanup_timeout_reached})
                    end
            after
                TTL ->
                    exit({error, async_call_cleanup_timeout_reached})
            end;
        TRpcError ->
            %% Wait for a yield request from the caller
            receive
                {YieldPid,Ref,yield} ->
                    YieldPid ! {self(), Ref, async_call, TRpcError}
            after
                TTL ->
                    exit({error, async_call_cleanup_timeout_reached})
            end
    end.

parse_multicall_results(Keys, Nodes, undefined) ->
    parse_multicall_results(Keys, Nodes, infinity);

parse_multicall_results(Keys, Nodes, Timeout) ->
    AsyncResults = [nb_yield(Key, Timeout) || Key <- Keys],
    {RealResults, RealBadNodes, _} = lists:foldl(fun
        ({value, {BadReply, _Reason}}, {Results, BadNodes, [Node|RestNodes]}) when BadReply =:= badrpc; BadReply =:= badtcp ->
            {Results, [Node|BadNodes], RestNodes};
        ({value, Value}, {Results, BadNodes, [_Node|RestNodes]}) ->
            {[Value|Results], BadNodes, RestNodes};
        (timeout, {Results, BadNodes, [Node|RestNodes]}) ->
            {Results, [Node|BadNodes], RestNodes}
    end, {[], [], Nodes}, AsyncResults),
    {RealResults, RealBadNodes}.

parse_sbcast_results(WorkerPids, Ref) ->
    Timeout = gen_rpc_helper:get_sbcast_receive_timeout(),
    parse_sbcast_results(WorkerPids, Ref, {[], []}, Timeout).

parse_sbcast_results([{_Pid,Node}|WorkerPids], Ref, {Good, Bad}, Timeout) ->
    receive
        {Ref, Node, error} ->
            parse_sbcast_results(WorkerPids, Ref, {Good, [Node|Bad]}, Timeout);
        {Ref, Node, success} ->
            parse_sbcast_results(WorkerPids, Ref, {[Node|Good], Bad}, Timeout)
    after
        Timeout ->
            parse_sbcast_results(WorkerPids, Ref, {Good, [Node|Bad]}, Timeout)
    end;

parse_sbcast_results([], _Ref, Results, _Timeout) ->
    Results.