Источник Ускорение с помощью ETS

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

В этой главе мы узнаем о ETS (Erlang Term Storage) и о том, как использовать его в качестве механизма кэширования.

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

ETS в качестве кэша

ETS позволяет нам хранить любые термы Elixir в таблице в оперативной памяти. Работа с таблицами ETS выполняется с помощью модуля Erlang :ets:

iex> table = :ets.new(:buckets_registry, [:set, :protected])
#Reference<0.1885502827.460455937.234656>
iex> :ets.insert(table, {"foo", self()})
true
iex> :ets.lookup(table, "foo")
[{"foo", #PID<0.41.0>}]

При создании таблицы ETS требуются два аргумента: имя таблицы и набор опций. Из доступных опций мы передали тип таблицы и правила доступа. Мы выбрали тип :set, что означает, что ключи не могут дублироваться. Мы также установили доступ к таблице :protected, что означает, что только процесс, создавший таблицу, может записывать в неё, но все процессы могут читать из неё. Возможные варианты управления доступом:

:public — Чтение/Запись доступны всем процессам.

:protected — Чтение доступно всем процессам. Запись только владельцу процесса. Это значение по умолчанию.

:private — Чтение/Запись ограничены процессом-владельцем.

Обратите внимание, что если ваш вызов Чтения/Записи нарушает правила доступа, операция вызовет ArgumentError. Наконец, поскольку :set и :protected являются значениями по умолчанию, мы их опустим.

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

iex> :ets.new(:buckets_registry, [:named_table])
:buckets_registry
iex> :ets.insert(:buckets_registry, {"foo", self()})
true
iex> :ets.lookup(:buckets_registry, "foo")
[{"foo", #PID<0.41.0>}]

Давайте изменим KV.Registry для использования таблиц ETS. Первое изменение — модификация нашего реестра для требования аргумента имени; мы будем использовать его для именования таблицы ETS и самого процесса реестра. Имена таблиц ETS и имена процессов хранятся в разных местах, поэтому конфликтов не возникнет.

Откройте lib/kv/registry.ex, и давайте изменим его реализацию. Мы добавили комментарии в исходный код, чтобы выделить внесенные нами изменения:

defmodule KV.Registry do
  use GenServer

  ## Client API

  @doc """
  Starts the registry with the given options.

  `:name` is always required.
  """
  def start_link(opts) do
    # 1. Pass the name to GenServer's init
    server = Keyword.fetch!(opts, :name)
    GenServer.start_link(__MODULE__, server, 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
    # 2. Lookup is now done directly in ETS, without accessing the server
    case :ets.lookup(server, name) do
      [{^name, pid}] -> {:ok, pid}
      [] -> :error
    end
  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

  ## Server callbacks

  @impl true
  def init(table) do
    # 3. We have replaced the names map by the ETS table
    names = :ets.new(table, [:named_table, read_concurrency: true])
    refs  = %{}
    {:ok, {names, refs}}
  end

  # 4. The previous handle_call callback for lookup was removed

  @impl true
  def handle_cast({:create, name}, {names, refs}) do
    # 5. Read and write to the ETS table instead of the map
    case lookup(names, name) do
      {:ok, _pid} ->
        {:noreply, {names, refs}}

      :error ->
        {:ok, pid} = DynamicSupervisor.start_child(KV.BucketSupervisor, KV.Bucket)
        ref = Process.monitor(pid)
        refs = Map.put(refs, ref, name)
        :ets.insert(names, {name, pid})
        {:noreply, {names, refs}}
    end
  end

  @impl true
  def handle_info({:DOWN, ref, :process, _pid, _reason}, {names, refs}) do
    # 6. Delete from the ETS table instead of the map
    {name, refs} = Map.pop(refs, ref)
    :ets.delete(names, name)
    {:noreply, {names, refs}}
  end

  @impl true
  def handle_info(_msg, state) do
    {:noreply, state}
  end
end

Обратите внимание, что до внесения изменений KV.Registry.lookup/2 отправлял запросы на сервер, но теперь он напрямую читает из таблицы ETS, которая совместно используется всеми процессами. В этом и заключается основная идея механизма кэширования, который мы реализуем.

Для работы механизма кэширования созданная таблица ETS должна иметь доступ :protected (значение по умолчанию), чтобы все клиенты могли читать из неё, а запись осуществлял только процесс KV.Registry. Мы также установили read_concurrency: true при запуске таблицы, оптимизируя её для общего сценария одновременных операций чтения.

Внесённые нами изменения нарушили наши тесты, потому что реестр требует опцию :name при запуске. Кроме того, некоторые операции реестра, такие как lookup/2, требуют указания имени в качестве аргумента вместо PID, чтобы можно было выполнить поиск по таблице ETS. Давайте изменим функцию настройки в test/kv/registry_test.exs, чтобы исправить обе проблемы:

  setup context do
    _ = start_supervised!({KV.Registry, name: context.test})
    %{registry: context.test}
  end

Поскольку у каждого теста уникальное имя, мы используем имя теста для именования наших реестров. Таким образом, нам больше не нужно передавать PID реестра, а мы можем идентифицировать его по имени теста. Также обратите внимание, что мы присвоили результат start_supervised! переменной нижнее подчеркивание (_). Этот приём часто используется для обозначения того, что нас не интересует результат start_supervised!.

После изменения setup некоторые тесты по-прежнему будут проваляться. Вы можете даже заметить, что результаты тестов непоследовательно проходят и проваливаются во время повторных запусков. Например, тест "создание бакетов":

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

может провалиться на этой строке:

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

Как эта строка может провалиться, если мы только что создали бакет в предыдущей строке?

Причина этих провалов заключается в том, что в учебных целях мы допустили две ошибки:

1. Мы слишком рано оптимизируем (добавляя этот слой кэширования)
2. Мы используем cast/2 (в то время как должны использовать call/2)

Условие гонки?

Разработка на Elixir не делает ваш код свободным от условий гонки. Однако абстракции Elixir, где по умолчанию ничего не делится, упрощают обнаружение причины условия гонки.

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

1. Мы вызываем KV.Registry.create(registry, "shopping")
2. Реестр создаёт бакет и обновляет таблицу кэша
3. Мы получаем информацию из таблицы с помощью KV.Registry.lookup(registry, "shopping")
4. Вышеприведённая команда возвращает {:ok, bucket}

Однако, так как KV.Registry.create/2 — операция преобразования, команда вернётся до того, как мы фактически запишем в таблицу! Другими словами, это происходит:

1. Мы вызываем KV.Registry.create(registry, "shopping")
2. Мы получаем информацию из таблицы с помощью KV.Registry.lookup(registry, "shopping")
3. Вышеприведённая команда возвращает :error
4. Реестр создаёт бакет и обновляет таблицу кэша

Для исправления провала нам нужно сделать KV.Registry.create/2 синхронным, используя call/2 вместо cast/2. Это гарантирует, что клиент будет продолжать только после внесения изменений в таблицу. Вернёмся к lib/kv/registry.ex и изменим функцию и её обратный вызов следующим образом:

def create(server, name) do
  GenServer.call(server, {:create, name})
end

@impl true
def handle_call({:create, name}, _from, {names, refs}) do
  case lookup(names, name) do
    {:ok, pid} ->
      {:reply, pid, {names, refs}}

    :error ->
      {:ok, pid} = DynamicSupervisor.start_child(KV.BucketSupervisor, KV.Bucket)
      ref = Process.monitor(pid)
      refs = Map.put(refs, ref, name)
      :ets.insert(names, {name, pid})
      {:reply, pid, {names, refs}}
  end
end

Мы изменили обратный вызов с handle_cast/2 на handle_call/3 и изменили его, чтобы он отвечал PID созданного бакета. Как правило, разработчики Elixir предпочитают использовать call/2 вместо cast/2, так как это также обеспечивает обратную связь — вы ждёте, пока не получите ответ. Использование cast/2 без необходимости также можно рассматривать как преждевременную оптимизацию.

Давайте снова запустим тесты. На этот раз мы передадим опцию --trace:

$ mix test --trace

Опция --trace полезна, когда ваши тесты зависают или возникают условия гонки, так как она запускает все тесты синхронно (async: true не имеет эффекта) и показывает подробную информацию о каждом тесте. Если вы запускаете тесты несколько раз, вы можете увидеть этот периодический провал:

  1) test removes buckets on exit (KV.RegistryTest)
     test/kv/registry_test.exs:19
     Assertion with == failed
     code:  assert KV.Registry.lookup(registry, "shopping") == :error
     left:  {:ok, #PID<0.109.0>}
     right: :error
     stacktrace:
       test/kv/registry_test.exs:23

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

В прошлый раз мы исправили условие гонки, заменив асинхронную операцию, cast, на синхронную call. К сожалению, обратный вызов handle_info/2, который мы используем для получения сообщения :DOWN и удаления записи из таблицы ETS, не имеет синхронного эквивалента. На этот раз нам нужно найти способ гарантировать, что реестр обработает уведомление :DOWN об окончании процесса бакета.

Простой способ сделать это — отправить синхронный запрос в реестр перед поиском бакета. Операция Agent.stop/2 является синхронной и возвращает только после завершения процесса бакета. Таким образом, после того как Agent.stop/2 вернётся, реестр получит сообщение :DOWN, но ещё может не обработать его. Чтобы гарантировать обработку сообщения :DOWN, мы можем сделать синхронный запрос. Поскольку сообщения обрабатываются в порядке, после того, как реестр ответит на синхронный запрос, сообщение :DOWN обязательно будет обработано.

Давайте сделаем это, создав "фиктивный" бакет, который представляет собой синхронный запрос, после Agent.stop/2 в обоих тестах "удаления" в test/kv/registry_test.exs:

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

    # Do a call to ensure the registry processed the DOWN message
    _ = KV.Registry.create(registry, "bogus")
    assert KV.Registry.lookup(registry, "shopping") == :error
  end

  test "removes bucket on crash", %{registry: registry} do
    KV.Registry.create(registry, "shopping")
    {:ok, bucket} = KV.Registry.lookup(registry, "shopping")

    # Stop the bucket with non-normal reason
    Agent.stop(bucket, :shutdown)

    # Do a call to ensure the registry processed the DOWN message
    _ = KV.Registry.create(registry, "bogus")
    assert KV.Registry.lookup(registry, "shopping") == :error
  end

Наши тесты теперь (всегда) должны проходить!

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

На практике, если вы столкнётесь с необходимостью реестра для динамических процессов, вы должны использовать модуль Registry, предоставляемый в рамках Elixir. Он предоставляет функциональность, аналогичную той, которую мы построили с помощью GenServer + :ets, а также позволяет выполнять одновременные операции записи и чтения. Он был протестирован на масштабируемость по всем ядрам, даже на машинах с 40 ядрами.

Далее давайте обсудим внешние и внутренние зависимости и как Mix помогает нам управлять большими кодовыми базами.