%% orbit-int worker (computing vertices and holding part of hash table) %% %% Author: Patrick Maier %% -module(worker). -export([init/3, distribute_vertices/3, send_image/4, verts_recvd_from_stat/1, credit_retd_from_stat/1, min_atomic_credit_from_stat/1, init_idle_from_stat/1, tail_idle_from_stat/1, max_idle_from_stat/1]). %% DATA (see module master) %% MESSAGES (see module master) %% counters/timers record -record(ct, {verts_recvd = 0, %% #vertices received by this server so far credit_retd = 0, %% #times server has returned credit to master min_atomic_credit = 0, %% minimal atomic credit received so far last_event = master:now(), %% time stamp [ms] of most recent event init_idle = -1, %% idle time [ms] between init recv first vertex tail_idle = -1, %% idle time [ms] between send last vertex and dump max_idle = -1}). %% max idle [ms] time between vertices %% initialise worker init(LocalTableSize, IdleTimeout, SpawnImgComp) -> Table = table:new(LocalTableSize), receive {init, StaticMachConf0} -> StatData = #ct{}, StaticMachConf1 = master:set_idle_timeout(StaticMachConf0, IdleTimeout), StaticMachConf = case SpawnImgComp of true -> StaticMachConf1; _Else -> master:clear_spawn_img_comp(StaticMachConf1) end, Credit = credit:zero(), vertex_server(StaticMachConf, Credit, Table, StatData) end. %% main worker loop: server handling vertex messages; %% StaticMachConf -- info about machine configuration %% Credit -- credit currently held by the server, %% Table -- hash table holding vertices %% StatData -- various counters and timers for gathering statistics vertex_server(StaticMachConf, Credit, Table, StatData) -> IdleTimeout = master:get_idle_timeout(StaticMachConf), receive {vertex, X, Slot, K} -> Credit_plus_K = credit:credit_atomic(K, Credit), Now = master:now(), {NewCredit, NewTable} = handle_vertex(StaticMachConf, X, Slot, Credit_plus_K, Table), VertsRecvd = StatData#ct.verts_recvd, MinAtomicCredit = StatData#ct.min_atomic_credit, LastEvent = StatData#ct.last_event, InitIdle = StatData#ct.init_idle, MaxIdle = StatData#ct.max_idle, NewStatData0 = StatData#ct{verts_recvd = VertsRecvd + 1, min_atomic_credit = max(MinAtomicCredit, K)}, NewStatData1 = if InitIdle < 0 -> NewStatData0#ct{init_idle = Now - LastEvent}; true -> NewStatData0#ct{max_idle = max(MaxIdle, Now - LastEvent)} end, NewStatData = NewStatData1#ct{last_event = master:now()}, vertex_server(StaticMachConf, NewCredit, NewTable, NewStatData); {dump} -> Now = master:now(), LastEvent = StatData#ct.last_event, NewStatData = StatData#ct{tail_idle = Now - LastEvent, last_event = master:now()}, dump_table(StaticMachConf, Table, NewStatData) after IdleTimeout -> CreditRetd = StatData#ct.credit_retd, NewCreditRetd = return_credit(StaticMachConf, Credit, CreditRetd), NewStatData = StatData#ct{credit_retd = NewCreditRetd}, vertex_server(StaticMachConf, credit:zero(), Table, NewStatData) end. %% handle_vertex checks whether vertex X is stored in Slot of Table; %% if not, it is in inserted there and the images of the generators %% are distributed among the workers. %% Precondition: Credit is non-zero. handle_vertex(StaticMachConf, X, Slot, Credit, Table) -> % check whether X is already in Table case table:is_member(X, Slot, Table) of true -> {Credit, Table}; % X already in table; do nothing _Else -> % X not in table % insert X at Slot NewTable = table:insert(X, Slot, Table), % distribute images of X under generators to their respective workers NewCredit = distribute_images(StaticMachConf, X, Credit), % return remaining credit and updated table {NewCredit, NewTable} end. %% return_credit sends non-zero Credit back to the master; %% returns number of times credit has been returned so far return_credit(StaticMachConf, Credit, CreditRetd) -> case credit:is_zero(Credit) of true -> CreditRetd; false -> MasterPid = master:get_master(StaticMachConf), MasterPid ! {done, Credit}, CreditRetd + 1 end. %% dump_table sends a list containing the local partial orbit to the master, %% together with some statistics on the distribution of vertices in the table. dump_table(StaticMachConf, Table, StatData) -> MasterPid = master:get_master(StaticMachConf), Stat = worker_stats(node(self()), table:get_freq(Table), StatData), PartialOrbit = table:to_list(Table), MasterPid ! {result, PartialOrbit, Stat}. %% distribute_images distributes the images of vertex X under the generators %% to the workers determined by the hash; some ore all of of the Credit is %% used to send the messages, the remaining credit is returned; %% computation and sending of vertices is actually done asynchronously. %% Precondition: Credit is non-zero. distribute_images(StaticMachConf, X, Credit) -> Gs = master:get_gens(StaticMachConf), do_distribute_images(StaticMachConf, X, Credit, Gs). do_distribute_images(_StaticMachConf, _X, Credit, []) -> Credit; do_distribute_images(StaticMachConf, X, Credit, [G]) -> {K, RemainingCredit} = credit:debit_atomic(Credit), case master:get_spawn_img_comp(StaticMachConf) of true -> spawn(worker, send_image, [StaticMachConf, X, G, K]), ok; _Else -> send_image(StaticMachConf, X, G, K) end, RemainingCredit; do_distribute_images(StaticMachConf, X, Credit, [G|Gs]) -> {K, NonZeroRemainingCredit} = credit:debit_atomic_nz(Credit), case master:get_spawn_img_comp(StaticMachConf) of true -> spawn(worker, send_image, [StaticMachConf, X, G, K]), ok; _Else -> send_image(StaticMachConf, X, G, K) end, do_distribute_images(StaticMachConf, X, NonZeroRemainingCredit, Gs). %% distribute_vertices distributes the list of vertices Xs to the workers %% determined by the hash; some ore all of of the Credit is used to send %% the messages, the remaining credit is returned. %% Precondition: If Xs is non-empty then Credit must be non-zero. distribute_vertices(_StaticMachConf, Credit, []) -> Credit; distribute_vertices(StaticMachConf, Credit, [X]) -> {K, RemainingCredit} = credit:debit_atomic(Credit), send_vertex(StaticMachConf, X, K), RemainingCredit; distribute_vertices(StaticMachConf, Credit, [X|Xs]) -> {K, NonZeroRemainingCredit} = credit:debit_atomic_nz(Credit), send_vertex(StaticMachConf, X, K), distribute_vertices(StaticMachConf, NonZeroRemainingCredit, Xs). %% send_image sends image of X under G to the worker determined by %% the hash of G(X); the message is tagged with atomic credit K. send_image(StaticMachConf, X, G, K) -> Y = G(X), send_vertex(StaticMachConf, Y, K). %% send_vertex hashes vertex X and sends it to the worker determined by %% the hash; the message is tagged with atomic credit K. send_vertex(StaticMachConf, X, K) -> {Pid, Slot} = hash_vertex(StaticMachConf, X), Pid ! {vertex, X, Slot, K}, ok. %% hash_vertex computes the two-dimensional hash table slot of vertex X where %% the first dim is a worker pid and the second a slot in that worker's table. hash_vertex(StaticMachConf, X) -> % get static info GlobalTableSize = master:get_global_table_size(StaticMachConf), Workers = master:get_workers(StaticMachConf), % compute raw hash and slot in global table Hash = erlang:phash2(X), GlobalSlot = Hash rem GlobalTableSize, % translate global slot into worker pid and local slot global_to_local_slot(Workers, GlobalSlot). %% global_to_local_slot traverses the list Workers sequentially to translate %% slot GlobSlot in the global hash table into a two-dimensional local slot %% {pid, slot}, where 'pid' is the PID of a worker and 'slot' a the slot %% in that worker's local hash table. %% Precondition: GlobSlot < sum of TableSize in Workers. %% Note: This procedure is horribly inefficient (linear in size of Workers); %% it should be log (size of Workers) at most. global_to_local_slot([{Pid, _, TabSize} | Workers], GlobSlot) -> if GlobSlot < TabSize -> {Pid, GlobSlot}; true -> global_to_local_slot(Workers, GlobSlot - TabSize) end. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% auxiliary functions %% produce readable statistics worker_stats(Node, Frequency, StatData) -> [{node, Node}, {vertices_recvd, StatData#ct.verts_recvd}, {credit_retd, StatData#ct.credit_retd}, {min_atomic_credit, StatData#ct.min_atomic_credit}, {init_idle_time, StatData#ct.init_idle}, {max_idle_time, StatData#ct.max_idle}, {tail_idle_time, StatData#ct.tail_idle} | table:freq_to_stat(Frequency)]. verts_recvd_from_stat(Stat) -> case lists:keyfind(vertices_recvd, 1, Stat) of {_, VertsRecvd} -> VertsRecvd; _Else -> false end. credit_retd_from_stat(Stat) -> case lists:keyfind(credit_retd, 1, Stat) of {_, CreditRetd} -> CreditRetd; _Else -> false end. min_atomic_credit_from_stat(Stat) -> case lists:keyfind(min_atomic_credit, 1, Stat) of {_, MinAtomicCredit} -> MinAtomicCredit; _Else -> false end. init_idle_from_stat(Stat) -> case lists:keyfind(init_idle_time, 1, Stat) of {_, InitIdle} -> InitIdle; _Else -> false end. tail_idle_from_stat(Stat) -> case lists:keyfind(tail_idle_time, 1, Stat) of {_, TailIdle} -> TailIdle; _Else -> false end. max_idle_from_stat(Stat) -> case lists:keyfind(max_idle_time, 1, Stat) of {_, MaxIdle} -> MaxIdle; _Else -> false end.