@@ -4,6 +4,8 @@ -export([start_link/1, init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). +-include("task.hrl"). + start_link(Nodes) -> error_logger:info_report([{"Fair scheduler starts"}]), case gen_server:start_link({local, scheduler}, @@ -17,7 +19,7 @@ init(Nodes) -> {ok, "fifo"} -> error_logger:info_report( [{"Scheduler uses fifo policy"}]), - fair_scheduler_fifo_policy:start_link(Nodes) + fair_scheduler_fifo_policy:start_link(Nodes); _ -> error_logger:info_report( [{"Scheduler uses fair policy"}]), @@ -33,7 +35,7 @@ handle_cast({update_nodes, NewNodes}, _) -> handle_cast({new_task, Task}, Nodes) -> JobName = Task#task.jobname, - JobPid = case ets:lookup(jobs, JobName) of + Job = case ets:lookup(jobs, JobName) of [] -> {ok, JobPid} = fair_scheduler_job:start( Task#task.jobname, Task#task.from), @@ -44,7 +46,7 @@ handle_cast({new_task, Task}, Nodes) -> JobPid; [{_, JobPid}] -> JobPid end, - gen_server:cast(JobPid, {new_task, Task}), + gen_server:cast(Job, {new_task, Task}), {noreply, Nodes}. handle_call({next_job, AvailableNodes}, _From, Nodes) -> @@ -56,13 +58,13 @@ next_job(AvailableNodes, Jobs, NotJobs) -> {ok, JobPid} -> case fair_scheduler_job:next_task( JobPid, Jobs, AvailableNodes) of - {ok, {Node, Task} = E} -> - {ok, E}; + {ok, Task} -> + {ok, Task}; none -> next_job(AvailableNodes, Jobs, [JobPid|NotJobs]) end; - nojobs -> nojobs; + nojobs -> nojobs end. job_cast(Msg) -> master/src/fair_scheduler.erl @@ -22,9 +22,9 @@ init(Nodes) -> register(fairy, spawn_link(fun() -> fairness_fairy(NumCores) end)), {ok, {gb_trees:empty(), [], NumCores}}. -% messages starting with _ are not part of the public policy api +% messages starting with 'priv' are not part of the public policy api -handle_cast({_update_priorities, Priorities}, _, {Jobs, _, NC}) -> +handle_cast({priv_update_priorities, Priorities}, {Jobs, _, NC}) -> % The Jobs tree may have changed while fairy was working. % Update only the elements that fairy knew about. NewJobs = lists:foldl(fun({JobPid, NewJob}, NJobs) -> @@ -42,12 +42,12 @@ handle_cast({_update_priorities, Priorities}, _, {Jobs, _, NC}) -> % Cluster topology has changed. Inform the fairy about the new total % number of cores available. -handle_cast({update_nodes, Nodes}, _, {Jobs, PrioQ, _}) -> +handle_cast({update_nodes, Nodes}, {Jobs, PrioQ, _}) -> NumCores = lists:sum([C || {_, C} <- Nodes]), fairy ! {update, NumCores}, {noreply, {Jobs, PrioQ, NumCores}}; -handle_cast({new_job, JobPid, JobName}, _, {Jobs, PrioQ, NC}) -> +handle_cast({new_job, JobPid, JobName}, {Jobs, PrioQ, NC}) -> InitialPrio = -1.0 / lists:max([gb_trees:size(Jobs), 1.0]), Job = #job{name = JobName, cputime = 0, prio = InitialPrio, @@ -62,13 +62,13 @@ handle_call({next_job, _}, _, {{0, _}, _, _} = S) -> % NotJobs lists all jobs that got 'none' reply from the fair_scheduler_job task % scheduler. We want to skip them. -handle_call({next_job, NotJobs} _, {Jobs, PrioQ, NC}) -> +handle_call({next_job, NotJobs}, _, {Jobs, PrioQ, NC}) -> {NextJob, RPrioQ} = dropwhile(PrioQ, [], NotJobs), {UJobs, UPrioQ} = bias_priority( gb_trees:get(NextJob), RPrioQ, Jobs, NC), {reply, {ok, NextJob}, {UJobs, UPrioQ, NC}}; -handle_call(_get_jobs, _, {Jobs, _, _} = S) -> +handle_call(priv_get_jobs, _, {Jobs, _, _} = S) -> {reply, {ok, Jobs}, S}. handle_info({'DOWN', _, _, JobPid, _}, {Jobs, PrioQ, NC}) -> @@ -78,7 +78,7 @@ handle_info({'DOWN', _, _, JobPid, _}, {Jobs, PrioQ, NC}) -> dropwhile([JobPid|R], H, NotJobs) -> case lists:member(JobPid, NotJobs) of false -> {JobPid, lists:reverse(H) ++ R}; - true -> dropwhile(R, [JobPid|H], NotJob) + true -> dropwhile(R, [JobPid|H], NotJobs) end. % Bias priority is a cheap trick to estimate a new priority for a job that @@ -95,8 +95,7 @@ bias_priority(Job, PrioQ, Jobs, NumCores) -> % Insert an item to an already sorted list prioq_insert(Item, R) -> prioq_insert(Item, R, []). -prioq_insert({Prio, _} = Item, [], H) -> - lists:reverse([Item|H]); +prioq_insert(Item, [], H) -> lists:reverse([Item|H]); prioq_insert({Prio, _} = Item, [{P, _} = E|R], H) when Prio > P -> prioq_insert(Item, R, [E|H]); prioq_insert(Item, L, H) -> @@ -114,11 +113,11 @@ fairness_fairy(NumCores) -> {ok, Alpha} = application:get_env(fair_scheduler_alpha), update_priorities(Alpha, NumCores), fairness_fairy(NumCores) - end + end. update_priorities(_, 0) -> ok; update_priorities(Alpha, NumCores) -> - {ok, Jobs} = gen_server:call(sched_policy, _get_jobs), + {ok, Jobs} = gen_server:call(sched_policy, priv_get_jobs), NumJobs = gb_trees:size(Jobs), % Get the status of each running job @@ -135,7 +134,7 @@ update_priorities(Alpha, NumCores) -> % Extra resources are shared equally among the needy ExtraShare = lists:sum(Extra) / (NumJobs - length(Extra)), - gen_server:cast(sched_policy, {_update_priorities, lists:map(fun + gen_server:cast(sched_policy, {priv_update_priorities, lists:map(fun ({Job, {NumTasks, NumRunning}}) -> MyShare = if NumTasks < Share -> @@ -151,18 +150,9 @@ update_priorities(Alpha, NumCores) -> Prio = Alpha * Deficit + (1 - Alpha) * Job#job.prio, {Job#job.pid, Job#job{prio = Prio, bias = 0, cputime = Job#job.cputime + NumRunning}} - end, Stats)}), - - - - - - - + end, Stats)}). +% callback stubs +terminate(_Reason, _State) -> {}. - - - - - +code_change(_OldVsn, State, _Extra) -> {ok, State}. master/src/fair_scheduler_fair_policy.erl @@ -16,10 +16,10 @@ start_link(Nodes) -> init(_) -> {ok, queue:new()}. -handle_cast({update_nodes, _}, _, Q) -> +handle_cast({update_nodes, _}, Q) -> {noreply, Q}; -handle_cast({new_job, JobPid, JobName}, _, Q) -> +handle_cast({new_job, JobPid, JobName}, Q) -> erlang:monitor(process, JobPid), {noreply, queue:in(JobPid, Q)}. @@ -32,7 +32,7 @@ dropwhile(Q, NotJobs) -> V = lists:member(Job, NotJobs), if V -> dropwhile(NQ, NotJobs); - false -> + true -> {ok, Job} end; {empty, _} -> nojobs @@ -43,7 +43,6 @@ handle_info({'DOWN', _, _, Job, _}, Q) -> % unused - terminate(_Reason, _State) -> {}. code_change(_OldVsn, State, _Extra) -> {ok, State}. master/src/fair_scheduler_fifo_policy.erl @@ -2,9 +2,11 @@ -module(fair_scheduler_job). -behaviour(gen_server). --export([start/0, init/1, next_task/3, handle_call/3, handle_cast/2, +-export([start/2, init/1, next_task/3, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). +-include("task.hrl"). + start(JobName, JobCoord) -> error_logger:info_report([{"JobProc starts for", JobName}]), gen_server:start(fair_scheduler_job, JobCoord, []). @@ -32,14 +34,14 @@ next_task(Job, Jobs, AvailableNodes) -> nonodes -> none end - end + end. % Return an often empty subset of AvailableNodes that don't have any tasks % assigned to them by any job. all_empty_nodes(_, []) -> []; all_empty_nodes([], AvailableNodes) -> AvailableNodes; all_empty_nodes([Job|Jobs], AvailableNodes) -> - get_empty(Jobs, gen_server:call(Job, + all_empty_nodes(Jobs, gen_server:call(Job, {get_empty_nodes, AvailableNodes})). % Assign a new task to this job. @@ -48,7 +50,7 @@ handle_cast({new_task, Task}, {Tasks, Running, Nodes}) -> {noreply, {NewTasks, Running, Nodes}}; % Cluster topology changed (see below). -handle_cast({update_nodes, NewNodes}, {Tasks, Running, Nodes}) -> +handle_cast({update_nodes, NewNodes}, {Tasks, Running, _}) -> NewTasks = reassign_tasks(Tasks, NewNodes), {noreply, {NewTasks, Running, NewNodes}}; @@ -69,7 +71,7 @@ handle_call({get_empty_nodes, AvailableNodes}, _, {Tasks, _, _} = S) -> case gb_trees:get(nopref, Tasks) of {0, _} -> {reply, empty_nodes(Tasks, AvailableNodes), S}; - _ => + _ -> {reply, [], S} end; @@ -79,7 +81,7 @@ handle_call({schedule_local, AvailableNodes}, _, {Tasks, Running, Nodes}) -> {reply, Reply, {UpdatedTasks, Running, Nodes}}; % Secondary scheduling policy (see below). -handle_call({schedule_remote, FreeNodes}, _, {Tasks, Running, Nodes} = S) -> +handle_call({schedule_remote, FreeNodes}, _, {Tasks, Running, Nodes}) -> {Reply, UpdatedTasks} = schedule_remote(Tasks, FreeNodes), {reply, Reply, {UpdatedTasks, Running, Nodes}}. @@ -108,7 +110,7 @@ schedule_local(Tasks, AvailableNodes) -> end; % Local tasks found. Choose an AvailableNode that has the % longest queue of tasks waiting. Pick the first task from it. - Nodes -> pop_busiest_node(Tasks, AvailableNodes) + Nodes -> pop_busiest_node(Tasks, Nodes) end. % Secondary task scheduling policy: @@ -136,11 +138,11 @@ pop_busiest_node(Tasks, Nodes) -> % return nodes that don't have any local tasks assigned to them empty_nodes(Tasks, AvailableNodes) -> - filter_nodes(Tasks, AvailableNodes, false); + filter_nodes(Tasks, AvailableNodes, false). % return nodes that have at least one local tasks assigned to them datalocal_nodes(Tasks, AvailableNodes) -> - filter_nodes(Tasks, AvailableNodes, true); + filter_nodes(Tasks, AvailableNodes, true). filter_nodes(Tasks, AvailableNodes, Local) -> lists:filter(fun(Node) -> @@ -158,8 +160,8 @@ filter_nodes(Tasks, AvailableNodes, Local) -> assign_task(Task, Tasks, Nodes) -> findpref(Task#task.input, Task, Tasks, Nodes). -findpref([], Task, Tasks, Nodes) -> - {N, L} = gb_trees:get(nopref, Tasks) +findpref([], Task, Tasks, _) -> + {N, L} = gb_trees:get(nopref, Tasks), gb_trees:update(nopref, {N + 1, [Task|L]}, Tasks); findpref([{_, Node}|R], Task, Tasks, Nodes) -> @@ -167,14 +169,14 @@ findpref([{_, Node}|R], Task, Tasks, Nodes) -> none -> ValidNode = lists:member(Node, Nodes), if ValidNode -> - gb_trees:insert(Node, {1, [Task]}, Tasks) + gb_trees:insert(Node, {1, [Task]}, Tasks); true -> - findpref(R, Task, Tasks) + findpref(R, Task, Tasks, Nodes) end; {value, {N, L}} -> IsBlack = lists:member(Node, Task#task.taskblack), if IsBlack -> - findpref(R, Task, Tasks); + findpref(R, Task, Tasks, Nodes); true -> gb_trees:update(Node, {N + 1, [Task|L]}, Tasks) end @@ -187,14 +189,14 @@ reassign_tasks(Tasks, NewNodes) -> case gb_trees:lookup(Node, OTasks) of {value, TList} -> {gb_trees:delete(Node, OTasks), - gb_trees:insert(Node, TList, NTasks)} + gb_trees:insert(Node, TList, NTasks)}; none -> - {OTasks, NTasks}; + {OTasks, NTasks} end end, {Tasks, gb_trees:empty()}, NewNodes), - lists:foldl(fun(Task, NTasks) -> - assing_task(Task, NTasks) + lists:foldl(fun(Task, NTasks0) -> + assign_task(Task, NTasks0, NewNodes) end, gb_trees:insert(nopref, {0, []}, NTasks), lists:flatten([L || {_, L} <- gb_trees:values(OTasks)])). master/src/fair_scheduler_job.erl 66e0ef22abf56c8445e3c0ce1c367c0d6f140dc7 Ville Tuulos tuulos@dxfront.(none) http://github.com/tuulos/disco/commit/28fc84effdcaad6d21dfdee6202dd6d0aab032fe 28fc84effdcaad6d21dfdee6202dd6d0aab032fe 2009-08-28T16:22:19-07:00 2009-08-28T16:22:19-07:00 scheduler compiles, not tested 521e4b2b7f1816b720d980fee3585d6317187fc0 Ville Tuulos tuulos@dxfront.(none)