Elixirで並行コマンド実行サーバーを作ったら感動した話

この記事はElixir Advent Calendar 2020の11日目です。

昨日は @zacky1972 さんの「[Elixir から Swift 5.3のコードを呼び出す方法(Autotoolsを使って / Apple Silicon M1チップにも対応)](https://qiita.com/zacky1972/items/4692e589bab7c84ef957)」でした。
明日は @MzRyuKa さんの「[書評：プログラミングElixir第2版](https://qiita.com/MzRyuKa/items/ed5c16c3ee1c7a102fba)」です。

---


**本記事では「Taskを使うとすごく簡単に並行処理が書ける！しかもサーバー化も簡単！Elixirすごい！」と感動したので、その作成過程を紹介しつつ、感動をお伝えできたらと思います。**

---

# Taskのおさらい

**[Task](https://hexdocs.pm/elixir/Task.html) 使ってますか？** ざっくり復習します。


## Task.asyncで別プロセスを生成できる
　
IExで、Process.sleepするとIExのプロセスがsleepするので入力できなくなります。

~~~elixir
iex()> Process.sleep(5000)
# 5秒間入力できない
~~~


Task.asyncを使うと別プロセスで処理ができる（つまり並行に処理できている）ので、入力が継続できます。

~~~elixir
iex()> require Logger
...()> Task.async(fn ->
...()>   Process.sleep(10000)
...()>   Logger.info("yey")
...()> end)
%Task{
  owner: #PID<0.111.0>,
  pid: #PID<0.144.0>,
  ref: #Reference<0.3890488730.554434561.238327>
}
...()> "saki ni yey" # 入力できます。
"saki ni yey"
iex()>
22:39:23.952 [info]  yey
~~~


## Task.awaitでTask.asyncの結果を受け取れる

Task.asyncが返すtask構造体をTask.awaitに渡すと、結果を受け取ることができます。

~~~elixir
iex()> task = Task.async(fn ->
...()>   Process.sleep(5000)
...()>   2+3
...()> end)
%Task{
  owner: #PID<0.111.0>,
  pid: #PID<0.137.0>,
  ref: #Reference<0.3890488730.554434561.238283>
}
iex()> Task.await(task)
5
~~~


# チョット工夫してみる

おさらいできたので、チョット工夫してみます。

Taskを使うと「mfa（module, function, arguments）のタプルのリスト」（以降コマンドリストと呼ぶ）を並行に実行してくれる関数を以下のように書くことができます。

~~~elixir
defmodule CommandExecutor do
  @spec do_async([tuple()]) :: [any()]
  def do_async(commands) when is_list(commands) do
    commands
    |> Enum.map(fn {m, f, a} -> Task.async(m, f, a) end) # コマンドリストは処理が開始されタスクのリストに
    |> Enum.map(fn task -> Task.await(task) end)         # タスクの結果を一つずつ回収
  end
end
# Task.async/1は匿名関数を、Task.async/3はmfaを処理することができます。
~~~


※コマンドの実行は並行でできますが、結果の回収は一番遅いtaskの処理に引っ張られます。
※do_asyncの呼び出し側は結果の回収が完了するまでブロックします。


# コマンド実行サーバーにしてみる

作ったdo_asyncを使って、メッセージとしてコマンドリストを投げたら、処理してくれるサーバーを作ってみます。
※リスト内のコマンドは並行処理しますが、リスト単位では逐次処理します。（GenServerは受け取ったメッセージを逐次処理する）

~~~elixir
defmodule CommandExecutor do
  use GenServer

  def start_link(state), do: GenServer.start_link(__MODULE__, state, name: __MODULE__)

  def init(state), do: {:ok, state}

  # 結果を返さない
  def handle_cast({:commands, commands, :async}, state) when is_list(commands) do
    do_async(commands)
    {:noreply, state}
  end

  # 結果を返す
  def handle_call({:commands, commands, :async}, _from, state) when is_list(commands) do
    {:reply, do_async(commands), state}
  end

  @spec do_async([tuple()]) :: [any()]
  def do_async(commands) when is_list(commands) do
    commands
    |> Enum.map(fn {m, f, a} -> Task.async(m, f, a) end)
    |> Enum.map(fn task -> Task.await(task) end)
  end
end
~~~


コマンドをぽいぽい投げ込めば、リスト単位で逐次処理してくれます。
※コマンドリストをたくさん投げる可能性がある場合は、GenServerのキュー（mailbox）にたまるので、その限界を知っておく必要があります。（知ってたら教えてくださいm(- -)m


# コマンド実行で例外が起きたら？

実行できないmfaが投げ込まれる可能性があるので対策しておきます。

~~~elixir
  @spec do_async([tuple()]) :: [any()]
  def do_async(commands) when is_list(commands) do
    commands
    |> Enum.map(fn {m, f, a} ->
         try do
           Task.async(m, f, a)
         rescue
           error in [UndefinedFunctionError] -> error
           other_error -> reraise("あちゃー", System.stacktrace)
         end
     end)
    |> Enum.map(fn task -> Task.await(task) end)
  end
~~~

でも、これだと予見できない例外でGenServerが落ちるかもしれません。


# 予見できない例外で落ちる場合はSupervisorに復帰してもらう

「僕に分かる対策はした、あとは頼んだよSupervisor君！！」という感じで、基本はお作法に従って書くだけです。

~~~elixir
defmodule CommandExecutorSupervisor do
  use Supervisor

  def start_link(start_state) do
    Supervisor.start_link(__MODULE__, start_state, name: __MODULE__)
  end

  def init(_start_state) do
    children = [{CommandExecutor, nil}]
    Supervisor.init(children, strategy: :one_for_one)
  end
end
~~~

※復帰動作の細かい設定をしたい場合は[ドキュメント](https://hexdocs.pm/elixir/Supervisor.html)を読みましょう。

# Applicationに組み込む

アプリ−ケーションの起動時からコマンド実行サーバーが動作するようにApplicationに組み込みます。

~~~elixir
defmodule Oreore.Application do
  use Application

  def start(_type, _args) do
    children = [..., {CommandExecutorSupervisor, []}]
    Supervisor.start_link(children, strategy: :one_for_one)
  end
end
~~~

できました。


**うぉお、めっちゃ簡単、並行コマンド実行サーバーできた！**

# おわり

雑にさらさら進めましたが、ポイントは以下でした。

* Taskを使うことで、スレッドを意識せずに並行処理が書ける！
* GenServerのメッセージキューに頼ることで、自身でキューの実装をせずにすむ！
* SupervisorにGenServerを監視してもらえるので自動復帰も余裕！

**Elixir, Erlang OTPすごいですね！！これからもElixirやっていきましょう♪♪**