Реестр

Локальное, децентрализованное и масштабируемое хранилище пар «ключ-значение» для процессов.

Он позволяет разработчикам искать один или несколько процессов по заданному ключу. Если реестр имеет :unique ключи, ключ указывает на 0 или 1 процесс. Если реестр допускает :duplicate ключи, один ключ может указывать на любое количество процессов. В обоих случаях разные ключи могут идентифицировать один и тот же процесс.

Каждая запись в реестре связана с процессом, который зарегистрировал ключ. Если процесс аварийно завершается, ключи, связанные с этим процессом, автоматически удаляются. Все сравнения ключей в реестре выполняются с помощью операции соответствия (===).

Реестр может использоваться для различных целей, таких как поиск по имени (используя :via опцию), хранение свойств, настройка правил диспетчеризации или реализация системы публикации-подписки. Мы рассмотрим некоторые из этих вариантов использования ниже.

Реестр также может быть прозрачно разбит на части, что обеспечивает более масштабируемое поведение для работы реестров в высококонкурентных средах с тысячами или миллионами записей.

Использование в :via

После запуска реестра с заданным именем (используя Registry.start_link/2) его можно использовать для регистрации и доступа к именованным процессам с помощью кортежа {:via, Registry, {registry, key}}.

{:ok, _} = Registry.start_link(:unique, Registry.ViaTest)
name = {:via, Registry, {Registry.ViaTest, "agent"}}
{:ok, _} = Agent.start_link(fn -> 0 end, name: name)
Agent.get(name, & &1)
#=> 0
Agent.update(name, & &1 + 1)
Agent.get(name, & &1)
#=> 1

Обычно реестр запускается как часть дерева надзора:

supervisor(Registry, [:unique, Registry.ViaTest])

Только реестры с уникальными ключами могут быть использованы в :via. Если имя уже занято, функция start_link, специфичная для данного случая (Agent.start_link/2 в примере выше), вернёт {:error, {:already_started, current_pid}}.

Использование в качестве диспетчера

Registry имеет механизм диспетчеризации, который позволяет разработчикам реализовывать пользовательскую логику диспетчеризации, вызываемую из вызывающей стороны. Например, предположим, что у нас есть дублируемый реестр, запущенный следующим образом:

{:ok, _} = Registry.start_link(:duplicate, Registry.DispatcherTest)

Вызывая register/3, разные процессы могут регистрироваться под заданным ключом и связывать любое значение с этим ключом. В этом случае давайте зарегистрируем текущий процесс под ключом "hello" и присоединим кортеж {IO, :inspect} к нему:

{:ok, _} = Registry.register(Registry.DispatcherTest, "hello", {IO, :inspect})

Теперь сущность, заинтересованная в диспетчеризации событий для данного ключа, может вызвать dispatch/3, передав ключ и обратный вызов. Этот обратный вызов будет вызван со списком всех значений, зарегистрированных под запрошенным ключом, вместе с pid процесса, который зарегистрировал каждое значение, в форме кортежей {pid, value}. В нашем примере value будет кортежем {module, function} в приведённом коде:

Registry.dispatch(Registry.DispatcherTest, "hello", fn entries ->
  for {pid, {module, function}} <- entries, do: apply(module, function, [pid])
end)
# Prints #PID<...> where the pid is for the process that called register/3 above
#=> :ok

Диспетчеризация происходит в процессе, который вызывает dispatch/3, либо последовательно, либо параллельно в случае нескольких разделов (через запущенные задачи). Зарегистрированные процессы не участвуют в диспетчеризации, если их участие не выполняется явно (например, отправкой им сообщения в обратном вызове).

Кроме того, если при диспетчеризации произойдёт сбой из-за неправильной регистрации, диспетчеризация всегда завершится неудачей, и зарегистрированный процесс не будет уведомлён. Поэтому давайте убедимся, что мы по крайней мере обернём и сообщим об этих ошибках:

