この章では KV.Bucket というモジュールを作成する。このモジュールはkey-value形式でデータを保存でき、複数のプロセスからの読み書きをできるように実装する。

The trouble with state

Elixirはshared nothingでimmutableな言語である。もし状態を持つ何かを作成して複数のプロセスが保存したり読み込んだりしたい場合、Getting StartではProcessを永続化する例をみた。ElixirはErlang OTPを使うことでstateの管理を抽象化したモジュールを提供している。チュートリアルではこれらを一つずつ見ていく。
- Agent
- GenServer
- GenEvent
- Task

Agents

Agentはシンプルにstateを管理することができ、単純にプロセスでstateを持ち続けたいだけであればとても使いやすい。iexを起動して試してみる。

# kv ディレクトリの下で
$ iex -S mix

# start_linkで空のListを持ったagentのプロセスが開始される
iex(1)> {:ok, agent} = Agent.start_link fn -> [] end
{:ok, #PID<0.94.0>}

# AgentのPIDを指定して中で持っているlistの先頭に"eggs"を追加する
iex(2)> Agent.update(agent, fn list -> ["eggs"|list] end)
:ok

# 取り出すと追加した "eggs" が返る
iex(3)> Agent.get(agent, fn list -> list end)
["eggs"]

# stopに渡すとAgentのプロセスは終了する
iex(4)> Agent.stop(agent)
:ok

# プロセスは死んでいるのでListは取り出せない
iex(5)> Agent.get(agent, fn list -> list end)
** (exit) exited in: GenServer.call(#PID<0.94.0>, {:get, #Function<6.54118792/1 in :erl_eval.expr/5>}, 5000)
    ** (EXIT) no process
    (elixir) lib/gen_server.ex:356: GenServer.call/3

次にAgentを KV.Bucket に組み込んでいく。その前にtest/kv/bucket_test.exs を作成してBucketのテストを書く。
Bucketの機能としてはKeyを指定してBucketから値を取り出す機能とBucketにKeyを指定してValueを保存する機能を実装するので、それを検証するテストを書けばよい。

defmodule KV.BucketTest do

  # asyncオプションは他のテストと並行で動くという指定。
  # ファイルへの書き込みやDBへの書き込みなど、競合する可能性がある場合は指定しない
  use ExUnit.Case, async: true

  test "stores values by key" do
    {:ok, bucket} = KV.Bucket.start_link

    # 存在しないkeyを指定したらnilが返ること
    assert KV.Bucket.get(bucket, "milk") == nil

    # keyを指定してvalueを保存して値を取り出せること
    KV.Bucket.put(bucket, "milk", 3)
    assert KV.Bucket.get(bucket, "milk") == 3
  end
end

次にAgentを実装したBucketを lib/kv/bucket.ex として作成する
Agentに保存する形式はMapではなくHashDictを使う。Mapは大量のKeyを扱う場合にパフォーマンスが悪い。
&HashDict.get(&1, key) は & をつけることでHashDict.get/3をキャプチャしている。

defmodule KV.Bucket do
  @doc """
  Starts a new bucket.
  """
  def start_link do
    Agent.start_link(fn -> HashDict.new end)
  end

  @doc """
  Gets a value from the `bucket` by `key`.
  """
  def get(bucket, key) do
    Agent.get(bucket, &HashDict.get(&1, key))
  end

  @doc """
  Puts the `value` for the given `key` in the `bucket`.
  """
  def put(bucket, key, value) do
    Agent.update(bucket, &HashDict.put(&1, key, value))
  end
end

ここまででデータを保存するBucketとそのテストができているので mix test を走らせてみる。通るようになっているはず。

ExUnit callbacks

次に進む前にExUnitのCallbackについて触れる。Bucketのテストには必ずプロセスが開始されていなければならない。テストの中では {:ok, bucket} = KV.Bucket.start_link の部分だが、これを毎回書くのはしんどいので、ExUnitはテスト前に必ず実行される処理を書くためのsetupというMacroを用意している。テストを以下のように書き直してみる。

defmodule KV.BucketTest do
  # asyncオプションは他のテストと並行で動くという指定。
  # ファイルへの書き込みやDBへの書き込みなど、競合する可能性がある場合は指定しない
  use ExUnit.Case, async: true

  # setup Macroは各テストの前に必ず実行される
  setup do
    {:ok, bucket} = KV.Bucket.start_link
    {:ok, bucket: bucket}
  end

  test "stores values by key", %{bucket: bucket} do
    # 存在しないkeyを指定したらnilが返ること
    assert KV.Bucket.get(bucket, "milk") == nil

    # keyを指定してvalueを保存して値を取り出せること
    KV.Bucket.put(bucket, "milk", 3)
    assert KV.Bucket.get(bucket, "milk") == 3
  end
end

テストの前にsetupが走り、開始された状態のbucketをテストが受け取るようになった。これでテストごとにstart_linkを呼ぶ必要が無くなった。このように繰り返し必要になる前処理はsetupで書いておくとよい。
setup以外のコールバックは ExUnit v1.0.5 Documentation を参照

Other agent actions

Bucketには更新以外にも削除の機能が必要なので、BucketのdeleteをAgent.get_and_update/2を使って実装する。
HashDict.pop/2を使っているのでkeyを削除すると同時にその時の値が返る。

@doc """
Deletes `key` from `bucket`.

Returns the current value of `key`, if `key` exists.
"""
def delete(bucket, key) do
  Agent.get_and_update(bucket, &HashDict.pop(&1, key))
end

テストも書いておく

  test "deletes values key", %{bucket: bucket} do

    # 存在しないキーを削除しようとしたらnilが返ること
    assert KV.Bucket.delete(bucket, "cheese") == nil

    # 値を保存
    KV.Bucket.put(bucket, "cheese", 1)

    # 指定したkeyで値を削除できること。削除の際、今持っている値が返ること
    assert KV.Bucket.delete(bucket, "cheese") == 1
    assert KV.Bucket.get(bucket, "cheese") == nil
  end

Client/Server in agents

次の章に行く前に今書いたBucketの課題について考えてみる。Bucketをサーバーとみたてると、複数クライアントからアクセスが来た場合、仮にkey-valueの操作に時間がかかる場合、クライアントの待ち時間が発生する可能性がある。delete関数を以下のように拡張してみる。

def delete(bucket, key) do
  :timer.sleep(1000) # puts client to sleep
  Agent.get_and_update(bucket, fn dict ->
    :timer.sleep(1000) # puts server to sleep
    HashDict.pop(dict, key)
  end)
end

こうするとdeleteのたびに2000milsecの待ち時間が発生し、アクセスの数だけ待ち時間が増え続けることになる。
次の章で GenServers を使い、サーバーとクライアントをより明確に分離してみる。

もやもやエンジニア

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

Elixir 入門 ~ MIX AND OTP ~ その2 - Agent

The trouble with state

Agents

ExUnit callbacks

Other agent actions

Client/Server in agents