Erlang/Elixir Erlang 的 RPC 模块代码分析

davidgao · 2017年07月31日 · 2063 次阅读

RPC的代码分析

进程创建

RPC模块本身是一个gen_server会随着kernel模块启动,也就是说,在Erlang/OTP启动后我们就免费获得了一个RPC进程。 RPC进程启动的时候,会在Erts中通过local注册一个名字rex的进程,这样没有经过修改的Erlang/OTP都会有这个名字在它的名字列表上。

RPC调用逻辑

不管是同步调用还是广播调用,在RPC模块中的调用都是依赖gen_server的相关方和erlang:send方法来完成。这样尽最大可能的重用代码,保证了整个OTP中对远程调用的表现的一致性。 并且RPC模块不单单可以调用远程节点的方法或进程,也可以调用本地节点的方法或进程,这样保证了整个RPC的系统位置透明性,并且RPC模块针对本地节点作了相关优化。

例如说call方法针对本地节点就采用了下面的方法:

local_call(M, F, A) when is_atom(M), is_atom(F), is_list(A) ->
    case catch apply(M, F, A) of
        {'EXIT',_} = V -> {badrpc, V};
        Other -> Other
    end.

call和block_call方法

这两个方法都是同步的调用,但是实现的细节非常不同,对rex进程的影响也是不同的。当然使用两个方法在并发执行的情况下,得到的结果是完全不同的。 不管是call也好,block_call也好,都会在执行阶段暂时的将被调用者进程的console输出重定向到调用者进程所在节点的group leader上。

call方法

在调用发起者一侧,RPC模块会立刻建立一个监控下的Erlang进程,并在该进程内通过gen_server:call方法来调用远程节点。

do_call(Node, Request, infinity) ->
    rpc_check(catch gen_server:call({?NAME,Node}, Request, infinity));
do_call(Node, Request, Timeout) ->
    Tag = make_ref(),
    {Receiver,Mref} =
    erlang:spawn_monitor(
      fun() ->
          process_flag(trap_exit, true),
          Result = gen_server:call({?NAME,Node}, Request, Timeout),
          exit({self(),Tag,Result})
      end),
    receive
    {'DOWN',Mref,_,_,{Receiver,Tag,Result}} ->
        rpc_check(Result);
    {'DOWN',Mref,_,_,Reason} ->
        rpc_check_t({'EXIT',Reason})
    end.

RPC模块会将超时或对方节点失去连接的情况处理为bad_rpc,让顶层逻辑发现并非业务本身引起的远程调用问题。

在被调用者一些,RPC模块也会立刻创建一个监控下的Erlang进程,并在该进程内处理调用者的call消息,同时会将相关信息保存在rex进程的进程上下文中。当新的进程完成了业务处理,就会把处理结果返回给被调用者节点的rex进程,然后再将结果返回给调用发起者。 我们可以仔细观察它的代码:

handle_call_call(Mod, Fun, Args, Gleader, To, S) ->
    RpcServer = self(),
    %% Spawn not to block the rpc server.
    {Caller,_} =
    erlang:spawn_monitor(
      fun () ->
          set_group_leader(Gleader),
          Reply = 
              case catch apply(Mod, Fun, Args) of
              {'EXIT', _} = Exit ->
                  {badrpc, Exit};
              Result ->
                  Result
              end,
          RpcServer ! {self(), {reply, Reply}}
      end),
    {noreply, gb_trees:insert(Caller, To, S)}.

block_call方法

于call方法一样,在调用发起者一侧,RPC模块会立刻建立一个监控下的Erlang进程,并在该进程内通过gen_server:call方法来调用远程节点。

但是在被调用者一些,RPC模块会选择使用被调用者所在节点的rex直接执行相关代码

handle_call({block_call, Mod, Fun, Args, Gleader}, _To, S) ->
    MyGL = group_leader(),
    set_group_leader(Gleader),
    Reply = 
    case catch apply(Mod,Fun,Args) of
        {'EXIT', _} = Exit ->
        {badrpc, Exit};
        Other ->
        Other
    end,
    group_leader(MyGL, self()), % restore
    {reply, Reply, S};

同步调用总结