require Logger
Registry.dispatch(Registry.DispatcherTest, "hello", fn entries ->
  for {pid, {module, function}} <- entries do
    try do
      apply(module, function, [pid])
    catch
      kind, reason ->
        formatted = Exception.format(kind, reason, System.stacktrace)
        Logger.error "Registry.dispatch/3 failed with #{formatted}"
    end
  end
end)
# Prints #PID<...>
#=> :ok

Вы также можете заменить всю систему apply явным отправлением сообщений. Это пример, который мы рассмотрим далее.

Использование в качестве PubSub

Реестры также могут использоваться для реализации локальной, нераспределённой, масштабируемой системы PubSub, опираясь на функцию dispatch/3, аналогично предыдущему разделу: в этом случае, однако, мы будем отправлять сообщения каждому ассоциированному процессу, а не вызывать заданную функцию модуля.

В этом примере мы также установим количество разделов равным количеству запущенных планировщиков, что сделает реестр более эффективным в высококонкурентных средах, так как каждый раздел запустит новый процесс, позволяя диспетчеризации происходить параллельно:

{:ok, _} = Registry.start_link(:duplicate, Registry.PubSubTest,
                               partitions: System.schedulers_online)
{:ok, _} = Registry.register(Registry.PubSubTest, "hello", [])
Registry.dispatch(Registry.PubSubTest, "hello", fn entries ->
  for {pid, _} <- entries, do: send(pid, {:broadcast, "world"})
end)
#=> :ok

В приведённом выше примере сообщение {:broadcast, "world"} было транслировано всем процессам, зарегистрированным под «темой» (или «ключом», как мы называли его до сих пор) "hello".

Третий аргумент, переданный в register/3, представляет собой значение, связанное с текущим процессом. Хотя в предыдущем разделе мы использовали его при диспетчеризации, в этом конкретном примере нас это не интересует, поэтому мы установили его в пустой список. При необходимости вы можете сохранить более значимое значение.

Регистрации

Поиск, диспетчеризация и регистрация эффективны и мгновенны за счёт отложенной отписки. Например, если процесс аварийно завершается, его ключи автоматически удаляются из реестра, но изменения могут не распространяться мгновенно. Это означает, что некоторые операции могут возвращать процессы, которые уже завершены. Когда это может произойти, это будет явно указано в документации функции.

Однако, имейте в виду, что такие случаи обычно не являются проблемой. В конце концов, процесс, на который ссылается pid, может аварийно завершиться в любое время, в том числе между получением значения из реестра и отправкой ему сообщения. Многие части стандартной библиотеки разработаны для обработки этого, например, Process.monitor/1, которая передаст сообщение :DOWN немедленно, если отслеживаемый процесс уже завершился, и Kernel.send/2, которая действует как команда бездействия для завершенных процессов.

ETS

Обратите внимание, что реестр использует одну таблицу ETS плюс две таблицы ETS на раздел.

Сводка

Типы

key()

Тип ключей, разрешённых при регистрации

kind()

Тип реестра

meta_key()

Тип ключей метаданных реестра

meta_value()

Тип значений метаданных реестра

registry()

Идентификатор реестра

value()

Тип значений, разрешённых при регистрации

Функции

dispatch(registry, key, mfa_or_fun)

Вызывает обратный вызов со всеми записями под key в каждом разделе для данного registry

keys(registry, pid)

Возвращает известные ключи для данного pid в registry в произвольном порядке

lookup(registry, key)

Находит пару {pid, value} для данного key в registry в произвольном порядке

match(registry, key, pattern)

Возвращает пары {pid, value} под заданным key в registry , которые соответствуют pattern

meta(registry, key)

Считывает метаданные реестра, заданные в start_link/3

put_meta(registry, key, value)

Хранит метаданные реестра

register(registry, key, value)

Регистрирует текущий процесс под заданным key в registry

start_link(kind, registry, options \\ [])

Запускает реестр как процесс надзирателя

unregister(registry, key)

Удаляет все записи для данного key , связанные с текущим процессом в registry

update_value(registry, key, callback)

Обновляет значение для key для текущего процесса в уникальном registry

Типы

key()

key() :: term()

Тип ключей, разрешённых при регистрации

