%% -*- erlang-indent-level: 2 -*- %%----------------------------------------------------------------------- %% %CopyrightBegin% %% %% Copyright Ericsson AB 2006-2011. 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% %% %%%------------------------------------------------------------------- %%% File : dialyzer_coordinator.erl %%% Authors : Stavros Aronis %%% %%% Description: %%% %%% The parallel version of Dialyzer's typesig analysis is spread over 4 modules %%% with the intention to both minimize the changes on the original code and use %%% a separate module for every kind of Erlang process that will be running. %%% %%% There are therefore 3 kinds of processes: %%% %%% - The original Dialyzer backend (in succ_typings module) %%% - The worker process for the typesig analysis (in typesig and %%% worker) %%% - A coordinator of the worker processes (in coordinator) %%% %%% Operation guidelines: %%% %%% - The backend requests from the coordinator to spawn a worker for each SCC %%% - The backend notifies the coordinator when all SCC have been spawned and %%% waits for the server to report that the PLT has been updated %%% - Each worker is responsible to notify all those who wait for it. %%% %%%------------------------------------------------------------------- -module(dialyzer_coordinator). %%% Exports for all possible uses of coordinator from main process -export([start/2, all_spawned/1]). %%% Exports for all possible workers -export([wait_activation/0, job_done/3]). %%% Export for the typesig, dataflow and warnings main process -export([scc_spawn/2]). %%% Export for the typesig and dataflow analysis main process -export([receive_not_fixpoint/0]). %%% Export for warning main process -export([receive_warnings/0]). %%% Exports for the typesig and dataflow analysis workers -export([request_activation/1, sccs_to_pids/1]). %%% Exports for the compilation main process -export([compiler_spawn/2, receive_compilation_data/0]). %%% Exports for the compilation workers -export([get_next_label/2, compilation_done/3]). -export_type([coordinator/0, mode/0]). -behaviour(gen_server). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). %%-------------------------------------------------------------------- -define(MAP, dialyzer_coordinator_map). -type coordinator() :: pid(). %%opaque -type scc() :: [mfa_or_funlbl()] | module(). -type mode() :: 'typesig' | 'dataflow' | 'compile' | 'warnings'. -type servers() :: dialyzer_succ_typings:servers() | dialyzer_analysis_callgraph:servers() | dialyzer_succ_typings:warning_servers(). -record(state, {parent :: pid(), mode :: mode(), spawn_count = 0 :: integer(), all_spawned = false :: boolean(), next_label :: integer(), result :: [mfa_or_funlbl()] | dialyzer_analysis_callgraph:result() | [dial_warning()], init_job_data :: servers(), tickets :: integer(), queue :: queue() }). -include("dialyzer.hrl"). %%-------------------------------------------------------------------- -spec start(mode(), servers()) -> coordinator(). start(Mode, Servers) -> {ok, Pid} = gen_server:start(?MODULE, {self(), Mode, Servers}, []), Pid. -spec scc_spawn(scc() | module(), coordinator()) -> ok. scc_spawn(SCC, Coordinator) -> cast({scc_spawn, SCC}, Coordinator). -spec sccs_to_pids([scc()]) -> {[dialyzer_worker:worker()], [scc()]}. sccs_to_pids(SCCs) -> lists:foldl(fun pid_partition/2, {[], []}, SCCs). pid_partition(SCC, {Pids, Unknown}) -> try ets:lookup_element(?MAP, SCC, 2) of Result -> {[Result|Pids], Unknown} catch _:_ -> {Pids, [SCC|Unknown]} end. -spec job_done(scc() | file:filename(), term(), coordinator()) -> ok. job_done(Job, Result, Coordinator) -> cast({done, Job, Result}, Coordinator). -spec compilation_done(file:filename(), dialyzer_analysis_callgraph:compilation_data(), coordinator()) -> ok. compilation_done(Filename, CompilationData, Coordinator) -> cast({done, Filename, CompilationData}, Coordinator). -spec all_spawned(coordinator()) -> ok. all_spawned(Coordinator) -> cast(all_spawned, Coordinator). send_done_to_parent(#state{mode = Mode, parent = Parent, result = Result, next_label = NextLabel}) -> Msg = case Mode of X when X =:= 'typesig'; X =:= 'dataflow' -> ets:delete(?MAP), {not_fixpoint, Result}; 'compile' -> {compilation_data, Result, NextLabel}; 'warnings' -> {warnings, Result} end, Parent ! Msg, ok. -spec receive_not_fixpoint() -> [mfa_or_funlbl()]. receive_not_fixpoint() -> receive {not_fixpoint, NotFixpoint} -> NotFixpoint end. -spec receive_compilation_data() -> {dialyzer_analysis_callgraph:result(), integer()}. receive_compilation_data() -> receive {compilation_data, CompilationData, NextLabel} -> {CompilationData, NextLabel} end. -spec receive_warnings() -> [dial_warning()]. receive_warnings() -> receive {warnings, Warnings} -> Warnings end. -spec compiler_spawn(file:filename(), coordinator()) -> ok. compiler_spawn(Filename, Coordinator) -> cast({compiler_spawn, Filename}, Coordinator). -spec get_next_label(integer(), coordinator()) -> integer(). get_next_label(EstimatedSize, Coordinator) -> call({get_next_label, EstimatedSize}, Coordinator). -spec request_activation(coordinator()) -> ok. request_activation(Coordinator) -> cast({request_activation, self()}, Coordinator). -spec wait_activation() -> ok. wait_activation() -> receive activate -> ok end. activate_pid(Pid) -> Pid ! activate. %%-------------------------------------------------------------------- -spec init({pid(), mode(), servers()}) -> {ok, #state{}}. init({Parent, Mode, InitJobData}) -> BaseTickets = erlang:system_info(logical_processors_available), Tickets = case Mode of 'compile' -> 4*BaseTickets; 'warnings' -> 4*BaseTickets; _ -> non_regulated end, InitState = #state{parent = Parent, mode = Mode, init_job_data = InitJobData, tickets = Tickets, queue = queue:new()}, State = case Mode of X when X =:= 'typesig'; X =:= 'dataflow' -> ?MAP = ets:new(?MAP, [named_table, {read_concurrency, true}]), InitState#state{result = []}; 'warnings' -> InitState#state{result = []}; 'compile' -> InitResult = dialyzer_analysis_callgraph:compile_coordinator_init(), InitState#state{result = InitResult, next_label = 0} end, {ok, State}. -spec handle_call(Query::term(), From::term(), #state{}) -> {reply, Reply::term(), #state{}}. handle_call({get_next_label, EstimatedSize}, _From, #state{next_label = NextLabel} = State) -> {reply, NextLabel, State#state{next_label = NextLabel + EstimatedSize}}. -spec handle_cast(Msg::term(), #state{}) -> {noreply, #state{}} | {stop, normal, #state{}}. handle_cast({done, _Job, NewData}, #state{mode = Mode, result = OldResult, init_job_data = Servers} = State) when Mode =:= 'typesig'; Mode =:= 'dataflow' -> FinalData = dialyzer_succ_typings:lookup_names(NewData, Servers), UpdatedState = State#state{result = FinalData ++ OldResult}, reduce_or_stop(UpdatedState); handle_cast({done, Job, NewData}, #state{mode = Mode, result = OldResult, tickets = Tickets, queue = Queue} = State) when Mode =:= 'compile'; Mode =:= 'warnings' -> {Waiting, NewQueue} = queue:out(Queue), NewTickets = case Waiting of empty -> Tickets+1; {value, Pid} -> activate_pid(Pid), Tickets end, NewResult = case Mode of 'compile' -> dialyzer_analysis_callgraph:add_to_result(Job, NewData, OldResult); 'warnings' -> NewData ++ OldResult end, UpdatedState = State#state{result = NewResult, tickets = NewTickets, queue = NewQueue}, reduce_or_stop(UpdatedState); handle_cast(all_spawned, #state{spawn_count = SpawnCount} = State) -> case SpawnCount of 0 -> send_done_to_parent(State), {stop, normal, State}; _ -> NewState = State#state{all_spawned = true}, {noreply, NewState} end; handle_cast({request_activation, Pid}, #state{tickets = Tickets, queue = Queue} = State) -> {NewTickets, NewQueue} = case Tickets of 0 -> {Tickets, queue:in(Pid, Queue)}; N -> activate_pid(Pid), {N-1, Queue} end, {noreply, State#state{tickets = NewTickets, queue = NewQueue}}; handle_cast({scc_spawn, SCC}, #state{mode = Mode, init_job_data = Servers, spawn_count = SpawnCount} = State) when Mode =:= 'typesig'; Mode =:= 'dataflow' -> Pid = dialyzer_worker:launch(Mode, SCC, Servers, self()), true = ets:insert(?MAP, {SCC, Pid}), {noreply, State#state{spawn_count = SpawnCount + 1}}; handle_cast({scc_spawn, SCC}, #state{mode = 'warnings', init_job_data = Servers, spawn_count = SpawnCount, tickets = Tickets, queue = Queue} = State) -> Pid = dialyzer_worker:launch('warnings', SCC, Servers, self()), {NewTickets, NewQueue} = case Tickets of 0 -> {Tickets, queue:in(Pid, Queue)}; N -> activate_pid(Pid), {N-1, Queue} end, {noreply, State#state{spawn_count = SpawnCount + 1, tickets = NewTickets, queue = NewQueue}}; handle_cast({compiler_spawn, Filename}, #state{mode = Mode, init_job_data = Servers, spawn_count = SpawnCount, tickets = Tickets, queue = Queue } = State) -> Pid = dialyzer_worker:launch(Mode, Filename, Servers, self()), NewSpawnCount = SpawnCount + 1, {NewTickets, NewQueue} = case Tickets of 0 -> {Tickets, queue:in(Pid, Queue)}; N -> activate_pid(Pid), {N-1, Queue} end, {noreply, State#state{spawn_count = NewSpawnCount, tickets = NewTickets, queue = NewQueue}}. -spec handle_info(term(), #state{}) -> {noreply, #state{}}. handle_info(_Info, State) -> {noreply, State}. -spec terminate(term(), #state{}) -> ok. terminate(_Reason, _State) -> ok. -spec code_change(term(), #state{}, term()) -> {ok, #state{}}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%-------------------------------------------------------------------- cast(Message, Coordinator) -> gen_server:cast(Coordinator, Message). call(Message, Coordinator) -> gen_server:call(Coordinator, Message, infinity). reduce_or_stop(#state{all_spawned = AllSpawned, spawn_count = SpawnCount} = State) -> Action = case AllSpawned of false -> reduce; true -> case SpawnCount of 1 -> finish; _ -> reduce end end, case Action of reduce -> NewState = State#state{spawn_count = SpawnCount - 1}, {noreply, NewState}; finish -> send_done_to_parent(State), {stop, normal, State} end.