Files
n8n-openai-adapter/lib/elixir/lib/kernel/parallel_compiler.ex
T

509 lines
17 KiB
Elixir

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`
"""
@doc since: "1.6.0"
def async(fun) when is_function(fun) do
if parent = :erlang.get(:elixir_compiler_pid) do
file = :erlang.get(:elixir_compiler_file)
dest = :erlang.get(:elixir_compiler_dest)
{: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)
dest != :undefined and :erlang.put(:elixir_compiler_dest, dest)
: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.
"""
@doc since: "1.6.0"
def compile(files, options \\ []) when is_list(options) do
spawn_workers(files, :compile, options)
end
@doc since: "1.6.0"
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
"""
@doc since: "1.6.0"
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
{:module, _} = :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)
:erlang.put(:elixir_compiler_dest, path)
:elixir_compiler.file_to_path(Path.expand(file), path)
:compile ->
:erlang.process_flag(:error_handler, Kernel.ErrorHandler)
:erlang.put(:elixir_compiler_dest, dest)
Code.compile_file(file)
:require ->
Code.require_file(file)
end
:ok
catch
kind, reason ->
{kind, reason, __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
# Single entry, just release it.
defp spawn_workers([], [_] = waiting, [_] = queued, result, warnings, state) do
[{_, _, ref, _, _}] = waiting
spawn_workers([{ref, :not_found}], waiting, queued, result, warnings, state)
end
# Multiple entries, try to release modules.
defp spawn_workers([], waiting, queued, result, warnings, state)
when length(waiting) == length(queued) do
# The goal of this function is to find leaves in the dependency graph,
# i.e. to find code that depends on code that we know is not being defined.
# Note we only release modules because those can be rescued. A missing
# struct is a guaranteed compile error, so we never release it and treat
# it exclusively a missing entry/deadlock.
pending =
for {pid, _, _, _} <- queued,
entry = waiting_on_without_definition(waiting, pid),
{kind, _, ref, on, _} = entry,
kind == :module,
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 #{kind} #{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(
"\nEnsure there are no compile-time dependencies between those files " <>
"and that the modules they reference exist and are correctly named\n"
)
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