kind()

kind() :: :unique | :duplicate

Тип реестра

meta_key()

meta_key() :: atom() | tuple()

Тип ключей метаданных реестра

meta_value()

meta_value() :: term()

Тип значений метаданных реестра

registry()

registry() :: atom()

Идентификатор реестра

value()

value() :: term()

Тип значений, разрешённых при регистрации

Функции

dispatch(registry, key, mfa_or_fun)

dispatch(registry(), key(), (entries :: [{pid(), value()}] -> term())) :: :ok

Вызывает обратный вызов со всеми записями под key в каждом разделе для данного registry.

Список entries представляет собой непустой список кортежей из двух элементов, где первый элемент — pid, а второй — значение, связанное с pid. Если для данного ключа нет записей, обратный вызов никогда не вызывается.

Если реестр не разделён, обратный вызов вызывается в процессе, который вызывает dispatch/3. Если реестр разделён, обратный вызов вызывается параллельно по разделам путём запуска задачи, связанной с вызывающим процессом. Однако обратный вызов вызывается только в том случае, если для этого раздела есть записи.

См. документацию модуля для примеров использования функции dispatch/3 для создания пользовательской системы диспетчеризации или системы публикации-подписки.

keys(registry, pid)

keys(registry(), pid()) :: [key()]

Возвращает известные ключи для заданного pid в registry в произвольном порядке.

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

Примеры

Регистрация под уникальным регистром не допускает нескольких записей:

iex> Registry.start_link(:unique, Registry.UniqueKeysTest)
iex> Registry.keys(Registry.UniqueKeysTest, self())
[]
iex> {:ok, _} = Registry.register(Registry.UniqueKeysTest, "hello", :world)
iex> Registry.register(Registry.UniqueKeysTest, "hello", :later) # registry is :unique
{:error, {:already_registered, self()}}
iex> Registry.keys(Registry.UniqueKeysTest, self())
["hello"]

Однако такое возможно для дублирующих регистров:

iex> Registry.start_link(:duplicate, Registry.DuplicateKeysTest)
iex> Registry.keys(Registry.DuplicateKeysTest, self())
[]
iex> {:ok, _} = Registry.register(Registry.DuplicateKeysTest, "hello", :world)
iex> {:ok, _} = Registry.register(Registry.DuplicateKeysTest, "hello", :world)
iex> Registry.keys(Registry.DuplicateKeysTest, self())
["hello", "hello"]

lookup(registry, key)

lookup(registry(), key()) :: [{pid(), value()}]

Находит пару {pid, value} для заданного key в registry в произвольном порядке.

Пустой список, если совпадений нет.

Для уникальных регистров необходим поиск по одной секции. Для дублирующих регистров необходимо выполнить поиск по всем секциям.

Примеры

В примере ниже мы регистрируем текущий процесс и ищем его как из него самого, так и из других процессов:

iex> Registry.start_link(:unique, Registry.UniqueLookupTest)
iex> Registry.lookup(Registry.UniqueLookupTest, "hello")
[]
iex> {:ok, _} = Registry.register(Registry.UniqueLookupTest, "hello", :world)
iex> Registry.lookup(Registry.UniqueLookupTest, "hello")
[{self(), :world}]
iex> Task.async(fn -> Registry.lookup(Registry.UniqueLookupTest, "hello") end) |> Task.await
[{self(), :world}]

То же самое относится к дублирующим регистрам:

iex> Registry.start_link(:duplicate, Registry.DuplicateLookupTest)
iex> Registry.lookup(Registry.DuplicateLookupTest, "hello")
[]
iex> {:ok, _} = Registry.register(Registry.DuplicateLookupTest, "hello", :world)
iex> Registry.lookup(Registry.DuplicateLookupTest, "hello")
[{self(), :world}]
iex> {:ok, _} = Registry.register(Registry.DuplicateLookupTest, "hello", :another)
iex> Enum.sort(Registry.lookup(Registry.DuplicateLookupTest, "hello"))
[{self(), :another}, {self(), :world}]

match(registry, key, pattern)

