dialyzer: Do not spawn all workers at once

Spawning all worker processes at once has the potential to increase
peak memory consumption. Therefore the implementation is now slightly
modified: `20 * dialyzer_utils:parallelism()' processes are running in
parallel. 20 i quite a big factor, but seems necessary to keep all
schedulers busy according to the Observer application's Load Charts,
with a 100 ms update interval.

4 files changed, 67 insertions, 50 deletions

diff --git a/lib/dialyzer/src/dialyzer_callgraph.erl b/lib/dialyzer/src/dialyzer_callgraph.erl
index 27249440e9..bb02bc8c0d 100644
--- a/lib/dialyzer/src/dialyzer_callgraph.erl
+++ b/lib/dialyzer/src/dialyzer_callgraph.erl
@@ -40,7 +40,7 @@
 	 module_postorder_from_funs/2,
 	 new/0,
 	 get_depends_on/2,
-	 get_required_by/2,
+	 %% get_required_by/2,
 	 in_neighbours/2,
 	 renew_race_info/4,
 	 renew_race_code/2,
@@ -250,12 +250,12 @@ get_depends_on(SCC, #callgraph{active_digraph = {'e', Out, _In, Maps}}) ->
 get_depends_on(SCC, #callgraph{active_digraph = {'d', DG}}) ->
   digraph:out_neighbours(DG, SCC).
 
--spec get_required_by(scc() | module(), callgraph()) -> [scc()].
+%% -spec get_required_by(scc() | module(), callgraph()) -> [scc()].
 
-get_required_by(SCC, #callgraph{active_digraph = {'e', _Out, In, Maps}}) ->
-  lookup_scc(SCC, In, Maps);
-get_required_by(SCC, #callgraph{active_digraph = {'d', DG}}) ->
-  digraph:in_neighbours(DG, SCC).
+%% get_required_by(SCC, #callgraph{active_digraph = {'e', _Out, In, Maps}}) ->
+%%   lookup_scc(SCC, In, Maps);
+%% get_required_by(SCC, #callgraph{active_digraph = {'d', DG}}) ->
+%%   digraph:in_neighbours(DG, SCC).
 
 lookup_scc(SCC, Table, Maps) ->
   case ets_lookup_dict({'scc', SCC}, Maps) of
diff --git a/lib/dialyzer/src/dialyzer_coordinator.erl b/lib/dialyzer/src/dialyzer_coordinator.erl
index de239f5c93..7c1bc1de5a 100644
--- a/lib/dialyzer/src/dialyzer_coordinator.erl
+++ b/lib/dialyzer/src/dialyzer_coordinator.erl
@@ -76,6 +76,8 @@
 		active     = 0 :: integer(),
 		result         :: result(),
 		next_label = 0 :: integer(),
+                jobs           :: [job()],
+                job_fun        :: fun(),
 		init_data      :: init_data(),
 		regulator      :: regulator(),
 		scc_to_pid     :: scc_to_pid()
@@ -108,16 +110,18 @@ spawn_jobs(Mode, Jobs, InitData, Timing) ->
       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
+  JobFun =
+    fun(Job) ->
+        Pid = dialyzer_worker:launch(Mode, Job, InitData, Coordinator),
+        case TypesigOrDataflow of
+          true  -> true = ets:insert(SCCtoPID, {Job, Pid});
+          false -> true
+        end
     end,
-  JobCount = lists:foldl(Fold, 0, Jobs),
+  JobCount = length(Jobs),
+  NumberOfInitJobs = min(JobCount, 20 * dialyzer_utils:parallelism()),
+  {InitJobs, RestJobs} = lists:split(NumberOfInitJobs, Jobs),
+  lists:foreach(JobFun, InitJobs),
   Unit =
     case Mode of
       'typesig'  -> "SCCs";
@@ -129,11 +133,13 @@ spawn_jobs(Mode, Jobs, InitData, Timing) ->
       '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}.
+  #state{mode = Mode, active = JobCount, result = InitResult,
+         next_label = 0, job_fun = JobFun, jobs = RestJobs,
+         init_data = InitData, regulator = Regulator, scc_to_pid = SCCtoPID}.
 
 collect_result(#state{mode = Mode, active = Active, result = Result,
 		      next_label = NextLabel, init_data = InitData,
+                      jobs = JobsLeft, job_fun = JobFun,
 		      regulator = Regulator, scc_to_pid = SCCtoPID} = State) ->
   receive
     {next_label_request, Estimation, Pid} ->
@@ -141,20 +147,35 @@ collect_result(#state{mode = Mode, active = Active, result = Result,
       collect_result(State#state{next_label = NextLabel + Estimation});
     {done, Job, Data} ->
       NewResult = update_result(Mode, InitData, Job, Data, Result),
+      TypesigOrDataflow = (Mode =:= 'typesig') orelse (Mode =:= 'dataflow'),
       case Active of
 	1 ->
 	  kill_regulator(Regulator),
 	  case Mode of
 	    'compile' ->
 	      {NewResult, NextLabel};
-	    X when X =:= 'typesig'; X =:= 'dataflow' ->
+	    _ when TypesigOrDataflow ->
 	      ets:delete(SCCtoPID),
 	      NewResult;
 	    'warnings' ->
 	      NewResult
 	  end;
 	N ->
-	  collect_result(State#state{result = NewResult, active = N - 1})
+          case TypesigOrDataflow of
+            true -> true = ets:delete(SCCtoPID, Job);
+            false -> true
+          end,
+          NewJobsLeft =
+            case JobsLeft of
+              [] -> [];
+              [NewJob|JobsLeft1] ->
+                JobFun(NewJob),
+                JobsLeft1
+            end,
+          NewState = State#state{result = NewResult,
+                                 jobs = NewJobsLeft,
+                                 active = N - 1},
+          collect_result(NewState)
       end
   end.
 
@@ -176,8 +197,12 @@ sccs_to_pids(SCCs, {_Collector, _Regulator, SCCtoPID}) ->
   Fold =
     fun(SCC, Pids) ->
         %% The SCCs that SCC depends on have always been started.
-	Result = ets:lookup_element(SCCtoPID, SCC, 2),
-        [Result|Pids]
+        try ets:lookup_element(SCCtoPID, SCC, 2) of
+          Pid when is_pid(Pid) ->
+            [Pid|Pids]
+        catch
+          _:_ -> Pids
+        end
     end,
   lists:foldl(Fold, [], SCCs).
 
diff --git a/lib/dialyzer/src/dialyzer_succ_typings.erl b/lib/dialyzer/src/dialyzer_succ_typings.erl
index 3c90f46e95..be685baf22 100644
--- a/lib/dialyzer/src/dialyzer_succ_typings.erl
+++ b/lib/dialyzer/src/dialyzer_succ_typings.erl
@@ -29,7 +29,7 @@
 -export([
 	 find_succ_types_for_scc/2,
 	 refine_one_module/2,
-	 find_required_by/2,
+	 %% find_required_by/2,
 	 find_depends_on/2,
 	 collect_warnings/2,
 	 lookup_names/2
@@ -236,10 +236,10 @@ refine_succ_typings(Modules, #st{codeserver = Codeserver,
 find_depends_on(SCC, {_Codeserver, Callgraph, _Plt, _Solvers}) ->
   dialyzer_callgraph:get_depends_on(SCC, Callgraph).
 
--spec find_required_by(scc() | module(), fixpoint_init_data()) -> [scc()].
+%% -spec find_required_by(scc() | module(), fixpoint_init_data()) -> [scc()].
 
-find_required_by(SCC, {_Codeserver, Callgraph, _Plt, _Solvers}) ->
-  dialyzer_callgraph:get_required_by(SCC, Callgraph).
+%% find_required_by(SCC, {_Codeserver, Callgraph, _Plt, _Solvers}) ->
+%%   dialyzer_callgraph:get_required_by(SCC, Callgraph).
 
 -spec lookup_names([label()], fixpoint_init_data()) -> [mfa_or_funlbl()].
 
diff --git a/lib/dialyzer/src/dialyzer_worker.erl b/lib/dialyzer/src/dialyzer_worker.erl
index 1a251b5307..af0f2e9e08 100644
--- a/lib/dialyzer/src/dialyzer_worker.erl
+++ b/lib/dialyzer/src/dialyzer_worker.erl
@@ -56,10 +56,14 @@ launch(Mode, Job, InitData, Coordinator) ->
 
 %%--------------------------------------------------------------------
 
-init(#state{job = SCC, mode = Mode, init_data = InitData} = State) when
+init(#state{job = SCC, mode = Mode, init_data = InitData,
+            coordinator = Coordinator} = State) when
     Mode =:= 'typesig'; Mode =:= 'dataflow' ->
-  DependsOn = dialyzer_succ_typings:find_depends_on(SCC, InitData),
-  ?debug("Deps ~p: ~p\n",[SCC, DependsOn]),
+  DependsOnSCCs = dialyzer_succ_typings:find_depends_on(SCC, InitData),
+  ?debug("~w: Deps ~p: ~p\n", [self(), SCC, DependsOnSCCs]),
+  Pids = dialyzer_coordinator:sccs_to_pids(DependsOnSCCs, Coordinator),
+  ?debug("~w: PidsDeps ~p\n", [self(), Pids]),
+  DependsOn = [{Pid, erlang:monitor(process, Pid)} || Pid <- Pids],
   loop(updating, State#state{depends_on = DependsOn});
 init(#state{mode = Mode} = State) when
     Mode =:= 'compile'; Mode =:= 'warnings' ->
@@ -67,7 +71,7 @@ init(#state{mode = Mode} = State) when
 
 loop(updating, #state{mode = Mode} = State) when
     Mode =:= 'typesig'; Mode =:= 'dataflow' ->
-  ?debug("Update: ~p\n",[State#state.job]),
+  ?debug("~w: Update: ~p\n", [self(), State#state.job]),
   NextStatus =
     case waits_more_success_typings(State) of
       true -> waiting;
@@ -76,11 +80,11 @@ loop(updating, #state{mode = Mode} = State) when
   loop(NextStatus, State);
 loop(waiting, #state{mode = Mode} = State) when
     Mode =:= 'typesig'; Mode =:= 'dataflow' ->
-  ?debug("Wait: ~p\n",[State#state.job]),
+  ?debug("~w: Wait: ~p\n", [self(), State#state.job]),
   NewState = wait_for_success_typings(State),
   loop(updating, NewState);
 loop(running, #state{mode = 'compile'} = State) ->
-  dialyzer_coordinator:request_activation(State#state.coordinator),
+  request_activation(State),
   ?debug("Compile: ~s\n",[State#state.job]),
   Result =
     case start_compilation(State) of
@@ -92,40 +96,28 @@ loop(running, #state{mode = 'compile'} = State) ->
     end,
   report_to_coordinator(Result, State);
 loop(running, #state{mode = 'warnings'} = State) ->
-  dialyzer_coordinator:request_activation(State#state.coordinator),
+  request_activation(State),
   ?debug("Warning: ~s\n",[State#state.job]),
   Result = collect_warnings(State),
   report_to_coordinator(Result, State);
 loop(running, #state{mode = Mode} = State) when
     Mode =:= 'typesig'; Mode =:= 'dataflow' ->
   request_activation(State),
-  ?debug("Run: ~p\n",[State#state.job]),
+  ?debug("~w: Run: ~p\n", [self(), State#state.job]),
   NotFixpoint = do_work(State),
-  ok = broadcast_done(State),
   report_to_coordinator(NotFixpoint, State).
 
 waits_more_success_typings(#state{depends_on = Depends}) ->
   Depends =/= [].
 
-broadcast_done(#state{job = SCC, init_data = InitData,
-		      coordinator = Coordinator}) ->
-  RequiredBy = dialyzer_succ_typings:find_required_by(SCC, InitData),
-  Callers = dialyzer_coordinator:sccs_to_pids(RequiredBy, Coordinator),
-  send_done(Callers, SCC).
-
-send_done(Callers, SCC) ->
-  ?debug("Sending ~p: ~p\n",[SCC, Callers]),
-  SendSTFun = fun(PID) -> PID ! {done, SCC} end,
-  lists:foreach(SendSTFun, Callers).
-
 wait_for_success_typings(#state{depends_on = DependsOn} = State) ->
   receive
-    {done, SCC} ->
-      ?debug("GOT ~p: ~p\n",[State#state.job, SCC]),
-      State#state{depends_on = DependsOn -- [SCC]}
+    {'DOWN', Ref, process, Pid, _Info} ->
+      ?debug("~w: ~p got DOWN: ~p\n", [self(), State#state.job, Pid]),
+      State#state{depends_on = DependsOn -- [{Pid, Ref}]}
   after
     5000 ->
-      ?debug("Still Waiting ~p: ~p\n",[State#state.job, DependsOn]),
+      ?debug("~w: Still Waiting ~p:\n   ~p\n",  [self(), State#state.job, DependsOn]),
       State
   end.
 
@@ -139,7 +131,7 @@ do_work(#state{mode = Mode, job = Job, init_data = InitData}) ->
   end.
 
 report_to_coordinator(Result, #state{job = Job, coordinator = Coordinator}) ->
-  ?debug("Done: ~p\n",[Job]),
+  ?debug("~w: Done: ~p\n",[self(), Job]),
   dialyzer_coordinator:job_done(Job, Result, Coordinator).
 
 start_compilation(#state{job = Job, init_data = InitData}) ->