path: root/lib/dialyzer/src/dialyzer_coordinator.erl

%% -*- 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 <[email protected]>
%%%
%%% 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.