もやもやエンジニア

IT系のネタで思ったことや技術系のネタを備忘録的に綴っていきます。フロント率高め。

Elixir 入門 ~ MIX AND OTP ~ その4 - Supervisor and Application

Elixir OTP Mix
  • この章ではElixirのポリシーである“fail fast” と “let it crash”を実現しているsupervisorについて学ぶ。

Our first supervisor

  • Supervisor behaviour を使ってsupervisorを実装したモジュール lib/kv/supervisor.ex を作成する。
defmodule KV.Supervisor do
  use Supervisor

  def start_link do
    Supervisor.start_link(__MODULE__, :ok)
  end

  # attributeで子プロセス名を定義
  @manager_name KV.EventManager
  @registry_name KV.Registry

  # GenServerと同様にstart_linkで呼ばれるコールバック
  def init(:ok) do
    # `supervisor`は`event manager` と`registry`の2つの子プロセスを管理している。
    # 子プロセスは他のプロセスがpidを知らなくてもアクセスできるように別名をつけておく。例えば子プロセスがクラッシュして立ち上げ直した場合にpidは変わるが別名をつけておけばそのままアクセスできる
    children = [
      worker(GenEvent, [[name: @manager_name]]),
      worker(KV.Registry, [@manager_name, [name: @registry_name]])
    ]

    # strategyは `:one_for_one` を指定してプロセスの管理を開始。この戦略はもし、子プロセスが死んだら新しいプロセスを作るというもの。
    supervise(children, strategy: :one_for_one)
  end
end
  • ファイルを作成したらiex -S mixで確認する
