Elixirライブラリ「Pattern」のDispatcherを読む

# はじめに
先日、@ryo33さんと [Pattern](https://gitlab.com/cizen/pattern) の機能拡張をペアプロで行いました。そのときの風景はYouTubeにて公開されています。

- [ペアプロYouTuber第一話「Elixirで書かれたやばいマクロを読む](https://www.youtube.com/watch?v=WEWZoZkWKmE)

この記事では上記の動画で触れなかった Pattern の Dispatcher を解説します。

# Patternとは
Pattern とは[Cizen](https://gitlab.com/cizen/cizen)から`Cizen.Filter`の機能を抽出し、リファクタリングや機能拡張を図ったものです。

## Cizenとは
Cizenはオートマトンと呼ばれるプロセスを使ってアプリケーションを構築するためのイベントハンドリングライブラリです。
@ryo33さんが書いたQiita記事、[Elixirでやばいマクロ（Pattern）を書いた理由【イベント購読】](https://qiita.com/ryo33/items/5baedf6dcb66d72f303c#cizenとは)から引用します。

> https://hexdocs.pm/cizen/readme.html
"Build highly concurrent, monitorable, and extensible applications with a collection of automata."

> めちゃくちゃ雑に説明すると、
1. プロセスをSagaという形式で統一的に管理し
2. Sagaはイベントを購読しイベントを配信することで他のSagaに影響を与えるような枠組みを提供するためのライブラリです。

>イベントの購読時にはCizen.Filterを使ってクエリを生成します。
近いうちにCizenからCizen.Filterを抽出したPatternでCizen.Filterを置き換える予定です。

## 使用例
Pattern は構造体や関数やパターンマッチが複合した柔軟な記法により、イベントをクエリすることができます。

イベントのクエリの前に単純なマッチの例を見ましょう。

```elixir
pattern = Pattern.new(%A{key1: "a"})
assert Pattern.match?(pattern, %A{key1: "a"})
refute Pattern.match?(pattern, %B{key1: "a"})
```

関数やパターンマッチも使ってみましょう。

```elixir
b = "b"
pattern = Pattern.new(fn
            %A{key1: a} when is_atom(a) -> a == :a
            %A{key1: ^b} -> true
          end)
refute Pattern.match?(pattern, %A{key1: "a"})
assert Pattern.match?(pattern, %A{key1: "b"})
assert Pattern.match?(pattern, %A{key1: :a})
```

次に Dispatcher を用いたイベントのクエリの例を見ましょう。

```elixir
Dispatcher.register(pid, Pattern.new(fn %Event{body: %TestEventA{}} -> true end), "ref1")
Dispatcher.register(pid, Pattern.new(fn %Event{body: %TestEventA{}} -> true end), "ref2")
Dispatcher.register(pid, Pattern.new(fn %Event{body: %TestEventB{}} -> true end), "ref3")
dispatch = Dispatcher.dispatch(pid, %Event{body: %TestEventA{}})
assert MapSet.new(["ref1", "ref2"]) == MapSet.new(dispatch)
```

`Dispatcher.dispatch/2`の第二引数で指定したクエリに対する適切な購読のリファレンスが帰ってきていることが確認できます。

Dispatcher の要件としてスケールすること、つまり購読数に対し線形探索よりも効率の良いアルゴリズムを用いる必要があります。実際の Dispatcher では`pattern`を前置記法で表現した命令のツリーに変換して格納しています。この設計の話は上司の書いたQiita記事にて解説されています。

- [Elixirでやばいマクロ（Pattern）を書いた理由【イベント購読】](https://qiita.com/ryo33/items/5baedf6dcb66d72f303c)

# Dispatcher のコード
解説はこのような流れで進めていきます。

1. `Pattern.Dispatcher.register/3`の第二引数で渡される`pattern`の正体
2. `Pattern.Dispatcher`がしていること
3. `Pattern.Dispatcher.Node`がしていること
    1. `Pattern.Dispatcher.Node`の構造
    2. `put`、`delete`したとき
    3. `get`したとき

## `Pattern.Dispatcher.register/3`の第二引数で渡される`pattern`の正体
Dispatcher本体の解説の前に`pattern`がどういうものなのかを見てみましょう。`pattern`とは`Pattern.new/1`したときに生成される構造体です。引数である構造体や無名関数は、木構造に入れやすく遅延評価できる前置記法に変換され、`code`フィールドに格納されます。

```elixir
iex> Pattern.new(%A{})                                               
%Pattern{
  code: {:and, [is_map: [access: []], ==: [{:access, [:__struct__]}, A]]}
}
iex> Pattern.new(%A{key1: "a"})                                                                                      
%Pattern{
  code: {:and,
   [
     and: [is_map: [access: []], ==: [{:access, [:__struct__]}, A]],
     ==: [{:access, [:key1]}, "a"]
   ]}
}
```

## `Pattern.Dispatcher`がしていること
`Pattern.Dispatcher`は`GenServer`を実装する形でAPIを提供しています。`init/1`初期値となるルートノードを生成し`register/3`、`unregister/3`、`dispatch/2`から`Pattern.Dispatch.Node`の関数を呼び出しています。

以下は`Pattern.Dispatcher`の抜粋です。

```elixir
defmodule Pattern.Dispatcher do
  # ...
  use GenServer

  @spec start_link([GenServer.option()]) :: GenServer.on_start()
  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts)
  end

  @spec register(GenServer.server(), Pattern.t(), any) :: :ok
  def register(name, pattern, ref) do
    GenServer.cast(name, {:put, pattern, ref})
  end
  # ...
  @impl GenServer
  def init(_) do
    {:ok, Node.new()}
  end

  @impl GenServer
  def handle_cast({:put, pattern, ref}, state) do
    {:noreply, Node.put(state, pattern.code, ref)}
  end
  # ...
  @impl GenServer
  def handle_call(:get, _from, state) do
    {:reply, state, state}
  end
end
```

## `Pattern.Dispatcher.Node`がしていること
### `Pattern.Dispatcher.Node`の構造
Nodeの型は次のように定義されています。

```elixir
@type t :: %__MODULE__{subscriptions: MapSet.t(), operations: %{Code.t() => %{term => t}}}
```

- `subscriptions` : ノードに紐づけされている購読のマップセット
- `operations` : `%{子ノードのキーを見つけるためのコード => %{キー => 子ノード}}`

いくつか`pattern`を追加してみます。

```elixir
iex> Dispatcher.register(pid, Pattern.new(%{}), "ref1") # => :ok
iex> Dispatcher.register(pid, Pattern.new(%A{foo: "a"}), "ref2") # => :ok
iex> Dispatcher.register(pid, Pattern.new(%A{foo: "b"}), "ref3") # => :ok
iex> Dispatcher.register(pid, Pattern.new(%B{bar: "c"}), "ref4") # => :ok
iex> Dispatcher.register(pid, Pattern.new(%B{bar: "c"}), "ref5") # => :ok
iex> Dispatcher.register(pid, Pattern.new(fn "e" -> true end), "ref6") # => :ok
iex> Dispatcher.register(pid, Pattern.any([Pattern.new(%A{}), Pattern.new(%B{})]), "ref7") # => :ok

iex> GenServer.call(pid, :get)                                      
%Pattern.Dispatcher.Node{
  operations: %{
    {:access, []} => %{
      "e" => %Pattern.Dispatcher.Node{
        operations: %{},
        subscriptions: #MapSet<["ref6"]>
      }
    },
    {:is_map, [access: []]} => %{
      true: %Pattern.Dispatcher.Node{
        operations: %{
          {:access, [:__struct__]} => %{
            A => %Pattern.Dispatcher.Node{
              operations: %{
                {:access, [:foo]} => %{
                  "a" => %Pattern.Dispatcher.Node{
                    operations: %{},
                    subscriptions: #MapSet<["ref2"]>
                  },
                  "b" => %Pattern.Dispatcher.Node{
                    operations: %{},
                    subscriptions: #MapSet<["ref3"]>
                  }
                }
              },
              subscriptions: #MapSet<["ref7"]>
            },
            B => %Pattern.Dispatcher.Node{
              operations: %{
                {:access, [:bar]} => %{
                  "c" => %Pattern.Dispatcher.Node{
                    operations: %{},
                    subscriptions: #MapSet<["ref4", "ref5"]>
                  }
                }
              },
              subscriptions: #MapSet<["ref7"]>
            }
          }
        },
        subscriptions: #MapSet<["ref1"]>
      }
    }
  },
  subscriptions: #MapSet<[]>
}
```

これを図で表現するとこのようになります。
![pattern_tree.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132844/cd11fa7f-5974-42b3-fe8f-dab814f99316.png)

### `put`、`delete`したとき
`Node.put/3`、`Node.delete/3`は[継続渡しスタイル (CPS: Continuation-passing style)](https://ja.wikipedia.org/wiki/%E7%B6%99%E7%B6%9A%E6%B8%A1%E3%81%97%E3%82%B9%E3%82%BF%E3%82%A4%E3%83%AB)で書かれた`run/2`を呼び出しています。

- [lib/pattern/dispatcher/node.ex](https://gitlab.com/cizen/pattern/-/blob/master/lib/pattern/dispatcher/node.ex)

#### `:and` と `:or`
コードが`{:and, [left, right]}`や`{:or, [left, right]}`の場合を見てみましょう。

```elixir
  defp run(node, {:update, {:and, [left, right]}, continuation}) do
    run(node, {:update, left, {:update, right, continuation}})
  end

  defp run(node, {:update, {:or, [left, right]}, continuation}) do
    node
    |> run({:update, left, continuation})
    |> run({:update, right, continuation})
  end
```
`:and`は`left`の子ノードに`right`を吊り下げることでandを実現しています。

```elixir
{:and, [is_map: [access: []], ==: [{:access, [:__struct__]}, A]]}
```
![pattern_tree_and.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132844/0c61208c-faaa-d161-892d-bc1c2fcefb0a.png)

`:or`は`left`と`right`それぞれに対し`next`を渡すことでorを実現しています。

```elixir
{:or,
  [
    and: [is_map: [access: []], ==: [{:access, [:__struct__]}, A]],
    and: [is_map: [access: []], ==: [{:access, [:__struct__]}, B]]
  ]}
```
![pattern_tree_or.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132844/d7ac3115-e6e2-17bc-609d-7abb6d226121.png)

#### `update_operation/4`
`update_operation/4`は子ノードに対し`continuation`を渡してアップデートします。

```elixir
  defp update_operation(node, operation, value, continuation) do
    values = Map.get(node.operations, operation, %{})

    next_node =
      values
      |> Map.get(value, new())
      |> run(continuation)

    node = put_in(node.operations[operation], values)
    put_in(node.operations[operation][value], next_node)
  end
end
```

```elixir:update_operationの例
%Pattern.Dispatcher.Node{
  operations: %{
    {:access, [:__struct__]} => %{
      A => %Pattern.Dispatcher.Node{
        operations: %{},
        subscriptions: #MapSet<["ref1"]>
      }
    }
  },
  subscriptions: #MapSet<[]>
} |> update_operation({:access, [:__struct__]}, A, {:delete_subscription, "ref1"})
# => ref1が削除される
%Pattern.Dispatcher.Node{
  operations: %{
    {:access, [:__struct__]} => %{
      A => %Pattern.Dispatcher.Node{
        operations: %{},
        subscriptions: #MapSet<[]>
      }
    }
  },
  subscriptions: #MapSet<[]>
}
```

### `get`したとき

`Node.get/2`は`operations`を評価しながらノードを走査し、その際にたどったノードに紐付いている`subscriptions`をマージしていくことで購読を得ます。

`Node.get/2`の手順は以下の通りです。

1. 今いるノードの`operations`1つに対し`Pattern.eval`を実行し、子ノードのキーを得る
2. `operation`に紐付いている`nodes`から1.で得たキーを使って子ノードを得る
3. 子ノードで`get/2`を実行する
4. 今いるノードの`subscriptions`と、子ノードの`get/2`で帰ってきた`subscriptions`を結合する
5. 以上の手順をすべての`operations`で行う

もう少し詳しく説明します。`Node.get/2`のコードは以下の通りです。

```elixir
  @spec get(t, struct) :: MapSet.t()
  def get(node, struct) do
    %__MODULE__{subscriptions: subscriptions, operations: operations} = node

    Enum.reduce(operations, subscriptions, fn {operation, nodes}, subscriptions ->
      value = Pattern.eval(operation, struct)
      node = Map.get(nodes, value, new())
      MapSet.union(subscriptions, get(node, struct))
    end)
  end
```

`operation`には`{:access, [:key1]}`のようなコードが格納されています。`Pattern.eval`でそのコードを`struct`に対して評価し、子ノードのキーを得ます。

```elixir:Pattern.evalの例
assert A == Pattern.eval({:access, [:__struct__]}, %A{key1: "a"})
```

実際のNodeではコードと子ノードはこのような配置になっています。

```elixir
%Pattern.Dispatcher.Node{
  operations: %{
    {:access, [:__struct__]} => %{ # == nodes
      A => %Pattern.Dispatcher.Node{ # 子ノード
        # ...
      },
      B => # ...
    },
    # ...
  },
  # ...
}
```

次の行では先ほど得た子ノードのキーをつかって`nodes`の中から子ノードを得ます。キーが見つからない場合は空のNodeを作成します。

```elixir
node = Map.get(nodes, value, new())
```

最後の行では`subscriptions`と「子ノードの`get/2`で得られる`subscriptions`」をマージします。

```elixir
MapSet.union(subscriptions, get(node, struct))
```

実際に下のようなクエリを実行すると、図のように探索します。

```elixir
iex> Dispatcher.dispatch(pid, %A{foo: "a"})
["ref1", "ref2", "ref7"]
```
![pattern_tree_get.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132844/a34e6a76-5e46-bc97-e0ee-934f2dcec483.png)

# 参考文献まとめ
- Pattern: [https://gitlab.com/cizen/pattern](https://gitlab.com/cizen/pattern)
- Cizen Project: [https://gitlab.com/cizen](https://gitlab.com/cizen)
- @ryo33さんの記事: [https://qiita.com/ryo33/items/5baedf6dcb66d72f303c](https://qiita.com/ryo33/items/5baedf6dcb66d72f303c)
- CPS: [https://ja.wikipedia.org/wiki/継続渡しスタイル](https://ja.wikipedia.org/wiki/%E7%B6%99%E7%B6%9A%E6%B8%A1%E3%81%97%E3%82%B9%E3%82%BF%E3%82%A4%E3%83%AB)