match(registry(), key(), match_pattern :: atom() | tuple()) :: [{pid(), term()}]

Возвращает пары {pid, value} под заданным key в registry , которые соответствуют pattern.

Шаблон должен быть атомом или кортежем, который будет соответствовать структуре значения, хранящегося в регистре. Атом :_ может использоваться для игнорирования заданного значения или элемента кортежа, а «$1» может использоваться для временного назначения части шаблона переменной для последующего сравнения.

Пустой список будет возвращен, если совпадений нет.

Для уникальных регистров необходим поиск по одной секции. Для дублирующих регистров необходимо выполнить поиск по всем секциям.

Примеры

В примере ниже мы регистрируем текущий процесс под тем же ключом в дублирующем регистре, но с разными значениями:

iex> Registry.start_link(:duplicate, Registry.MatchTest)
iex> {:ok, _} = Registry.register(Registry.MatchTest, "hello", {1, :atom, 1})
iex> {:ok, _} = Registry.register(Registry.MatchTest, "hello", {2, :atom, 2})
iex> Registry.match(Registry.MatchTest, "hello", {1, :_, :_})
[{self(), {1, :atom, 1}}]
iex> Registry.match(Registry.MatchTest, "hello", {2, :_, :_})
[{self(), {2, :atom, 2}}]
iex> Registry.match(Registry.MatchTest, "hello", {:_, :atom, :_}) |> Enum.sort()
[{self(), {1, :atom, 1}}, {self(), {2, :atom, 2}}]
iex> Registry.match(Registry.MatchTest, "hello", {:"$1", :_, :"$1"}) |> Enum.sort()
[{self(), {1, :atom, 1}}, {self(), {2, :atom, 2}}]

meta(registry, key)

meta(registry(), meta_key()) :: {:ok, meta_value()} | :error

Считывает метаданные регистра, заданные на start_link/3.

В качестве ключей разрешены атомы и кортежи.

Примеры

iex> Registry.start_link(:unique, Registry.MetaTest, meta: [custom_key: "custom_value"])
iex> Registry.meta(Registry.MetaTest, :custom_key)
{:ok, "custom_value"}
iex> Registry.meta(Registry.MetaTest, :unknown_key)
:error

put_meta(registry, key, value)

put_meta(registry(), meta_key(), meta_value()) :: :ok

Хранит метаданные регистра.

В качестве ключей разрешены атомы и кортежи.

Примеры

iex> Registry.start_link(:unique, Registry.PutMetaTest)
iex> Registry.put_meta(Registry.PutMetaTest, :custom_key, "custom_value")
:ok
iex> Registry.meta(Registry.PutMetaTest, :custom_key)
{:ok, "custom_value"}
iex> Registry.put_meta(Registry.PutMetaTest, {:tuple, :key}, "tuple_value")
:ok
iex> Registry.meta(Registry.PutMetaTest, {:tuple, :key})
{:ok, "tuple_value"}

register(registry, key, value)

register(registry(), key(), value()) ::
  {:ok, pid()} |
  {:error, {:already_registered, pid()}}

Регистрирует текущий процесс под заданным key в registry.

Также должно быть указано значение, которое будет ассоциировано с этой регистрацией. Это значение будет извлекаться при отправке или выполнении поиска по ключу.

Эта функция возвращает {:ok, owner} или {:error, reason}. owner — это PID секции регистра, ответственной за PID. Владелец автоматически привязывается к вызывающему.

Если регистр имеет уникальные ключи, он вернёт {:ok, owner} , если ключ уже связан с PID, в противном случае он вернёт {:error, {:already_registered, pid}}.

Если регистр имеет дублирующие ключи, несколько регистраций текущего процесса под одним ключом разрешены.

Примеры

Регистрация под уникальным регистром не допускает нескольких записей:

iex> Registry.start_link(:unique, Registry.UniqueRegisterTest)
iex> {:ok, _} = Registry.register(Registry.UniqueRegisterTest, "hello", :world)
iex> Registry.register(Registry.UniqueRegisterTest, "hello", :later)
{:error, {:already_registered, self()}}
iex> Registry.keys(Registry.UniqueRegisterTest, self())
["hello"]

