defmodule Kernel.ParallelCompiler do @moduledoc """ A module responsible for compiling and requiring files in parallel. """ @doc """ Starts a task for parallel compilation. If you have a file that needs to compile other modules in parallel, the spawned processes need to be aware of the compiler environment. This function allows a developer to create a task that is aware of those environments. See `Task.async/1` for more information. The task spawned must be always awaited on by calling `Task.await/1` """ def async(fun) when is_function(fun) do if parent = :erlang.get(:elixir_compiler_pid) do file = :erlang.get(:elixir_compiler_file) {:error_handler, error_handler} = :erlang.process_info(self(), :error_handler) Task.async(fn -> :erlang.put(:elixir_compiler_pid, parent) :erlang.put(:elixir_compiler_file, file) :erlang.process_flag(:error_handler, error_handler) fun.() end) else raise ArgumentError, "cannot spawn parallel compiler task because " <> "the current file is not being compiled/required" end end @doc """ Compiles the given files. Those files are compiled in parallel and can automatically detect dependencies between them. Once a dependency is found, the current file stops being compiled until the dependency is resolved. It returns `{:ok, modules, warnings}` or `{:error, errors, warnings}`. Both errors and warnings are a list of three element tuples containing the file, line and the formatted error/warning. ## Options * `:each_file` - for each file compiled, invokes the callback passing the file * `:each_long_compilation` - for each file that takes more than a given timeout (see the `:long_compilation_threshold` option) to compile, invoke this callback passing the file as its argument * `:each_module` - for each module compiled, invokes the callback passing the file, module and the module bytecode * `:each_cycle` - after the given files are compiled, invokes this function that return a list with potentially more files to compile * `:long_compilation_threshold` - the timeout (in seconds) after the `:each_long_compilation` callback is invoked; defaults to `15` * `:dest` - the destination directory for the BEAM files. When using `files/2`, this information is only used to properly annotate the BEAM files before they are loaded into memory. If you want a file to actually be written to `dest`, use `compile_to_path/3` instead. """ def compile(files, options \\ []) when is_list(options) do spawn_workers(files, :compile, options) end def compile_to_path(files, path, options \\ []) when is_binary(path) and is_list(options) do spawn_workers(files, {:compile, path}, options) end @doc """ Requires the given files in parallel. Opposite to compile, dependencies are not attempted to be automatically solved between files. It returns `{:ok, modules, warnings}` or `{:error, errors, warnings}`. Both errors and warnings are a list of three element tuples containing the file, line and the formatted error/warning. ## Options * `:each_file` - for each file compiled, invokes the callback passing the file * `:each_module` - for each module compiled, invokes the callback passing the file, module and the module bytecode """ def require(files, options \\ []) when is_list(options) do spawn_workers(files, :require, options) end # TODO: Deprecate on Elixir v1.8 @doc false def files(files, options \\ []) when is_list(options) do case spawn_workers(files, :compile, options) do {:ok, modules, _} -> modules {:error, _, _} -> exit({:shutdown, 1}) end end # TODO: Deprecate on Elixir v1.8 @doc false def files_to_path(files, path, options \\ []) when is_binary(path) and is_list(options) do case spawn_workers(files, {:compile, path}, options) do {:ok, modules, _} -> modules {:error, _, _} -> exit({:shutdown, 1}) end end defp spawn_workers(files, output, options) do true = Code.ensure_loaded?(Kernel.ErrorHandler) compiler_pid = self() :elixir_code_server.cast({:reset_warnings, compiler_pid}) schedulers = max(:erlang.system_info(:schedulers_online), 2) result = spawn_workers(files, [], [], [], [], %{ dest: Keyword.get(options, :dest), each_cycle: Keyword.get(options, :each_cycle, fn -> [] end), each_file: Keyword.get(options, :each_file, fn _file -> :ok end), each_long_compilation: Keyword.get(options, :each_long_compilation, fn _file -> :ok end), each_module: Keyword.get(options, :each_module, fn _file, _module, _binary -> :ok end), output: output, long_compilation_threshold: Keyword.get(options, :long_compilation_threshold, 15), schedulers: schedulers }) # In case --warning-as-errors is enabled and there was a warning, # compilation status will be set to error. compilation_status = :elixir_code_server.call({:compilation_status, compiler_pid}) case {result, compilation_status} do {{:ok, _, warnings}, :error} -> message = "Compilation failed due to warnings while using the --warnings-as-errors option" IO.puts(:stderr, message) {:error, warnings, []} {{:error, errors, warnings}, :error} -> {:error, errors ++ warnings, []} _ -> result end end # We already have n=schedulers currently running, don't spawn new ones defp spawn_workers(files, waiting, queued, result, warnings, %{schedulers: schedulers} = state) when length(queued) - length(waiting) >= schedulers do wait_for_messages(files, waiting, queued, result, warnings, state) end # Release waiting processes defp spawn_workers([{ref, found} | t], waiting, queued, result, warnings, state) do waiting = case List.keytake(waiting, ref, 2) do {{_kind, pid, ^ref, _on, _defining}, waiting} -> send(pid, {ref, found}) waiting nil -> waiting end spawn_workers(t, waiting, queued, result, warnings, state) end defp spawn_workers([file | files], waiting, queued, result, warnings, state) do %{output: output, long_compilation_threshold: threshold, dest: dest} = state parent = self() {pid, ref} = :erlang.spawn_monitor(fn -> :erlang.put(:elixir_compiler_pid, parent) :erlang.put(:elixir_compiler_file, file) result = try do _ = case output do {:compile, path} -> :erlang.process_flag(:error_handler, Kernel.ErrorHandler) :elixir_compiler.file_to_path(file, path) :compile -> :erlang.process_flag(:error_handler, Kernel.ErrorHandler) :elixir_compiler.file(file, dest) :require -> Code.require_file(file) end :ok catch kind, reason -> {kind, reason, System.stacktrace()} end send(parent, {:file_done, self(), file, result}) exit(:shutdown) end) timer_ref = Process.send_after(self(), {:timed_out, pid}, threshold * 1000) queued = [{pid, ref, file, timer_ref} | queued] spawn_workers(files, waiting, queued, result, warnings, state) end # No more files, nothing waiting, queue is empty, this cycle is done defp spawn_workers([], [], [], result, warnings, state) do case state.each_cycle.() do [] -> modules = for {:module, mod} <- result, do: mod warnings = Enum.reverse(warnings) {:ok, modules, warnings} more -> spawn_workers(more, [], [], result, warnings, state) end end # Queued x, waiting for x: POSSIBLE ERROR! Release processes so we get the failures defp spawn_workers([], waiting, queued, result, warnings, state) when length(waiting) == length(queued) do pending = for {pid, _, _, _} <- queued, entry = waiting_on_without_definition(waiting, pid), {_, _, ref, on, _} = entry, do: {on, {ref, :not_found}} # Instead of releasing all files at once, we release them in groups # based on the module they are waiting on. We pick the module being # depended on with less edges, as it is the mostly likely source of # error (for example, someone made a typo). This may not always be # true though: for example, if there is a macro injecting code into # multiple modules and such code becomes faulty, now multiple modules # are waiting on the same module required by the faulty code. However, # since we need to pick something to be first, the one with fewer edges # sounds like a sane choice. pending |> Enum.group_by(&elem(&1, 0), &elem(&1, 1)) |> Enum.sort_by(&length(elem(&1, 1))) |> case do [{_on, refs} | _] -> spawn_workers(refs, waiting, queued, result, warnings, state) [] -> errors = handle_deadlock(waiting, queued) {:error, errors, warnings} end end # No more files, but queue and waiting are not full or do not match defp spawn_workers([], waiting, queued, result, warnings, state) do wait_for_messages([], waiting, queued, result, warnings, state) end defp waiting_on_without_definition(waiting, pid) do {_, ^pid, _, on, _} = entry = List.keyfind(waiting, pid, 1) if Enum.any?(waiting, fn {_, _, _, _, defining} -> on in defining end) do nil else entry end end # Wait for messages from child processes defp wait_for_messages(files, waiting, queued, result, warnings, state) do %{output: output} = state receive do {:struct_available, module} -> available = for {:struct, _, ref, waiting_module, _defining} <- waiting, module == waiting_module, do: {ref, :found} result = [{:struct, module} | result] spawn_workers(available ++ files, waiting, queued, result, warnings, state) {:module_available, child, ref, file, module, binary} -> state.each_module.(file, module, binary) # Release the module loader which is waiting for an ack send(child, {ref, :ack}) available = for {:module, _, ref, waiting_module, _defining} <- waiting, module == waiting_module, do: {ref, :found} cancel_waiting_timer(queued, child) result = [{:module, module} | result] spawn_workers(available ++ files, waiting, queued, result, warnings, state) # If we are simply requiring files, we do not add to waiting. {:waiting, _kind, child, ref, _on, _defining} when output == :require -> send(child, {ref, :not_found}) spawn_workers(files, waiting, queued, result, warnings, state) {:waiting, kind, child, ref, on, defining} -> # Oops, we already got it, do not put it on waiting. # Alternatively, we're waiting on ourselves, # send :found so that we can crash with a better error. waiting = if :lists.any(&match?({^kind, ^on}, &1), result) or on in defining do send(child, {ref, :found}) waiting else [{kind, child, ref, on, defining} | waiting] end spawn_workers(files, waiting, queued, result, warnings, state) {:timed_out, child} -> case List.keyfind(queued, child, 0) do {^child, _, file, _} -> state.each_long_compilation.(file) _ -> :ok end spawn_workers(files, waiting, queued, result, warnings, state) {:warning, file, line, message} -> file = file && Path.absname(file) message = :unicode.characters_to_binary(message) warning = {file, line, message} wait_for_messages(files, waiting, queued, result, [warning | warnings], state) {:file_done, child_pid, file, :ok} -> discard_down(child_pid) state.each_file.(file) cancel_waiting_timer(queued, child_pid) # Sometimes we may have spurious entries in the waiting # list because someone invoked try/rescue UndefinedFunctionError new_files = List.delete(files, child_pid) new_queued = List.keydelete(queued, child_pid, 0) new_waiting = List.keydelete(waiting, child_pid, 1) spawn_workers(new_files, new_waiting, new_queued, result, warnings, state) {:file_done, child_pid, file, {kind, reason, stack}} -> discard_down(child_pid) print_error(file, kind, reason, stack) cancel_waiting_timer(queued, child_pid) terminate(queued) {:error, [to_error(file, kind, reason, stack)], warnings} {:DOWN, ref, :process, _pid, reason} -> case handle_down(queued, ref, reason) do :ok -> wait_for_messages(files, waiting, queued, result, warnings, state) {:error, errors} -> {:error, errors, warnings} end end end defp discard_down(pid) do receive do {:DOWN, _, :process, ^pid, _} -> :ok end end defp handle_down(_queued, _ref, :normal) do :ok end defp handle_down(queued, ref, reason) do case List.keyfind(queued, ref, 1) do {_child, ^ref, file, _timer_ref} -> print_error(file, :exit, reason, []) terminate(queued) {:error, [to_error(file, :exit, reason, [])]} _ -> :ok end end defp handle_deadlock(waiting, queued) do deadlock = for {pid, _, file, _} <- queued do {:current_stacktrace, stacktrace} = Process.info(pid, :current_stacktrace) Process.exit(pid, :kill) {_kind, ^pid, _, on, _} = List.keyfind(waiting, pid, 1) description = "deadlocked waiting on module #{inspect(on)}" error = CompileError.exception(description: description, file: nil, line: nil) print_error(file, :error, error, stacktrace) {file, on, description} end IO.puts(""" Compilation failed because of a deadlock between files. The following files depended on the following modules: """) max = deadlock |> Enum.map(&(&1 |> elem(0) |> String.length())) |> Enum.max() for {file, mod, _} <- deadlock do IO.puts([" ", String.pad_leading(file, max), " => " | inspect(mod)]) end IO.puts("") for {file, _, description} <- deadlock, do: {Path.absname(file), nil, description} end defp terminate(queued) do for {pid, _, _, _} <- queued do Process.exit(pid, :kill) end end defp print_error(file, kind, reason, stack) do IO.write([ "\n== Compilation error in file #{Path.relative_to_cwd(file)} ==\n", Kernel.CLI.format_error(kind, reason, stack) ]) end defp cancel_waiting_timer(queued, child_pid) do case List.keyfind(queued, child_pid, 0) do {^child_pid, _ref, _file, timer_ref} -> Process.cancel_timer(timer_ref) # Let's flush the message in case it arrived before we canceled the # timeout. receive do {:timed_out, ^child_pid} -> :ok after 0 -> :ok end nil -> :ok end end defp to_error(file, kind, reason, stack) do line = get_line(file, reason, stack) file = Path.absname(file) message = :unicode.characters_to_binary(Kernel.CLI.format_error(kind, reason, stack)) {file, line, message} end defp get_line(_file, %{line: line}, _stack) when is_integer(line) and line > 0 do line end defp get_line(file, :undef, [{_, _, _, []}, {_, _, _, info} | _]) do if Keyword.get(info, :file) == to_charlist(Path.relative_to_cwd(file)) do Keyword.get(info, :line) end end defp get_line(file, _reason, [{_, _, _, info} | _]) do if Keyword.get(info, :file) == to_charlist(Path.relative_to_cwd(file)) do Keyword.get(info, :line) end end defp get_line(_, _, _) do nil end end