%%
%% %CopyrightBegin%
%%
%% Copyright Ericsson AB 1996-2011. 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).
%% pg provides a process group facility. Messages
%% can be multicasted to all members in the group
-export([create/1,
create/2,
standby/2,
join/2,
send/2,
esend/2,
members/1,
name_to_pid/1,
master/1]).
%% Create a brand new empty process group with the master residing
%% at the local node
-spec create(PgName) -> 'ok' | {'error', Reason} when
PgName :: term(),
Reason :: 'already_created' | term().
create(PgName) ->
catch begin check(PgName),
Pid = spawn(pg,master,[PgName]),
global:register_name(PgName,Pid),
ok end.
%% Create a brand new empty process group with the master
%% residing at Node
-spec create(PgName, Node) -> 'ok' | {'error', Reason} when
PgName :: term(),
Node :: node(),
Reason :: 'already_created' | term().
create(PgName, Node) ->
catch begin check(PgName),
Pid = spawn(Node,pg,master,[PgName]),
global:register_name(PgName,Pid),
ok end.
%% Have a process on Node that will act as a standby for the process
%% group manager. So if the node where the manager runs fails, the
%% process group will continue to function.
-spec standby(term(), node()) -> 'ok'.
standby(_PgName, _Node) ->
ok.
%% Tell process group PgName that Pid is a new member of the group
%% synchronously return a list of all old members in the group
-spec join(PgName, Pid) -> Members when
PgName :: term(),
Pid :: pid(),
Members :: [pid()].
join(PgName, Pid) when is_atom(PgName) ->
global:send(PgName, {join,self(),Pid}),
receive
{_P,{members,Members}} ->
Members
end.
%% Multi cast Mess to all members in the group
-spec send(PgName, Msg) -> 'ok' when
PgName :: term(),
Msg :: term().
send(PgName, Mess) when is_atom(PgName) ->
global:send(PgName, {send, self(), Mess}),
ok;
send(Pg, Mess) when is_pid(Pg) ->
Pg ! {send,self(),Mess},
ok.
%% multi cast a message to all members in the group but ourselves
%% If we are a member
-spec esend(PgName, Msg) -> 'ok' when
PgName :: term(),
Msg :: term().
esend(PgName, Mess) when is_atom(PgName) ->
global:send(PgName, {esend,self(),Mess}),
ok;
esend(Pg, Mess) when is_pid(Pg) ->
Pg ! {esend,self(),Mess},
ok.
%% Return the members of the group
-spec members(PgName) -> Members when
PgName :: term(),
Members :: [pid()].
members(PgName) when is_atom(PgName) ->
global:send(PgName, {self() ,members}),
receive
{_P,{members,Members}} ->
Members
end;
members(Pg) when is_pid(Pg) ->
Pg ! {self,members},
receive
{_P,{members,Members}} ->
Members
end.
-spec name_to_pid(atom()) -> pid() | 'undefined'.
name_to_pid(PgName) when is_atom(PgName) ->
global:whereis_name(PgName).
-spec master(term()) -> no_return().
master(PgName) ->
process_flag(trap_exit, true),
master_loop(PgName, []).
master_loop(PgName,Members) ->
receive
{send,From,Message} ->
send_all(Members,{pg_message,From,PgName,Message}),
master_loop(PgName,Members);
{esend,From,Message} ->
send_all(lists:delete(From,Members),
{pg_message,From,PgName,Message}),
master_loop(PgName,Members);
{join,From,Pid} ->
link(Pid),
send_all(Members,{new_member,PgName,Pid}),
From ! {self(),{members,Members}},
master_loop(PgName,[Pid|Members]);
{From,members} ->
From ! {self(),{members,Members}},
master_loop(PgName,Members);
{'EXIT',From,_} ->
L =
case lists:member(From,Members) of
true ->
NewMembers = lists:delete(From,Members),
send_all(NewMembers, {crashed_member,PgName,From}),
NewMembers;
false ->
Members
end,
master_loop(PgName,L)
end.
send_all([], _) -> ok;
send_all([P|Ps], M) ->
P ! M,
send_all(Ps, M).
%% Check if the process group already exists
check(PgName) ->
case global:whereis_name(PgName) of
Pid when is_pid(Pid) ->
throw({error,already_created});
undefined ->
ok
end.
