Packages
otpbp
4.72.2
7.18.0
7.16.1
7.15.0
7.14.0
7.13.0
7.12.0
7.11.0
7.10.2
7.9.3
7.9.2
7.9.1
7.8.7
7.8.6
7.8.5
7.8.4
7.7.1
7.7.0
7.6.3
7.6.2
7.6.1
7.6.0
7.5.1
7.5.0
7.4.0
7.3.4
7.3.3
7.3.2
7.3.1
7.3.0
7.2.0
7.1.1
7.1.0
7.0.4
7.0.3
7.0.2
7.0.1
6.26.0
6.25.0
6.24.0
6.23.0
6.22.2
6.21.3
6.21.2
6.21.1
6.20.2
6.19.1
6.19.0
6.18.3
6.18.2
6.18.1
6.18.0
6.17.1
6.17.0
6.16.0
6.15.4
6.15.3
6.15.2
6.15.1
6.15.0
6.14.0
6.13.5
6.13.4
6.13.3
6.13.2
6.13.1
6.13.0
6.12.0
6.11.2
6.11.1
6.10.0
6.9.2
6.9.1
6.9.0
6.8.0
6.7.1
6.7.0
6.6.4
6.6.3
6.6.2
6.6.1
6.6.0
6.5.0
6.4.0
6.3.3
6.3.2
6.3.1
6.3.0
6.2.1
6.2.0
6.1.0
6.0.0
5.39.0
5.38.0
5.37.0
5.36.0
5.35.2
5.34.3
5.34.2
5.34.1
5.33.1
5.32.1
5.32.0
5.31.3
5.31.2
5.31.1
5.31.0
5.30.1
5.30.0
5.29.0
5.28.4
5.28.3
5.28.2
5.28.1
5.28.0
5.27.0
5.26.5
5.26.4
5.26.3
5.26.2
5.26.1
5.26.0
5.25.0
5.24.2
5.24.1
5.23.0
5.22.2
5.22.1
5.22.0
5.21.0
5.20.1
5.20.0
5.19.4
5.19.3
5.19.2
5.19.1
5.19.0
5.18.0
5.17.0
5.16.3
5.16.1
5.16.0
5.15.1
5.15.0
5.14.0
5.13.1
5.13.0
5.12.1
5.12.0
5.11.2
5.11.1
5.10.1
5.10.0
5.9.0
5.8.0
5.7.0
5.6.1
5.6.0
5.5.0
5.4.2
5.4.1
5.4.0
5.3.0
5.2.2
5.2.1
5.2.0
5.1.2
5.1.1
4.79.0
4.78.0
4.77.0
4.76.2
4.75.3
4.75.2
4.75.1
4.74.0
4.73.1
4.73.0
4.72.3
4.72.2
4.72.1
4.72.0
4.71.1
4.71.0
4.70.0
4.69.4
4.69.3
4.69.2
4.69.1
4.69.0
4.68.0
4.67.5
4.67.4
4.67.3
4.67.2
4.67.0
4.66.0
4.65.2
4.65.1
4.64.0
4.63.2
4.63.1
4.63.0
4.62.0
4.61.1
4.61.0
4.60.4
4.60.3
4.60.2
4.60.1
4.60.0
4.59.0
4.58.0
4.57.2
4.57.1
4.57.0
4.56.2
4.56.1
4.56.0
4.55.0
4.54.0
4.53.0
4.52.1
4.52.0
4.51.3
4.51.2
4.51.0
4.50.0
4.49.3
4.49.2
OTP backports
Current section
Files
Jump to
Current section
Files
src/otpbp_pg2.erl
-module(otpbp_pg2).
-ifndef(HAVE_pg2__create_1).
% OTP < 24.0
-export([create/1]).
-endif.
-ifndef(HAVE_pg2__delete_1).
% OTP < 24.0
-export([delete/1]).
-endif.
-ifndef(HAVE_pg2__get_closest_pid_1).
% OTP < 24.0
-export([get_closest_pid/1]).
-endif.
-ifndef(HAVE_pg2__get_local_members_1).
% OTP < 24.0
-export([get_local_members/1]).
-endif.
-ifndef(HAVE_pg2__get_members_1).
% OTP < 24.0
-export([get_members/1]).
-endif.
-ifndef(HAVE_pg2__join_2).
% OTP < 24.0
-export([join/2]).
-endif.
-ifndef(HAVE_pg2__leave_2).
% OTP < 24.0
-export([leave/2]).
-endif.
-ifndef(HAVE_pg2__start_0).
% OTP < 24.0
-export([start/0]).
-endif.
-ifndef(HAVE_pg2__start_link_0).
% OTP < 24.0
-export([start_link/0]).
-endif.
-ifndef(HAVE_pg2__which_groups_0).
% OTP < 24.0
-export([which_groups/0]).
-endif.
-ifndef(HAVE_pg2__create_1).
-ifdef(HAVE_pg2__start_0).
-import(pg2, [start/0]).
-endif.
-endif.
-ifndef(HAVE_pg2__delete_1).
-ifdef(HAVE_pg2__start_0).
-import(pg2, [start/0]).
-endif.
-endif.
-ifndef(HAVE_pg2__get_closest_pid_1).
-ifdef(HAVE_pg2__get_local_members_1).
-import(pg2, [get_local_members/1]).
-endif.
-ifdef(HAVE_pg2__get_members_1).
-import(pg2, [get_members/1]).
-endif.
-endif.
-ifndef(HAVE_pg2__get_local_members_1).
-ifdef(HAVE_pg2__start_0).
-import(pg2, [start/0]).
-endif.
-endif.
-ifndef(HAVE_pg2__get_members_1).
-ifdef(HAVE_pg2__start_0).
-import(pg2, [start/0]).
-endif.
-endif.
-ifndef(HAVE_pg2__join_2).
-ifdef(HAVE_pg2__start_0).
-import(pg2, [start/0]).
-endif.
-endif.
-ifndef(HAVE_pg2__leave_2).
-ifdef(HAVE_pg2__start_0).
-import(pg2, [start/0]).
-endif.
-endif.
-ifndef(HAVE_pg2__which_groups_0).
-ifdef(HAVE_pg2__start_0).
-import(pg2, [start/0]).
-endif.
-endif.
-ifndef(HAVE_pg2__init_1).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
-behaviour(gen_server).
-define(GEN_SERVER_MODULE, ?MODULE).
-ifndef(NEED_record__local_member).
-define(NEED_record__local_member, true).
-endif.
-else.
-define(GEN_SERVER_MODULE, pg2).
-endif.
-define(SERVICE, pg2).
-ifndef(HAVE_pg2__get_local_members_1).
-ifndef(NEED_record__local_member).
-define(NEED_record__local_member, true).
-endif.
-endif.
-ifdef(NEED_record__local_member).
-record(local_member, {name, pid}).
-endif.
-ifndef(HAVE_pg2__create_1).
create(Name) ->
start(),
ets:member(pg2_table, {group, Name}) orelse
global:trans({{?SERVICE, Name}, self()}, fun() -> gen_server:multi_call(?SERVICE, {create, Name}) end),
ok.
-endif.
-ifndef(HAVE_pg2__delete_1).
delete(Name) ->
start(),
global:trans({{?SERVICE, Name}, self()}, fun() -> gen_server:multi_call(?SERVICE, {delete, Name}) end),
ok.
-endif.
-ifndef(HAVE_pg2__get_closest_pid_1).
get_closest_pid(Name) ->
case get_local_members(Name) of
[Pid] -> Pid;
[] ->
case get_members(Name) of
[] -> {error, {no_process, Name}};
Members -> random_element(Members)
end;
Members when is_list(Members) -> random_element(Members);
Else -> Else
end.
random_element(List) -> lists:nth(abs(erlang:monotonic_time() bxor erlang:unique_integer()) rem length(List) + 1, List).
-endif.
-ifndef(HAVE_pg2__get_local_members_1).
get_local_members(Name) ->
start(),
case ets:member(pg2_table, {group, Name}) of
true -> local_group_members(Name);
_false -> {error, {no_such_group, Name}}
end.
-compile({inline, local_group_members/1}).
local_group_members(Name) ->
[P || [Pid] <- ets:match(pg2_table, {#local_member{name = Name, pid = '$1'}}), P <- member_in_group(Pid, Name)].
-ifndef(NEED_member_in_group_2).
-define(NEED_member_in_group_2, true).
-endif.
-endif.
-ifndef(HAVE_pg2__get_members_1).
get_members(Name) ->
start(),
case ets:member(pg2_table, {group, Name}) of
true -> group_members(Name);
_false -> {error, {no_such_group, Name}}
end.
-ifndef(NEED_group_members_1).
-define(NEED_group_members_1, true).
-endif.
-endif.
-ifndef(HAVE_pg2__join_2).
join(Name, Pid) -> act(Name, Pid, join).
-ifndef(NEED_act_3).
-define(NEED_act_3, true).
-endif.
-endif.
-ifndef(HAVE_pg2__leave_2).
leave(Name, Pid) -> act(Name, Pid, leave).
-ifndef(NEED_act_3).
-define(NEED_act_3, true).
-endif.
-endif.
-ifndef(HAVE_pg2__start_0).
start() ->
case whereis(?SERVICE) of
undefined ->
supervisor:start_child(kernel_safe_sup,
#{id => pg2, start => {?GEN_SERVER_MODULE, start_link, []}, shutdown => 1000});
Pid -> {ok, Pid}
end.
-endif.
-ifndef(HAVE_pg2__start_link_0).
start_link() -> gen_server:start_link({local, ?SERVICE}, ?GEN_SERVER_MODULE, [], []).
-endif.
-ifndef(HAVE_pg2__which_groups_0).
which_groups() ->
start(),
all_groups().
-ifndef(NEED_all_groups_0).
-define(NEED_all_groups_0, true).
-endif.
-endif.
-ifndef(HAVE_pg2__init_1).
-record(state, {}).
-record(group, {name}).
-record(ref, {v}).
init([]) ->
ok = net_kernel:monitor_nodes(true),
lists:foreach(fun(N) ->
{?SERVICE, N} ! {new_pg2, node()},
self() ! {nodeup, N}
end, nodes()),
pg2_table = ets:new(pg2_table, [ordered_set, protected, named_table]),
{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) ->
lists:foreach(fun(Pid) -> leave_group(Name, Pid) end, group_members(Name)),
true = ets:delete(pg2_table, #group{name = Name}),
{reply, ok, S};
handle_call(Request, From, S) ->
error_logger:warning_msg("The pg2 server received an unexpected message:\nhandle_call(~tp, ~tp, _)\n",
[Request, From]),
{noreply, S}.
handle_cast({exchange, _Node, Names}, S) ->
lists:foreach(fun([Name, Members]) ->
assure_group(Name) andalso lists:foreach(fun(P) -> join_group(Name, P) end,
Members -- group_members(Name));
(_) -> ok
end,
Names),
{noreply, S};
handle_cast(_, S) -> {noreply, S}.
-record(pid, {pid, name}).
handle_info({'DOWN', MonitorRef, process, _Pid, _Info}, S) ->
[{_, Pid}] = ets:lookup(pg2_table, #ref{v = MonitorRef}),
Names = all_matches({#pid{pid = Pid, name = '$1'}}),
_ = [leave_group(Name, P) || Name <- Names, P <- member_in_group(Pid, Name)],
lists:foreach(fun(Name) -> gen_server:abcast(nodes(), ?SERVICE, {del_member, Name, Pid}) end, Names),
{noreply, S};
handle_info({T, Node}, S) when T =:= nodeup; T =:= new_pg2 ->
gen_server:cast({?SERVICE, Node}, {exchange, node(), [[G, group_members(G)] || G <- all_groups()]}),
{noreply, S};
handle_info(_, S) -> {noreply, S}.
terminate(_Reason, _S) -> ets:delete(pg2_table).
assure_group(Name) ->
Key = #group{name = Name},
ets:member(pg2_table, Key) orelse true =:= ets:insert(pg2_table, {Key}).
-record(member, {name, pid}).
join_group(Name, Pid) ->
Ref_Pid = #ref{v = Pid},
try ets:update_counter(pg2_table, Ref_Pid, {4, 1}) of
_ -> true
catch
_:_ ->
{RPid, Ref} = case lists:member(node(Pid), nodes([this, visible])) of
true -> {Pid, monitor(process, Pid)};
_false ->
spawn_monitor(fun() ->
Ref = monitor(process, Pid),
receive
{'DOWN', Ref, process, Pid, _Info} -> exit(normal)
end
end)
end,
true = ets:insert(pg2_table, [{Ref_Pid, RPid, Ref, 1}, {#ref{v = Ref}, Pid}])
end,
Member = #member{name = Name, pid = Pid},
try
ets:update_counter(pg2_table, Member, {2, 1})
catch _:_ ->
true = ets:insert(pg2_table, [{Member, 1}, {#pid{pid = Pid, name = Name}}]),
node(Pid) =:= node() andalso ets:insert(pg2_table, {#local_member{name = Name, pid = Pid}})
end.
leave_group(Name, Pid) ->
Member = #member{name = Name, pid = Pid},
try ets:update_counter(pg2_table, Member, {2, -1}) of
N ->
N =:= 0 orelse
begin
true = ets:delete(pg2_table, #pid{pid = Pid, name = Name}),
node(Pid) =:= node() andalso ets:delete(pg2_table, #local_member{name = Name, pid = Pid}),
true = ets:delete(pg2_table, Member)
end,
Ref_Pid = #ref{v = Pid},
case ets:update_counter(pg2_table, Ref_Pid, {4, -1}) of
0 ->
[{#ref{}, RPid, Ref, 0}] = ets:lookup(pg2_table, Ref_Pid),
true = ets:delete(pg2_table, #ref{v = Ref}),
true = ets:delete(pg2_table, Ref_Pid),
true = demonitor(Ref, [flush]),
RPid =:= Pid orelse exit(RPid, kill);
_ -> ok
end
catch _:_ -> ok
end.
-ifndef(NEED_member_in_group_2).
-define(NEED_member_in_group_2, true).
-endif.
-ifndef(NEED_all_matches_1).
-define(NEED_all_matches_1, true).
-endif.
-ifndef(NEED_all_groups_0).
-define(NEED_all_groups_0, true).
-endif.
-ifndef(NEED_group_members_1).
-define(NEED_group_members_1, true).
-endif.
-endif.
-ifdef(NEED_act_3).
act(Name, Pid, Action) when is_pid(Pid) ->
start(),
case ets:member(pg2_table, {group, Name}) of
false -> {error, {no_such_group, Name}};
true ->
global:trans({{?SERVICE, Name}, self()},
fun() -> gen_server:multi_call(?SERVICE, {Action, Name, Pid}) end),
ok
end.
-endif.
-ifdef(NEED_group_members_1).
group_members(Name) ->
[P || [P, N] <- ets:match(pg2_table, {#member{name = Name, pid = '$1'}, '$2'}), _ <- lists:seq(1, N)].
-endif.
-ifdef(NEED_member_in_group_2).
member_in_group(Pid, Name) ->
case ets:lookup(pg2_table, #member{name = Name, pid = Pid}) of
[] -> [];
[{#member{}, N}] -> lists:duplicate(N, Pid)
end.
-endif.
-ifdef(NEED_all_groups_0).
all_groups() -> all_matches({#group{name = '$1'}}).
-ifndef(NEED_all_matches_1).
-define(NEED_all_matches_1, true).
-endif.
-endif.
-ifdef(NEED_all_matches_1).
all_matches(Spec) -> [N || [N] <- ets:match(pg2_table, Spec)].
-endif.