% -*- indent-tabs-mode:nil; -*- %%%------------------------------------------------------------------- %%% @author Fred Youhanaie %%% @copyright (C) 2017, Fred Youhanaie %%% @doc %%% This is the server that manages the Tuple Space. %%% %%% It is a `gen_server' that sits between the `espace' clients and %%% the rest of the espace application. %%% %%% Upon receiving the `worker' request, the worker supervisor will be %%% triggered to start a new child process to run the supplied %%% function. %%% %%% Upon receiving the `eval' request, the worker supervisor will be %%% triggered to start a new child process to evaluate the supplied %%% tuple. %%% %%% For the tuple space data operations, i.e. `in', `rd', `inp', `rdp' %%% and `out', the request is passed to the `espace_tspace_srv' %%% server, which in turn will process the request. %%% %%% The patterns supplied with the input operators should be %%% compatible with `ets:match/2,3' match patterns. When match is %%% found the return results are in the form of `{List, Tuple}', where %%% `List' is the, possibaly empty, list of the matched `$N' pattern %%% variables, and `Tuple' is the entire data tuple. %%% %%% @end %%% Created : 9 Dec 2017 by Fred Youhanaie %%%------------------------------------------------------------------- -module(espace_tspool_srv). -behaviour(gen_server). %% API -export([start_link/1, espace_eval/2, espace_out/2, espace_in/2, espace_rd/2, espace_inp/2, espace_rdp/2, espace_worker/2]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]). -define(SERVER, ?MODULE). -record(state, {inst_name, workersup}). %%%=================================================================== %%% API %%%=================================================================== %%-------------------------------------------------------------------- %% @doc %% Starts the server %% %% @end %%-------------------------------------------------------------------- -spec start_link(atom()) -> ignore | {error, {already_started, pid()} | term()}| {ok, pid()}. start_link(Inst_name) -> gen_server:start_link({local, espace_util:inst_to_name(?SERVER, Inst_name)}, ?MODULE, Inst_name, []). %%-------------------------------------------------------------------- %% @doc %% perform an `eval' operation. %% %% A request is sent to the worker supervisor to start a new child %% process, which will evaluate the elements of the supplied `Tuple'. %% %% @end %%-------------------------------------------------------------------- -spec espace_eval(atom(), tuple()) -> pid(). espace_eval(Inst_name, Tuple) -> gen_server:call(espace_util:inst_to_name(?SERVER, Inst_name), {espace_eval, Tuple}). %%-------------------------------------------------------------------- %% @doc %% start a worker process. %% %% A request is sent to the worker supervisor to start a new child %% process based on the `MFA' parameter. `MFA' can be one of `{Mod, %% Func, Args}' triple, or an anonymous function/args pair `{Func, %% Args}', where `Func' can be a `fun' expression or string %% represtation of a `fun' expression. The string version can be used %% when reading espace terms from a file. %% %% @end %%-------------------------------------------------------------------- -spec espace_worker(atom(), tuple()) -> pid(). espace_worker(Inst_name, MFA) -> gen_server:call(espace_util:inst_to_name(?SERVER, Inst_name), {espace_worker, MFA}). %%-------------------------------------------------------------------- %% @doc %% Perform an `out' operation. %% %% Sends supplied `Tuple' to `espace_tspace_srv' to be stored in the %% ETS table. %% %% `espace_tspace_srv' will in turn trigger `espace_tspatt_srv' to %% inform any `in'/`rd' clients that may be blocking on such %% tuple. The blocking mechanisim is implemented internally in this %% module, see `espace_op/3'. %% %% @end %%-------------------------------------------------------------------- -spec espace_out(atom(), tuple()) -> done. espace_out(Inst_name, Tuple) -> gen_server:call(espace_util:inst_to_name(?SERVER, Inst_name), {espace_out, Tuple}). %%-------------------------------------------------------------------- %% @doc %% Perform an `in' operation. %% %% The request is passed to `espace_tspace_srv' to search for %% `Pattern', if found, the data is returned, otherwise we block until %% a matching tuple is added to the tuple space via `out'. The matched %% tuple will be removed from the tuple space. %% %% @end %%-------------------------------------------------------------------- -spec espace_in(atom(), tuple()) -> {list(), tuple()}. espace_in(Inst_name, Pattern) -> espace_op(Inst_name, espace_in, Pattern). %%-------------------------------------------------------------------- %% @doc %% Perform a `rd' operation. %% %% Similar to `in', except that the matched tuple remains in the tuple %% space. %% %% @end %%-------------------------------------------------------------------- -spec espace_rd(atom(), tuple()) -> {list(), tuple()}. espace_rd(Inst_name, Pattern) -> espace_op(Inst_name, espace_rd, Pattern). %%-------------------------------------------------------------------- %% @doc %% Perform an `inp' operation. %% %% Similar to `in' but does not block if no match found. In case of no %% match, the `nomatch' atom is returned. %% %% @end %%-------------------------------------------------------------------- -spec espace_inp(atom(), tuple()) -> nomatch | {list(), tuple()}. espace_inp(Inst_name, Pattern) -> espace_op(Inst_name, espace_inp, Pattern). %%-------------------------------------------------------------------- %% @doc %% Perform a `rdp' operation. %% %% Similar to `inp', but does not remove the matched tuple. %% %% @end %%-------------------------------------------------------------------- -spec espace_rdp(atom(), tuple()) -> nomatch | {list(), tuple()}. espace_rdp(Inst_name, Pattern) -> espace_op(Inst_name, espace_rdp, Pattern). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %%-------------------------------------------------------------------- %% @private %% @doc %% Initializes the server %% %% @end %%-------------------------------------------------------------------- -spec init(atom()) -> {ok, #state{inst_name::atom(), workersup::atom()}}. init(Inst_name) -> process_flag(trap_exit, true), {ok, #state{inst_name=Inst_name, workersup=espace_util:inst_to_name(espace_worker_sup, Inst_name)}}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling call messages %% %% @end %%-------------------------------------------------------------------- -spec handle_call({atom(), tuple()}, {pid(), term()}, term()) -> {reply, {nomatch} | {nomatch, reference()} | {match, {list(), tuple()}}, term() }. handle_call({espace_in, Pattern}, From, State) -> handle_espace_op(espace_in, Pattern, From, State); handle_call({espace_inp, Pattern}, From, State) -> handle_espace_op(espace_inp, Pattern, From, State); handle_call({espace_rd, Pattern}, From, State) -> handle_espace_op(espace_rd, Pattern, From, State); handle_call({espace_rdp, Pattern}, From, State) -> handle_espace_op(espace_rdp, Pattern, From, State); handle_call({espace_out, Tuple}, _From, State) -> Reply = espace_tspace_srv:add_tuple(State#state.inst_name, Tuple), {reply, Reply, State}; handle_call({espace_eval, Tuple}, _From, State) -> Reply = handle_espace_eval(State#state.inst_name, State#state.workersup, Tuple), {reply, Reply, State}; handle_call({espace_worker, {M, F, A}}, _From, State) -> Reply = handle_espace_worker(State#state.workersup, {M, F, A}), {reply, Reply, State}; handle_call({espace_worker, {Fun, Args}}, _From, State) -> Reply = handle_espace_worker(State#state.workersup, {Fun, Args}), {reply, Reply, State}; handle_call(Request, From, State) -> logger:warning("~p:handle_call: Unexpected request=~p, from pid=~p, ignored.", [?SERVER, Request, From]), Reply = ok, {reply, Reply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling cast messages %% %% @end %%-------------------------------------------------------------------- -spec handle_cast(term(), term()) -> {noreply, term()}. handle_cast(Request, State) -> logger:warning("~p:handle_cast: Unexpected request=~p, ignored.", [?SERVER, Request]), {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling all non call/cast messages %% %% @end %%-------------------------------------------------------------------- -spec handle_info(timeout() | term(), term()) -> {noreply, term(), hibernate} | {noreply, term(), timeout()} | {noreply, term()} | {stop, normal | term(), term()}. handle_info(Info, State) -> logger:warning("~p:handle_info: Unexpected message=~p, ignored.", [?SERVER, Info]), {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% This function is called by a gen_server when it is about to %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_server terminates %% with Reason. The return value is ignored. %% %% @end %%-------------------------------------------------------------------- -spec terminate(atom(), term()) -> ok. terminate(Reason, State) -> logger:notice("~p: terminating, reason=~p, state=~p.", [?SERVER, Reason, State]), ok. %%%=================================================================== %%% Internal functions %%%=================================================================== %%-------------------------------------------------------------------- %% @doc %% unified function for "in" and "rd" client API functions %% %% @end %%-------------------------------------------------------------------- -spec espace_op(atom(), espace_in | espace_inp | espace_rd | espace_rdp, tuple()) -> {list(), tuple()} | nomatch. espace_op(Inst_name, Espace_Op, Pattern) -> Reply = gen_server:call(espace_util:inst_to_name(?SERVER, Inst_name), {Espace_Op, Pattern}), case Reply of {match, Match} -> Match; {nomatch} -> %% only from the inp and rdp operations nomatch; {nomatch, Cli_ref} -> %% only from the in and rd operations receive Cli_ref -> espace_op(Inst_name, Espace_Op, Pattern) % our tuple has arrived, try again! end end. %%-------------------------------------------------------------------- %% @doc %% unified call handler for "in", "inp", "rd" and "rdp" %% %% @end %%-------------------------------------------------------------------- -spec handle_espace_op(espace_in | espace_inp | espace_rd | espace_rdp, tuple(), pid(), term()) -> {reply, {nomatch} | {nomatch, reference()} | {match, {list(), tuple()}}, term() }. handle_espace_op(Espace_Op, Pattern, From, State) -> case espace_tspace_srv:get_tuple(State#state.inst_name, Pattern) of {nomatch} -> case Espace_Op of espace_inp -> {reply, {nomatch}, State}; espace_rdp -> {reply, {nomatch}, State}; _ -> %% only "in" and "rd" should block on no match {Cli_pid, _} = From, %% we use the pid to notify the client Cli_ref = make_ref(), %% the client should wait for this ref espace_tspatt_srv:add_pattern(State#state.inst_name, Cli_ref, Pattern, Cli_pid), {reply, {nomatch, Cli_ref}, State} end; {match, {TabKey, Fields, Tuple}} -> Reply = {match, {Fields, Tuple}}, case Espace_Op of %% "in" and "inp" should remove the tuple espace_in -> espace_tspace_srv:del_tuple(State#state.inst_name, TabKey); espace_inp -> espace_tspace_srv:del_tuple(State#state.inst_name, TabKey); _ -> ok end, {reply, Reply, State} end. %%-------------------------------------------------------------------- %% @private %% @doc handle the espace_worker call. %% %% @end %%-------------------------------------------------------------------- -spec handle_espace_worker(atom(), tuple()) -> pid(). handle_espace_worker(Worker_sup, Arg) -> {ok, Pid} = supervisor:start_child(Worker_sup, erlang:tuple_to_list(Arg)), Pid. %%-------------------------------------------------------------------- %% @private %% @doc handle the espace_eval call. %% %% We start a worker child, and let `espace_util:eval_out' to evaluate %% the tuple and `out' the result. %% %% @end %%-------------------------------------------------------------------- -spec handle_espace_eval(atom(), atom(), tuple()) -> pid(). handle_espace_eval(Inst_name, Worker_sup, Tuple) -> {ok, Pid} = supervisor:start_child(Worker_sup, [espace_util, eval_out, [Inst_name, Tuple]]), Pid.