-module(tf_minion_serv). -ifdef(E48). -moduledoc false. -endif. -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}]). -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() }). -type state() :: #minion_state{}. -if(?OTP_RELEASE =< 24). -type gen_server_from() :: {pid(), reference()}. -else. -type gen_server_from() :: gen_server:from(). -endif. %%%=================================================================== %%% API functions %%%=================================================================== -spec start_link(pid()) -> {ok, pid()}. start_link(MasterPid) -> gen_server:start_link(?MODULE, [MasterPid], []). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== -spec init([pid(), ...]) -> {ok, state()}. init([MasterPid]) -> State0 = #minion_state{ master_pid = MasterPid, master_monitor = monitor(process, MasterPid) }, gen_server:cast(self(), get_task), {ok, State0}. -spec handle_call(term(), gen_server_from(), state()) -> {noreply, state()}. handle_call(_Request, _From, #minion_state{} = State) -> {noreply, State}. -spec handle_cast(term(), state()) -> {noreply, state()} | {stop, normal, state()}. handle_cast(get_task, #minion_state{exec_state = idle} = State) -> MasterPid = State#minion_state.master_pid, try 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 = run_task(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} catch _Class:_Reason -> {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}. -spec handle_info(term(), state()) -> {noreply, state()} | {stop, normal, state()}. 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}. -spec terminate(term(), state()) -> ok. terminate(_Reason, #minion_state{} = _State) -> ok. -spec code_change(term(), state(), term()) -> {ok, state()}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== run_task(TaskFun, TaskArgs) -> try apply(TaskFun, TaskArgs) catch error:Reason:Stacktrace -> {'EXIT', {Reason, Stacktrace}}; % exit:Reason -> {'EXIT', Reason}; % throw:Reason -> Reason end.