Cats Effect 使ってみた

公式チュートリアル をやってみた程度の人ですが、推測で補足しながら書きました。自分が面白い・新鮮と感じたところを共有できれば嬉しいです。

Cats Effectとは

The pure asynchronous runtime for Scala -- https://typelevel.org/cats-effect/

Cats Effect はホームページの冒頭に、「Scala 言語のための、純粋・非同期 ランタイム」と紹介しています。この紹介にあったキーワードを個人がこう理解しています：

1. 純粋
   関数型プログラミング界隈によく出る「純粋さ」のこと。副作用ある・なしコードの分離、イミュータブル性、参照透過性などと関わる。

2. 非同期
   非同期処理と同期処理を IO といった、統一したMonad風のAPIで処理できる。

3. ランタイム
   IO[A] で定義した一般的な処理を走らせるための、スケジューリングをはじめとするインフラみたいなものである。（普段より特化した目的を持つ）ライブラリやフレームワークではない。

IO[A] とは

IO[A] はCats Effectにおける、「A型の結果を得る処理」を表す型です。Cats Effect使っていれば、この型が一番よく使うことになります。

IO[A] 型のオブジェクト作るや渡すことに、副作用がありません。いわゆる「pure / 純粋」です。

※もちろんCPUが回る、時間が進むようなことがついてくるが、ここで言う副作用は時間の経過などを除いた「私達が関心を持つ副作用」と理解していいと思います。

IO[A] 内部の副作用を持つ処理を走らせるには、IO オブジェクトをCats Effectに投げて、実行してもらうしかない。
Cats Effect内部がスレッドプールなどの構造を持って、スケジューリングを管理します。ここがCats Effectが「ランタイム」と自称した最大の理由だと思います。
実行中の処理は fiber という構造で管理される。この構造ではJVM/OSスレッド切り替わりを減らして、効率よく多くの並行処理を回すことができます。

IO で書いたHello World

IOによる副作用管理を見せるためのデモプログラム。

import cats.effect.{ExitCode, IO, IOApp}

object HelloWorldIOApp extends IOApp {

  // sayHello() は「helloをプリントする」という処理をIOで返す。
  // プリントは副作用であるが、「副作用を含めたIOオブジェクトを作って返す」は副作用なし。
  def sayHello(target: String): IO[Unit] = IO {
    println(s"hello $target")
  }

  // sayHello() でIOオブジェクト複数回作っても、プリントの実行にならない
  val unsaid1: IO[Unit] = sayHello("not printing this")
  val unsaid2: IO[Unit] = sayHello("or this")

  // run メソッドは親クラスIOAppが定義したエントリポイント
  // ここから返したIO[ExitCode]が IOApp によって実行される
  override def run(args: List[String]): IO[ExitCode] = {
    val say = sayHello("world")

    // 「sayを3回実行して、ExitCode.Successを返す」処理を新しいIOオブジェクトで返す
    for (
      said1 <- say;
      said2 <- say;
      said3 <- say
    ) yield ExitCode.Success
  }
}

このプログラム実行すると、 run() が返したIOオブジェクトだけが実行されて、hello world を3回プリントします。

IO[A] はキャンセル可能

伝統的な () => A 型で処理を表す場合、最後の結果は成功 (return) / 失敗 (throw) のいずれかになります。

仮にこんなモデルで「処理aと処理bを同時に始めて、先に終わったほうの結果を使う」ようなプログラムを書けば、aが先に終わってもまだ実行中のbを止めることができなくて、リソースの浪費となります。（こんな浪費をなくす書き方はなくはないが、だいたいメンタルコストかかることになります）

そしてCats Effectの IO とスケジューリングは「最後まで行かなくて、途中でキャンセルされる」を実行結果の一種として定義したため、こんなケースをより簡単に対応できます。

※もちろん始まった処理がキャンセルになっても、すでに行われた副作用が自動で取り消されるわけではない。必要以上に行かせないで、CPUを含めて各種リソースを早く解放することが目的のようです。

IO で書いたFizzBuzz

fiberオブジェクトを利用して、無限再帰をキャンセルして止めるデモプログラム。

import cats.effect.{ExitCode, FiberIO, IO, IOApp}

import scala.concurrent.duration._

object FizzBuzzIOApp extends IOApp {

  // 副作用がない純粋な同期計算でも IO[A] を利用できる
  def computeFizzBuzz(n: Int): IO[String] = IO {
    if (n % 15 == 0) s"${n}: FizzBuzz"
    else if (n % 3 == 0) s"${n}: Fizz"
    else if (n % 5 == 0) s"${n}: Buzz"
    else n.toString
  }

  // sinceからのFizzBuzz序列を再帰的に、0.5s間隔で無限にプリントする
  // computeFizzBuzzと相対的、こちらは非同期 (sleep) 、副作用あり (println) IOオブジェクトになります。
  // (IO[A].flatMap() を利用した組み合わせがはJVMスタックサイズを無限に消費しないため、StackOverflowにならない)
  def printFizzBuzzSeries(since: Int): IO[Unit] = {
    for (
      line <- computeFizzBuzz(since);
      _    <- IO.println(line);
      _    <- IO.sleep(500.millisecond);
      _    <- printFizzBuzzSeries(since + 1)
    ) yield ()
  }

  override def run(args: List[String]): IO[ExitCode] = {
    for (
      // FizzBuzz序列のプリントを始めて、終了を待たない。
      // printingFizzBuzz変数に、開始した処理のfiberオブジェクトを格納する。
      printingFizzBuzz <- printFizzBuzzSeries(1).start;
      _ <- IO.sleep(10.second);
      // 10秒待った後、実行中のfiberをキャンセルする
      _ <- printingFizzBuzz.cancel;
      _ <- IO.println("fizz buzz should stop now")
    ) yield ExitCode.Success
  }
}

このプログラム実行すると、終了条件がない再帰を含めた printFizzBuzzSeries でも無限に実行することにならず、20までプリントしたところで終了します。

...
11
12: Fizz
13
14
15: FizzBuzz
16
17
18: Fizz
19
20: Buzz
fizz buzz should stop now

Process finished with exit code 0

どんな応用に向くか

Cats Effect自体は特定の目的に紐づかないが、書き方と特徴からみて、個人はこんな場合にいい選択だと思います：

• 副作用隔離して、コードをメインテナンスしやすくしたい時
• 効率よく並行処理を回したい場合