iex> KV.Supervisor.start_link
{:ok, #PID<0.100.0>}
iex> KV.Registry.create(KV.Registry, "shopping")
:ok
iex> KV.Registry.lookup(KV.Registry, "shopping")
{:ok, #PID<0.104.0>}
  • supervisorを開始すると同時に event manager と registry のプロセスが同時に開始された。なのでsuoervisor.start_linkのあとにbucketのcreateやlookupができている。

Understanding applications

  • いま、.appファイルは_build/dev/lib/kv/ebin/kv.ap に見つけることができる。
  • このファイルはErlangのシンタックスで書かれており、ErlangのカーネルやElixir自身、mix.exsで定義した依存関係にあるモジュールやそのバージョンなどの情報が含まれている。
  • mix.exsのapplication/0で返された値を使って.appをカスタマイズすることもできる。

Starting applications

  • iex -S mix run --no-start を使ってアプリケーションの開始・停止を手動で行ってみる。
iex(1)> Application.start(:kv)
:ok
iex(2)> Application.stop(:kv)
:ok

14:19:09.629 [info]  Application kv exited: :stopped
iex(3)> Application.stop(:logger)
:ok

=INFO REPORT==== 14-Sep-2015::14:19:16 ===
    application: logger
    exited: stopped
    type: temporary
iex(4)> Application.start(:kv)
{:error, {:not_started, :logger}}
  • 最初のstartはiex起動時に --no-startオプションがないと二重に起動しているという旨のエラーが返る。
  • 4行目のstartが失敗するのは依存関係にあるloggerが3行目で止められているから。このような場合は Application.ensure_all_started/1を使えば丸ごと立ち上げることができる。

The application callback

  • アプリケーションは起動時にcallbackを指定できる。callbackの戻り値は {:ok, pid}で、pidはsupervisorのpidでなければいけない。
  • callbackを実装するためにmix.exsを開いて以下のように編集する。
def application do
  [applications: [],
   mod: {KV, []}]
end
  • :modオプションがcallback時に起動するモジュールを指しており、そのモジュールは Application behaviour を実装していなければいけない。:mod に KV を指定したので次に KVに Application behaviour を実装する。lib/kv.exを開き、以下のように編集する。
defmodule KV do
  use Application

  # Application behaviourを実装するためには start/2 を実装する必要がある。
  def start(_type, _args) do
    KV.Supervisor.start_link
  end

  #  stop/1 を定義して停止の際にカスタムを入れることもできる
end
  • ここまでできたら iex -S mix を起動する。startですでにsupervisorのプロセスが開始されているのでKV.Supervisor.start_link をたたかなくてもすでにbucketの操作ができるようになっているはず。

Projects or applications?

  • Mixはproject と Application を区別する。mix.exsの内容に基づき、我々は :kv applicationを定義したMix Project を持っていると言える。あとの章ではどのような Applicationも定義しない Projectがでてくる。
  • チュートリアルの中で Project について話すときはMixについて考えるべきである。Mix は Projectを管理するツールで、どのようにProjectを編集し、テストし、関連した Applicationを開始するかを知っている。
  • ApplicationではOTPについて考えるべきである。Applicationはランタイムで開始して終了するエンティティで mix help compile.app で def application のオプションについて学ぶことができる。

Simple one for one supervisors

  • bucketのプロセスとregistryのプロセスはリンクしていてbucketのクラッシュはregistryのクラッシュと同意義である。registryがクラッシュした場合はsupervisorがregistryを復活させることが保証されているが、bucketはすべてなくなってしまう。bucketがクラッシュしてもregistryは生き続けるようにしてみよう。
  • 次のテストを追加する。
  test "removes bucket on crash", %{registry: registry} do
    KV.Registry.create(registry, "shopping")
    {:ok, bucket} = KV.Registry.lookup(registry, "shopping")

    # Kill the bucket and wait for the notification
    Process.exit(bucket, :shutdown)
    assert_receive {:exit, "shopping", ^bucket}
    assert KV.Registry.lookup(registry, "shopping") == :error
  end
  • “removes bucket on exit” テストと似ているが、Agent.stop/1 の代わりにshutdownの命令を送ってbucketのプロセス自体を破棄している。bucketのプロセスはregistryのプロセスとリンクしているのでこの行為はregistryのプロセスを破棄することに等しく、すなわちテストも失敗する。
  • この問題を解決するために新しい戦略 :simple_one_for_one を持ったsupervisorを定義する。このsupervisorはすべてのbucketを生み出し、管理する役目を負う。lib/kv/bucket/supervisor.exを新しく作成する。
defmodule KV.Bucket.Supervisor do
  use Supervisor

  def start_link(opts \\ []) do
    Supervisor.start_link(__MODULE__, :ok, opts)
  end

  # 受け取ったsupervisorの子プロセスとしてbucketを開始する関数
  # KV.Bucket.start_linkの代わりに呼び出すようになる
  def start_bucket(supervisor) do
    Supervisor.start_child(supervisor, [])
  end

  def init(:ok) do
    # restart: :temporaryはbucketが死んでも自動で再開しないことを明示している。
    # bucketはregistryを通してのみ管理されるようにする = start_bucket をでしか開始されない
    children = [
      worker(KV.Bucket, [], restart: :temporary)
    ]

    supervise(children, strategy: :simple_one_for_one)
  end
end
  • iex -S mixでBucket.Supervisorの動きを確認する。
iex(1)> {:ok, sup} = KV.Bucket.Supervisor.start_link
{:ok, #PID<0.90.0>}
iex(2)> {:ok, bucket} = KV.Bucket.Supervisor.start_bucket(sup)
{:ok, #PID<0.92.0>}
iex(3)> KV.Bucket.put(bucket, "eggs", 3)
:ok
iex(4)> KV.Bucket.get(bucket, "eggs")
3
  • 次に test/kv/registry_test.exs のsetupを編集して期待する動作を書く
  setup do
    # Bucketのsupervisorを開始（今回追加）
    {:ok, sup} = KV.Bucket.Supervisor.start_link
    # Event Managerを起動
    {:ok, manager} = GenEvent.start_link
    # EventManagerとBucketのsupervisorをレジストリに渡して起動（bucketのsupervisorのpidをargmentsに追加）
    {:ok, registry} = KV.Registry.start_link(manager, sup)

    GenEvent.add_mon_handler(manager, Forwarder, self())
    {:ok, registry: registry}
  end
  • 合わせて KV.Registry をテストが通るように編集
 ## Client API

  @doc """
  Starts the registry.
  """
  def start_link(event_manager, buckets, opts \\ []) do
    # 1 bucketのsupervisorのプロセスを受け取れるようにする
    GenServer.start_link(__MODULE__, {event_manager, buckets}, opts)
  end
  ## Server callbacks

  def init({events, buckets}) do
    names = HashDict.new
    refs  = HashDict.new
    # 2 start_linkで追加したbucketのsupervisorのプロセスをstateに追加する。
    {:ok, %{names: names, refs: refs, events: events, buckets: buckets}}
  end

  def handle_cast({:create, name}, state) do
    if HashDict.get(state.names, name) do
      {:noreply, state}
    else
      # 3 bucketのsupervisor経由で新しいbucketが作成されるようにする
      {:ok, pid} = KV.Bucket.Supervisor.start_bucket(state.buckets)
      ref = Process.monitor(pid)
      refs = HashDict.put(state.refs, ref, name)
      names = HashDict.put(state.names, name, pid)
      GenEvent.sync_notify(state.events, {:create, name, pid})
      {:noreply, %{state | names: names, refs: refs}}
    end
  end
  • ここまで追加できたらtestを通してみる。正しく動くようになっているはず。

Supervision trees

  • Bucketのsupervisorをアプリケーションで使うためには、supervisorを監督するsupervisorを作りsupervision treesを形成する必要がある。lib/kv/supervisor.exを編集する。
  @manager_name KV.EventManager
  @registry_name KV.Registry
  # bucketのsupervisorの別名を追加
  @bucket_sup_name KV.Bucket.Supervisor

  def init(:ok) do
    # childrenにbucketのsupervisorを追加し、registryのworkerのargumentsにbucketのsupervisorのpid(別名)を追加
    # bucketのsupervisorはregistryのworkerより前に定義しなければいけない。
    children = [
      worker(GenEvent, [[name: @manager_name]]),
      supervisor(KV.Bucket.Supervisor, [[name: @bucket_sup_name]]),
      worker(KV.Registry, [@manager_name, @bucket_sup_name, [name: @registry_name]])
    ]

    # この時点での :one_for_one 戦略は正しい。registryがクラッシュしたらbucketのsupervisorも死ぬべきだし、その逆も同じ。
    supervise(children, strategy: :one_for_one)
  end