Injecting and Discovering Dependencies in OTP Supervisors

Our task

We need siblings of supervisors to find out about each other. Preferably we also want our supervisor tree to be "hermetic": the possibility to start multiple tree instances with different options.

Example: a pipeline for processing network data which consists of several intercommunicating processes that consume and transform data. We would like to start different pipelines for different data sources in an ad-hoc fashion so they should not interfere with each other and be "hermetic".

There are several ways to do it.

Our options

Hardcoded process name registration

The easiest and most common way is to simply register your processes using global names.

defmodule PipelineSupervisor do
  use Supervisor

  @impl Supervisor
  def init([]) do
    children = [
      {SomeConsumer, name: ConsumerGlobalName}
      {SomeWorker, %{consumer: ConsumerGlobalName}},
    ]

    Supervisor.init(children, strategy: :rest_for_one)
  end
end

defmodule SomeWorker do
  use GenServer

  @impl GenServer
  def init(%{consumer: consumer} = options) do
    {:ok, options}
  end

  @impl GenServer
  def handle_call({:process_data_chunk, data_chunk}, _from, state) do
    {:reply, :ok, do_process_data_chunk(data_chunk, state)}
  end

  defp do_process_data_chunk(data_chunk, state) do
    case parse_data_somehow(data_chunk, state) do
      {:ok, parsed_message, new_state} ->
        Process.send(state.consumer, {:message_received, parsed_message})
        new_state

      {:more_data_necessary, new_state} ->
        new_state
    end
  end
end

defmodule SomeConsumer do
  use GenServer

  def start_link(%{name: name}) do
    GenServer.start_link(__MODULE__, [], name: name)
  end

  @impl GenServer
  def init([]) do
    {:ok, :undefined_state}
  end

  @impl GenServer
  def handle_info({:message_received, parsed_message}, state) do
    # ...
    {:noreply, state}
  end
end

Instead of implicit name registration provided by erlang, you can use some process registry that will serve basically as DI container: a way to register and lookup dependencies (process pid) by some arbitrary id.

These include:

• Using a well-known and reliable registry, for instance, Elixir.Registry
• Using some self-written registry that stores its state (process pid mappings) as GenServer State or in ETS

When the process starts, it registers itself. When its pid is required by other processes they ask this DI registry for necessary dependency.

But the problem is still there. It is a chicken and egg dilemma: To call registry you need in turn to know its pid or name.

This solution is easy and common, but it prevents us from running the supervision tree multiple times because of the hardcoded process name of Consumer.

Prefixed process name registration

To make our supervision tree hermetic, we can prefix all processes with some common key and make it an init parameter of our supervisor.

defmodule PipelineSupervisor do
  use Supervisor

  @impl Supervisor
  def init(%{name_prefix: prefix}) do
    consumer_name = Module.concat([__MODULE__, prefix, Consumer])

    children = [
      {SomeConsumer, name: consumer_name}
      {SomeWorker, %{consumer: consumer_name}},
    ]

    Supervisor.init(children, strategy: :rest_for_one)
  end
end

Although it gets the job done, it clutters the process namespace and creates API that might seem awkward ("why do you need me to provide some prefix?").

Implicit registry using Supervisor.which_children function

If we pay closer attention to Supervisor we can notice that any supervisor is basically a registry itself. It naturally knows pids of all its processes and all processes have unique ids local to the supervisor.

Unfortunately (or luckily) supervisors are single-purpose building blocks and therefore lack convenient apis to lookup their children. However, we still have the way to use supervisors as registers using Supervisor.which_children function.

defmodule PipelineSupervisor do
  use Supervisor

  @impl Supervisor
  def init([]) do
    supervisor_pid = self()

    children = [
      SomeConsumer,
      {SomeWorker, %{parent_supervisor: supervisor_pid}},
    ]

    Supervisor.init(children, strategy: :rest_for_one)
  end

  def child_pid!(supervisor_pid, child_id) do
    spec = Supervisor.which_children(supervisor_pid)

    case List.keyfind(spec, child_id, 0) do
      {_id, pid, _type, _modules} when is_pid(pid) ->
        pid

      nil ->
        raise "no started child with id:#{id} in supervisor spec:#{inspect(spec)}"
    end
  end
