Parallel compilation of files under analysis

4 files changed, 263 insertions, 146 deletions

diff --git a/lib/dialyzer/src/dialyzer_analysis_callgraph.erl b/lib/dialyzer/src/dialyzer_analysis_callgraph.erl
index 7060028d17..e009b8f43f 100644
--- a/lib/dialyzer/src/dialyzer_analysis_callgraph.erl
+++ b/lib/dialyzer/src/dialyzer_analysis_callgraph.erl
@@ -30,6 +30,11 @@
 
 -export([start/3]).
 
+-export([compile_coordinator_init/0,
+	 add_to_result/3,
+	 start_compilation/2,
+	 continue_compilation/2]).
+
 -include("dialyzer.hrl").
 
 -record(analysis_state,
@@ -212,33 +217,15 @@ compile_and_store(Files, #analysis_state{codeserver = CServer,
   Includes = [{i, D} || D <- Dirs],
   Defines = [{d, Macro, Val} || {Macro, Val} <- Defs],
   Callgraph = dialyzer_callgraph:new(),
-  Fun = case StartFrom of
-	  src_code ->
-	    fun(File, {TmpCG, TmpCServer, TmpFailed, TmpNoWarn, TmpMods}) ->
-		case compile_src(File, Includes, Defines, TmpCG,
-				 TmpCServer, UseContracts) of
-		  {error, Reason} ->
-		    {TmpCG, TmpCServer, [{File, Reason}|TmpFailed], TmpNoWarn,
-                     TmpMods};
-		  {ok, NewCG, NoWarn, NewCServer, Mod} ->
-		    {NewCG, NewCServer, TmpFailed, NoWarn++TmpNoWarn,
-                     [Mod|TmpMods]}
-		end
-	    end;
-	  byte_code ->
-	    fun(File, {TmpCG, TmpCServer, TmpFailed, TmpNoWarn, TmpMods}) ->
-		case compile_byte(File, TmpCG, TmpCServer, UseContracts) of
-		  {error, Reason} ->
-		    {TmpCG, TmpCServer, [{File, Reason}|TmpFailed], TmpNoWarn,
-                     TmpMods};
-		  {ok, NewCG, NoWarn, NewCServer, Mod} ->
-		    {NewCG, NewCServer, TmpFailed, NoWarn++TmpNoWarn,
-                     [Mod|TmpMods]}
-		end
-	    end
-	end,
-  {NewCallgraph1, NewCServer, Failed, NoWarn, Modules} =
-    lists:foldl(Fun, {Callgraph, CServer, [], [], []}, Files),
+  Servers = {Callgraph, CServer, StartFrom, Includes, Defines, UseContracts},
+  Coordinator = dialyzer_coordinator:start(compile, Servers),
+  Spawner = fun(F) -> dialyzer_coordinator:compiler_spawn(F, Coordinator) end,
+  lists:foreach(Spawner, Files),
+  dialyzer_coordinator:all_spawned(Coordinator),
+  {{V, E, Failed, NoWarn, Modules}, NextLabel} =
+    dialyzer_coordinator:receive_compilation_data(),
+  CServer2 = dialyzer_codeserver:set_next_core_label(NextLabel, CServer),
+  Callgraph = dialyzer_callgraph:add_edges(E, V, Callgraph),
   case Failed =:= [] of
     true ->
       NewFiles = lists:zip(lists:reverse(Modules), Files),
@@ -254,11 +241,41 @@ compile_and_store(Files, #analysis_state{codeserver = CServer,
   {T2, _} = statistics(runtime),
   Msg1 = io_lib:format("done in ~.2f secs\nRemoving edges... ", [(T2-T1)/1000]),
   send_log(Parent, Msg1),
-  NewCallgraph2 = cleanup_callgraph(State, NewCServer, NewCallgraph1, Modules),
+  Callgraph = cleanup_callgraph(State, CServer2, Callgraph, Modules),
   {T3, _} = statistics(runtime),
   Msg2 = io_lib:format("done in ~.2f secs\n", [(T3-T2)/1000]),
   send_log(Parent, Msg2),
-  {NewCallgraph2, sets:from_list(NoWarn), NewCServer}.
+  {Callgraph, sets:from_list(NoWarn), CServer2}.
+
+-type servers()     :: term().
+-type result()      :: term().
+-type file_result() :: term().
+-type data()        :: term().
+
+-spec compile_coordinator_init() -> result().
+
+compile_coordinator_init() -> {[], [], [], [], []}.
+
+-spec add_to_result(file:filename(), file_result(), result()) -> result().
+
+add_to_result(File, NewData, {V, E, Failed, NoWarn, Mods}) ->
+  case NewData of
+    {error, Reason} ->
+      {[{File, Reason}|Failed], NoWarn, Mods};
+    {ok, NV, NE, NewNoWarn, Mod} ->
+      {NV ++ V, NE ++ E, Failed, NewNoWarn ++ NoWarn, [Mod|Mods]}
+  end.
+
+-spec start_compilation(file:filename(), servers()) -> data().
+
+start_compilation(File, {Callgraph, Codeserver, StartFrom,
+			 Includes, Defines, UseContracts}) ->
+  case StartFrom of
+    src_code ->
+      compile_src(File, Includes, Defines, Callgraph, Codeserver, UseContracts);
+    byte_code ->
+      compile_byte(File, Callgraph, Codeserver, UseContracts)
+  end.
 
 cleanup_callgraph(#analysis_state{plt = InitPlt, parent = Parent,
 				  codeserver = CodeServer
@@ -348,10 +365,17 @@ compile_common(File, AbstrCode, CompOpts, Callgraph, CServer, UseContracts) ->
 store_core(Mod, Core, NoWarn, Callgraph, CServer) ->
   Exp = get_exports_from_core(Core),
   ExpTypes = get_exported_types_from_core(Core),
-  CServer1 = dialyzer_codeserver:insert_exports(Exp, CServer),
-  CServer2 = dialyzer_codeserver:insert_temp_exported_types(ExpTypes, CServer1),
-  {LabeledCore, CServer3} = label_core(Core, CServer2),
-  store_code_and_build_callgraph(Mod, LabeledCore, Callgraph, CServer3, NoWarn).
+  CServer = dialyzer_codeserver:insert_exports(Exp, CServer),
+  CServer = dialyzer_codeserver:insert_temp_exported_types(ExpTypes, CServer),
+  CoreTree = cerl:from_records(Core),
+  {cerl_trees:size(CoreTree), {Mod, CoreTree, NoWarn, Callgraph, CServer}}.
+
+-spec continue_compilation(integer(), data()) -> {integer(), file_result()}.
+
+continue_compilation(NextLabel, {Mod, CoreTree, NoWarn, Callgraph, CServer}) ->
+  {LabeledTree, _NewNextLabel} = cerl_trees:label(CoreTree, NextLabel),
+  LabeledCore = cerl:to_records(LabeledTree),
+  store_code_and_build_callgraph(Mod, LabeledCore, Callgraph, NoWarn, CServer).
 
 abs_get_nowarn(Abs, M) ->
   Opts = lists:flatten([C || {attribute, _, compile, C} <- Abs]),
@@ -384,18 +408,11 @@ get_exports_from_core(Core) ->
   M = cerl:atom_val(cerl:module_name(Tree)),
   [{M, F, A} || {F, A} <- Exports2].
 
-label_core(Core, CServer) ->
-  NextLabel = dialyzer_codeserver:get_next_core_label(CServer),
-  CoreTree = cerl:from_records(Core),
-  {LabeledTree, NewNextLabel} = cerl_trees:label(CoreTree, NextLabel),
-  {cerl:to_records(LabeledTree),
-   dialyzer_codeserver:set_next_core_label(NewNextLabel, CServer)}.
-
-store_code_and_build_callgraph(Mod, Core, Callgraph, CServer, NoWarn) ->
+store_code_and_build_callgraph(Mod, Core, Callgraph, NoWarn, CServer) ->
   CoreTree = cerl:from_records(Core),
-  NewCallgraph = dialyzer_callgraph:scan_core_tree(CoreTree, Callgraph),
-  CServer2 = dialyzer_codeserver:insert(Mod, CoreTree, CServer),
-  {ok, NewCallgraph, NoWarn, CServer2, Mod}.
+  {Vertices, Edges} = dialyzer_callgraph:scan_core_tree(CoreTree, Callgraph),
+  CServer = dialyzer_codeserver:insert(Mod, CoreTree, CServer),
+  {ok, Vertices, Edges, NoWarn, Mod}.
 
 %%--------------------------------------------------------------------
 %% Utilities
diff --git a/lib/dialyzer/src/dialyzer_callgraph.erl b/lib/dialyzer/src/dialyzer_callgraph.erl
index 9eba64dd57..bf939655d0 100644
--- a/lib/dialyzer/src/dialyzer_callgraph.erl
+++ b/lib/dialyzer/src/dialyzer_callgraph.erl
@@ -28,6 +28,7 @@
 -module(dialyzer_callgraph).
 
 -export([add_edges/2,
+	 add_edges/3,
 	 all_nodes/1,
 	 delete/1,
 	 finalize/1,
@@ -358,10 +359,10 @@ scan_core_tree(Tree, #callgraph{calls = ETSCalls,
 				name_map = ETSNameMap,
 				rec_var_map = ETSRecVarMap,
 				rev_name_map = ETSRevNameMap,
-				self_rec = ETSSelfRec} = CG) ->
+				self_rec = ETSSelfRec}) ->
   %% Build name map and recursion variable maps.
   build_maps(Tree, ETSRecVarMap, ETSNameMap, ETSRevNameMap),
-  
+
   %% First find the module-local dependencies.
   {Deps0, EscapingFuns, Calls} = dialyzer_dep:analyze(Tree),
   true = ets:insert(ETSCalls, dict:to_list(Calls)),
@@ -399,7 +400,7 @@ scan_core_tree(Tree, #callgraph{calls = ETSCalls,
   NewNamedEdges1 =
     [E || {From, To} = E <- NamedEdges1, From =/= top, To =/= top],
   NamedEdges3 = NewNamedEdges1 ++ NewNamedEdges2,
-  add_edges(NamedEdges3, Names3, CG).
+  {Names3, NamedEdges3}.
 
 build_maps(Tree, ETSRecVarMap, ETSNameMap, ETSRevNameMap) ->
   %% We only care about the named (top level) functions. The anonymous
diff --git a/lib/dialyzer/src/dialyzer_coordinator.erl b/lib/dialyzer/src/dialyzer_coordinator.erl
index fa78670883..d8a3ef2bd2 100644
--- a/lib/dialyzer/src/dialyzer_coordinator.erl
+++ b/lib/dialyzer/src/dialyzer_coordinator.erl
@@ -46,15 +46,26 @@
 
 -module(dialyzer_coordinator).
 
--export([
-	 all_spawned/1,
-	 scc_done/3,
-	 scc_spawn/2,
+%%% Exports for all possible uses of coordinator
+-export([start/2,
+	 all_spawned/1]).
+
+%%% Exports for the typesig and dataflow analysis main process
+-export([scc_spawn/2,
+	 receive_not_fixpoint/0]).
+
+%%% Exports for the typesig and dataflow analysis workers
+-export([scc_done/3,
 	 sccs_to_pids_reply/0,
-	 sccs_to_pids_request/2,
-	 start/2,
-	 receive_not_fixpoint/0
-	]).
+	 sccs_to_pids_request/2]).
+
+%%% Exports for the compilation main process
+-export([compiler_spawn/2,
+	 receive_compilation_data/0]).
+
+%%% Exports for the compilation workers
+-export([compilation_done/3,
+	 get_next_label/2]).
 
 -behaviour(gen_server).
 
@@ -64,129 +75,192 @@
 -type coordinator() :: pid().
 -type map() :: dict().
 -type scc() :: [mfa_or_funlbl()].
--type mode()   :: typesig | dataflow.
+-type mode()   :: 'typesig' | 'dataflow' | 'compile'.
 
 -record(state, {parent                   :: pid(),
 		mode                     :: mode(),
 		spawn_count  = 0         :: integer(),
 		all_spawned  = false     :: boolean(),
-		scc_to_pid   = new_map() :: map(),
-		not_fixpoint = []        :: [mfa_or_funlbl()],
-		servers                  :: dialyzer_typesig:servers()
+		job_to_pid               :: map(),
+		next_label               :: integer(),
+		result                   :: [mfa_or_funlbl()] |
+					    dialyzer_analysis_callgraph:result(),
+		init_job_data            :: dialyzer_typesig:servers()
 	       }).
 
 -include("dialyzer.hrl").
 
 %%--------------------------------------------------------------------
 
--spec start(mode(), dialyzer_typesig:servers()) -> pid().
+-spec start('typesig' | 'dataflow', dialyzer_typesig:servers()) -> pid();
+	   ('compile', dialyzer_analysis_callgraph:servers()) -> pid().
 
 start(Mode, Servers) ->
-    {ok, Pid} = gen_server:start(?MODULE, {self(), Mode, Servers}, []),
-    Pid.
+  {ok, Pid} = gen_server:start(?MODULE, {self(), Mode, Servers}, []),
+  Pid.
 
 -spec scc_spawn(scc(), coordinator()) -> ok.
 
 scc_spawn(SCC, Coordinator) ->
-    cast({scc_spawn, SCC}, Coordinator).
+  cast({scc_spawn, SCC}, Coordinator).
 
 -spec sccs_to_pids_request([scc()], coordinator()) -> ok.
 
 sccs_to_pids_request(SCCs, Coordinator) ->
-    cast({sccs_to_pids, SCCs, self()}, Coordinator).
+  cast({sccs_to_pids, SCCs, self()}, Coordinator).
 
 scc_to_pids_request_handle(Worker, SCCs, SCCtoPID) ->
-    Pids = [fetch_map(SCC, SCCtoPID) || SCC <- SCCs],
-    Worker ! {sccs_to_pids, Pids},
-    ok.
+  Pids = [fetch_map(SCC, SCCtoPID) || SCC <- SCCs],
+  Worker ! {sccs_to_pids, Pids},
+  ok.
 
 -spec sccs_to_pids_reply() -> [dialyzer_worker:worker()].
 
 sccs_to_pids_reply() ->
-    receive {sccs_to_pids, Pids} -> Pids end.
+  receive {sccs_to_pids, Pids} -> Pids end.
 
 -spec scc_done(scc(), scc(), coordinator()) -> ok.
 
 scc_done(SCC, NotFixpoint, Coordinator) ->
-    cast({scc_done, SCC, NotFixpoint}, Coordinator).
+  cast({done, SCC, NotFixpoint}, 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{parent = Parent, not_fixpoint = NotFixpoint}) ->
-    Parent ! {not_fixpoint, NotFixpoint}.
+  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' -> {not_fixpoint, Result};
+      'compile' ->  {compilation_data, Result, NextLabel}
+    end,
+  Parent ! Msg.
 
 -spec receive_not_fixpoint() -> dialyzer_plt:plt().
 
 receive_not_fixpoint() ->
-    receive {not_fixpoint, NotFixpoint} -> NotFixpoint end.
+  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 compiler_spawn(file:filename(), coordinator()) -> ok.
 
--spec init({pid(), mode(), dialyzer_succ_typings:servers()}) -> {ok, #state{}}.
+compiler_spawn(Filename, Coordinator) ->
+  cast({compiler_spawn, Filename}, Coordinator).
 
-init({Parent, Mode, Servers}) ->
-    {ok, #state{parent = Parent, mode = Mode, servers = Servers}}.
+-spec get_next_label(integer(), coordinator()) -> integer().
+
+get_next_label(EstimatedSize, Coordinator) ->
+  call({get_next_label, EstimatedSize}, Coordinator).
+
+%%--------------------------------------------------------------------
+
+-spec init({pid(), mode(), dialyzer_succ_typings:servers() |
+	    dialyzer_analysis_callgraph:servers()}) -> {ok, #state{}}.
+
+init({Parent, Mode, InitJobData}) ->
+  InitState = #state{parent = Parent, mode = Mode, init_job_data = InitJobData},
+  State =
+    case Mode of
+      X when X =:= 'typesig'; X =:= 'dataflow' ->
+	InitState#state{result = [], job_to_pid = new_map()};
+      '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{}}.
+		     {reply, Reply::term(), #state{}}.
 
-handle_call(_Request, _From, State) ->
-    {reply, ok, 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{}}.
+		     {noreply, #state{}} | {stop, normal, #state{}}.
 
-handle_cast({scc_done, _SCC, NotFixpoint},
-	    #state{spawn_count = SpawnCount,
+handle_cast({done, Job, NewData},
+	    #state{mode = Mode,
+		   spawn_count = SpawnCount,
 		   all_spawned = AllSpawned,
-		   not_fixpoint = OldNotFixpoint
+		   result = OldResult
 		  } = State) ->
-    NewNotFixpoint = ordsets:union(OldNotFixpoint, NotFixpoint),
-    UpdatedState = State#state{not_fixpoint = NewNotFixpoint},
-    Action =
-	case AllSpawned of
-	    false -> reduce;
-	    true ->
-		case SpawnCount of
-		    1 -> finish;
-		    _ -> reduce
-		end
-	end,
-    case Action of
-	reduce ->
-	    NewState = UpdatedState#state{spawn_count = SpawnCount - 1},
-	    {noreply, NewState};
-	finish ->
-	    send_done_to_parent(UpdatedState),
-	    {stop, normal, State}
-    end;
+  NewResult =
+    case Mode of
+      X when X =:= 'typesig'; X =:= 'dataflow' ->
+	ordsets:union(OldResult, NewData);
+      'compile' ->
+	dialyzer_analysis_callgraph:add_to_result(Job, NewData, OldResult)
+    end,
+  UpdatedState = State#state{result = NewResult},
+  Action =
+    case AllSpawned of
+      false -> reduce;
+      true ->
+	case SpawnCount of
+	  1 -> finish;
+	  _ -> reduce
+	end
+    end,
+  case Action of
+    reduce ->
+      NewState = UpdatedState#state{spawn_count = SpawnCount - 1},
+      {noreply, NewState};
+    finish ->
+      send_done_to_parent(UpdatedState),
+      {stop, normal, State}
+  end;
 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;
+  case SpawnCount of
+    0 ->
+      send_done_to_parent(State),
+      {stop, normal, State};
+    _ ->
+      NewState = State#state{all_spawned = true},
+      {noreply, NewState}
+  end;
 handle_cast({sccs_to_pids, SCCs, Worker},
-	    #state{scc_to_pid = SCCtoPID} = State) ->
-    scc_to_pids_request_handle(Worker, SCCs, SCCtoPID),
-    {noreply, State};
+	    #state{job_to_pid = SCCtoPID} = State) ->
+  scc_to_pids_request_handle(Worker, SCCs, SCCtoPID),
+  {noreply, State};
 handle_cast({scc_spawn, SCC},
 	    #state{mode = Mode,
-		   servers = Servers,
+		   init_job_data = Servers,
 		   spawn_count = SpawnCount,
-		   scc_to_pid = SCCtoPID
+		   job_to_pid = SCCtoPID
 		  } = State) ->
-    Pid = dialyzer_worker:launch(Mode, SCC, Servers),
-    {noreply,
-     State#state{spawn_count = SpawnCount + 1,
-		 scc_to_pid = store_map(SCC, Pid, SCCtoPID)}
-    }.
+  Pid = dialyzer_worker:launch(Mode, SCC, Servers),
+  {noreply,
+   State#state{spawn_count = SpawnCount + 1,
+	       job_to_pid = store_map(SCC, Pid, SCCtoPID)}
+  };
+handle_cast({compiler_spawn, Filename},
+	    #state{mode = Mode,
+		   init_job_data = Servers,
+		   spawn_count = SpawnCount
+		  } = State) ->
+  dialyzer_worker:launch(Mode, Filename, Servers),
+  {noreply,
+   State#state{spawn_count = SpawnCount + 1}
+  }.
 
 -spec handle_info(term(), #state{}) -> {noreply, #state{}}.
 
@@ -206,13 +280,16 @@ code_change(_OldVsn, State, _Extra) ->
 %%--------------------------------------------------------------------
 
 cast(Message, Coordinator) ->
-    gen_server:cast(Coordinator, Message).
+  gen_server:cast(Coordinator, Message).
+
+call(Message, Coordinator) ->
+  gen_server:call(Coordinator, Message).
 
 new_map() ->
-    dict:new().
+  dict:new().
 
 store_map(Key, Value, Map) ->
-    dict:store(Key, Value, Map).
+  dict:store(Key, Value, Map).
 
 fetch_map(Key, Map) ->
-    dict:fetch(Key, Map).
+  dict:fetch(Key, Map).
diff --git a/lib/dialyzer/src/dialyzer_worker.erl b/lib/dialyzer/src/dialyzer_worker.erl
index a2d30c27d3..b24103c5a8 100644
--- a/lib/dialyzer/src/dialyzer_worker.erl
+++ b/lib/dialyzer/src/dialyzer_worker.erl
@@ -26,10 +26,11 @@
 
 -record(state, {
 	  mode             :: dialyzer_coordinator:mode(),
-	  scc         = [] :: mfa_or_funlbl(),
+	  job              :: mfa_or_funlbl() | file:filename(),
 	  depends_on  = [] :: list(),
 	  coordinator      :: dialyzer_coordinator:coordinator(),
-	  servers          :: dialyzer_typesig:servers(),
+	  servers          :: dialyzer_typesig:servers() |
+			      dialyzer_analysis_callgraph:servers(),
 	  scc_data         :: dialyzer_typesig:scc_data()
 	 }).
 
@@ -48,17 +49,22 @@
 -spec launch(dialyzer_coordinator:mode(), [mfa_or_funlbl()],
 	     dialyzer_typesig:servers()) -> worker().
 
-launch(Mode, SCC, Servers) ->
+launch(Mode, Job, Servers) ->
   State = #state{mode        = Mode,
-		 scc         = SCC,
+		 job         = Job,
 		 servers     = Servers,
 		 coordinator = self()},
-  spawn(fun() -> loop(initializing, State) end).
+  InitState =
+    case Mode of
+      X when X =:= 'typesig'; X =:= 'dataflow' -> initializing;
+      'compile' -> running
+    end,
+  spawn(fun() -> loop(InitState, State) end).
 
 %%--------------------------------------------------------------------
 
 loop(updating, State) ->
-  ?debug("Update: ~p\n",[State#state.scc]),
+  ?debug("Update: ~p\n",[State#state.job]),
   NextStatus =
     case waits_more_success_typings(State) of
       true -> waiting;
@@ -73,20 +79,27 @@ loop(updating, State) ->
 	end
     end,
   loop(NextStatus, State);
-loop(initializing, #state{scc = SCC, servers = Servers} = State) ->
+loop(initializing, #state{job = SCC, servers = Servers} = State) ->
   DependsOn = dialyzer_succ_typings:find_depends_on(SCC, Servers),
   WithoutSelf = DependsOn -- [SCC],
-  ?debug("Deps ~p: ~p\n",[State#state.scc, WithoutSelf]),
+  ?debug("Deps ~p: ~p\n",[State#state.job, WithoutSelf]),
   loop(updating, State#state{depends_on = WithoutSelf});
 loop(waiting, State) ->
-  ?debug("Wait: ~p\n",[State#state.scc]),
+  ?debug("Wait: ~p\n",[State#state.job]),
   NewState = wait_for_success_typings(State),
   loop(updating, NewState);
 loop(getting_data, State) ->
-  ?debug("Data: ~p\n",[State#state.scc]),
+  ?debug("Data: ~p\n",[State#state.job]),
   loop(updating, get_data(State));
-loop(running, State) ->
-  ?debug("Run: ~p\n",[State#state.scc]),
+loop(running, #state{mode = 'compile'} = State) ->
+  ?debug("Compile: ~s\n",[State#state.job]),
+  {EstimatedSize, Data} = start_compilation(State),
+  Label = ask_coordinator_for_label(EstimatedSize, State),
+  Result = continue_compilation(Label, Data),
+  report_to_coordinator(Result, State);
+loop(running, #state{mode = Mode} = State) when
+    Mode =:= 'typesig'; Mode =:= 'dataflow' ->
+  ?debug("Run: ~p\n",[State#state.job]),
   ok = ask_coordinator_for_callers(State),
   NotFixpoint = do_work(State),
   Callers = get_callers_reply_from_coordinator(),
@@ -106,7 +119,7 @@ has_data(#state{scc_data = Data}) ->
     _ -> true
   end.
 
-get_data(#state{mode = Mode, scc = SCC, servers = Servers} = State) ->
+get_data(#state{mode = Mode, job = SCC, servers = Servers} = State) ->
   Data =
     case Mode of
       typesig -> dialyzer_succ_typings:collect_scc_data(SCC, Servers);
@@ -114,7 +127,7 @@ get_data(#state{mode = Mode, scc = SCC, servers = Servers} = State) ->
     end,
   State#state{scc_data = Data}.
 
-ask_coordinator_for_callers(#state{scc = SCC,
+ask_coordinator_for_callers(#state{job = SCC,
 				   servers = Servers,
 				   coordinator = Coordinator}) ->
   RequiredBy = dialyzer_succ_typings:find_required_by(SCC, Servers),
@@ -125,7 +138,7 @@ ask_coordinator_for_callers(#state{scc = SCC,
 get_callers_reply_from_coordinator() ->
   dialyzer_coordinator:sccs_to_pids_reply().
 
-broadcast_done(#state{scc = SCC}, Callers) ->
+broadcast_done(#state{job = SCC}, Callers) ->
   ?debug("Sending ~p: ~p\n",[SCC, Callers]),
   SendSTFun = fun(PID) -> PID ! {done, SCC} end,
   lists:foreach(SendSTFun, Callers).
@@ -133,11 +146,11 @@ broadcast_done(#state{scc = SCC}, Callers) ->
 wait_for_success_typings(#state{depends_on = DependsOn} = State) ->
   receive
     {done, SCC} ->
-      ?debug("GOT ~p: ~p\n",[State#state.scc, SCC]),
+      ?debug("GOT ~p: ~p\n",[State#state.job, SCC]),
       State#state{depends_on = DependsOn -- [SCC]}
   after
     5000 ->
-      ?debug("Still Waiting ~p: ~p\n",[State#state.scc, DependsOn]),
+      ?debug("Still Waiting ~p: ~p\n",[State#state.job, DependsOn]),
       State
   end.
 
@@ -148,6 +161,15 @@ do_work(#state{mode = Mode, scc_data = SCCData}) ->
   end.
 
 report_to_coordinator(NotFixpoint,
-		      #state{scc = SCC, coordinator = Coordinator}) ->
-  ?debug("Done: ~p\n",[SCC]),
-  dialyzer_coordinator:scc_done(SCC, NotFixpoint, Coordinator).
+		      #state{job = Job, coordinator = Coordinator}) ->
+  ?debug("Done: ~p\n",[Job]),
+  dialyzer_coordinator:scc_done(Job, NotFixpoint, Coordinator).
+
+start_compilation(#state{job = Job, servers = Servers}) ->
+  dialyzer_analysis_callgraph:start_compilation(Job, Servers).
+
+ask_coordinator_for_label(EstimatedSize, #state{coordinator = Coordinator}) ->
+  dialyzer_coordinator:get_next_label(EstimatedSize, Coordinator).
+
+continue_compilation(Label, Data) ->
+  dialyzer_analysis_callgraph:continue_compilation(Label, Data).