% vim: set expandtab softtabstop=4 shiftwidth=4: %% @hidden -module(tf_minion_serv). -behaviour(gen_server). %% API functions -export([start_link/1]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -ignore_xref([{start_link, 1}]). %% @headerfile "../include/taskforce.hrl" -include("include/taskforce.hrl"). -record(task_ref, { id :: any(), pid :: pid(), timeout_timer :: reference(), tag :: reference() }). -type task_ref() :: #task_ref{}. -type exec_state() :: idle | {running, task_ref()}. -record(minion_state, { master_pid :: pid(), master_monitor :: reference(), exec_state = 'idle' :: exec_state() }). %%%=================================================================== %%% API functions %%%=================================================================== %%-------------------------------------------------------------------- %% @doc %% Starts the server %% %% @spec start_link(MasterPid) -> {ok, Pid} | ignore | {error, Error} %% @end %%-------------------------------------------------------------------- start_link(MasterPid) -> gen_server:start_link(?MODULE, [MasterPid], []). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %%-------------------------------------------------------------------- %% @private %% @doc %% Initializes the server %% %% @spec init(Args) -> {ok, State} | %% {ok, State, Timeout} | %% ignore | %% {stop, Reason} %% @end %%-------------------------------------------------------------------- init([MasterPid]) -> State0 = #minion_state{master_pid = MasterPid, master_monitor = monitor(process, MasterPid)}, gen_server:cast(self(), get_task), {ok, State0}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling call messages %% %% @spec handle_call(Request, From, State) -> %% {reply, Reply, State} | %% {reply, Reply, State, Timeout} | %% {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, Reply, State} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_call(_Request, _From, #minion_state{}=State) -> {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling cast messages %% %% @spec handle_cast(Msg, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_cast(get_task, #minion_state{ exec_state=idle }=State) -> MasterPid = State#minion_state.master_pid, case catch gen_server:call(MasterPid, consume_task) of {ok, Task} -> #tf_task{id=TaskId, fun_ref=TaskFun, args=TaskArgs, timeout=Timeout }=Task, MinionPid = self(), TaskTag = make_ref(), TimeoutTimer = erlang:send_after(Timeout, MinionPid, {task_timeout, TaskTag}), HandlerFun = fun() -> Result = (catch apply(TaskFun, TaskArgs)), gen_server:cast(MinionPid, {{task_result, TaskTag}, Result}) end, TaskPid = spawn(HandlerFun), TaskRef = #task_ref{id = TaskId, pid = TaskPid, timeout_timer = TimeoutTimer, tag = TaskTag}, NewExecState = {running, TaskRef}, NewState = State#minion_state{ exec_state=NewExecState }, {noreply, NewState}; {error, no_more_tasks} -> {stop, normal, State}; {'EXIT', _} -> {stop, normal, State} end; handle_cast({{task_result, TaskTag}, TaskResult}, #minion_state{ exec_state=({running, (#task_ref{ tag=TaskTag }=TaskRef)}) }=State) -> #task_ref{id = TaskId, timeout_timer = TimeoutTimer }=TaskRef, _ = erlang:cancel_timer(TimeoutTimer), MasterPid = State#minion_state.master_pid, gen_server:cast(MasterPid, {{task_completed, TaskId}, TaskResult}), gen_server:cast(self(), get_task), NewState = State#minion_state{ exec_state=idle }, {noreply, NewState}; handle_cast({{task_result, _TaskTag}, _TaskResult}, #minion_state{}=State) -> {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling all non call/cast messages %% %% @spec handle_info(Info, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_info({'DOWN', Reference, process, _Pid, _Reason}, #minion_state{ master_monitor=Reference }=State) -> {stop, normal, State}; handle_info({task_timeout, TaskTag}, #minion_state{ exec_state={running, (#task_ref{ tag=TaskTag }=TaskRef)} }=State) -> MasterPid = State#minion_state.master_pid, TaskId = TaskRef#task_ref.id, gen_server:cast(MasterPid, {{task_timeout, TaskId}}), gen_server:cast(self(), get_task), NewState = State#minion_state{ exec_state=idle }, {noreply, NewState}; handle_info({task_timeout, _TaskTag}, #minion_state{}=State) -> {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. %% %% @spec terminate(Reason, State) -> void() %% @end %%-------------------------------------------------------------------- terminate(_Reason, #minion_state{}=_State) -> ok. %%-------------------------------------------------------------------- %% @private %% @doc %% Convert process state when code is changed %% %% @spec code_change(OldVsn, State, Extra) -> {ok, NewState} %% @end %%-------------------------------------------------------------------- code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%===================================================================