代码之家 › 专栏 › 技术社区 › Dmitry Poroh

分布式Erlang:多呼叫超过请求的超时

erlang-otp rpc distributed-computing erlang

Dmitry Poroh · 技术社区 · 7 年前

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

此处调用示例:

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

我预计这个调用返回的结果是超时毫秒。但在网络拆分的情况下,它不会。等待约8秒。

我发现multi_call请求在call中额外暂停了5秒 erlang:monitor(process, {Name, Node}) 发送请求之前。

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

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

2 回复 | 直到 7 年前

Jr0 7 年前

我不确定我是否完全理解您试图解决的问题,但如果要在X时间内获得所有可以检索到的答案,而忽略其余部分,您可以尝试async\u调用和nb\u 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 了解更多详细信息。

Dmitry Poroh 7 年前

我对这个问题的解决方案。

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

为了控制超时,我计算超时过期时的截止时间,然后将其作为参考点来计算 after 在里面 receive

实施

主要功能是:

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) }.

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

调用一个节点是从派生进程执行的。它捕获了 gen_服务器:调用 如果出现错误。

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 超时可以用明显的方式处理