Распределенные задачи и конфигурация

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

Слой маршрутизации будет получать таблицу маршрутизации в следующем формате:

[{?a..?m, :"foo@computer-name"},
 {?n..?z, :"bar@computer-name"}]

Маршрутизатор будет искать первый байт имени корзины в таблице и передавать запрос нужному узлу. Например, если имя начинается с буквы «a» (?a представляет код буквы «a» в Юникоде), запрос будет передан узлу foo@computer-name.

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

Вы можете удивиться, почему мы не запрашиваем у найденного в таблице узла выполнение запроса напрямую, а вместо этого посылаем запрос на дальнейшую маршрутизацию этому узлу. Пока таблица маршрутизации такая простая, как показано выше, её логично использовать для всех узлов, но пересылка запросов маршрутизации позволяет гораздо проще разделить таблицу маршрутизации на небольшие части, когда приложение начинает расти. Возможно в какой-то момент foo@computer-name будет отвечать только за маршрутизацию запросов к хранилищу, а корзины будут храниться на разных узлах. При этом bar@computer-name не должен ничего знать о таких изменениях.

Замечание: Мы будем использовать оба узла на одной машине в этой главе. Вы можете использовать две (или больше) разных машины в одной сети, но для этого нужно будет сделать некоторые приготовления. Для начала нужно убедиться, что на машинах есть файл ~/.erlang.cookie с одинаковым значением. Далее необходимо, чтобы epmd был запущен на незаблокированном порту (вы можете запустить epmd -d для получения отладочной информации). И наконец, если вы хотите больше узнать о распределенности, мы рекомендуем замечательную главу «Distribunomicon» из книги «Learn You Some Erlang».

Наш первый распределённый код

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

Чтобы запустить распределённый код, нам нужно запустить виртуальную машину, задав ей имя. Имя может быть коротким (при расположении узлов в одной сети) или длинным (включающим полный адрес машины). Запустим новую IEx-сессию:

$ iex --sname foo

Вы можете увидеть, что приглашение строки ввода отличается и показывает имя узла после имени машины:

Interactive Elixir – press Ctrl+C to exit (type h() ENTER for help)
iex(foo@jv)1>

Компьютер в данном примере назван jv, поэтому можно увидеть foo@jv в примере выше, но вы получите другой результат. Мы будем использовать foo@computer-name в дальнейших примерах, а вам нужно будет изменить их в соответствии с именем своей машины для запуска кода.

Объявим модуль Hello в этом терминале:

iex> defmodule Hello do
...>   def world, do: IO.puts "hello world"
...> end

Если у вас есть другой компьютер в той же сети с установленными Эрлангом и Эликсиром, вы можете запустить на нём вторую IEx-сессию. Если нет, запустите её в другом терминале. В обоих случаях дайте ей короткое имя bar:

$ iex --sname bar

Обратите внимание, что внутри этой новой сессии у нас нет доступа к функции Hello.world/0:

iex> Hello.world
** (UndefinedFunctionError) undefined function: Hello.world/0
    Hello.world()

Однако, мы можем порождать новые процессы на foo@computer-name, находясь на bar@computer-name! Попробуйте (только укажите вместо @computer-name то имя, которое видите у себя):

iex> Node.spawn_link :"foo@computer-name", fn -> Hello.world end
#PID<9014.59.0>
hello world

Эликсир породил процесс на другом узле и вернул его pid. Затем код выполнился на другом узле, где существует функция Hello.world/0 и вызвал эту функцию. Обратите внимание, что результат «hello world» был выведен на текущем узле bar, но не на foo. Другими словами, сообщение было отправлено назад с foo на bar. Это происходит, потому что процесс, порождённый на другом узле (foo) всё ещё имеет лидера группы на текущем узле (bar). Мы кратко говорили о лидерах групп в главе «Ввод/вывод и файловая система».

Мы также можем посылать и принимать сообщения на pid, возвращённый Node.spawn_link/2, как обычно. Попробуем простой ping-pong пример:

iex> pid = Node.spawn_link :"foo@computer-name", fn ->
...>   receive do
...>     {:ping, client} -> send client, :pong
...>   end
...> end
#PID<9014.59.0>

