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