Исходный код Ускорение с помощью ETS

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

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

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

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

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

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, некоторые тесты по-прежнему будут проваливаться. Вы даже можете заметить, что тесты проходят и проваливаются непоследовательно между запусками. Например, тест "spawns buckets":

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 помогает нам управлять большими кодовыми базами.