Однако такое возможно для дублирующих регистров:

iex> Registry.start_link(:duplicate, Registry.DuplicateRegisterTest)
iex> {:ok, _} = Registry.register(Registry.DuplicateRegisterTest, "hello", :world)
iex> {:ok, _} = Registry.register(Registry.DuplicateRegisterTest, "hello", :world)
iex> Registry.keys(Registry.DuplicateRegisterTest, self())
["hello", "hello"]

start_link(kind, registry, options \\ [])

start_link(kind(), registry(), options) ::
  {:ok, pid()} |
  {:error, term()} when options: [partitions: pos_integer(), listeners: [atom()], meta: [{meta_key(), meta_value()}]]

Запускает регистр как процесс-надзиратель.

Вручную его можно запустить так:

Registry.start_link(:unique, MyApp.Registry)

В дереве надзирателей вы напишите:

supervisor(Registry, [:unique, MyApp.Registry])

Для интенсивных задач регистр также может быть разбит на секции (указав опцию :partitions). Если требуется разбиение на секции, хорошим значением по умолчанию является количество секций, равное количеству доступных планировщиков:

Registry.start_link(:unique, MyApp.Registry, partitions: System.schedulers_online())

или:

supervisor(Registry, [:unique, MyApp.Registry, [partitions: System.schedulers_online()]])

Параметры

Регистр поддерживает следующие параметры:

• :partitions - количество секций в регистре. По умолчанию 1.
• :listeners - список именованных процессов, которые уведомляются о событиях :register и :unregister. Регистрируемый процесс должен контролироваться слушателем, если слушатель хочет быть уведомлён о сбоях зарегистрированного процесса.
• :meta - ключевой список метаданных, которые будут добавлены к регистру.

unregister(registry, key)

unregister(registry(), key()) :: :ok

Удаляет все записи для заданного key , связанного с текущим процессом в registry.

Всегда возвращает :ok и автоматически разрывает связь текущего процесса с владельцем, если больше нет ключей, связанных с текущим процессом. См. также register/3, чтобы узнать больше о «владельце».

Примеры

Удаляет все записи для key для уникальных регистров:

iex> Registry.start_link(:unique, Registry.UniqueUnregisterTest)
iex> Registry.register(Registry.UniqueUnregisterTest, "hello", :world)
iex> Registry.keys(Registry.UniqueUnregisterTest, self())
["hello"]
iex> Registry.unregister(Registry.UniqueUnregisterTest, "hello")
:ok
iex> Registry.keys(Registry.UniqueUnregisterTest, self())
[]

Так же, как и для дублирующих регистров:

iex> Registry.start_link(:duplicate, Registry.DuplicateUnregisterTest)
iex> Registry.register(Registry.DuplicateUnregisterTest, "hello", :world)
iex> Registry.register(Registry.DuplicateUnregisterTest, "hello", :world)
iex> Registry.keys(Registry.DuplicateUnregisterTest, self())
["hello", "hello"]
iex> Registry.unregister(Registry.DuplicateUnregisterTest, "hello")
:ok
iex> Registry.keys(Registry.DuplicateUnregisterTest, self())
[]

update_value(registry, key, callback)

update_value(registry(), key(), (value() -> value())) ::
  {new_value :: term(), old_value :: term()} |
  :error

Обновляет значение для key для текущего процесса в уникальном registry.

Возвращает кортеж {new_value, old_value} или :error , если для текущего процесса такой ключ не назначен.

Если задан не уникальный регистр, возникает ошибка.

Примеры

iex> Registry.start_link(:unique, Registry.UpdateTest)
iex> {:ok, _} = Registry.register(Registry.UpdateTest, "hello", 1)
iex> Registry.lookup(Registry.UpdateTest, "hello")
[{self(), 1}]
iex> Registry.update_value(Registry.UpdateTest, "hello", & &1 + 1)
{2, 1}
iex> Registry.lookup(Registry.UpdateTest, "hello")
[{self(), 2}]