% vim: set expandtab softtabstop=4 shiftwidth=4: %% @hidden -module(tf_master_serv). -behaviour(gen_server). %% API functions -export([start_link/2]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -ignore_xref([{start_link, 2}]). %% @headerfile "../include/taskforce.hrl" -include("include/taskforce.hrl"). -type exec_state() :: idle | {running, Bidder::any()}. -record(master_state, { patron_pid :: pid(), patron_monitor :: reference(), max_minion_count :: pos_integer(), exec_state = 'idle' :: exec_state(), tasks = [] :: [tf_task()], consumed_task_ids = [] :: [any()], completed = [] :: [any()], timedout_task_ids = [] :: [any()] }). %%%=================================================================== %%% API functions %%%=================================================================== %%-------------------------------------------------------------------- %% @doc %% Starts the server %% %% @spec start_link(PatronPid, MaxMinionCount) -> {ok, Pid} | ignore | {error, Error} %% @end %%-------------------------------------------------------------------- start_link(PatronPid, MaxMinionCount) -> gen_server:start_link(?MODULE, [PatronPid, MaxMinionCount], []). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %%-------------------------------------------------------------------- %% @private %% @doc %% Initializes the server %% %% @spec init(Args) -> {ok, State} | %% {ok, State, Timeout} | %% ignore | %% {stop, Reason} %% @end %%-------------------------------------------------------------------- init([PatronPid, MaxMinionCount]) -> State0 = #master_state{patron_pid = PatronPid, patron_monitor = monitor(process, PatronPid), max_minion_count = MaxMinionCount}, {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({do_my_bidding, #tf_bidding{}=Bidding}, Bidder, #master_state{ exec_state=idle }=State) -> #tf_bidding{tasks=Tasks, timeout=BiddingTimeout }=Bidding, MaxMinionCount = State#master_state.max_minion_count, MinionCount = min(length(Tasks), MaxMinionCount), ok = spawn_minions(MinionCount), NewExecState = {running, Bidder}, NewState = State#master_state{exec_state = NewExecState, tasks = Tasks}, case length(Tasks) == 0 of true -> {stop, normal, NewState}; false -> erlang:send_after(BiddingTimeout, self(), bidding_timeout), {noreply, NewState} end; handle_call(consume_task, {_MinionPid, _}, #master_state{ exec_state={running, _}, tasks=[Task | Remaining] }=State) -> NewState = State#master_state{tasks = Remaining, consumed_task_ids = [Task#tf_task.id | State#master_state.consumed_task_ids]}, {reply, {ok, Task}, NewState}; handle_call(consume_task, {_MinionPid, _}, #master_state{ exec_state={running, _}, tasks=[] }=State) -> {reply, {error, no_more_tasks}, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling cast messages %% %% @spec handle_cast(Msg, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_cast({{task_completed, TaskId}, TaskResult}, #master_state{ exec_state={running, _} }=State) -> PrevCompleted = State#master_state.completed, NewCompleted = [{TaskId, TaskResult} | PrevCompleted], NewState = State#master_state{completed = NewCompleted}, ConsumedIds = NewState#master_state.consumed_task_ids, Tasks = NewState#master_state.tasks, Timeouts = NewState#master_state.timedout_task_ids, case {length(Tasks), length(NewCompleted) + length(Timeouts), length(ConsumedIds)} of {0, V, V} -> {stop, normal, NewState}; _ -> {noreply, NewState} end; handle_cast({{task_timeout, TaskId}}, #master_state{ exec_state={running, _} }=State) -> PrevTimeouts = State#master_state.timedout_task_ids, NewTimeouts = [TaskId | PrevTimeouts], NewState = State#master_state{timedout_task_ids = NewTimeouts}, ConsumedIds = NewState#master_state.consumed_task_ids, Tasks = NewState#master_state.tasks, Completed = NewState#master_state.completed, case {length(Tasks), length(NewTimeouts) + length(Completed), length(ConsumedIds)} of {0, V, V} -> {stop, normal, NewState}; _ -> {noreply, NewState} end. %%-------------------------------------------------------------------- %% @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}, #master_state{ patron_monitor=Reference }=State) -> {stop, {shutdown, patron_death}, State}; handle_info(bidding_timeout, #master_state{ exec_state={running, _} }=State) -> {stop, {shutdown, bidding_timeout}, 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, #master_state{ exec_state=idle }=_State) -> ok; terminate(_Reason, #master_state{ exec_state={running, Bidder} }=State) -> #master_state{tasks=NeverConsumed, consumed_task_ids=ConsumedIds, completed=Completed}=State, NeverConsumedIds = [Task#tf_task.id || Task <- NeverConsumed], CompletedIds = [TaskId || {TaskId, _} <- Completed], IndividualTimeouts = (ConsumedIds -- CompletedIds), GlobalTimeouts = NeverConsumedIds, BiddingResults = #tf_bidding_result{completed = Completed, individual_timeouts = IndividualTimeouts, global_timeouts = GlobalTimeouts}, gen_server:reply(Bidder, {ok, BiddingResults}), 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 %%%=================================================================== spawn_minions(MaxMinionCount) -> Self = self(), lists:foreach( fun (_Id) -> {ok, _MinionPid} = supervisor:start_child(tf_minion_sup, [Self]) end, lists:seq(1, MaxMinionCount) ).