[Elixir] DynamicSupervisor 및 Registry를 사용하는 프로세스의 동적 생성 및 관리

Elixir의 주요 용례는 실시간 메시지 발송이라고 생각하지만, 이것을 갑작스러운 필요조건으로 한다면, 예를 들면 다음과 같다.

웹 소켓 연결에 메시지 보내기

MQTT 주제에 대한 PubSub

채팅방에 올라온 소식을 참가자에게 배포

이러한 실시 방식을 고려하면 어느 것이든 다음과 같은 처리가 필요하다.

세션으로 메시지를 전달해야 하는 시간에 대응하는 과정

생성된 프로세스를 {세션}의 ID를 키로 하고 값으로 설정된 Elixir의 KVS 등록

을 처리합니다.

ID를 키로 하는 명령을 받았을 때 KVS에서 대응하는 과정을 받아서 이 과정으로 메시지를 전달한다

세션}에 메시지를 보낼 필요가 없는 시간에 대응하는 처리를 종료

프로세스의 정상적인 예외와 상관없이 종료 시점에 KVS에서 제거

{세션} "WebSocket 연결", "MQTT 화제", "채팅방"모두 적용할 수 있습니다.
이 처리를 실현하기 위해 Elixir가 제공하는 표준 라이브러리는 Dynamic Supervisor와 Registry입니다.이 파일에는 샘플 코드를 쓰면서 다이나믹 슈퍼바이저와 리지스트리 사용법을 들여다본다.샘플 코드의 전체는https://github.com/keshihoriuchi/samples/tree/master/elixir/registry_and_dynamic_supervisor에 있다.Elixir1.12로 동작을 확인하고 있습니다.

이루어지다

우선 - sup 옵션을 사용하여 항목을 생성합니다.

$ mix new --sup session_manager
* creating README.md
* creating .formatter.exs
* creating .gitignore
* creating mix.exs
* creating lib
* creating lib/session_manager.ex
* creating lib/session_manager/application.ex
* creating test
* creating test/test_helper.exs
* creating test/session_manager_test.exs

Your Mix project was created successfully.
You can use "mix" to compile it, test it, and more:

    cd session_manager
    mix test

Run "mix help" for more commands.

다음 세 가지 이름을 정의하는 과정입니다.
이름:
개요
SessionManager.SessionWorker
앞의 예에서 생성 종료 대상 GenServer.
SessionManager.SessionSupervisor
Session Worker의 DynamicSupervisor를 모니터링합니다.
SessionManager.SessionRegistry
위의 예에서 등록 프로세스 KVS.Registry 를 사용합니다.lib/session_manager/application.ex에서 start/2의children를 다음과 같이 개작한다.
lib/session_manager/application.ex

def start(_type, _args) do
  children = [
    {DynamicSupervisor, name: SessionManager.SessionSupervisor, strategy: :one_for_one},
    {Registry, keys: :unique, name: SessionManager.SessionRegistry}
  ]

  opts = [strategy: :one_for_all, name: SessionManager.Supervisor]
  Supervisor.start_link(children, opts)
end

이것이 바로 Session Manager입니다.Session Supervisor 및 Session ManagerSession Registry는 Session Manager입니다.Supervisor 밑에서 생성됩니다.Elixir에서는 슈퍼바이저{<モジュール名>, <引数>} 원조를 주면<モジュール名>.start_link(<引数>) 호칭이 붙는다.Registrykeys: 옵션에 :unique 옵션이 지정되었지만 :duplicate 옵션도 지정할 수 있습니다.:unique와 같은 키로 등록할 수 있는 과정은 단일한 것으로 제한되며:duplicate일 경우 여러 개를 허용한다.
또opts의strategy:를 :one_for_all로 변경했다.다이나믹 슈퍼바이저와 리지스트리는 서로 일치성을 유지해야 하기 때문에 한쪽이 무너지고 다시 시작하면 다른 한쪽도 다시 시작하기를 원하기 때문이다.
다음은 Session Worker를 쓰십시오.
lib/session_manager/session_worker.ex

