ZEAM開発ログv0.1.0 Flow / GenStage による並列プログラミング入門

（この記事は、[「Elixir or Phoenix Advent Calendar 2017」](https://qiita.com/advent-calendar/2017/elixir-or-phoenix)の6日目です）

昨日は @twinbee さんの[ElixirのGenStageに入門する #1](https://qiita.com/twinbee/private/12a61863ceef794996e0)でした。

[「ZEAM開発ログ 目次」はこちら](https://qiita.com/zacky1972/items/70593ab2b70d192813df)

# はじめに
 
ZACKY です。

今日は「ZEAM開発ログv0.1.0 Flow / GenStage による並列プログラミング入門」です。先日の[「fukuoka.ex #8 ～2018年 春のElixir入学式～」](https://fukuokaex.connpass.com/event/85038/)で発表した[「fukuoka.ex 入学式 / ZEAM開発ログ 第2回: 並列Elixirプログラミングことはじめ〜ZEAMで目指すことは何か」](https://zeam-vm.github.io/zeam-fukuoka.ex-20180420/#/)を発展させた内容を連載コラム化していきます。


@piacere さんから福岡Elixirコミュニティ[fukuoka.ex](https://fukuokaex.connpass.com)に誘われて，2018年の2月にElixirプログラミングを始めました。そこでElixirに大いなる可能性に魅力を感じ，大学に務める研究者としての10年くらいかけて取り組む研究テーマとして選んだ次第です。
![fuuoka.ex](https://camo.qiitausercontent.com/f2d8fd672d9ae47ef45876a8f896222558b12857/68747470733a2f2f71696974612d696d6167652d73746f72652e73332e616d617a6f6e6177732e636f6d2f302f3135353432332f35346164633038632d343830362d356138372d376631662d6338616135353562386235302e706e67)

fukuoka.ex のデビュー作はこちらです。ZEAM(ZACKY's Elixir Abstract Machine)という処理系の構想について語りました。

[![fukuoka.ex#5-ZEAM.png](https://qiita-image-store.s3.amazonaws.com/0/55223/8959be0e-54bb-b0a6-2f2f-6b14f0e43092.png)](https://zeam-vm.github.io/zeam-fukuoka.ex-20170223/#/)

ではさっそく本題に入っていきましょう。

# 並行と並列の違い

まず**並行(concurrent)**と**並列(parallel)**の違いについて，押さえておきましょう。

![並行と並列の違い](https://qiita-image-store.s3.amazonaws.com/0/55223/dbbda4a0-4de7-6586-4af0-1898b4119621.png)

上図を見てください。

* 並列というのは，物理的にCPUやコアが複数あって，同時に複数のタスクが動いている状態を指します(上図の下)。この状態は並行でもあります。
* 一方，並列ではないが並行な状態というのは，物理的にはCPUやコアが1つだけで，ある瞬間はタスク1を，次の瞬間にはタスク2を実行するというように，高速でタスクを切り替えながら同時に複数のタスクをこなしているように見せている状態を指します(上図の上)。
* ちなみに並列でも並行でもない，タスクを1つずつ順番に動かす状態のことを逐次処理と言います。

今回 Elixir で行いたいのは，タスクを**並列**で動作させること，すなわち**並列プログラミング**です。

# Flow / GenStage を使って並列プログラミングしてみよう

Elixir で並列プログラミングを行う方法はいくつかありますが，今回はそのうちの Flow / GenStage を使う方法を紹介します。

```elixir
1..10
|> Flow.from_enumerable(max_demand: 1, stages: 10)
  # stages で並列実行する最大数を指定する
  # max_demand を指定しない，もしくは大きくしすぎると，並列実行してくれない
|> Flow.map(fn i -> Process.sleep(500); i end)
|> Enum.to_list
|> Enum.sort
```

* まず，`1..10`というのは1〜10の範囲を表します。これは`[1,2,3,4,5,6,7,8,9,10]`というリストと等価です。
* `Flow.from_enumerable`によって並列処理の単位に分割します。パラメータ`max_demand`と`stages`が肝心です。
  * `stages`は並列実行する最大数を指定します。`10`を指定しているので，最大10並列で動作します。
  * `max_demand`は Flow がリストを分割する際の1つあたりの要素の最大数を指定します。`1`を指定しているので，リストを1つずつに分割して並列処理します。
  * もし `max_demand` を指定しない場合には，Flow は要素をある程度まとめて分割しようとするので，入力が`1..10`のように10個しか要素が含まれないと，並列ではなく逐次処理になってしまいます。
  * 何でそんな動作をするかというと，やみくもに並列に実行するよりも，ある程度処理をまとめて逐次に処理した方が実行時間が短いことが多いからなんですね。
* `Flow.map` で処理を分配します。ここでは，`Process.sleep(500)`つまり0.5秒待つ処理を行います。
  * そのため，もし`max_demand`を指定しないと逐次処理となって全体で5秒待つ処理になります。
  * 一方，`max_demand`を指定すると並列処理となって全体で0.5秒待つだけで済みます。
* `Enum.to_list` でバラした処理をひとまとめのリストにします。
  * 並列処理すると実行順番がバラバラになるので，このときには順番がバラバラになります。
* そこで，`Enum.sort`でソートして一意に揃えます。

実行してみると，`max_demand`を指定した状態だと一瞬で処理が終了しますが，`max_demand`を外すと5秒待って終了します。

![Process.png](https://qiita-image-store.s3.amazonaws.com/0/55223/c3928e0f-a79a-7d80-4b45-cd52c6d6e40a.png)

Flowが実際にどのように並列処理をしているかを模式的に表したのが上図です。

1. まず`Flow.from_enumerable`で処理を分配します(一番上の階層)。
2. `Flow.map`を実行した時にGenStage に処理を引き渡します(上から2番目の階層)。
3. GenStage は1つ1つのタスクを Erlang VM のプロセスに割り当てます(上から3番目の階層)。
4. Erlang VM のプロセスは OS で CPU やコアに割り当てられます(一番下の階層)。

このように Flow はバックエンドで GenStage を使っているのですね。GenStage については昨日の @twinbee さんの[ElixirのGenStageに入門する #1](https://qiita.com/twinbee/items/12a61863ceef794996e0)に紹介されていますので，読んでみてください。

と，ここまでが[fukuoka.ex#8](https://fukuokaex.connpass.com/event/85038/)で語った内容です。

[![fukuoka.ex#8-ZEAM.png](https://qiita-image-store.s3.amazonaws.com/0/55223/4b84bc11-b0e2-c9ac-0f4f-8d3f9adf0261.png)](https://zeam-vm.github.io/zeam-fukuoka.ex-20180420/#/)


# Flow / GenStage を使ったベンチマーク〜整数によるロジスティック写像演算

ここからが新しい内容です。

スリープではなく[ロジスティック写像を使ってCPUで整数演算処理をぶん回すベンチマーク](https://github.com/zeam-vm/logistic_map)を作ってみました。

[ロジスティック写像についてはWikipediaの説明をご覧ください。](https://ja.wikipedia.org/wiki/ロジスティック写像)

![ロジスティック写像の分岐図](https://qiita-image-store.s3.amazonaws.com/0/55223/1625cd32-ea71-a4e6-cea3-3648df4eadf2.png)

ここでは整数演算にするために次の漸化式を用いました。

```math
X_{i+1} = \mu_{p}X_{i}(X_i+1) \mod p
```

$p$は任意の素数です。`mod`はあまりを求める演算です。$\mu_p$は1〜$p-1$の任意の数です。$X_1$を1〜$p-1$の任意の数で始め，再帰的に$X_n$を計算します。

出典は次の通りです。ちなみに著者は大学の同僚です。

T. Miyazaki, S. Araki, S. Uehara, and Y. Nogami, “A Study of an Automorphism on the Logistic Maps over Prime Fields,” Proc. of The 2014 International Symposium on Information Theory and its Applications (ISITA2014), pp.727-731, 2014.

ちなみにこの方法で求めた値をグラフにプロットしても下図のようにはなりません。これは，この方法はロジスティック写像ではあるものの，オリジナルの方法と幾何学的性質が異なるからです。

![ロジスティック写像の分岐図](https://qiita-image-store.s3.amazonaws.com/0/55223/1625cd32-ea71-a4e6-cea3-3648df4eadf2.png)

ではElixirで実装してみましょう。

```math
X_{i+1} = \mu_{p}X_{i}(X_i+1) \mod p
```

上の漸化式をElixirで実装すると次のようになります。

```elixir
def calc(x, p, mu) do
  rem(mu * x * (x + 1), p) 
end
```

これを末尾再帰呼び出しで繰り返す関数を用意します。

```elixir
def loopCalc(num, x, p, mu) do
  if num <= 0 do
    x
  else
    loopCalc(num - 1, calc(x, p, mu), p, mu)
  end
end
```

【追記】 @piacere_ex さんの意見を受けて，次のコードも試しました。

```elixir
def loopCalc(num, x, p, mu) when num <= 0 do x end
def loopCalc(num, x, p, mu) do
  new_num = num - 1
  new_calc = calc( x, p, mu )
  loopCalc( new_num, new_calc, p, mu )
end
```

そしてこれを`Flow.map`で並列に呼び出します。

```elixir
def mapCalc(list, num, p, mu, stages) do
  list
  |> Flow.from_enumerable(stages: stages)
  |> Flow.map(& loopCalc(num, &1, p, mu))
  |> Enum.to_list
end
```

単体のベンチマークでは，漸化式を10回繰り返します。素数$p$として6,700,417を採用しました。1〜0x2000000の範囲でFlowで並列処理を行います。`IO.puts`で並列数と実行時間を表示します。

```elixir
def benchmark(stages) do
  IO.puts "stages: #{stages}"
  IO.puts (
    :timer.tc(fn -> mapCalc(1..0x2000000, 10, 6_700_417, 22, stages) end)
    |> elem(0)
    |> Kernel./(1000000)
  )
end
```

ベンチマーク全体では，並列数は1,2,4,8,16,32,64,128の8通り設定しています。

```elixir
def benchmarks() do
  [1, 2, 4, 8, 16, 32, 64, 128]
  |> Enum.map(& benchmark(&1))
  |> Enum.to_list
end
```

ベンチマークの実行方法は次の通りです。

```bash
$ mix run -e "LogisticMap.benchmarks"
```

なお，ベンチマークのバリエーションとして他に2つ用意しました。

`benchmarks2` は `Flow.map(& calc(&1, p, mu))` をインライン展開して10回呼び出します。下記の`mapCalc2`を呼び出します。

```elixir
def mapCalc2(list, p, mu, stages) do
  list
  |> Flow.from_enumerable(stages: stages)
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Flow.map(& calc(&1, p, mu))
  |> Enum.to_list
end
```

`benchmark3` は `Flow.map` の中で `calc` のパイプラインでインライン展開して10回呼び出します。下記の`mapCalc3`を呼び出します。

```elixir
def mapCalc3(list, p, mu, stages) do
  list
  |> Flow.from_enumerable(stages: stages)
  |> Flow.map(& (&1
    |> calc(p, mu)
    |> calc(p, mu)
    |> calc(p, mu)
    |> calc(p, mu)
    |> calc(p, mu)
    |> calc(p, mu)
    |> calc(p, mu)
    |> calc(p, mu)
    |> calc(p, mu)
    |> calc(p, mu)
    ))
  |> Enum.to_list
end
```

[ソースコードの全体はGitHubに公開しました。](https://github.com/zeam-vm/logistic_map)

また，[Hexで公開しました。](https://hex.pm/packages/logistic_map)

# ベンチマークによる検証

検証に用いたマシンの仕様は次の通りです。物理コア数4，ハイパースレッディングが有効なので8並列で動作します。

> Mac Pro (Mid 2010)
> Processor 2.8GHz Quad-Core Intel Xeon
> Memory 16GB
> ATI Radeon HD 5770 1024MB

実行結果は次の通りです。

```bash
$ mix run -e "LogisticMap.benchmarks"
Compiling 1 file (.ex)
stages: 1
52.79562
stages: 2
24.716176
stages: 4
15.016131
stages: 8
12.664873
stages: 16
12.807277
stages: 32
12.841774
stages: 64
13.158978
stages: 128
13.21785
```

* Mac Pro は最大8並列で動作することから，並列数が8のときが最も実行時間が短かったです。
* 並列数1から2，2から4の時の短縮度合いの方が並列数4から8のときより大きいのは，物理コアとハイパースレッディングの違いによるものでしょう。他のマシンで試しても同様の傾向が得られました。
* 並列数を8より大きく増やすと若干実行時間が延びます。これは並列数が多すぎることでタスク切り替えが発生し実行効率が悪くなるためでしょう。

それにしても Elixir では物理コア数に準じて速度がきれいに向上しますね！ これは Elixir が関数型言語であるために，変数の書き換えが発生しないことにより，並列で動作させてもプロセス間通信がほとんど発生しないのでオーバーヘッドが生じないことによるものです。Elixir の可能性を感じますね！

異なるアプローチでインライン展開した `benchmark2` と `benchmark3` を比較してみましょう。

```bash
$ mix run -e "LogisticMap.benchmarks2"
stages: 1
54.697525
stages: 2
25.297751
stages: 4
15.763084
stages: 8
13.366235
stages: 16
13.611112
stages: 32
14.007026
stages: 64
14.013323
stages: 128
13.422258
```

```bash
$ mix run -e "LogisticMap.benchmarks3"
stages: 1
44.455119
stages: 2
20.67561
stages: 4
13.610704
stages: 8
11.308742
stages: 16
11.411827
stages: 32
11.714803
stages: 64
11.898896
stages: 128
11.914322
```

【追記】 @piacere_ex さんの意見を受けて試した結果です。

```bash
$ mix run -e "LogisticMap.benchmarks4"
stages: 1
53.469383
stages: 2
26.011707
stages: 4
15.299365
stages: 8
12.827463
stages: 16
12.947229
stages: 32
12.968043
stages: 64
13.230172
stages: 128
13.201316
```


`Flow.map` ごと展開した `benchmark2` より `Flow.map` の中でパイプラインを形成して展開した `benchmark3` の方が速いです！ `benchmark2` は `benchmark` より遅いですね。また，`benchmark4`は`benchmark`とあまり変わらないみたいです。

このことから，`Flow.map` をパイプラインで数珠繋ぎにしたい時には，`Flow.map` は1回に集約させて，`Flow.map` の中にパイプラインを作って数珠繋ぎにした方が高速であるということが言えます。要は下記みたいな感じにしましょうということです。

```elixir
list
|> Flow.from_enumerable
|> Flow.map(& (&1
  |> foo
  |> bar
  |> hoge
  ))
|> Enum.to_list
```

また，Elixirでは`benchmark`のコードでも，末尾再帰の最適化はしてくれているみたいです。

実験結果を表にまとめておきましょう。

|stages(数)|benchmark(秒)|benchmark2(秒)|benchmark3(秒)|benchmark4(秒)|備考|
|----:|--------:|--------:|--------:|--------:|:---|
|    1|52.795620|54.697525|44.455119|53.469383||
|    2|24.716176|25.297751|20.675610|26.011707||
|    4|15.016131|15.763084|13.610704|15.299365|物理コア数|
|    8|12.664873|13.366235|11.308742|12.827463|論理コア数(HT込みコア数)，最速|
|   16|12.807277|13.611112|11.411827|12.947229||
|   32|12.841774|14.007026|11.714803|12.968043||
|   64|13.158978|14.013323|11.898896|13.230172||
|  128|13.217850|13.422258|11.914322|13.201316||
|備考 |mapの中で再帰ループ|mapを展開|mapの中に展開|末尾再帰に配慮||

いずれのベンチマークも `stages` がHT(ハイパースレッディング)込みのコア数の時に最速となっています。ちなみにFlowでは `stages` のデフォルト値は論理コア数(HT込みのコア数)になっています。そのおかげでFlowはいい感じで並列処理をしてくれるというわけです。

# おわりに

1. Elixir では Flow / GenStage を用いて並列プログラミングをすることが容易にできます。
2. 思ったように並列性を稼げていないようならば，`Flow.from_enumerable` の `max_demand` を調整しましょう。
3. `Flow.map` をパイプラインで数珠繋ぎにしたい時には，`Flow.map` を1回に集約させて，`Flow.map` の中にパイプラインを作って数珠繋ぎにした方が高速です。
4. Flowでは `stages` のデフォルト値はHT込みのコア数となっているので，stagesは多くの場合最適に設定されています。

今回のベンチマークは [GitHub](https://github.com/zeam-vm/logistic_map)と[Hex](https://hex.pm/packages/logistic_map)で公開しています。ぜひご利用ください。

しかし，2,3についていちいち意識するのは面倒ですね。そこで私たちが開発しているZEAM(ZACKY's Elixir Abstract Machine)という処理系では，この辺りを最適化する仕組みを導入したいなと思っています。

次回[「ZEAM開発ログv0.1.1 AI/MLを爆速にしたい！ Flow / GenStage でGPUを駆動できないの？」](https://qiita.com/zacky1972/items/140d2380dfdf727b22bc)では Elixir の可能性をもっと高められないか，ZEAMでどのような最適化を取り入れると効果的か，考察してみましょう。お楽しみに！

明日は，[@takasehideki](https://qiita.com/takasehideki)さんの[「ElixirでIoT#1：IoTボードへのElixir環境の構築とEEloTツールキットの紹介」](https://qiita.com/takasehideki/items/fc570ae92a895caed213)です！