Add API for listening to concurrent mix compilations (#13896)

Author: jonatanklosko

lib/iex/lib/iex/helpers.ex

@@ -136,7 +136,14 @@ defmodule IEx.Helpers do
 
         if is_nil(project) or
              project.__info__(:compile)[:source] == String.to_charlist(Path.absname("mix.exs")) do
-          do_recompile(options)
+          Mix.Project.with_build_lock(fn ->
+            purge_result = IEx.MixListener.purge()
+
+            case do_recompile(options) do
+              :noop -> purge_result
+              compile_result -> compile_result
+            end
+          end)
         else
           message = "Cannot recompile because the current working directory changed"
           IO.puts(IEx.color(:eval_error, message))

lib/iex/lib/iex/mix_listener.ex

@@ -0,0 +1,55 @@
+defmodule IEx.MixListener do
+  @moduledoc false
+
+  use GenServer
+
+  @name __MODULE__
+
+  @spec start_link(keyword) :: GenServer.on_start()
+  def start_link(_opts) do
+    GenServer.start_link(__MODULE__, {}, name: @name)
+  end
+
+  @doc """
+  Unloads all modules invalidated by external compilations.
+  """
+  @spec purge :: :ok | :noop
+  def purge do
+    GenServer.call(@name, :purge, :infinity)
+  end
+
+  @impl true
+  def init({}) do
+    {:ok, %{to_purge: MapSet.new()}}
+  end
+
+  @impl true
+  def handle_call(:purge, _from, state) do
+    for module <- state.to_purge do
+      :code.purge(module)
+      :code.delete(module)
+    end
+
+    status = if Enum.empty?(state.to_purge), do: :noop, else: :ok
+
+    {:reply, status, %{state | to_purge: MapSet.new()}}
+  end
+
+  @impl true
+  def handle_info({:modules_compiled, info}, state) do
+    if info.os_pid == System.pid() do
+      # Ignore compilations from ourselves, because the modules are
+      # already updated in memory
+      {:noreply, state}
+    else
+      %{changed: changed, removed: removed} = info.modules_diff
+      state = update_in(state.to_purge, &Enum.into(changed, &1))
+      state = update_in(state.to_purge, &Enum.into(removed, &1))
+      {:noreply, state}
+    end
+  end
+
+  def handle_info(_message, state) do
+    {:noreply, state}
+  end
+end

lib/mix/lib/mix.ex

@@ -402,7 +402,7 @@ defmodule Mix do
   def start(_type, []) do
     Mix.Local.append_archives()
     Mix.Local.append_paths()
-    children = [Mix.State, Mix.TasksServer, Mix.ProjectStack]
+    children = [Mix.Sync.PubSub, Mix.State, Mix.TasksServer, Mix.ProjectStack]
     opts = [strategy: :one_for_one, name: Mix.Supervisor, max_restarts: 0]
     Supervisor.start_link(children, opts)
   end

lib/mix/lib/mix/compilers/elixir.ex

@@ -202,15 +202,24 @@ defmodule Mix.Compilers.Elixir do
 
           put_compile_env(sources)
           all_warnings = previous_warnings ++ runtime_warnings ++ compile_warnings
-          unless_previous_warnings_as_errors(previous_warnings, opts, {:ok, all_warnings})
+
+          lazy_modules_diff = fn ->
+            modules_diff(modules, removed_modules, all_modules, timestamp)
+          end
+
+          unless_previous_warnings_as_errors(
+            previous_warnings,
+            opts,
+            {:ok, all_warnings, lazy_modules_diff}
+          )
 
         {:error, errors, %{runtime_warnings: r_warnings, compile_warnings: c_warnings}, state} ->
           # In case of errors, we show all previous warnings and all new ones.
           {_, _, sources, _, _, _} = state
           errors = Enum.map(errors, &diagnostic/1)
           warnings = Enum.map(r_warnings ++ c_warnings, &diagnostic/1)
           all_warnings = Keyword.get(opts, :all_warnings, errors == [])
-          {:error, previous_warnings(sources, all_warnings) ++ warnings ++ errors}
+          {:error, previous_warnings(sources, all_warnings) ++ warnings ++ errors, nil}
       after
         Code.compiler_options(previous_opts)
       end
@@ -247,7 +256,12 @@ defmodule Mix.Compilers.Elixir do
 
       all_warnings = Keyword.get(opts, :all_warnings, true)
       previous_warnings = previous_warnings(sources, all_warnings)
-      unless_previous_warnings_as_errors(previous_warnings, opts, {status, previous_warnings})
+
+      unless_previous_warnings_as_errors(
+        previous_warnings,
+        opts,
+        {status, previous_warnings, nil}
+      )
     end
   end
 
@@ -989,16 +1003,38 @@ defmodule Mix.Compilers.Elixir do
     File.rm(manifest <> ".checkpoint")
   end
 
-  defp unless_previous_warnings_as_errors(previous_warnings, opts, {status, all_warnings}) do
+  defp unless_previous_warnings_as_errors(
+         previous_warnings,
+         opts,
+         {status, all_warnings, modules_diff}
+       ) do
     if previous_warnings != [] and opts[:warnings_as_errors] do
       message = "Compilation failed due to warnings while using the --warnings-as-errors option"
       IO.puts(:stderr, message)
-      {:error, all_warnings}
+      {:error, all_warnings, modules_diff}
     else
-      {status, all_warnings}
+      {status, all_warnings, modules_diff}
     end
   end
 
+  defp modules_diff(compiled_modules, removed_modules, all_modules, timestamp) do
+    {changed, added} =
+      compiled_modules
+      |> Map.keys()
+      |> Enum.split_with(&Map.has_key?(all_modules, &1))
+
+    # Note that removed_modules may also include changed modules
+    removed =
+      for {module, _} <- removed_modules, not Map.has_key?(compiled_modules, module), do: module
+
+    %{
+      added: added,
+      changed: changed,
+      removed: removed,
+      timestamp: timestamp
+    }
+  end
+
   ## Compiler loop
   # The compiler is invoked in a separate process so we avoid blocking its main loop.
 

lib/mix/lib/mix/project.ex

@@ -896,7 +896,7 @@ defmodule Mix.Project do
       Mix.shell().info("Waiting for lock on the build directory (held by process #{os_pid})")
     end
 
-    Mix.Lock.with_lock(build_path, fun, on_taken: on_taken)
+    Mix.Sync.Lock.with_lock(build_path, fun, on_taken: on_taken)
   end
 
   @doc false
@@ -910,7 +910,7 @@ defmodule Mix.Project do
       Mix.shell().info("Waiting for lock on the deps directory (held by process #{os_pid})")
     end
 
-    Mix.Lock.with_lock(deps_path, fun, on_taken: on_taken)
+    Mix.Sync.Lock.with_lock(deps_path, fun, on_taken: on_taken)
   end
 
   # Loads mix.exs in the current directory or loads the project from the

lib/mix/lib/mix/pubsub.ex

@@ -0,0 +1,107 @@
+defmodule Mix.PubSub do
+  @moduledoc false
+
+  # The Mix pub/sub is responsible for notifying other OS processes
+  # about relevant concurrent events.
+  #
+  # The pub/sub consists of a local subscriber process that receives
+  # events from other OS processes, and a listener supervisor, which
+  # holds all listener processes that have been configured. Whenever
+  # the subscriber receives an event, it sends a message to each of
+  # the listeners.
+  #
+  # Inherently, a compilation may be required before the listener
+  # modules are available, so we start the local subscriber process
+  # separately with `start/0`, and then start the listeners later
+  # with `start_listeners/0`. The subscriber is going to accumulate
+  # events and reply them once the listenres are started.
+
+  @spec start :: :ok
+  def start do
+    # Avoid calling the supervisor, if already started
+    if Process.whereis(Mix.PubSub) do
+      :ok
+    else
+      case Supervisor.start_child(Mix.Supervisor, Mix.PubSub) do
+        {:ok, _pid} ->
+          :ok
+
+        {:error, {:already_started, _pid}} ->
+          :ok
+
+        {:error, reason} ->
+          raise RuntimeError, "failed to start Mix.PubSub, reason: #{inspect(reason)}"
+      end
+    end
+  end
+
+  @spec child_spec(term) :: Supervisor.child_spec()
+  def child_spec(_opts) do
+    children = [
+      Mix.PubSub.Subscriber
+    ]
+
+    opts = [strategy: :one_for_one, name: Mix.PubSub]
+
+    %{
+      id: Mix.PubSub,
+      start: {Supervisor, :start_link, [children, opts]},
+      type: :supervisor
+    }
+  end
+
+  @spec start_listeners :: :ok
+  def start_listeners do
+    # Avoid calling the supervisor, if already started
+    if Process.whereis(Mix.PubSub.ListenerSupervisor) do
+      :ok
+    else
+      case Supervisor.start_child(Mix.PubSub, listener_supervisor()) do
+        {:ok, _pid} ->
+          Mix.PubSub.Subscriber.flush()
+          :ok
+
+        {:error, {:already_started, _pid}} ->
+          :ok
+
+        {:error, reason} ->
+          raise RuntimeError,
+                "failed to start Mix.PubSub.ListenerSupervisor, reason: #{inspect(reason)}"
+      end
+    end
+  end
+
+  defp listener_supervisor do
+    children = configured_listeners() ++ built_in_listeners()
+
+    children = Enum.map(children, &Supervisor.child_spec(&1, []))
+
+    opts = [strategy: :one_for_one, name: Mix.PubSub.ListenerSupervisor]
+
+    %{
+      id: Mix.PubSub.ListenerSupervisor,
+      start: {Supervisor, :start_link, [children, opts]},
+      type: :supervisor
+    }
+  end
+
+  defp configured_listeners do
+    config = Mix.Project.config()
+
+    listeners =
+      Application.get_env(:mix, :listeners, []) ++
+        Keyword.get(config, :listeners, [])
+
+    Enum.uniq(listeners)
+  end
+
+  defp built_in_listeners do
+    iex_started? = Process.whereis(IEx.Supervisor) != nil
+
+    if iex_started? do
+      [IEx.MixListener]
+    else
+      []
+    end
+  end
+end

lib/mix/lib/mix/pubsub/subscriber.ex

@@ -0,0 +1,51 @@
+defmodule Mix.PubSub.Subscriber do
+  @moduledoc false
+
+  use GenServer
+
+  @name __MODULE__
+
+  @spec start_link(keyword) :: GenServer.on_start()
+  def start_link(_opts) do
+    GenServer.start_link(__MODULE__, {}, name: @name)
+  end
+
+  @spec flush :: :ok
+  def flush do
+    GenServer.cast(@name, :flush)
+  end
+
+  @impl true
+  def init({}) do
+    build_path = Mix.Project.build_path()
+    Mix.Sync.PubSub.subscribe(build_path)
+    {:ok, %{acc: []}}
+  end
+
+  @impl true
+  def handle_info(message, %{acc: nil} = state) do
+    notify_listeners([message])
+    {:noreply, state}
+  end
+
+  def handle_info(message, state) do
+    # Accumulate messages until the flush
+    {:noreply, update_in(state.acc, &[message | &1])}
+  end
+
+  @impl true
+  def handle_cast(:flush, state) do
+    notify_listeners(Enum.reverse(state.acc))
+    {:noreply, %{state | acc: nil}}
+  end
+
+  defp notify_listeners(messages) do
+    children = Supervisor.which_children(Mix.PubSub.ListenerSupervisor)
+
+    for message <- messages do
+      for {_, pid, _, _} <- children, is_pid(pid) do
+        send(pid, message)
+      end
+    end
+  end
+end