defmodule SessionManager.SessionWorker do
  use GenServer, restart: :temporary

  # (Dynamic)Supervisor向けのAPI
  def start_link({k, v}) do
    name = {:via, Registry, {SessionManager.SessionRegistry, k}}
    GenServer.start_link(__MODULE__, {k, v}, name: name)
  end

  # 以下2つClient向けのAPI
  def get_value(pid) do
    GenServer.call(pid, :getvalue)
  end

  def stop(pid) do
    GenServer.cast(pid, :stop)
  end

  # 以下3つGenServerのコールバック実装
  @impl true
  def init({k, v}) do
    {:ok, {k, v}}
  end

  @impl true
  def handle_call(:getvalue, _from, {k, v}) do
    {:reply, v, {k, v}}
  end

  @impl true
  def handle_cast(:stop, state) do
    {:stop, :normal, state}
  end
end

use GenServer, restart: :temporary의:temporary과정이 끝난 후 어떠한 상황에서도 다시 시작하지 않는다는 뜻이다.다이나믹 슈퍼바이저의 감시 대상인 GenServer는 대체로 이 옵션이 좋다고 생각합니다.기본값이면 어떤 경우에도 재부팅:permanent되므로 주의해야 합니다.
주다GenServer.start_link/3name:{:via, Registry, {SessionManager.SessionRegistry, k}}.즉, 생성 프로세스 중에 Session Registry에 k 키로 로그인합니다.
대체적으로 실현되었기 때문에 사용 방법을 나타내는 테스트 코드를 쓴다.
test/session_manager_test.exs

defmodule SessionManager.SessionManagerTest do
  use ExUnit.Case

  alias SessionManager.SessionSupervisor
  alias SessionManager.SessionRegistry
  alias SessionManager.SessionWorker

  test "regsiter and unregister" do
    {:ok, pid} = DynamicSupervisor.start_child(SessionSupervisor, {SessionWorker, {"k1", "v1"}})
    assert [{pid, nil}] == Registry.lookup(SessionRegistry, "k1")
    ref = Process.monitor(pid)
    assert SessionWorker.get_value(pid) == "v1"
    SessionWorker.stop(pid)
    assert_receive({:DOWN, ^ref, :process, ^pid, :normal})
    :ok = wait_until_process_removed(20)
  end

  defp wait_until_process_removed(0) do
    :error
  end

  defp wait_until_process_removed(n) do
    case Registry.lookup(SessionRegistry, "k1") do
      [] ->
        :ok

      [{_pid, nil}] ->
        Process.sleep(50)
        wait_until_process_removed(n - 1)
    end
  end
end

{:ok, pid} = DynamicSupervisor.start_child(SessionSupervisor, {SessionWorker, {"k1", "v1"}})에서 Session Supervisor의 감시 대상SessionWorker.start_link/1으로 부여{"k1", "v1"}됩니다.Session Manager.슈퍼바이저의 경우와 마찬가지로 여기서 주는application.ex의 원조도 {<モジュール名>, <引数>}라고 부른다.
이렇게 하면 <モジュール名>.start_link(<引数>)에서 k1을 키로 생성된 프로세스의pid를 얻을 수 있다.
프로세스를 중지하고 모니터를 받은 후 다시 [{pid, nil}] = Registry.lookup(SessionRegistry, "k1")하면Registry.lookup(SessionRegistry, "k1") 답장을 원하지만 그렇게 되지 않습니다.프로세스 중지와 Registry 제거 사이에는 대기 상태가 있습니다.이 때문에 테스트 코드는 고통스러웠지만 순환 후 줄곧 기다렸다[].
정지된pid를 lookup으로 줄일 때 처리는 프로그램의 요구에 의존합니다. 예를 들어 핑 메시지를 저장하고 pong 메시지에 답장하기 전에,receive가 모니터의 DOWN을 가져오면 lookup에서 다시 시작합니다. 이런 처리가 필요할 수도 있습니다.

참고 자료

공식 가이드 Mixand OPP

Dynamic supervisors - The Elixir programming language

에도 감시 대상 과정의 restart를 []로 설정한 해설

이 있다.

ETS - The Elixir programming language

이 절까지 ETS를 사용하는 과정 KVS의 실현 방법에 대해 설명하고 이 절의 마지막 공식 공연에서 Registry를 사용하는 것이 좋다는 취지를 적었다
In practice, if you find yourself in a position where you need a process registry for dynamic processes, you should use the Registry module provided as part of Elixir.

API 참조

DynamicSupervisor — Elixir v1.12.2

Registry — Elixir v1.12.2

Registrations 부분에 프로세스 정지와 Registry 삭제 사이에 시체가 존재한다고 적혀 있다