Исходный код Взаимодействие клиент-сервер с 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, bucket} = 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. Вызовы и броски выполняются только через GenServer API, поэтому неизвестное сообщение, скорее всего, является ошибкой разработчика.

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

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

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

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

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

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

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