rename pg2 to pg_local, prior to more substantial modifications

1 files changed, 376 insertions, 0 deletions

diff --git a/src/pg_local.erl b/src/pg_local.erl
new file mode 100644
index 00000000..58a3b798
--- /dev/null
+++ b/src/pg_local.erl
@@ -0,0 +1,376 @@
+%%
+%% %CopyrightBegin%
+%% 
+%% Copyright Ericsson AB 1997-2009. All Rights Reserved.
+%% 
+%% The contents of this file are subject to the Erlang Public License,
+%% Version 1.1, (the "License"); you may not use this file except in
+%% compliance with the License. You should have received a copy of the
+%% Erlang Public License along with this software. If not, it can be
+%% retrieved online at http://www.erlang.org/.
+%% 
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and limitations
+%% under the License.
+%% 
+%% %CopyrightEnd%
+%%
+-module(pg_local).
+
+-export([create/1, delete/1, join/2, leave/2]).
+-export([get_members/1, get_local_members/1]).
+-export([get_closest_pid/1, which_groups/0]).
+-export([start/0,start_link/0,init/1,handle_call/3,handle_cast/2,handle_info/2,
+         terminate/2]).
+
+%%% As of R13B03 monitors are used instead of links.
+
+%%%
+%%% Exported functions
+%%%
+
+-spec start_link() -> {'ok', pid()} | {'error', term()}.
+
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+-spec start() -> {'ok', pid()} | {'error', term()}.
+
+start() ->
+    ensure_started().
+
+-spec create(term()) -> 'ok'.
+
+create(Name) ->
+    ensure_started(),
+    case ets:member(pg2_table, {group, Name}) of
+        false ->
+            global:trans({{?MODULE, Name}, self()},
+                         fun() ->
+                                 gen_server:multi_call(?MODULE, {create, Name})
+                         end),
+            ok;
+        true ->
+            ok
+    end.
+
+-type name() :: term().
+
+-spec delete(name()) -> 'ok'.
+
+delete(Name) ->
+    ensure_started(),
+    global:trans({{?MODULE, Name}, self()},
+                 fun() ->
+                         gen_server:multi_call(?MODULE, {delete, Name})
+                 end),
+    ok.
+
+-spec join(name(), pid()) -> 'ok' | {'error', {'no_such_group', term()}}.
+
+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.
+
+-spec leave(name(), pid()) -> 'ok' | {'error', {'no_such_group', name()}}.
+
+leave(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,
+                                                       {leave, Name, Pid})
+                         end),
+            ok
+    end.
+
+-type get_members_ret() :: [pid()] | {'error', {'no_such_group', name()}}.
+
+-spec get_members(name()) -> get_members_ret().
+   
+get_members(Name) ->
+    ensure_started(),
+    case ets:member(pg2_table, {group, Name}) of
+        true ->
+            group_members(Name);
+        false ->
+            {error, {no_such_group, Name}}
+    end.
+
+-spec get_local_members(name()) -> get_members_ret().
+
+get_local_members(Name) ->
+    ensure_started(),
+    case ets:member(pg2_table, {group, Name}) of
+        true ->
+            local_group_members(Name);
+        false ->
+            {error, {no_such_group, Name}}
+    end.
+
+-spec which_groups() -> [name()].
+
+which_groups() ->
+    ensure_started(),
+    all_groups().
+
+-type gcp_error_reason() :: {'no_process', term()} | {'no_such_group', term()}.
+
+-spec get_closest_pid(term()) -> pid() | {'error', gcp_error_reason()}.
+
+get_closest_pid(Name) ->
+    case get_local_members(Name) of
+        [Pid] ->
+            Pid;
+        [] ->
+            {_,_,X} = erlang:now(),
+            case get_members(Name) of
+                [] -> {error, {no_process, Name}};
+                Members ->
+                    lists:nth((X rem length(Members))+1, Members)
+            end;
+        Members when is_list(Members) ->
+            {_,_,X} = erlang:now(),
+            lists:nth((X rem length(Members))+1, Members);
+        Else ->
+            Else
+    end.
+
+%%%
+%%% Callback functions from gen_server
+%%%
+
+-record(state, {}).
+
+-spec init([]) -> {'ok', #state{}}.
+
+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{}}.
+
+-type call() :: {'create', name()}
+              | {'delete', name()}
+              | {'join', name(), pid()}
+              | {'leave', name(), pid()}.
+
+-spec handle_call(call(), _, #state{}) -> 
+        {'reply', 'ok', #state{}}.
+
+handle_call({create, Name}, _From, S) ->
+    assure_group(Name),
+    {reply, ok, S};
+handle_call({join, Name, Pid}, _From, S) ->
+    ets:member(pg2_table, {group, Name}) andalso join_group(Name, Pid),
+    {reply, ok, S};
+handle_call({leave, Name, Pid}, _From, S) ->
+    ets:member(pg2_table, {group, Name}) andalso leave_group(Name, Pid),
+    {reply, ok, S};
+handle_call({delete, Name}, _From, S) ->
+    delete_group(Name),
+    {reply, ok, S};
+handle_call(Request, From, S) ->
+    error_logger:warning_msg("The pg2 server received an unexpected message:\n"
+                             "handle_call(~p, ~p, _)\n", 
+                             [Request, From]),
+    {noreply, S}.
+
+-type all_members() :: [[name(),...]].
+-type cast() :: {'exchange', node(), all_members()}
+              | {'del_member', name(), pid()}.
+
+-spec handle_cast(cast(), #state{}) -> {'noreply', #state{}}.
+
+handle_cast({exchange, _Node, List}, S) ->
+    store(List),
+    {noreply, S};
+handle_cast(_, S) ->
+    %% Ignore {del_member, Name, Pid}.
+    {noreply, S}.
+
+-spec handle_info(tuple(), #state{}) -> {'noreply', #state{}}.
+
+handle_info({'DOWN', MonitorRef, process, _Pid, _Info}, S) ->
+    member_died(MonitorRef),
+    {noreply, S};
+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};
+handle_info(_, S) ->
+    {noreply, S}.
+
+-spec terminate(term(), #state{}) -> 'ok'.
+
+terminate(_Reason, _S) ->
+    true = ets:delete(pg2_table),
+    ok.
+
+%%%
+%%% Local functions
+%%%
+
+%%% One ETS table, pg2_table, is used for bookkeeping. The type of the
+%%% table is ordered_set, and the fast matching of partially
+%%% instantiated keys is used extensively.
+%%%
+%%% {{group, Name}}
+%%%    Process group Name.
+%%% {{ref, Pid}, RPid, MonitorRef, Counter}
+%%% {{ref, MonitorRef}, Pid}
+%%%    Each process has one monitor. Sometimes a process is spawned to
+%%%    monitor the pid (RPid). Counter is incremented when the Pid joins
+%%%    some group.
+%%% {{member, Name, Pid}, GroupCounter}
+%%% {{local_member, Name, Pid}}
+%%%    Pid is a member of group Name, GroupCounter is incremented when the
+%%%    Pid joins the group Name.
+%%% {{pid, Pid, Name}}
+%%%    Pid is a member of group Name.
+
+store(List) ->
+    _ = [assure_group(Name) andalso [join_group(Name, P) || P <- Members] ||
+            [Name, Members] <- List],
+    ok.
+
+assure_group(Name) ->
+    Key = {group, Name},
+    ets:member(pg2_table, Key) orelse true =:= ets:insert(pg2_table, {Key}).
+
+delete_group(Name) ->
+    _ = [leave_group(Name, Pid) || Pid <- group_members(Name)],
+    true = ets:delete(pg2_table, {group, Name}),
+    ok.
+
+member_died(Ref) ->
+    [{{ref, Ref}, Pid}] = ets:lookup(pg2_table, {ref, Ref}),
+    Names = member_groups(Pid),
+    _ = [leave_group(Name, P) || 
+            Name <- Names,
+            P <- member_in_group(Pid, Name)],
+    %% Kept for backward compatibility with links. Can be removed, eventually.
+    _ = [gen_server:abcast(nodes(), ?MODULE, {del_member, Name, Pid}) ||
+            Name <- Names],
+    ok.
+
+join_group(Name, Pid) ->
+    Ref_Pid = {ref, Pid}, 
+    try _ = ets:update_counter(pg2_table, Ref_Pid, {4, +1})
+    catch _:_ ->
+            {RPid, Ref} = do_monitor(Pid),
+            true = ets:insert(pg2_table, {Ref_Pid, RPid, Ref, 1}),
+            true = ets:insert(pg2_table, {{ref, Ref}, Pid})
+    end,
+    Member_Name_Pid = {member, Name, Pid},
+    try _ = ets:update_counter(pg2_table, Member_Name_Pid, {2, +1})
+    catch _:_ ->
+            true = ets:insert(pg2_table, {Member_Name_Pid, 1}),
+            _ = [ets:insert(pg2_table, {{local_member, Name, Pid}}) ||
+                    node(Pid) =:= node()],
+            true = ets:insert(pg2_table, {{pid, Pid, Name}})
+    end.
+
+leave_group(Name, Pid) ->
+    Member_Name_Pid = {member, Name, Pid},
+    try ets:update_counter(pg2_table, Member_Name_Pid, {2, -1}) of
+        N ->
+            if 
+                N =:= 0 ->
+                    true = ets:delete(pg2_table, {pid, Pid, Name}),
+                    _ = [ets:delete(pg2_table, {local_member, Name, Pid}) ||
+                            node(Pid) =:= node()],
+                    true = ets:delete(pg2_table, Member_Name_Pid);
+                true ->
+                    ok
+            end,
+            Ref_Pid = {ref, Pid}, 
+            case ets:update_counter(pg2_table, Ref_Pid, {4, -1}) of
+                0 ->
+                    [{Ref_Pid,RPid,Ref,0}] = ets:lookup(pg2_table, Ref_Pid),
+                    true = ets:delete(pg2_table, {ref, Ref}),
+                    true = ets:delete(pg2_table, Ref_Pid),
+                    true = erlang:demonitor(Ref, [flush]),
+                    kill_monitor_proc(RPid, Pid);
+                _ ->
+                    ok
+            end
+    catch _:_ ->
+            ok
+    end.
+
+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)].
+
+local_group_members(Name) ->
+    [P || 
+        [Pid] <- ets:match(pg2_table, {{local_member, Name, '$1'}}),
+        P <- member_in_group(Pid, Name)].
+
+member_in_group(Pid, Name) ->
+    [{{member, Name, Pid}, N}] = ets:lookup(pg2_table, {member, Name, Pid}),
+    lists:duplicate(N, Pid).
+
+member_groups(Pid) ->
+    [Name || [Name] <- ets:match(pg2_table, {{pid, Pid, '$1'}})].
+
+all_groups() ->
+    [N || [N] <- ets:match(pg2_table, {{group,'$1'}})].
+
+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.
+
+
+kill_monitor_proc(RPid, Pid) ->
+    RPid =:= Pid orelse exit(RPid, kill).
+
+%% When/if erlang:monitor() returns before trying to connect to the
+%% other node this function can be removed.
+do_monitor(Pid) ->
+    case (node(Pid) =:= node()) orelse lists:member(node(Pid), nodes()) of
+        true ->
+            %% Assume the node is still up
+            {Pid, erlang:monitor(process, Pid)};
+        false ->
+            F = fun() -> 
+                        Ref = erlang:monitor(process, Pid),
+                        receive 
+                            {'DOWN', Ref, process, Pid, _Info} ->
+                                exit(normal)
+                        end
+                end,
+            erlang:spawn_monitor(F)
+    end.