分布式Erlang：多调用超过请求的超时

Posted 2023-03-15

【中文标题】分布式Erlang：多调用超过请求的超时

【英文标题】：Distributed Erlang: multicall exceeds requested timeout

【发布时间】：2018-04-25 07:42:52

【问题描述】：

我们使用分布式 erlang 集群，现在我在网络分裂的情况下对其进行测试。

为了从集群的所有节点获取信息，我使用 gen_server:multicall/4 并定义了超时。我需要的是尽快从可用节点获取信息。所以超时不会太大（大约3000毫秒）。 这里调用示例：

Timeout = 3000
Nodes = AllConfiguredNodes
gen_server:multi_call(Nodes, broker, get_score, Timeout)

我希望此调用以 Timeout ms 为单位返回结果。但在净拆分的情况下，它不会。它等待大约。 8 秒。

在发送请求之前，我发现 multi_call 请求在 call erlang:monitor(process, Name, Node) 中被暂停 5 秒。

我真的不在乎某些节点不回复、忙或不可用，我可以使用任何其他节点，但是由于这种暂停，我不得不等到 Erlang VM 尝试建立与死/不可用节点的新连接。

问题是：您知道可以防止这种停止的解决方案吗？或者可能是另一个适合我情况的 RPC。

【问题讨论】：

看看更新的答案，我认为它解决了你的问题

【参考方案1】：

我不确定我是否完全理解您要解决的问题，但如果要获得可以在 X 时间内检索到的所有答案并忽略其余部分，您可以尝试 async_call 的组合和 nb_yield。

可能是这样的

somefun() ->
    SmallTimeMs = 50,
    Nodes = AllConfiguredNodes,
    Promises = [rpc:async_call(N, some_mod, some_fun, ArgList) || N <- Nodes],
    get_results([], Promises, SmallTimeMs).


get_results(Results, _Promises, _SmallTimeMs) when length(Results) > 1 ->   % Replace 1 with whatever is the minimum acceptable number of results
    lists:flatten(Results);
get_results(Results, Promises, SmallTimeMs) ->
    Rs = get_promises(Promises, SmallTimeMs)
    get_results([Results|Rs], Promises, SmallTimeMs)).


get_promise(Promises, WaitMs) ->
    [rpc:nb_yield(Key, WaitMs) || Key <- Promises].

有关详细信息，请参阅：http://erlang.org/doc/man/rpc.html#async_call-4。

【讨论】：

在您的解决方案中，如果第一个节点不可用，则 rpc:nb_yeild() 等待超时，如果第二个节点也不可用，则等待 2*Timeout 等。我想收集在一个超时内到达的所有答案.而且我也不确定，但我希望 rpc:async_call 在发送消息之前也会调用 erlang:monitor(node)。

查看更新的建议。只需使用超短超时并重复，直到获得 1 个或足够的结果。我给你一个我认为应该可行的方向，而不是完整的解决方案。因为这是一个promise，我不希望它阻止任何事情 - asynch_call/4 应该立即返回

【参考方案2】：

我对问题的解决方案。

我自己实现了使用gen_server:call 的多重调用 基本思想是在单独的进程中使用 gen_server:call() 调用所有节点。并收集这些调用的结果。通过从调用进程的邮箱接收消息进行收集。

为了控制超时，我计算超时到期时的截止日期，然后将其用作参考点来计算 receive 中 after 的超时。

实施

主要功能是：

multicall(Nodes, Name, Req, Timeout) ->
    Refs = lists:map(fun(Node) -> call_node(Node, Name, Req, Timeout) end, Nodes),
    Results = read_all(Timeout, Refs),
    PosResults = [  Node, Result  ||  ok,  ok,  Node, Result    <- Results ],
     PosResults, calc_bad_nodes(Nodes, PosResults) .

这里的想法是调用所有节点并在一个 Timeout 内等待所有结果。

调用一个节点是从派生的进程中执行的。它会捕获 gen_server:call 使用的出口以防出错。

call_node(Node, Name, Req, Timeout) ->
    Ref = make_ref(),
    Self = self(),
    spawn_link(fun() ->
                       try
                           Result = gen_server:call(Name,Node,Req,Timeout),
                           Self !  Ref,  ok,  Node, Result   
                       catch
                           exit:Exit ->
                               Self !  Ref,  error,  'EXIT', Exit   
                       end
               end),
    Ref.

错误节点被计算为那些在超时时间内没有响应的节点

calc_bad_nodes(Nodes, PosResults) ->
     GoodNodes, _  = lists:unzip(PosResults),
    [ BadNode || BadNode <- Nodes, not lists:member(BadNode, GoodNodes) ].

通过超时读取邮箱来收集结果

read_all(ReadList, Timeout) ->
    Now = erlang:monotonic_time(millisecond),
    Deadline = Now + Timeout,
    read_all_impl(ReadList, Deadline, []).

在截止日期不发生之前执行读取

read_all_impl([], _, Results) ->
    lists:reverse(Results);
read_all_impl([ W | Rest ], expired, Results) ->
    R = read(0, W),
    read_all_impl(Rest, expired, [R | Results ]);
read_all_impl([ W | Rest ] = L, Deadline, Results) ->
    Now = erlang:monotonic_time(millisecond),
    case Deadline - Now of
        Timeout when Timeout > 0 ->
            R = read(Timeout, W),
            case R of
                 ok, _  ->
                    read_all_impl(Rest, Deadline, [ R | Results ]);
                 error,  read_timeout, _   ->
                    read_all_impl(Rest, expired, [ R | Results ])
            end;
        Timeout when Timeout =< 0 ->
            read_all_impl(L, expired, Results)
    end.

一次读取只是从带有超时的邮箱接收。

read(Timeout, Ref) ->
    receive
         Ref, Result  ->
             ok, Result 
    after Timeout ->
             error,  read_timeout, Timeout  
    end.

进一步改进：

rpc 模块生成单独的进程以避免迟到答案的垃圾。所以在这个多重调用函数中做同样的事情会很有用 infinity 超时可以用明显的方式处理