Исходный код Взаимодействие клиент-сервер с GenServer

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

CREATE shopping
OK

PUT shopping milk 1
OK

GET shopping milk
1
OK

В приведенном выше сеансе мы взаимодействовали с корзиной "shopping".

Поскольку агенты являются процессами, каждая корзина имеет идентификатор процесса (PID), но корзины не имеют имени. В главе о процессах мы узнали, что можем регистрировать процессы в Elixir, присваивая им имена атомов:

iex> Agent.start_link(fn -> %{} end, name: :shopping)
{:ok, #PID<0.43.0>}
iex> KV.Bucket.put(:shopping, "milk", 1)
:ok
iex> KV.Bucket.get(:shopping, "milk")
1

Однако именование динамических процессов атомами — плохая идея! Если мы используем атомы, нам нужно будет преобразовать имя корзины (часто получаемое от внешнего клиента) в атомы, и никогда не следует преобразовывать пользовательский ввод в атомы. Это связано с тем, что атомы не подлежат сборке мусора. После создания атома он никогда не возвращается в свободное пространство. Генерация атомов из пользовательского ввода означает, что пользователь может ввести достаточно разных имен, чтобы исчерпать память нашей системы!

На практике, скорее всего, вы достигнете ограничения Erlang VM для максимального количества атомов, прежде чем исчерпаете память, что приведет к сбою вашей системы независимо.

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

Регистр должен гарантировать, что он всегда является актуальным. Например, если один из процессов корзины аварийно завершится из-за ошибки, регистр должен заметить это изменение и избегать предоставления устаревших записей. В Elixir мы говорим, что регистр должен отслеживать каждую корзину. Поскольку нашему регистру нужно быть способным получать и обрабатывать произвольные сообщения от системы, API Agent недостаточно.

Мы будем использовать GenServer для создания процесса-регистра, который может отслеживать процессы корзин. GenServer предоставляет функциональные возможности промышленного уровня для построения серверов как в Elixir, так и в OTP.

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

Обратные вызовы GenServer

GenServer — это процесс, который вызывает ограниченный набор функций при определенных условиях. Когда мы использовали Agent, мы бы сохраняли и клиентский, и серверный код рядом, как это:

def put(bucket, key, value) do
  Agent.update(bucket, &Map.put(&1, key, value))
end

Давайте разделим этот код на части:

def put(bucket, key, value) do
  # Here is the client code
  Agent.update(bucket, fn state ->
    # Here is the server code
    Map.put(state, key, value)
  end)
  # Back to the client code
end

В приведенном выше коде у нас есть процесс, который мы называем «клиентом», отправляющий запрос агенту «серверу». Запрос содержит анонимную функцию, которую должен выполнить сервер.

В GenServer этот код будет разделен на две отдельные функции, примерно так:

def put(bucket, key, value) do
  # Send the server a :put "instruction"
  GenServer.call(bucket, {:put, key, value})
end

# Server callback

def handle_call({:put, key, value}, _from, state) do
  {:reply, :ok, Map.put(state, key, value)}
end

В коде GenServer немного больше формальностей, но, как мы увидим, это приносит и определенные преимущества.

Сейчас мы напишем только обратные вызовы сервера для нашей логики регистрации корзин, не предоставляя надлежащего API, что мы сделаем позже.

Создайте новый файл в lib/kv/registry.ex со следующим содержимым:

defmodule KV.Registry do
  use GenServer

  ## Missing Client API - will add this later

  ## Defining GenServer Callbacks

  @impl true
  def init(:ok) do
    {:ok, %{}}
  end

  @impl true
  def handle_call({:lookup, name}, _from, names) do
    {:reply, Map.fetch(names, name), names}
  end

  @impl true
  def handle_cast({:create, name}, names) do
    if Map.has_key?(names, name) do
      {:noreply, names}
    else
      {:ok, bucket} = KV.Bucket.start_link([])
      {:noreply, Map.put(names, name, bucket)}
    end
  end
end

Существует два типа запросов, которые вы можете отправить GenServer: вызовы и броски. Вызовы являются синхронными, и сервер обязательно должен вернуть ответ на такие запросы. Пока сервер вычисляет ответ, клиент ждет. Броски асинхронны: сервер не вернет ответ, и, следовательно, клиент не будет ждать его. Оба запроса являются сообщениями, отправленными на сервер, и будут обработаны последовательно. В приведенной выше реализации мы используем шаблонную сопоставление сообщений :create, чтобы обрабатывать их как броски, и на сообщения :lookup, чтобы обрабатывать их как вызовы.

Для вызова приведенных выше обратных вызовов нам нужно пройти соответствующие функции GenServer. Давайте запустим реестр, создадим именованную корзину, а затем найдем ее:

iex> {:ok, registry} = GenServer.start_link(KV.Registry, :ok)
{:ok, #PID<0.136.0>}
iex> GenServer.cast(registry, {:create, "shopping"})
:ok
iex> {:ok, bk} = GenServer.call(registry, {:lookup, "shopping"})
{:ok, #PID<0.174.0>}

Наш KV.Registry процесс получил бросок с {:create, "shopping"} и вызов с {:lookup, "shopping"}, в таком порядке. GenServer.cast немедленно вернётся, как только сообщение будет отправлено на registry. В GenServer.call, с другой стороны, мы будем ждать ответа, предоставленного выше обратным вызовом KV.Registry.handle_call.

Вы, возможно, также заметили, что мы добавили @impl true перед каждым обратным вызовом. @impl true информирует компилятор о том, что наше намерение для последующего определения функции — определить обратный вызов. Если по какой-то причине мы допустим ошибку в имени функции или количестве аргументов, например, определим handle_call/2, компилятор предупредит нас, что нет никакого handle_call/2 для определения, и предоставит нам полный список известных обратных вызовов для модуля GenServer.

Это хорошо, но мы все равно хотим предложить нашим пользователям API, который позволит нам скрыть детали нашей реализации.

API клиента

GenServer реализуется в двух частях: API клиента и обратных вызовах сервера. Вы можете либо объединить обе части в один модуль, либо разделить их на клиентский и серверный модули. Клиент — это любой процесс, который вызывает клиентскую функцию. Сервер — это всегда идентификатор процесса или имя процесса, которое мы будем явно передавать в качестве аргумента API клиента. Здесь мы будем использовать один модуль как для обратных вызовов сервера, так и для API клиента.

Отредактируйте файл в lib/kv/registry.ex, заполнив пробелы для API клиента:

  ## Client API

  @doc """
  Starts the registry.
  """
  def start_link(opts) do
    GenServer.start_link(__MODULE__, :ok, opts)
  end

  @doc """
  Looks up the bucket pid for `name` stored in `server`.

  Returns `{:ok, pid}` if the bucket exists, `:error` otherwise.
  """
  def lookup(server, name) do
    GenServer.call(server, {:lookup, name})
  end

  @doc """
  Ensures there is a bucket associated with the given `name` in `server`.
  """
  def create(server, name) do
    GenServer.cast(server, {:create, name})
  end

Первая функция — start_link/1, которая запускает новый GenServer, передавая список опций. start_link/1 вызывает GenServer.start_link/3, который принимает три аргумента:

1. Модуль, в котором реализованы обратные вызовы сервера, в данном случае __MODULE__ (то есть текущий модуль)

2. Аргументы инициализации, в данном случае атом :ok

3. Список опций, которые можно использовать для указания таких вещей, как имя сервера. На данный момент мы передаем список опций, которые мы получаем в start_link/1 в GenServer.start_link/3

Следующие две функции, lookup/2 и create/2, отвечают за отправку этих запросов на сервер. В данном случае мы использовали {:lookup, name} и {:create, name} соответственно. Запросы часто задаются в виде кортежей, как это, чтобы обеспечить более одного «аргумента» в этом первом аргументе. Обычно действие, которое запрашивается, указывается как первый элемент кортежа, а аргументы для этого действия — в оставшихся элементах. Обратите внимание, что запросы должны совпадать с первым аргументом handle_call/3 или handle_cast/2.

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

Первым является обратный вызов init/1, который получает второй аргумент, переданный GenServer.start_link/3, и возвращает {:ok, state}, где состояние — это новая карта. Мы уже можем заметить, как API GenServer делает разделение клиент/сервер более очевидным. start_link/3 происходит на стороне клиента, в то время как init/1 — это соответствующий обратный вызов, который выполняется на сервере.

Для запросов call/2 мы реализуем обратный вызов handle_call/3, который получает request, процесс, с которого мы получили запрос (_from), и текущее состояние сервера (names). Обратный вызов handle_call/3 возвращает кортеж в формате {:reply, reply, new_state}. Первый элемент кортежа, :reply, указывает, что сервер должен отправить ответ клиенту. Второй элемент, reply, — это то, что будет отправлено клиенту, а третий, new_state — новое состояние сервера.

Для запросов cast/2 мы реализуем обратный вызов handle_cast/2, который получает request и текущее состояние сервера (names). Обратный вызов handle_cast/2 возвращает кортеж в формате {:noreply, new_state}. Обратите внимание, что в реальном приложении мы, вероятно, реализовали бы обратный вызов для :create с синхронным вызовом вместо асинхронного броска. Мы делаем это таким образом, чтобы проиллюстрировать, как реализовать обратный вызов броска.

Существуют и другие форматы кортежей, которые могут возвращать как обратные вызовы handle_call/3, так и handle_cast/2. Существуют и другие обратные вызовы, такие как terminate/2 и code_change/3, которые мы могли бы реализовать. Вы можете изучить полную документацию GenServer, чтобы узнать больше об этом.

Сейчас давайте напишем некоторые тесты, чтобы гарантировать, что наш GenServer работает как ожидается.

Тестирование GenServer

Тестирование GenServer мало чем отличается от тестирования агента. Мы запустим сервер в обратном вызове настройки и будем использовать его в наших тестах. Создайте файл в test/kv/registry_test.exs со следующим содержимым:

defmodule KV.RegistryTest do
  use ExUnit.Case, async: true

  setup do
    registry = start_supervised!(KV.Registry)
    %{registry: registry}
  end

  test "spawns buckets", %{registry: registry} do
    assert KV.Registry.lookup(registry, "shopping") == :error

    KV.Registry.create(registry, "shopping")
    assert {:ok, bucket} = KV.Registry.lookup(registry, "shopping")

    KV.Bucket.put(bucket, "milk", 1)
    assert KV.Bucket.get(bucket, "milk") == 1
  end
end

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

Существует одно важное различие между блоком setup, который мы написали для KV.Registry, и блоком, который мы написали для KV.Bucket. Вместо запуска реестра вручную с помощью вызова KV.Registry.start_link/1, мы вместо этого вызвали функцию ExUnit.Callbacks.start_supervised!/2, передав модуль KV.Registry.

Функция start_supervised! была встроена в наш тестовый модуль с помощью use ExUnit.Case. Она выполняет задачу запуска процесса KV.Registry, вызывая его функцию start_link/1. Преимущество использования start_supervised! заключается в том, что ExUnit гарантирует, что процесс реестра будет остановлен до начала следующего теста. Другими словами, это помогает гарантировать, что состояние одного теста не повлияет на следующий тест в случае, если они зависят от общих ресурсов.

При запуске процессов во время тестов мы всегда должны отдавать предпочтение start_supervised!. Рекомендуем изменить блок setup в bucket_test.exs на использование start_supervised! также.

Запустите тесты, и они должны пройти!

Необходимость отслеживания

Всё, что мы сделали до сих пор, можно было реализовать с помощью Agent. В этом разделе мы рассмотрим один из многих способов использования GenServer, который недоступен с Agent.

Начнём с теста, описывающего, как мы хотим, чтобы реестр вел себя, если бакетирование останавливается или терпит крах:

test "removes buckets on exit", %{registry: registry} do
  KV.Registry.create(registry, "shopping")
  {:ok, bucket} = KV.Registry.lookup(registry, "shopping")
  Agent.stop(bucket)
  assert KV.Registry.lookup(registry, "shopping") == :error
end

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

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

Сначала поработаем с мониторингом, запустив новую консоль с iex -S mix:

iex> {:ok, pid} = KV.Bucket.start_link([])
{:ok, #PID<0.66.0>}
iex> Process.monitor(pid)
#Reference<0.0.0.551>
iex> Agent.stop(pid)
:ok
iex> flush()
{:DOWN, #Reference<0.0.0.551>, :process, #PID<0.66.0>, :normal}

Обратите внимание, что Process.monitor(pid) возвращает уникальную ссылку, которая позволяет нам сопоставлять будущие сообщения с этой ссылкой для отслеживания. После остановки агента мы можем flush/0 все сообщения и заметить сообщение :DOWN с точной ссылкой, возвращённой monitor, уведомляющее о завершении процесса бакета с причиной :normal.

Давайте переиспользуем обратные вызовы сервера, чтобы исправить ошибку и заставить тест пройти. Сначала мы изменим состояние GenServer на два словаря: один, содержащий name -> pid, и другой, содержащий ref -> name. Затем нам нужно отслеживать бакеты в handle_cast/2 и реализовать обратный вызов handle_info/2 для обработки сообщений мониторинга. Полная реализация обратных вызовов сервера показана ниже:

## Server callbacks

@impl true
def init(:ok) do
  names = %{}
  refs = %{}
  {:ok, {names, refs}}
end

@impl true
def handle_call({:lookup, name}, _from, state) do
  {names, _} = state
  {:reply, Map.fetch(names, name), state}
end

@impl true
def handle_cast({:create, name}, {names, refs}) do
  if Map.has_key?(names, name) do
    {:noreply, {names, refs}}
  else
    {:ok, bucket} = KV.Bucket.start_link([])
    ref = Process.monitor(bucket)
    refs = Map.put(refs, ref, name)
    names = Map.put(names, name, bucket)
    {:noreply, {names, refs}}
  end
end

@impl true
def handle_info({:DOWN, ref, :process, _pid, _reason}, {names, refs}) do
  {name, refs} = Map.pop(refs, ref)
  names = Map.delete(names, name)
  {:noreply, {names, refs}}
end

@impl true
def handle_info(msg, state) do
  require Logger
  Logger.debug("Unexpected message in KV.Registry: #{inspect(msg)}")
  {:noreply, state}
end

Обратите внимание, что мы смогли значительно изменить реализацию сервера, не изменяя при этом ни одного из API-клиентов. Это одно из преимуществ явного разделения сервера и клиента.

Наконец, в отличие от других обратных вызовов, мы определили «всеобъемлющий» пункт для handle_info/2, который отбрасывает и регистрирует любые неизвестные сообщения. Чтобы понять почему, перейдём к следующему разделу.

call, cast или info?

До сих пор мы использовали три обратных вызова: handle_call/3, handle_cast/2 и handle_info/2. Вот что нужно учитывать при выборе каждого из них:

1. handle_call/3 необходимо использовать для синхронных запросов. Это должен быть выбор по умолчанию, так как ожидание ответа сервера — это полезный механизм обратной связи.

2. handle_cast/2 необходимо использовать для асинхронных запросов, когда вы не заботитесь о ответе. Бросок не гарантирует, что сервер получил сообщение, и по этой причине его следует использовать экономно. Например, функция create/2, которую мы определили в этой главе, должна была использовать call/2. Мы использовали cast/2 для дидактических целей.

3. handle_info/2 необходимо использовать для всех других сообщений, которые может получить сервер, которые не отправляются через GenServer.call/2 или GenServer.cast/2, включая обычные сообщения, отправленные с помощью send/2. Пример таких сообщений — сообщения мониторинга :DOWN.

Так как любое сообщение, включая те, что отправлены через send/2, попадает в handle_info/2, существует вероятность получения неожиданных сообщений сервером. Поэтому, если мы не определим всеобъемлющий пункт, эти сообщения могут привести к сбою реестра, так как ни одна строка не будет соответствовать. Тем не менее, о таких случаях не нужно беспокоиться для handle_call/3 и handle_cast/2. Вызовы и броски выполняются только через API GenServer, поэтому неизвестное сообщение, скорее всего, является ошибкой разработчика.

Чтобы помочь разработчикам запомнить различия между вызовами, бросками и информацией, поддерживаемые значения возврата и многое другое, у нас есть небольшая сводка по GenServer.

Мониторы или ссылки?

Мы ранее узнали о ссылках в главе Процессы. Теперь, когда реестр завершён, вы можете задаться вопросом: когда использовать мониторы, а когда ссылки?

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

Вернувшись к нашей реализации handle_cast/2, вы можете видеть, что реестр связывает и отслеживает бакеты:

{:ok, bucket} = KV.Bucket.start_link([])
ref = Process.monitor(bucket)

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