call方法可以保证,同一调用者的远程请求按序列执行,但是不保证多个调用者的远程请求按序列执行。 block_call方法保证,多个调用者的远程请求按序列执行。 不管是call还是block_call的方法都会给调用者带来大量的进程创建的压力(Erlang创建进程很快,但不代表没有代价)。 call方法还会给被调用者节点带来大量的进程创建压力。

cast方法

RPC模块的cast方法直接依赖于gen_sever:cast,并没有做更多的事情。

针对本地节点,cast方法会在调用者节点内创建一个进程来执行相关代码:

cast(Node, Mod, Fun, Args) when Node =:= node() ->
    catch spawn(Mod, Fun, Args),
    true;
cast(Node, Mod, Fun, Args) ->
    gen_server:cast({?NAME,Node}, {cast,Mod,Fun,Args,group_leader()}),
    true.

被调用者接收到消息后会立刻创建进程执行相关代码:

handle_cast({cast, Mod, Fun, Args, Gleader}, S) ->
    spawn(fun() ->
          set_group_leader(Gleader),
          apply(Mod, Fun, Args)
      end),
    {noreply, S};

cast方法总结

cast方法是非常简单的。和call方法一样,会给被调用者节点带来大量的进程创建压力。 同样不要忘记了,cast方法也会将新创建的进程的console输出重新定向调用者所在节点的group leader上。

abcast和sbcast

这两个方法都是通过erlang:send将调用者的消息发送到被调用者节点上。

abcast

abcast采用的是纯异步,发出去就不管了,直接将消息不经过rex进程直接发送到目标进程上


abcast(Name, Mess) ->
    abcast([node() | nodes()], Name, Mess).

abcast([Node|Tail], Name, Mess) ->
    Dest = {Name,Node},
    %这么做的好处是不会让进程被trap
    %从而保证了异步性
    case catch erlang:send(Dest, Mess, [noconnect]) of
    noconnect -> spawn(erlang, send, [Dest,Mess]), ok;
    _ -> ok
    end,
    abcast(Tail, Name, Mess);
abcast([], _,_) -> abcast.

此处abcast完全是异步的,如果发现了目标节点是没有连接的时候,直接创建一个新的进程来进行消息发送,完全不会进入Trap状态等待节点连接。

sbcast

sbcast算是同步的广播方式,发送后会回收广播结果,并且当节点没有完成连接的时候,会进入Trap状态等待节点连接完成

sbcast(Name, Mess) ->
    sbcast([node() | nodes()], Name, Mess).

sbcast(Nodes, Name, Mess) ->
    Monitors = send_nodes(Nodes, ?NAME, {sbcast, Name, Mess}, []),
    rec_nodes(?NAME, Monitors).

send_nodes([Node|Tail], Name, Msg, Monitors) when is_atom(Node) ->
    Monitor = start_monitor(Node, Name),
    %% Handle non-existing names in rec_nodes.
    catch {Name, Node} ! {self(), Msg},
    send_nodes(Tail, Name, Msg, [Monitor | Monitors]);
send_nodes([_Node|Tail], Name, Msg, Monitors) ->
    %% Skip non-atom _Node
    send_nodes(Tail, Name, Msg, Monitors);
send_nodes([], _Name,  _Req, Monitors) -> 
    Monitors.

rec_nodes(Name, Nodes) -> 
    rec_nodes(Name, Nodes, [], []).

rec_nodes(_Name, [],  Badnodes, Replies) ->
    {Replies, Badnodes};
rec_nodes(Name, [{N,R} | Tail], Badnodes, Replies) ->
    receive
    {'DOWN', R, _, _, _} ->
        rec_nodes(Name, Tail, [N|Badnodes], Replies);
    {?NAME, N, {nonexisting_name, _}} ->  
        %% used by sbcast()
        erlang:demonitor(R, [flush]),
        rec_nodes(Name, Tail, [N|Badnodes], Replies);
    {Name, N, Reply} ->  %% Name is bound !!!
        erlang:demonitor(R, [flush]),
        rec_nodes(Name, Tail, Badnodes, [Reply|Replies])
    end.

转载自TechTalk

暂无回复。
需要 登录 后方可回复, 如果你还没有账号请点击这里 注册