Packages

For registering processes

Current section

Files

Jump to
regis src regis_pstore_ets.erl
Raw

src/regis_pstore_ets.erl

-module(regis_pstore_ets).
-export([ new/0
, delete/1
, register/4
, unregister/3
, unregister_all/2
, whereis_job/2
, jobs/2
, num_registered_pids/1
, fold_names/3
, fold_pids/3
]).
-record(pstore, {proc_store, proc_pids}).
-opaque pstore() :: #pstore{}.
-export_type([pstore/0]).
new() ->
JobsId = ets:new(pscomp_proc_jobs, [public, named_table]),
PidsId = ets:new(pscomp_proc_pids, [public, named_table]),
{ok, _State=#pstore{proc_store=JobsId, proc_pids=PidsId}}.
delete(_State=#pstore{proc_store=JobsId, proc_pids=PidsId}) ->
true = ets:delete(JobsId),
true = ets:delete(PidsId),
{ok, #pstore{proc_store=undefined, proc_pids=undefined}}.
-spec register(pstore(), term(), term(), pid()) -> {ok, pstore()} | {error, exists_proc_store, none_proc_pids} | {error, none_proc_store, exists_proc_pids}.
register(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta={_App, _CallBack, _ProcId}, VNodeId, Pid) when is_pid(Pid) ->
register_( State
, JobMeta
, _VnodeItem={VNodeId, Pid}
, { ets:lookup(JobsId, JobMeta), ets:lookup(PidsId, Pid) }
).
register_(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta, VnodeItem={_VNodeId, Pid}, {[], []}) ->
link(Pid),
true = ets:insert(JobsId, {JobMeta, [ VnodeItem ]}),
true = ets:insert(PidsId, {Pid, [ JobMeta ]}),
{ok, State};
register_(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta, VnodeItem={_VNodeId, Pid}, {[{_JobMeta, VnodeItems}], [{_Pid, JobMetas}]}) ->
VnodeItem1 = [VnodeItem] ++ [ VI || VI <- VnodeItems, VI/=VnodeItem],
JobMeta1 = [JobMeta] ++ [ PI || PI <- JobMetas, PI/=JobMeta],
true = ets:insert(JobsId, {JobMeta, VnodeItem1}),
true = ets:insert(PidsId, {Pid, JobMeta1}),
{ok, State};
register_(_State, _JobMeta, _VnodeItem, {[{_JobMeta, _VnodeItems}], []}) ->
{error, exists_proc_store, none_proc_pids};
register_(_State, _JobMeta, _VnodeItem, {[], [{_Pid, _JobMetas}]}) ->
{error, none_proc_store, exists_proc_pids}.
-spec unregister(pstore(), term(), pid()) -> {ok, pstore()} | {error, exists_proc_store, none_proc_pids} | {error, none_proc_store, exists_proc_pids}.
unregister(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta={_App, _CallBack, _ProcId}, Pid) when is_pid(Pid) ->
unregister_( State
, JobMeta
, Pid
, { ets:lookup(JobsId, JobMeta), ets:lookup(PidsId, Pid) }
).
unregister_(State, _JobMeta, _Pid, {[], []}) ->
{ok, State};
unregister_(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta={_App, _CallBack, _ProcId}, Pid, {[{_JobMeta, VnodeItems}], [{_Pid, JobMetas}]}) ->
VnodeItemsFiltered = [{VNodeId, P} || {VNodeId, P} <- VnodeItems, P/=Pid],
JobMetaFiltered = JobMetas -- [ JobMeta ],
case {VnodeItemsFiltered, JobMetaFiltered} of
{[], []} ->
unlink(Pid),
true = ets:delete(JobsId, JobMeta),
true = ets:delete(PidsId, Pid),
{ok, State};
{[], JobMetas1} ->
true = ets:delete(JobsId, JobMeta),
true = ets:insert(PidsId, {Pid, JobMetas1}),
{ok, State};
{VnodeItems1, []} ->
true = ets:insert(JobsId, {JobMeta, VnodeItems1}),
true = ets:delete(PidsId, Pid),
{ok, State};
{VnodeItems2, JobMetas2} ->
true = ets:insert(JobsId, {JobMeta, VnodeItems2}),
true = ets:insert(PidsId, {Pid, JobMetas2}),
{ok, State}
end;
unregister_(_State, _JobMeta, _Pid, {[{_JobMeta, _VnodeItems}], []}) ->
{error, exists_proc_store, none_proc_pids};
unregister_(_State, _JobMeta, _Pid, {[], [{_Pid, _JobMetas}]}) ->
{error, none_proc_store, exists_proc_pids}.
-spec unregister_all(pstore(), pid()) -> {ok, pstore()}.
unregister_all(State=#pstore{proc_store=_JobsId, proc_pids=PidsId}, Pid) ->
case ets:lookup(PidsId, Pid) of
[] ->
{ok, State};
[{_Pid, JobMetas}] ->
[ unregister(State, JobMeta, Pid) || JobMeta <- JobMetas ],
{ok, State}
end.
-spec num_registered_pids(pstore()) -> integer().
num_registered_pids(_State=#pstore{proc_store=_JobsId, proc_pids=PidsId}) ->
ets:info(PidsId, size).
-spec whereis_job(pstore() | atom(), term()) -> pid() | undefined.
whereis_job(_State=#pstore{proc_store=JobsId, proc_pids=_PidsId}, JobMeta) ->
whereis_job(JobsId, JobMeta);
whereis_job(JobsId, JobMeta) ->
case jobs(JobsId, JobMeta) of
[{_, Pid}] -> Pid;
_ -> undefined
end.
-spec jobs(pstore() | atom(), term()) -> [{term(), pid()}].
jobs(_State=#pstore{proc_store=JobsId, proc_pids=_PidsId}, JobMeta) ->
jobs(JobsId, JobMeta);
jobs(JobsId, JobMeta) ->
case ets:lookup(JobsId, JobMeta) of
[] -> [];
[{_JobMeta, VnodeItems}] ->
[{VNodeId, Pid} || {VNodeId, Pid} <- VnodeItems]
end.
-spec fold_names(function(), term(), pstore()) -> term().
fold_names(UserFun, Acc0, _State=#pstore{proc_store=JobsId, proc_pids=_PidsId}) when is_function(UserFun, 3) ->
Fun = fun({JobMeta, Values}, Acc) -> UserFun(JobMeta, Values, Acc) end,
ets:foldl(Fun, Acc0, JobsId).
-spec fold_pids(function(), term(), pstore()) -> term().
fold_pids(UserFun, Acc0, _State=#pstore{proc_store=_JobsId, proc_pids=PidsId}) when is_function(UserFun, 3) ->
Fun = fun({Pid, JobMetas}, Acc) -> UserFun(Pid, JobMetas, Acc) end,
ets:foldl(Fun, Acc0, PidsId).
%% ===================================================================
%% EUnit tests
%% ===================================================================
-include_lib("eunit/include/eunit.hrl").
-define(setup(F), {setup, fun start/0, fun stop/1, F}).
%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% TESTS DESCRIPTIONS %%%
%%%%%%%%%%%%%%%%%%%%%%%%%%
start_stop_test_() ->
{"1. Модуль должен реализовывать ETS-реестр процессов, который можно создать при запуске процесса и удалить при остановке",
?setup(fun is_registered/1)}.
register_test_() ->
[
{"1 Register. Должна иметься возможность регистрации Pid и Job из одной функции", ?setup(fun register_t/1)},
{"2 UnRegister. Должна иметься возможность снятия с регистрации Pid и Job из одной функции обладая полями (JobMeta и Pid)", ?setup(fun unregister_t/1)},
{"3 Search. Должна быть возможность определения Pid по имени Job, извлечения всех зарегистрированных джобов и количества зарегистрированных Pid", ?setup(fun search_t/1)}
].
%%%%%%%%%%%%%%%%%%%%%%%
%%% SETUP FUNCTIONS %%%
%%%%%%%%%%%%%%%%%%%%%%%
start() ->
{ok, State} = new(),
State.
stop(State) ->
{ok, State1} = delete(State),
State1.
%%%%%%%%%%%%%%%%%%%%
%%% ACTUAL TESTS %%%
%%%%%%%%%%%%%%%%%%%%
is_registered(State) ->
#pstore{proc_store=JobsId, proc_pids=PidsId} = State,
[
?_assertEqual(pscomp_proc_jobs, JobsId),
?_assertEqual(pscomp_proc_pids, PidsId)
].
register_t(State) ->
Self = self(),
JobMeta1 = {pscomp1, sample1, Self},
VnodeItem1 = {111111111112234455666, Self},
#pstore{proc_store=JobsId, proc_pids=PidsId}=State,
[
?_assertMatch({ok, _State}, register(State, JobMeta1, VnodeItem1, Self) ),
?_assertEqual(1, num_items(JobsId, JobMeta1) ),
?_assertEqual(1, num_items(PidsId, Self) )
].
unregister_t(State) ->
Self = self(),
JobMeta1 = {pscomp1, sample1, Self},
VnodeItem1 = {111111111112234455666, Self},
VnodeItem2 = {212313123486898080821, Self},
#pstore{proc_store=JobsId, proc_pids=PidsId}=State,
[ register(State, JobMeta1, VnodeItem, Self) || VnodeItem <- [ VnodeItem1, VnodeItem2 ] ],
[
?_assertEqual(2, num_items(JobsId, JobMeta1) ),
?_assertEqual(1, num_items(PidsId, Self) ),
?_assertMatch({ok, _State}, unregister(State, JobMeta1, Self) ),
?_assertEqual(0, num_items(JobsId, JobMeta1) ),
?_assertEqual(0, num_items(PidsId, Self) )
].
search_t(State) ->
Self = self(),
JobMeta1 = {pscomp1, sample1, Self},
VnodeItem1 = {111111111112234455666, Self},
VnodeItem2 = {212313123486898080821, Self},
[
?_assertMatch({ok, _State}, register(State, JobMeta1, VnodeItem1, Self) ),
?_assertEqual(true, is_pid( whereis_job(State, JobMeta1) ) ),
?_assertMatch({ok, _State}, register(State, JobMeta1, VnodeItem2, Self) ),
?_assertEqual(1, num_registered_pids(State) ),
?_assertEqual(2, lists:sum( lists:map(fun(_X) -> 1 end, jobs(State, JobMeta1) ) ) )
].
%%%%%%%%%%%%%%%%%%%%%%%%
%%% HELPER FUNCTIONS %%%
%%%%%%%%%%%%%%%%%%%%%%%%
num_items(TabId, LookupItem) ->
case ets:lookup(TabId, LookupItem) of
[] ->
0;
[{_Pid, Items}] ->
lists:sum( lists:map(fun(_X) -> 1 end, Items) )
end.