aboutsummaryrefslogtreecommitdiffstats
path: root/lib/dialyzer/src/dialyzer_coordinator.erl
diff options
context:
space:
mode:
Diffstat (limited to 'lib/dialyzer/src/dialyzer_coordinator.erl')
-rw-r--r--lib/dialyzer/src/dialyzer_coordinator.erl250
1 files changed, 250 insertions, 0 deletions
diff --git a/lib/dialyzer/src/dialyzer_coordinator.erl b/lib/dialyzer/src/dialyzer_coordinator.erl
new file mode 100644
index 0000000000..87cbd25b30
--- /dev/null
+++ b/lib/dialyzer/src/dialyzer_coordinator.erl
@@ -0,0 +1,250 @@
+%% -*- erlang-indent-level: 2 -*-
+%%-----------------------------------------------------------------------
+%% %CopyrightBegin%
+%%
+%% Copyright Ericsson AB 2006-2016. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%% http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%
+%% %CopyrightEnd%
+%%
+
+%%%-------------------------------------------------------------------
+%%% File : dialyzer_coordinator.erl
+%%% Authors : Stavros Aronis <[email protected]>
+%%%-------------------------------------------------------------------
+
+-module(dialyzer_coordinator).
+
+%%% Export for dialyzer main process
+-export([parallel_job/4]).
+
+%%% Export for all possible workers
+-export([job_done/3]).
+
+%%% Exports for the typesig and dataflow analysis workers
+-export([sccs_to_pids/2, request_activation/1]).
+
+%%% Exports for the compilation workers
+-export([get_next_label/2]).
+
+-export_type([coordinator/0, mode/0, init_data/0, result/0, job/0]).
+
+%%--------------------------------------------------------------------
+
+-type collector() :: pid().
+-type regulator() :: pid().
+-type scc_to_pid() :: ets:tid() | 'unused'.
+
+-opaque coordinator() :: {collector(), regulator(), scc_to_pid()}.
+-type timing() :: dialyzer_timing:timing_server().
+
+-type scc() :: [mfa_or_funlbl()].
+-type mode() :: 'typesig' | 'dataflow' | 'compile' | 'warnings'.
+
+-type compile_job() :: file:filename().
+-type typesig_job() :: scc().
+-type dataflow_job() :: module().
+-type warnings_job() :: module().
+
+-type job() :: compile_job() | typesig_job() | dataflow_job() | warnings_job().
+
+-type compile_init_data() :: dialyzer_analysis_callgraph:compile_init_data().
+-type typesig_init_data() :: dialyzer_succ_typings:typesig_init_data().
+-type dataflow_init_data() :: dialyzer_succ_typings:dataflow_init_data().
+-type warnings_init_data() :: dialyzer_succ_typings:warnings_init_data().
+
+-type compile_result() :: dialyzer_analysis_callgraph:compile_result().
+-type typesig_result() :: [mfa_or_funlbl()].
+-type dataflow_result() :: [mfa_or_funlbl()].
+-type warnings_result() :: [dial_warning()].
+
+-type init_data() :: compile_init_data() | typesig_init_data() |
+ dataflow_init_data() | warnings_init_data().
+
+-type result() :: compile_result() | typesig_result() |
+ dataflow_result() | warnings_result().
+
+-type job_result() :: dialyzer_analysis_callgraph:one_file_mid_error() |
+ dialyzer_analysis_callgraph:one_file_result_ok() |
+ typesig_result() | dataflow_result() | warnings_result().
+
+-record(state, {mode :: mode(),
+ active = 0 :: integer(),
+ result :: result(),
+ next_label = 0 :: integer(),
+ init_data :: init_data(),
+ regulator :: regulator(),
+ scc_to_pid :: scc_to_pid()
+ }).
+
+-include("dialyzer.hrl").
+
+%%--------------------------------------------------------------------
+
+-spec parallel_job('compile', [compile_job()], compile_init_data(), timing()) ->
+ {compile_result(), integer()};
+ ('typesig', [typesig_job()], typesig_init_data(), timing()) ->
+ typesig_result();
+ ('dataflow', [dataflow_job()], dataflow_init_data(),
+ timing()) -> dataflow_result();
+ ('warnings', [warnings_job()], warnings_init_data(),
+ timing()) -> warnings_result().
+
+parallel_job(Mode, Jobs, InitData, Timing) ->
+ State = spawn_jobs(Mode, Jobs, InitData, Timing),
+ collect_result(State).
+
+spawn_jobs(Mode, Jobs, InitData, Timing) ->
+ Collector = self(),
+ Regulator = spawn_regulator(),
+ TypesigOrDataflow = (Mode =:= 'typesig') orelse (Mode =:= 'dataflow'),
+ SCCtoPID =
+ case TypesigOrDataflow of
+ true -> ets:new(scc_to_pid, [{read_concurrency, true}]);
+ false -> unused
+ end,
+ Coordinator = {Collector, Regulator, SCCtoPID},
+ Fold =
+ fun(Job, Count) ->
+ Pid = dialyzer_worker:launch(Mode, Job, InitData, Coordinator),
+ case TypesigOrDataflow of
+ true -> true = ets:insert(SCCtoPID, {Job, Pid}), ok;
+ false -> ok
+ end,
+ Count + 1
+ end,
+ JobCount = lists:foldl(Fold, 0, Jobs),
+ Unit =
+ case Mode of
+ 'typesig' -> "SCCs";
+ _ -> "modules"
+ end,
+ dialyzer_timing:send_size_info(Timing, JobCount, Unit),
+ InitResult =
+ case Mode of
+ 'compile' -> dialyzer_analysis_callgraph:compile_init_result();
+ _ -> []
+ end,
+ #state{mode = Mode, active = JobCount, result = InitResult, next_label = 0,
+ init_data = InitData, regulator = Regulator, scc_to_pid = SCCtoPID}.
+
+collect_result(#state{mode = Mode, active = Active, result = Result,
+ next_label = NextLabel, init_data = InitData,
+ regulator = Regulator, scc_to_pid = SCCtoPID} = State) ->
+ receive
+ {next_label_request, Estimation, Pid} ->
+ Pid ! {next_label_reply, NextLabel},
+ collect_result(State#state{next_label = NextLabel + Estimation});
+ {done, Job, Data} ->
+ NewResult = update_result(Mode, InitData, Job, Data, Result),
+ case Active of
+ 1 ->
+ kill_regulator(Regulator),
+ case Mode of
+ 'compile' ->
+ {NewResult, NextLabel};
+ X when X =:= 'typesig'; X =:= 'dataflow' ->
+ ets:delete(SCCtoPID),
+ NewResult;
+ 'warnings' ->
+ NewResult
+ end;
+ N ->
+ collect_result(State#state{result = NewResult, active = N - 1})
+ end
+ end.
+
+update_result(Mode, InitData, Job, Data, Result) ->
+ case Mode of
+ 'compile' ->
+ dialyzer_analysis_callgraph:add_to_result(Job, Data, Result,
+ InitData);
+ X when X =:= 'typesig'; X =:= 'dataflow' ->
+ dialyzer_succ_typings:lookup_names(Data, InitData) ++ Result;
+ 'warnings' ->
+ Data ++ Result
+ end.
+
+-spec sccs_to_pids([scc() | module()], coordinator()) ->
+ {[dialyzer_worker:worker()], [scc() | module()]}.
+
+sccs_to_pids(SCCs, {_Collector, _Regulator, SCCtoPID}) ->
+ Fold =
+ fun(SCC, {Pids, Unknown}) ->
+ try ets:lookup_element(SCCtoPID, SCC, 2) of
+ Result -> {[Result|Pids], Unknown}
+ catch
+ _:_ -> {Pids, [SCC|Unknown]}
+ end
+ end,
+ lists:foldl(Fold, {[], []}, SCCs).
+
+-spec job_done(job(), job_result(), coordinator()) -> ok.
+
+job_done(Job, Result, {Collector, Regulator, _SCCtoPID}) ->
+ Regulator ! done,
+ Collector ! {done, Job, Result},
+ ok.
+
+-spec get_next_label(integer(), coordinator()) -> integer().
+
+get_next_label(EstimatedSize, {Collector, _Regulator, _SCCtoPID}) ->
+ Collector ! {next_label_request, EstimatedSize, self()},
+ receive
+ {next_label_reply, NextLabel} -> NextLabel
+ end.
+
+-spec wait_activation() -> ok.
+
+wait_activation() ->
+ receive activate -> ok end.
+
+activate_pid(Pid) ->
+ Pid ! activate.
+
+-spec request_activation(coordinator()) -> ok.
+
+request_activation({_Collector, Regulator, _SCCtoPID}) ->
+ Regulator ! {req, self()},
+ wait_activation().
+
+spawn_regulator() ->
+ InitTickets = dialyzer_utils:parallelism(),
+ spawn_link(fun() -> regulator_loop(InitTickets, queue:new()) end).
+
+regulator_loop(Tickets, Queue) ->
+ receive
+ {req, Pid} ->
+ case Tickets of
+ 0 ->
+ regulator_loop(0, queue:in(Pid, Queue));
+ N ->
+ activate_pid(Pid),
+ regulator_loop(N-1, Queue)
+ end;
+ done ->
+ {Waiting, NewQueue} = queue:out(Queue),
+ NewTickets =
+ case Waiting of
+ empty -> Tickets + 1;
+ {value, Pid} ->
+ activate_pid(Pid),
+ Tickets
+ end,
+ regulator_loop(NewTickets, NewQueue);
+ stop -> ok
+ end.
+
+kill_regulator(Regulator) ->
+ Regulator ! stop.
generated by cgit v1.2.3 (git 2.25.1) at 2024-10-31 05:34:37 +0000