%%
%% %CopyrightBegin%
%%
%% Copyright Ericsson AB 2010-2017. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
%% %CopyrightEnd%
%%
%%
%% A simple term -> pid registry.
%%
-module(diameter_reg).
-behaviour(gen_server).
-export([add/1,
add_new/1,
remove/1,
match/1,
wait/1,
subscribe/2]).
-export([start_link/0]).
%% gen_server callbacks
-export([init/1,
terminate/2,
handle_call/3,
handle_cast/2,
handle_info/2,
code_change/3]).
%% test
-export([pids/0,
terms/0,
subs/0,
waits/0]).
%% debug
-export([state/0,
uptime/0]).
-define(SERVER, ?MODULE).
-define(TABLE, ?MODULE).
-type key() :: term().
-type from() :: {pid(), term()}.
-type rcvr() :: [pid() | term()] %% subscribe
| from(). %% wait
-type pattern() :: term().
-record(state, {id = diameter_lib:now(),
notify = #{} :: #{pattern() => [rcvr()]},
monitors = sets:new() :: sets:set(pid())}).
%% The ?TABLE bag contains the Key -> Pid mapping, as {Key, Pid}
%% tuples. Each pid is stored in the monitors set to ensure only one
%% monitor for each pid: more are harmless, but unnecessary. A pattern
%% is added to notify a result of calls to wait/1 or subscribe/2:
%% changes to ?TABLE causes processes to be notified as required.
%% ===========================================================================
%% # add(T)
%%
%% Associate the specified term with self(). The list of pids having
%% this or other assocations can be retrieved using match/1.
%%
%% An association is removed when the calling process dies or as a
%% result of calling remove/1. Adding the same term more than once is
%% equivalent to adding it once.
%%
%% Note that since match/1 takes a pattern as argument, specifying a
%% term that contains match variables is probably not a good idea
%% ===========================================================================
-spec add(key())
-> true.
add(T) ->
call({add, false, T}).
%% ===========================================================================
%% # add_new(T)
%%
%% Like add/1 but only one process is allowed to have the the
%% association, false being returned if an association already exists.
%% ===========================================================================
-spec add_new(key())
-> boolean().
add_new(T) ->
call({add, true, T}).
%% ===========================================================================
%% # remove(Term)
%%
%% Remove any existing association of Term with self().
%% ===========================================================================
-spec remove(key())
-> true.
remove(T) ->
call({remove, T}).
%% ===========================================================================
%% # match(Pat)
%%
%% Return the list of associations whose Term, as specified to add/1
%% or add_new/1, matches the specified pattern.
%%
%% Note that there's no guarantee that the returned processes are
%% still alive. (Although one that isn't will soon have its
%% associations removed.)
%% ===========================================================================
-spec match(pattern())
-> [{key(), pid()}].
match(Pat) ->
match(Pat, '_').
%% match/2
match(Pat, Pid) ->
ets:match_object(?TABLE, {Pat, Pid}).
%% ===========================================================================
%% # wait(Pat)
%%
%% Like match/1 but return only when the result is non-empty or fails.
%% It's up to the caller to ensure that the wait won't be forever.
%% ===========================================================================
-spec wait(pattern())
-> [{key(), pid()}].
wait(Pat) ->
_ = match(Pat), %% ensure match can succeed
call({wait, Pat}).
%% ===========================================================================
%% # subscribe(Pat, T)
%%
%% Like match/1, but additionally receive messages of the form
%% {T, add|remove, {term(), pid()}} when associations are added
%% or removed.
%% ===========================================================================
-spec subscribe(Pat :: any(), T :: term())
-> [{term(), pid()}].
subscribe(Pat, T) ->
_ = match(Pat), %% ensure match can succeed
call({subscribe, Pat, T}).
%% ===========================================================================
start_link() ->
ServerName = {local, ?SERVER},
Options = [{spawn_opt, diameter_lib:spawn_opts(server, [])}],
gen_server:start_link(ServerName, ?MODULE, [], Options).
state() ->
call(state).
uptime() ->
call(uptime).
%% pids/0
-spec pids()
-> [{pid(), [key()]}].
pids() ->
append(ets:select(?TABLE, [{{'$1','$2'}, [], [{{'$2', '$1'}}]}])).
append(Pairs) ->
dict:to_list(lists:foldl(fun({K,V}, D) -> dict:append(K, V, D) end,
dict:new(),
Pairs)).
%% terms/0
-spec terms()
-> [{key(), [pid()]}].
terms() ->
append(ets:tab2list(?TABLE)).
%% subs/0
-spec subs()
-> [{pattern(), [{pid(), term()}]}].
subs() ->
#state{notify = Dict} = state(),
[{K, Ts} || {K,Ps} <- maps:to_list(Dict),
Ts <- [[{P,T} || [P|T] <- Ps]]].
%% waits/0
-spec waits()
-> [{pattern(), [from()]}].
waits() ->
#state{notify = Dict} = state(),
[{K, Ts} || {K,Ps} <- maps:to_list(Dict),
Ts <- [[T || {_,_} = T <- Ps]]].
%% ----------------------------------------------------------
%% # init/1
%% ----------------------------------------------------------
init(_) ->
ets:new(?TABLE, [bag, named_table]),
{ok, #state{}}.
%% ----------------------------------------------------------
%% # handle_call/3
%% ----------------------------------------------------------
handle_call({add, Uniq, Key}, {Pid, _}, S) ->
Rec = {Key, Pid},
NS = flush(Uniq, Rec, S), %% before insert
{Res, New} = insert(Uniq, Rec),
{reply, Res, notify(add, New andalso Rec, if New ->
add_monitor(Pid, NS);
true ->
NS
end)};
handle_call({remove, Key}, {Pid, _}, S) ->
Rec = {Key, Pid},
{reply, true, try
notify(remove, Rec, S)
after
ets:delete_object(?TABLE, Rec)
end};
handle_call({wait, Pat}, {Pid, _} = From, S) ->
NS = add_monitor(Pid, S),
case match(Pat) of
[_|_] = Recs ->
{reply, Recs, NS};
[] ->
{noreply, queue(Pat, From, NS)}
end;
handle_call({subscribe, Pat, T}, {Pid, _}, S) ->
{reply, match(Pat), queue(Pat, [Pid | T], add_monitor(Pid, S))};
handle_call(state, _, S) ->
{reply, S, S};
handle_call(uptime, _, #state{id = Time} = S) ->
{reply, diameter_lib:now_diff(Time), S};
handle_call(_Req, _From, S) ->
{reply, nok, S}.
%% ----------------------------------------------------------
%% # handle_cast/2
%% ----------------------------------------------------------
handle_cast(_Msg, S)->
{noreply, S}.
%% ----------------------------------------------------------
%% # handle_info/2
%% ----------------------------------------------------------
handle_info({'DOWN', _MRef, process, Pid, _}, S) ->
{noreply, down(Pid, S)};
handle_info(_Info, S) ->
{noreply, S}.
%% ----------------------------------------------------------
%% # terminate/2
%% ----------------------------------------------------------
terminate(_Reason, _State)->
ok.
%% ----------------------------------------------------------
%% # code_change/3
%% ----------------------------------------------------------
code_change(_, State, "2.1") ->
{ok, lists:foldl(fun add_monitor/2,
State,
ets:select(?TABLE, [{{'_', '$1'}, [], ['$1']}]))};
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%% ===========================================================================
%% insert/2
insert(false, Rec) ->
Spec = [{'$1', [{'==', '$1', {const, Rec}}], ['$_']}],
X = '$end_of_table' /= ets:select(?TABLE, Spec, 1), %% entry exists?
X orelse ets:insert(?TABLE, Rec),
{true, not X};
insert(true, Rec) ->
B = ets:insert_new(?TABLE, Rec), %% entry inserted?
{B, B}.
%% add_monitor/2
%%
%% Only add a single monitor for any given process, since there's no
%% use to more.
add_monitor(Pid, #state{monitors = Ps} = S) ->
case sets:is_element(Pid, Ps) of
false ->
monitor(process, Pid),
S#state{monitors = sets:add_element(Pid, Ps)};
true ->
S
end.
%% notify/3
notify(_, false, S) ->
S;
notify(Op, {_,_} = Rec, #state{notify = Dict} = S) ->
S#state{notify = maps:fold(fun(P,Rs,D) -> notify(Op, Rec, P, Rs, D) end,
Dict,
Dict)}.
%% notify/5
notify(Op, {_, Pid} = Rec, Pat, Rcvrs, Dict) ->
case lists:member(Rec, match(Pat, Pid)) of
true ->
reset(Pat, Dict, [P || P <- Rcvrs, send(P, Op, Rec)]);
false ->
Dict
end.
%% send/3
send([Pid | T], Op, Rec) ->
Pid ! {T, Op, Rec},
true;
%% No processes wait on remove: they receive notification immediately
%% or at add, by construction.
send({_,_} = From, add, Rec) ->
gen_server:reply(From, [Rec]),
false.
%% down/2
down(Pid, #state{monitors = Ps} = S) ->
Recs = match('_', Pid),
Acc0 = flush(Pid, S#state{monitors = sets:del_element(Pid, Ps)}),
try
lists:foldl(fun(R,NS) -> notify(remove, R, NS) end, Acc0, Recs)
after
ets:match_delete(?TABLE, {'_', Pid})
end.
%% flush/3
%% Remove any processes that are dead but for which we may not have
%% received 'DOWN' yet, to ensure that add_new can be used to register
%% a unique name each time a registering process restarts.
flush(true, {Key, Pid}, S) ->
Spec = [{{'$1', '$2'},
[{'andalso', {'==', '$1', {const, Key}},
{'/=', '$2', Pid}}],
['$2']}],
lists:foldl(fun down/2, S, [P || P <- ets:select(?TABLE, Spec),
not is_process_alive(P)]);
flush(false, _, S) ->
S.
%% flush/2
%% Process has died and should no longer receive messages/replies.
flush(Pid, #state{notify = Dict} = S) ->
S#state{notify = maps:fold(fun(P,Rs,D) -> flush(Pid, P, Rs, D) end,
Dict,
Dict)}.
%% flush/4
flush(Pid, Pat, Rcvrs, Dict) ->
reset(Pat, Dict, [T || T <- Rcvrs, Pid /= head(T)]).
%% head/1
head([P|_]) ->
P;
head({P,_}) ->
P.
%% reset/3
reset(Key, Map, []) ->
maps:remove(Key, Map);
reset(Key, Map, List) ->
maps:put(Key, List, Map).
%% queue/3
queue(Pat, Rcvr, #state{notify = Dict} = S) ->
S#state{notify = maps:put(Pat, [Rcvr | maps:get(Pat, Dict, [])], Dict)}.
%% call/1
call(Request) ->
gen_server:call(?SERVER, Request, infinity).
