fs2 の並行処理 〜 Balance, Broadcast編

Scala の Streaming I/O ライブラリ fs2 の並行処理についての記事。今回は Balance と Broardcastにについて。

## はじめに
fs2 の *concurrency primitives* には、これまでに紹介した [Queue](https://qiita.com/yasuabe2613/items/731cbf9fc991dda24c10)、[Topic、Signal](https://qiita.com/yasuabe2613/items/3c32e530d24bc3610c5c)の他に、比較的新しい[^new_feature]`Balance`、`Broadcast` といったものもある。

[^new_feature]: [v1.0.0](https://github.com/functional-streams-for-scala/fs2/releases/tag/v1.0.0) 以降でサポート

## 概要
どちらもファンアウト(*fan-out*）系[^fanout]で、一本の源流のストリームを複数のワーカー[^worker]につなげるものだが、以下のような違いがある。

- **Balance**: 上から流れてきた要素を、複数のワーカーからどれか一つ適当なもの選んで流す[^worker_selection]
- **Broadcast**: 上から流れてきた要素を、全てのワーカーが受け取る

これらを扱うための `balance*`系や`broadCast*`系などのメソッドが`Stream` クラスで提供されていて、直接 `Balance` や `Broadcast` を使うより簡単なので、今回はそれを使う。

[^fanout]: 扇状に一本から複数に分岐する形や流れ
[^worker_selection]: ワーカーの選出基準の詳細などはここでは触れない。APIドキュメントに詳しい。

[^worker]: ワーカーは `Pipe` や `Sink` として、ソース・ストリームにつなげる。
[^outer]: 正式な用語ではないが、ここではソース・ストリームや、`balanceThrough` での合流後などのストリームを、アウターストリームと読んでいる。

## お題
以下のような標準入力由来のストリームのファンアウトを考えてみる。

1. ソース・ストリームでは、標準入力からのバイト列を、(1)文字列にして、(2)行で分割して、(3)トリムして、(4)空行を取り除いて、後続のストリームに流す。
2. 下記のようなワーカーをソース・ストリームにつなげる
 1. 識別 id と秒数を指定してパイプを生成する
 2. 各ワーカーは指定秒数だけ待ってから、流れてきた文字列を修飾して標準出力に書き出す
 3. 結果として文字列の長さを整数として返す
3. 全体のストリーム構成としては、ソース・ストリームを Balance または Broadcast を介してワーカーに接続し、流れてきた文字列長を適当に修飾して表示する。

（fs2 は v2.0.0 を使った。動くコードは[ここ](https://gist.github.com/yasuabe/6fcbbdd77beec0a6f2b85a5b18d2dc59)に置いた）

## 実装
以下のような共通コードを使う。

```scala
def readlineStream[F[_]: Sync : ContextShift](implicit bl: Blocker): Stream[F, String] =
  io.stdin[F](4096, bl)
    .through(text.utf8Decode)
    .through(text.lines)
    .map(_.trim)
    .filter(_.nonEmpty)
    .takeWhile(_ != ":q", takeFailure = true)

def printlnSink[F[_]: Sync : ContextShift](implicit bl: Blocker): Pipe[F, String, Unit] =
  _.map(s => s"$s\n")
    .through(text.utf8Encode)
    .through(io.stdout[F](bl))
```
`readlineStream` は、コンソール入力を「お題」のように加工している。`printlnSink` は、上流ストリームから流れてきた文字列に改行を付けて標準出力に書き出している。

Balance と Broadcast で共用するワーカーは、以下のようなパイプとした。

```scala
def worker[F[_]: Sync: ContextShift: Timer](id: String, sec: Int)
          (implicit bl: Blocker): Pipe[F, String, Int] = _ flatMap { s =>
  val sleep = Stream.sleep_[F](sec.second)
  val print = Stream(s"worker#$id processing '$s' (${sec}s)") through printlnSink
  (sleep ++ print) zipRight Stream(s).map(_.length)
}
```
パイプをファンアウトするには、`Stream` の `balanceThrough` か `broadcastThrough` を使う。[^usage_with_sink]

（どちらについても、最大並列数を指定して、ある一つのワーカーから複数のインナーストリームを生成する版[^worker_with_maxconcurrent]もあるが、今回は異なる複数のワーカーを個別に指定する版を使う。）

[^worker_with_maxconcurrent]: [balanceThrough](https://static.javadoc.io/co.fs2/fs2-core_2.12/1.0.2/fs2/Stream.html#balanceThrough[F2[x]%3E:F[x],O2](chunkSize:Int,maxConcurrent:Int)(pipe:fs2.Pipe[F2,O,O2])(implicitevidence$14:cats.effect.Concurrent[F2]):fs2.Stream[F2,O2]), [broadcastThrough](https://static.javadoc.io/co.fs2/fs2-core_2.12/1.0.2/fs2/Stream.html#broadcastThrough[F2[x]%3E:F[x],O2](maxConcurrent:Int)(pipe:fs2.Pipe[F2,O,O2])(implicitevidence$6:cats.effect.Concurrent[F2]):fs2.Stream[F2,O2])
[^usage_with_sink]: シンクにつなげるには、`balanceTo` や `broadcastTo` を使う。

### Balance
Balance では、以下のように `balanceThrough` メソッドを使って３つのワーカーにファンアウトする。

```scala
def stream[F[_] : Concurrent : Timer : ContextShift]
  (implicit bl: Blocker): Stream[F, Unit] =
    readlineStream[F]
      .balanceThrough(1)(worker("A", 1), worker("B", 2), worker("C", 3))
      .map(n => s"length: $n")
      .through(printlnSink)
```

#### 結果
動かしてみると以下のようになる。（コメントは後で書き足したもの）

```
hello                            // hello と入力
worker#A processing 'hello' (1s) // ワーカーAが一秒待って文字列長をエミット
length: 5                        // アウター側で表示
bye                              // bye と入力
worker#B processing 'bye' (2s)   // 同様にワーカーB
length: 3
1                                // "1" と "1234" を立て続けに入力
1234
worker#A processing '1234' (1s)  // 順序としてはワーカーCだが、待ち時間が小さいワーカーAが先に終わる
length: 4                        // アウター側も終わった順にストリーム要素を受け取る
worker#C processing '1' (3s)     // 遅れてワーカーC が終了
length: 1                        // アウター側は同様に先行ワーカーが終わり次第処理
```
A, B, C、A, B, C ...の順にファンアウトされ、アウターストリームは終ったもの順に処理を続行する。

### Broadcast
Broadcast では以下のように `broadcastThrough` を使って、３つのワーカーを接続する。

```scala
def stream[F[_] : Concurrent : Timer : ContextShift]
  (implicit bl: Blocker): Stream[F, Unit] =
    readlineStream[F]
      .broadcastThrough(worker("A", 1), worker("B", 2), worker("C", 3))
      .map(n => s"length: $n")
      .through(printlnSink)
```

#### 結果
Balance とは違って、３つのワーカーそれぞれがアウターストリームから流れてきた要素を処理する。

```
hello                             // ”hello” と入力
worker#A processing 'hello' (1s)  // ワーカーが3つともアウターからのストリームを処理
length: 5                         // アウターでは終わった順に処理
worker#B processing 'hello' (2s)
length: 5
worker#C processing 'hello' (3s)
length: 5
```

ストリーム中の一つの要素（あるいはイベント）を複数のワーカー（あるいは購読者）が処理するという点で、なんとなくトピックに似ている気もするが、`Topic` のようにオブジェクトを明示的には作らなくても `Stream` のメソッドで直接つなげられる点や、ワーカーの出力がアウターに戻ってくる点などが違う。