Исходный код Взаимодействие клиент-сервер с 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 агентов недостаточно.

Мы будем использовать 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)

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