diff options
Diffstat (limited to 'lib/stdlib/src/pool.erl')
| -rw-r--r-- | lib/stdlib/src/pool.erl | 212 |
1 files changed, 212 insertions, 0 deletions
diff --git a/lib/stdlib/src/pool.erl b/lib/stdlib/src/pool.erl new file mode 100644 index 0000000000..7f5f23e26d --- /dev/null +++ b/lib/stdlib/src/pool.erl @@ -0,0 +1,212 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 1996-2009. All Rights Reserved. +%% +%% The contents of this file are subject to the Erlang Public License, +%% Version 1.1, (the "License"); you may not use this file except in +%% compliance with the License. You should have received a copy of the +%% Erlang Public License along with this software. If not, it can be +%% retrieved online at http://www.erlang.org/. +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and limitations +%% under the License. +%% +%% %CopyrightEnd% +%% +-module(pool). + +%% Supplies a computational pool of processors. +%% The chief user interface function here is get_node() +%% Which returns the name of the nodes in the pool +%% with the least load !!!! +%% This function is callable from any node including the master +%% That is part of the pool +%% nodes are scheduled on a per usgae basis and per load basis, +%% Whenever we use a node, we put at the end of the queue, and whenever +%% a node report a change in load, we insert it accordingly + +% User interface Exports ... +-export([start/1, + start/2, + stop/0, + get_nodes/0, + get_nodes_and_load/0, + get_node/0, + pspawn/3, + attach/1, + pspawn_link/3]). + +%% Internal Exports +-export([statistic_collector/0, + do_spawn/4, + init/1, + handle_call/3, + handle_cast/2, + handle_info/2, + terminate/2]). + +%% User interface + +%% Start up using the .hosts.erlang file + +-spec start(atom()) -> [node()]. +start(Name) -> + start(Name,[]). + +-spec start(atom(), string()) -> [node()]. +start(Name, Args) when is_atom(Name) -> + gen_server:start({global, pool_master}, pool, [], []), + Hosts = net_adm:host_file(), + Nodes = start_nodes(Hosts, Name, Args), + lists:foreach(fun attach/1, Nodes), + Nodes. + +%% +%% Interface functions ... +%% +-spec get_nodes() -> [node()]. +get_nodes() -> + get_elements(2, get_nodes_and_load()). + +-spec attach(node()) -> 'already_attached' | 'attached'. +attach(Node) -> + gen_server:call({global, pool_master}, {attach, Node}). + +get_nodes_and_load() -> + gen_server:call({global, pool_master}, get_nodes). + +-spec get_node() -> node(). +get_node() -> + gen_server:call({global, pool_master}, get_node). + +-spec pspawn(module(), atom(), [term()]) -> pid(). +pspawn(M, F, A) -> + gen_server:call({global, pool_master}, {spawn, group_leader(), M, F, A}). + +-spec pspawn_link(module(), atom(), [term()]) -> pid(). +pspawn_link(M, F, A) -> + P = pspawn(M, F, A), + link(P), + P. + +start_nodes([], _, _) -> []; +start_nodes([Host|Tail], Name, Args) -> + case slave:start(Host, Name, Args) of + {error, R} -> + io:format("Can't start node on host ~w due to ~w~n",[Host, R]), + start_nodes(Tail, Name, Args); + {ok, Node} -> + [Node | start_nodes(Tail, Name, Args)] + end. + +-spec stop() -> 'stopped'. +stop() -> + gen_server:call({global, pool_master}, stop). + +get_elements(_Pos,[]) -> []; +get_elements(Pos,[E|T]) -> [element(Pos,E) | get_elements(Pos,T)]. + +stop_em([]) -> stopped; +stop_em([N|Tail]) -> + rpc:cast(N, erlang, halt, []), + stop_em(Tail). + +init([]) -> + process_flag(trap_exit, true), + spawn_link(pool, statistic_collector, []), + {ok,[{0,node()}]}. + +handle_call(get_nodes, _From, Nodes)-> + {reply, Nodes, Nodes}; +handle_call(get_node, _From, [{Load,N}|Tail]) -> + {reply, N, Tail++[{Load+1, N}]}; +handle_call({attach, Node}, _From, Nodes) -> + case lists:keymember(Node, 2, Nodes) of + true -> + {reply, already_attached, Nodes}; + false -> + erlang:monitor_node(Node, true), + spawn_link(Node, pool, statistic_collector, []), + {reply, attached, Nodes++[{999999,Node}]} + end; +handle_call({spawn, Gl, M, F, A}, _From, Nodes) -> + [{Load,N}|Tail] = Nodes, + Pid = spawn(N, pool, do_spawn, [Gl, M, F, A]), + {reply, Pid, Tail++[{Load+1, N}]}; +handle_call(stop, _From, Nodes) -> + %% clean up in terminate/2 + {stop, normal, stopped, Nodes}. + +handle_cast(_, Nodes) -> + {noreply, Nodes}. + +handle_info({Node,load,Load}, Nodes) -> + Nodes2 = insert_node({Load,Node}, Nodes), + {noreply, Nodes2}; +handle_info({nodedown, Node}, Nodes) -> + {noreply, lists:keydelete(Node, 2, Nodes)}; +handle_info(_, Nodes) -> %% The EXIT signals etc.etc + {noreply, Nodes}. + +terminate(_Reason, Nodes) -> + N = lists:delete(node(), get_elements(2, Nodes)), + stop_em(N), + ok. + +-spec do_spawn(pid(), module(), atom(), [term()]) -> term(). +do_spawn(Gl, M, F, A) -> + group_leader(Gl, self()), + apply(M, F, A). + +insert_node({Load,Node},[{L,Node}|Tail]) when Load > L -> + %% We have a raised load here + pure_insert({Load,Node},Tail); +insert_node({Load,Node},[{L,N}|Tail]) when Load =< L -> + %% Move forward in the list + T = lists:keydelete(Node,2,[{L,N}|Tail]), + [{Load,Node} | T]; +insert_node(Ln,[H|T]) -> + [H | insert_node(Ln,T)]; +insert_node(X,[]) -> % Can't happen + error_logger:error_msg("Pool_master: Bad node list X=~w\n", [X]), + exit(crash). + +pure_insert({Load,Node},[]) -> + [{Load,Node}]; +pure_insert({Load,Node},[{L,N}|Tail]) when Load < L -> + [{Load,Node}, {L,N} | Tail]; +pure_insert(L,[H|T]) -> [H|pure_insert(L,T)]. + +%% Really should not measure the contributions from +%% the back ground processes here .... which we do :-( +%% We don't have to monitor the master, since we're slaves anyway + +statistic_collector() -> + statistic_collector(5). + +statistic_collector(0) -> exit(normal); +statistic_collector(I) -> + sleep(300), + case global:whereis_name(pool_master) of + undefined -> + statistic_collector(I-1); + M -> + stat_loop(M, 999999) + end. + +%% Do not tell the master about our load if it has not changed + +stat_loop(M, Old) -> + sleep(2000), + case statistics(run_queue) of + Old -> + stat_loop(M, Old); + NewLoad -> + M ! {node(), load, NewLoad}, %% async + stat_loop(M, NewLoad) + end. + +sleep(I) -> receive after I -> ok end. |