iex> send pid, {:ping, self()}
{:ping, #PID<0.73.0>}

iex> flush()
:pong
:ok

Из нашего небольшого исследования мы можем заключить, что нужно использовать функцию Node.spawn_link/2 для порождения процессов на удалённом узле каждый раз, когда нам нужны распределенные вычисления. Однако, мы также выяснили в этом руководстве, что порождение процессов вне дерева супервизора следует максимально избегать, поэтому нам нужно найти другие варианты.

Есть три лучших альтернативы Node.spawn_link/2, которые можно использовать в нашем примере:

1. Можно использовать модуль :rpc из Эрланга для выполнения функций на удалённом узле. В оболочке bar@computer-name выше, вы можете вызвать :rpc.call(:"foo@computer-name", Hello, :world, []) и получите «hello world».

2. Можно запустить сервер на другом узле и посылать запросы через API модуля GenServer. Например, можно осуществить вызов к другому серверу через GenServer.call({name, node}, arg) или отправку PID удалённого процесса в качестве первого аргумента.

3. Можно исользовать задачи, которые мы изучили в предыдущей главе, т. к. они могут быть порождены и на локальном, и на удалённом узле.

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

На нашем уровне маршрутизации мы будем использовать задачи, но вы также можете попробовать остальные альтернативы.

Шаблон async/await

До сих пор мы использовали задачи, которые запускаются и работают в изоляции, игнорируя возвращаемые ими значения. Однако, иногда полезно запустить задачу для вычисления значения и прочитать этот результат после. Для этого в задачах есть шаблон async/await:

task = Task.async(fn -> compute_something_expensive end)
res  = compute_something_else()
res + Task.await(task)

Шаблон async/await предоставляет очень простой механизм параллельного вычисления значений. Кроме того, шаблон async/await может быть использован с тем же модулем Task.Supervisor, который мы использовали в предыдущих главах. Достаточно вызвать Task.Supervisor.async/2 вместо Task.Supervisor.start_child/2 и использовать Task.await/2 для чтения результата.

Распределённые задачи

Распределённые задачи – ровно то же самое, что контролируемые супервизором задачи. Единственная разница в том, что мы передаём имя узла супервизору при порождении задачи. Откройте файл lib/kv/supervisor.ex из приложения :kv. Давайте добавим супервизор задач как последнего потомка в дереве:

{Task.Supervisor, name: KV.RouterTasks},

Теперь запустим два именованных узла снова, но внутри приложения :kv:

$ iex --sname foo -S mix
$ iex --sname bar -S mix

Изнутри bar@computer-name мы теперь можем порождать задачи прямо на другом узле через супервизор:

iex> task = Task.Supervisor.async {KV.RouterTasks, :"foo@computer-name"}, fn ->
...>   {:ok, node()}
...> end
%Task{owner: #PID<0.122.0>, pid: #PID<12467.88.0>, ref: #Reference<0.0.0.400>}

iex> Task.await(task)
{:ok, :"foo@computer-name"}

Наша первая распределённая задача получает имя узла, на котором запускать задачу. Обратите внимание, что мы передали анонимную функцию в Task.Supervisor.async/2, но для распределённых случаев предпочтительно передавать модуль, функцию и аргументы явно:

iex> task = Task.Supervisor.async {KV.RouterTasks, :"foo@computer-name"}, Kernel, :node, []
%Task{owner: #PID<0.122.0>, pid: #PID<12467.89.0>, ref: #Reference<0.0.0.404>}

iex> Task.await(task)
:"foo@computer-name"

Разница в том, что анонимная функция обязывает иметь одинаковый код на узле выполнения и узле, который осуществляет вызов. Использование модуля, функции и аргументов более надёжный вариант, вам достаточно найти функцию, которая подходит по арности в переданном модуле.

С этими знаниями можно наконец написать код маршрутизации.

Слой маршрутизации

Создайте файл lib/kv/router.ex со следующим содержимым:

defmodule KV.Router do
  @doc """
  Направленяет данный `mod`, `fun`, `args` запрос
  на нужную ноду, основываясь на корзине `bucket`.
  """
  def route(bucket, mod, fun, args) do
    # Берёт первый байт из двоичных данных
    first = :binary.first(bucket)

    # Пытается найти запись в table() или выкидывает ошибку
    entry =
      Enum.find(table(), fn {enum, _node} ->
        first in enum
      end) || no_entry_error(bucket)

    # Если мы уже на искомой ноде
    if elem(entry, 1) == node() do
      apply(mod, fun, args)
    else
      {KV.RouterTasks, elem(entry, 1)}
      |> Task.Supervisor.async(KV.Router, :route, [bucket, mod, fun, args])
      |> Task.await()
    end
  end

  defp no_entry_error(bucket) do
    raise "could not find entry for #{inspect bucket} in table #{inspect table()}"
  end

  @doc """
  Таблица маршрутизации.
  """
  def table do
    # Замените имя компьютера на свои данные.
    [{?a..?m, :"foo@computer-name"},
     {?n..?z, :"bar@computer-name"}]
  end
end

Давайте напишем тест, чтобы убедиться, что наш маршрутизатор работает. Создайте файл test/kv/router_test.exs, содержащий:

defmodule KV.RouterTest do
  use ExUnit.Case, async: true

  test "route requests across nodes" do
    assert KV.Router.route("hello", Kernel, :node, []) ==
           :"foo@computer-name"
    assert KV.Router.route("world", Kernel, :node, []) ==
           :"bar@computer-name"
  end

  test "raises on unknown entries" do
    assert_raise RuntimeError, ~r/could not find entry/, fn ->
      KV.Router.route(<<0>>, Kernel, :node, [])
    end
  end
end

Первый тест вызывает функцию Kernel.node/0, которая возвращает имя текущего узла, основываясь на именах корзин «hello» и «world». Согласно нашей таблице маршрутизации, мы должны получить foo@computer-name и bar@computer-name в качестве ответов, соответственно.

Второй тест проверяет, что код падает на вводе неизвестных значений.

Для запуска первого теста нам нужно два запущенных узла. Перейдём в apps/kv и перезапустим узел с именем bar, чтобы использовать его в тестах.

$ iex --sname bar -S mix

И запустим тест следующим образом:

$ elixir --sname foo -S mix test

Тест должен пройти без ошибок.

Фильтры тестов и теги

Хотя наши тесты проходят, структура тестов становится более сложной. А именно, запуск тестов командой mix test приведёт к ошибкам, т. к. наш тест предусматривает подключение к другому узлу.

К счастью, в ExUnit есть средства для тегирования тестов, которые позволяют запускать определённые обратные вызовы или даже фильтровать тесты по этим тегам. Мы уже использовали тег :capture_log в предыдущей главе, семантику которого определяет сам ExUnit.

Теперь давайте добавим тег :distributed в файл test/kv/router_test.exs:

@tag :distributed
test "route requests across nodes" do

Формулировка @tag :distributed эквивалентна @tag distributed: true.

Когда тесты отмечены нужными тегами, мы можем проверить, запущен ли узел в сети, и, если нет, мы можем исключить все распределённые тесты. Откройте test/test_helper.exs внутри приложения :kv и добавьте следующее:

exclude =
  if Node.alive?, do: [], else: [distributed: true]

ExUnit.start(exclude: exclude)

Теперь выполните команду mix test:

$ mix test
Excluding tags: [distributed: true]

.......

Finished in 0.1 seconds (0.1s on load, 0.01s on tests)
7 tests, 0 failures, 1 skipped

На этот раз все тесты прошл и ExUnit предупреждает нас, что распределённые тесты были исключены. Если вы запустите $ elixir --sname foo -S mix test, один дополнительный тест должен запуститься и проходить, пока узел bar@computer-name доступен.

Команда mix test также позволяет нам динамически включать и исключать теги. Например, мы можем запустить $ mix test --include distributed для запуска распределенных тестов независимо от значения в test/test_helper.exs. Мы также можем передать --exclude чтобы исключить тег. Наконец, --only можно использовать для запуска только тестов с определённым тегом:

$ elixir --sname foo -S mix test --only distributed

Вы можете прочитать больше о фильтрах, тегах и стандартных тегах в документации модуля ExUnit.Case.

Окружение приложения и конфигурация

До сих пор мы задавали таблицу маршрутизации статически в модуле KV.Router. Однако, мы бы хотели сделать таблицу динамической. Это позволит нам не только сконфигурировать окружение для разработки/тестирования/продакшена, но также позволит разным узлам запускаться с разными значениями в таблице маршрутизации. В OTP есть возможность задавать окружение приложения.

Каждое приложение имеет окружение, которое хранит конфигурацию приложения по ключу. Например, мы могли бы сохранить таблицу маршрутизации в окружении :kv, задать ему стандартные значения и позволить другим приложениям изменять таблицу, если необходимо.

Откройте apps/kv/mix.exs и измените функцию application/0:

def application do
  [extra_applications: [:logger],
   env: [routing_table: []],
   mod: {KV, []}]
end

Мы добавили новый ключ :env в приложение. Он возвращает стандартное окружение приложения, которое имеет ключ :routing_table и пустой список в качестве значения. Есть смысл в том, чтобы приложение изначально было с пустой таблицей, т. к. вид таблицы будет зависеть от структуры тестирования/разработки.

Для использования окружения в нашем коде, нам нужно заменить KV.Router.table/0 на код ниже:

@doc """
Таблица маршрутизации.
"""
def table do
  Application.fetch_env!(:kv, :routing_table)
end

Мы используем функцию Application.fetch_env!/2 для чтения записи :routing_table в окружени :kv. Вы можете найти больше информации и другие функции для работы с окружением приложения в модуле Application.

Т. к. наша таблица маршрутизации теперь пуста, распределённые тесты должны упасть. Перезапустите приложение и тесты, чтобы увидеть падение:

$ iex --sname bar -S mix
$ elixir --sname foo -S mix test --only distributed

Интересная особенность окружении приложения состоит в том, что оно может быть настроено не только для текущего приложения, но и для всех приложений. Такая конфигурация осуществляется в файле config/config/exs. Например, мы можем изменить стандартную IEx-консоль на другое значение. Просто откройте файл apps/kv/config/config.exs и добавьте в конец:

config :iex, default_prompt: ">>>"

Запустите IEx командой iex -S mis и вы сможете увидеть, что приглашение строки ввода изменилось.

Это значит, что мы также можем настроить нашу :routing_table прямо в файле apps/kv/config/config.exs:

# Замените имя компьютера на свои данные.
config :kv, :routing_table,
       [{?a..?m, :"foo@computer-name"},
        {?n..?z, :"bar@computer-name"}]

Перезапустите все узлы и запустите распределённые тесты снова. Теперь они снова должны пройти.

Начиная с Эликсира версии 1.2, все зонтичные приложения имеют общую конфигурацию, благодаря этой строке в config/config.exs в корне зонтичного проекта, которая загружает конфигурацию всем потомкам:

import_config "../apps/*/config/config.exs"

Команда mix run также принимает флаг --config, который позволяет файлам конфигурации быть переданными по требованию. Это можно использовать для запуска разных узлов, каждый из которых имеет отличающуюся конфигурацию (например, разные таблицы маршрутизации).

Более того, встроенная возможность конфигурировать приложения и тот факт, что мы сделали наш проект зонтичным, позволяет нам по-разному деплоить его. Мы можем:

• деплоить зонтичное приложение на узел, который будет и TCP-сервером, и хранилищем ключ-значение;

• деплоить приложение :kv_server для работы только в качестве TCP-сервера, а в таблице маршрутизации указать другие узлы;

• деплоить только приложение :kv, когда мы хотим узел с хранилищем (без TCP-доступа).

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

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

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

• изменить приложение :kv_server для использования порта из окружения, вместо использования жестко заданного 4040;

• изменить и настроить приложение :kv_server для использования маршрутизации вместо прямых вызовов локального KV.Registry. Для тестирования :kv_server вы можете создать таблицу маршрутизации, которая указывает на текущий узел.

Заключение

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

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

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

Приятного кодинга!