Надзорщик, запускающий несколько партиций одного и того же дочернего процесса.

В больших системах некоторые процессы могут стать узкими местами. Если состояние таких процессов можно тривиально разбить на партиции, при этом между ними нет зависимостей, то они могут использовать PartitionSupervisor для создания нескольких изолированных и независимых партиций.

После запуска PartitionSupervisor можно отправлять сообщения его дочерним процессам, используя {:via, PartitionSupervisor, {name, key}}, где name — имя PartitionSupervisor, а ключ используется для маршрутизации.

Этот модуль был представлен в Elixir v1.14.0.

Простой пример

Начнём с примера, который сам по себе не очень полезен, но показывает, как запускаются партиции и как сообщения маршрутизируются к ним.

Вот игрушечный GenServer, который просто собирает полученные сообщения. Он выводит их для наглядности.

defmodule Collector do
  use GenServer

  def start_link(args) do
    GenServer.start_link(__MODULE__, args)
  end

  def init(args) do
    IO.inspect([__MODULE__, " got args ", args, " in ", self()])
    {:ok, _initial_state = []}
  end

  def collect(server, msg) do
    GenServer.call(server, {:collect, msg})
  end

  def handle_call({:collect, msg}, _from, state) do
    new_state = [msg | state]
    IO.inspect(["current messages:", new_state, " in process", self()])
    {:reply, :ok, new_state}
  end
end

Чтобы запустить несколько таких процессов, мы можем запустить их под PartitionSupervisor, поместив это в нашу иерархию надзора:

{PartitionSupervisor,
  child_spec: Collector.child_spec([some: :arg]),
  name: MyApp.PartitionSupervisor
}

Мы можем отправлять им сообщения, используя "кортеж в качестве канала":

# The key is used to route our message to a particular instance.
key = 1
Collector.collect({:via, PartitionSupervisor, {MyApp.PartitionSupervisor, key}}, :hi)
# ["current messages:", [:hi], " in process", #PID<0.602.0>]
:ok
Collector.collect({:via, PartitionSupervisor, {MyApp.PartitionSupervisor, key}}, :ho)
# ["current messages:", [:ho, :hi], " in process", #PID<0.602.0>]
:ok

# With a different key, the message will be routed to a different instance.
key = 2
Collector.collect({:via, PartitionSupervisor, {MyApp.PartitionSupervisor, key}}, :a)
# ["current messages:", [:a], " in process", #PID<0.603.0>]
:ok
Collector.collect({:via, PartitionSupervisor, {MyApp.PartitionSupervisor, key}}, :b)
# ["current messages:", [:b, :a], " in process", #PID<0.603.0>]
:ok

Теперь перейдём к полезному примеру.

DynamicSupervisor Пример

DynamicSupervisor — это единственный процесс, отвечающий за запуск других процессов. В некоторых приложениях DynamicSupervisor может стать узким местом. Чтобы решить эту проблему, вы можете запустить несколько экземпляров DynamicSupervisor через PartitionSupervisor, а затем выбрать "случайный" экземпляр для запуска дочернего процесса.

Вместо запуска одного DynamicSupervisor:

children = [
  {DynamicSupervisor, name: MyApp.DynamicSupervisor}
]

Supervisor.start_link(children, strategy: :one_for_one)

и запуска дочерних процессов на этом динамическом надзорщике напрямую:

DynamicSupervisor.start_child(MyApp.DynamicSupervisor, {Agent, fn -> %{} end})

Вы можете запустить динамические надзорщики под PartitionSupervisor:

children = [
  {PartitionSupervisor,
   child_spec: DynamicSupervisor,
   name: MyApp.DynamicSupervisors}
]

Supervisor.start_link(children, strategy: :one_for_one)

и затем:

DynamicSupervisor.start_child(
  {:via, PartitionSupervisor, {MyApp.DynamicSupervisors, self()}},
  {Agent, fn -> %{} end}
)

В приведенном коде мы запускаем надзорщика партиций, который по умолчанию запустит динамического надзорщика для каждого ядра вашего компьютера. Затем, вместо вызова DynamicSupervisor по имени, вы вызываете его через надзорщика партиций с помощью формата {:via, PartitionSupervisor, {name, key}}. Мы выбрали self() в качестве ключа маршрутизации, что означает, что каждый процесс будет назначен одному из существующих динамических надзорщиков. Смотрите start_link/1, чтобы увидеть все поддерживаемые варианты PartitionSupervisor.

Примечания по реализации

PartitionSupervisor использует таблицу ETS или Registry для управления всеми партициями. Под капотом PartitionSupervisor генерирует спецификацию дочернего процесса для каждой партиции, а затем действует как обычный надзорщик. Идентификатор каждой спецификации дочернего процесса — номер партиции.

Для маршрутизации используются две стратегии. Если key — целое число, оно маршрутизируется, используя rem(abs(key), partitions), где partitions — количество партиций. В противном случае используется :erlang.phash2(key, partitions). Конкретная стратегия маршрутизации может измениться в будущем, поэтому на ней не стоит полагаться. Если вы хотите получить конкретный PID для определённого ключа, вы можете использовать GenServer.whereis({:via, PartitionSupervisor, {name, key}}).