<7>pg2 分析

网上看到erlang的pg2模块似乎没人推荐使用，但是还是有不少使用者，自己也感觉使用跨节点通信时候，使用它来管理各个近点的进程也是不错的选择， 使用过程中也有不少疑问，也感觉接口不够丰富，简单分析下：

pg2为什么允许一个进程加入两次呢？，并且也要退出两次，使用时候可以自己加些判断，
join(Pid， Group) ->         
      ok =  case lists:member(Pid, pg2:get_members(Group)) of
            true -> ok;
               _ -> pg2:join(Group, Pid)
             end;

加接口，
获取某个节点的所有进程
获取某个节点的某个进程，进程id随机
get_node_members(Node, Group) -> [pid()] | {error, {no_such_group, Group}}
get_node_pid(Node, Group) -> pid() | {error, Reason}

get_node_members(Node, Group) ->
    case pg2:get_members(Group) of
        {error, Reason} ->
            {error, Reason};
        AllMembers ->
            Fun = fun(X) ->
                node(X) =:= Node end,
            lists:filter(Fun, AllMembers)
    end.

get_node_pid(Node, Group) ->
    case get_node_members(Node, Group) of
        [Pid] ->
            Pid;
        [] ->
            [];
        Members when is_list(Members) ->
            {_,_,X} = erlang:now(),
            lists:nth((X rem length(Members))+1, Members);
        Else ->
            Else
    end.

网上还看到一种获取最优的pid
get_best_pid(Group) ->
  Members = pg2:get_members(Group),
  Members1 = lists:map(fun(Pid) ->
      [{message_queue_len, Messages}] = erlang:process_info(Pid, [message_queue_len]),
                                                                 {Pid, Messages}
                       end, Members),
case lists:keysort(2, Members1) of
  [{Pid, _} | _] -> Pid;
  [] -> {error, empty_process_group}
end.
不过只能获得本地进程的信息，不能跨节点获取，
Failure: badarg if Pid is not a local process, or if Item is not a valid Item.
所以这种方法只适用于本地进程


那么pg2 是怎么实现节点信息存取的?
其实就是各个节点注册了一个相同名称的pg2， 通过{pg2, Node} ! Msg来实现
pg2.erl

start() ->
    ensure_started().
ensure_started() ->
    case whereis(?MODULE) of
        undefined ->
            C = {pg2, {?MODULE, start_link, []}, permanent,
                 1000, worker, [?MODULE]},
            supervisor:start_child(kernel_safe_sup, C);
        Pg2Pid ->
            {ok, Pg2Pid}
    end.

初始化的时候，给其他可见的node发送消息{new_pg2, node()}, 并且给自己发送{nodeup, Node}，   并创建ets表
init([]) ->
    Ns = nodes(),
    net_kernel:monitor_nodes(true),
    lists:foreach(fun(N) ->
                          {?MODULE, N} ! {new_pg2, node()},
                          self() ! {nodeup, N}
                  end, Ns),
    pg2_table = ets:new(pg2_table, [ordered_set, protected, named_table]),
    {ok, #state{}}.

当自己收到nodeup， 向Node发送信息
当收到nodeup或者new_pg2都会返回信息{exchange, node(), all_members()}
all_members 就是所有的group和pid信息

handle_info({nodeup, Node}, S) ->
    gen_server:cast({?MODULE, Node}, {exchange, node(), all_members()}),
    {noreply, S};

handle_info({new_pg2, Node}, S) ->
    gen_server:cast({?MODULE, Node}, {exchange, node(), all_members()}),
    {noreply, S};

all_members() ->
    [[G, group_members(G)] || G <- all_groups()].

group_members(Name) ->
    [P ||
        [P, N] <- ets:match(pg2_table, {{member, Name, '$1'},'$2'}),
        _ <- lists:seq(1, N)].

当收到exchange信息，就会进行存储
handle_cast({exchange, _Node, List}, S) ->
    store(List),
    {noreply, S};

store(List) ->
    _ = [(assure_group(Name)
          andalso
          [join_group(Name, P) || P <- Members -- group_members(Name)]) ||
            [Name, Members] <- List],
    ok.
all_members() ->
    [[G, group_members(G)] || G <- all_groups()].

group_members(Name) ->
    [P || 
        [P, N] <- ets:match(pg2_table, {{member, Name, '$1'},'$2'}),
        _ <- lists:seq(1, N)].

存储就是将其他节点的进程，全部加入本节点，如果本几点没有该进程的话


那么在join时候会发生什么呢？
join(Name, Pid) when is_pid(Pid) ->
    ensure_started(),
    case ets:member(pg2_table, {group, Name}) of
        false ->
            {error, {no_such_group, Name}};
        true ->
            global:trans({{?MODULE, Name}, self()},
                         fun() ->
                                 gen_server:multi_call(?MODULE,
                                                       {join, Name, Pid})
                         end),
            ok 
    end.
就是告诉所有的节点，有一个进程要加入,
在告诉其他节点的时候，会对这个节点加锁 ,利用函数 global:trans/2
Sets a lock on Id (using set_lock/3). If this succeeds, Fun() is evaluated and the result Res is returned. Returns aborted if the lock attempt failed. If Retries is set to infinity, the transaction will not abort.
infinity is the default setting and will be used if no value is given for Retries.
就是说默认是是无限次加锁，如果加锁失败就会重新尝试，直到成功，
再看set_lock/3,
Sets a lock on the specified nodes (or on all nodes if none are specified) on ResourceId for LockRequesterId. If a lock already exists on ResourceId for another requester than LockRequesterId, and Retries is not equal to 0, the process sleeps for a while and will try to execute the action later. When Retries attempts have been made, false is returned, otherwise true. If Retries is infinity, true is eventually returned
This function is completely synchronous.
This function does not address the problem of a deadlock. A deadlock can never occur as long as processes only lock one resource at a time. But if some processes try to lock two or more resources, a deadlock may occur. It is up to the application to detect and rectify a deadlock.
pg2/leave/2和join做了相同的操作，
加锁过程完全同步，直到加锁成功，并且有可能会造成死锁，
如果是global注册的就只有一个锁，而pg2是本地注册的，
，并且有可能造成死锁的危险，看来pg2的代价真的是不小，而如果只采用注册本地进程来发消息，就不用这么多资源的使用， 或许很多人不用pg2也是有原因的吧，但是只要不是频繁的调用函数leave, join还是可以使用， 使用时候，最好用异步来调用leave/2 和join/2函数，
至于pg2可以实现进程挂掉主动从pg2移除，也是用leave实现的