end

defmodule SomeWorker do
  use GenServer

  @impl GenServer
  def init(%{supervisor_pid: supervisor_pid} = options) do
    {:ok, options}
  end

  defp do_process_data_chunk(data_chunk, state) do
    # ...
    consumer_pid = PipelineSupervisor.child_pid!(state.supervisor_pid, SomeConsumer)
    Process.send(consumer_pid, {:message_received, parsed_message})
    # ...
    end
  end
end

defmodule SomeConsumer do
  use GenServer

  def start_link([]) do
    GenServer.start_link(__MODULE__, [])
  end

end

In some cases calling a supervisor on each message might create a performance bottleneck. You can fix this by calling child_pid! once in Worker.init function and saving Consumer pid in GenServer state.

defmodule SomeWorker do
  use GenServer

  @impl GenServer
  def init(%{supervisor_pid: supervisor_pid} = options) do
    consumer_pid = PipelineSupervisor.child_pid!(state.supervisor_pid, SomeConsumer)
    {:ok, %{consumer_pid: consumer_pid}}
  end

  defp do_process_data_chunk(data_chunk, state) do
    # ...
    Process.send(state.consumer_pid, {:message_received, parsed_message})
    # ...
    end
  end
end

But when we save process pids, we should always be careful, and in our case, think about what would happen if Consumer crashes. In our case, everything will be fine because of the rest_for_one supervisor strategy. If Consumer crashes, its pid saved in Worker will become invalid. But our supervisor is configured with rest_for_one strategy, and Consumer is higher in its children list than Worker. Therefore when Consumer crashes, Worker will also be restarted and will discover pid of new Consumer.

Starting dependencies ad-hoc

Also, we can exploit the fact that Supervisor returns pid of started process in Supervisor.start_child function.

defmodule PipelineSupervisor do
  use Supervisor

  @impl Supervisor
  def init([]) do
    supervisor_pid = self()

    children = [
      {SomeWorker, %{parent_supervisor: supervisor_pid}},
    ]

    Supervisor.init(children, strategy: :rest_for_one)
  end

  def start_consumer do
    Supervisor.start_child(SomeConsumer)
  end
end

defmodule SomeWorker do
  use GenServer

  @impl GenServer
  def init(%{supervisor_pid: supervisor_pid} = options) do
    {:ok, options, {:continue, :start_worker}}
  end

  def handle_continue(:start_worker, state) do
    {:ok, consumer_pid} = PipelineSupervisor.start_consumer()
    {:noreply, Map.put(state, :consumer_pid, consumer_pid)}
  end

  defp do_process_data_chunk(data_chunk, state) do
    # ...
    Process.send(state.consumer_pid, {:message_received, parsed_message})
    # ...
    end
  end
end

defmodule SomeConsumer do
  use GenServer

  def start_link([]) do
    GenServer.start_link(__MODULE__, [])
  end

end

Note that we have to use :handle_continue callback to avoid deadlocking the supervisor. Otherwise, it would not be able to reply to our start_child call because it would be waiting init function to return.

This method of DI is convoluted and probably not worth it just for the sake of itself. But it can make more sense in more complicated scenarios where Worker instead of just sending messages to Consumer also needs to manage its lifetime (kill or restart).

Conclusion

Some of the options that I showed may seem complex and unnecessary. And in many cases that is entirely true. But my point is that using OTP is much more than just writing GenServers and occasionally some custom Supervisor or two. OTP gives us a pretty basic but very smart combination of base elements that we can use to form some higher-level usage patterns. Our task as a community is to discover those patterns and make them simple to use.

To find out more about supervisors and their usage patterns look at this great blogpost Using Supervisors to Organize Your Elixir Application