World 実装の基礎実験。詳しくは Wiki を参照のこと。

Readme.md

Future と Promise を使って World の基本原理を実装している。

基本機能として、

• 定期的に与えられた計算を実施する。
• 規定回数の計算を終えたら終了のメッセージを表示する。
• 計算中に例外が発生した場合は、スタックトレースを表示する。
• DoomsDay 例外を発生した場合は、World を停止して終了し、異常とは扱わない。

ややこしい実装になっているのは、busy waiting を避けるため。

next(n: Int, p: Promise[Int]): Unit = ...

next 関数の主な働きは1ステップ分の計算を実施することである。以下が引数の意味。

• n: World の状態変数。ここでは 0 からインクリメントする自然数。
• p: World を生成した main 関数が用意した Promise。main 関数はこの Promise からの返事を Await.ready(p.future, ...) で待っている。
• 返り値の型 (Unit) を明示しているのは、この関数が再帰的に定義されているため。

next 関数は最初に Future により1ステップ分の計算を実施する並行スレッドを生成する。

この Future は、以下の処理を実施する。

• n を表示し、
• n が既定値に達した場合には正常終了し、 // p.success(n)
• TICK_MSマイクロ秒だけ眠り、 blocking { Thread.sleep(TICK_MS) }
• 次の状態を n + 1 に設定する。

この処理全体を try { ... } catch { case e: Exception => { p.failure(e); n } } で覆うことで実行時エラー (e) を捉え、それを p.failure(e) によって main 関数内の処理に通知するためである。dooms_day(message) も例外で実装されており、通常の例外と DoomsDay は main 関数で区別して扱っている。

Thread.sleep(TICK_MS) を blocking { ... } で覆っているのは、ForkJoinPool の制限を逃れるためである。FJP はブロックするスレッド数がプロセッサ数を越えると破綻するが、blocking 節を用いれば、この制限が及ばないということだ。本例題ではブロックするスレッド数は高々2なのだが、念のため blocking 節を用いている。

main 関数

main 関数は並行計算のために promise と next(n, p) を実行する future を用意し、その結果を p.future.onComplete 節で設定されたハンドラーで受け取る。

最後の Await.ready(p.future, Duration.Inf) は正常値が返るか、あるいは例外が発生するまで永遠に待ち続けることを意味している。

bigbang.scala

package prg1.world.test

// 公式ドキュメントでの Futures and Promises の解説 - https://docs.scala-lang.org/overviews/core/futures.html#futures

import scala.concurrent._
import scala.concurrent.duration._
import scala.util.{Success, Failure}
import ExecutionContext.Implicits.global

/**
  * World 実装の基礎実験。
  * 苦節の跡は lx/scala3/bigbang.scala を参照。
  */

class DoomsDay(message: String) extends RuntimeException(message)
class Done() extends RuntimeException

object World {
  enum Result { case Success, Failure, Doom }
  val R = Result.Doom

  val TICK_MS = 100
  val N_TICKS = 8

  def dooms_day(message: String) = { throw new DoomsDay(message) }
  
  def next(n: Int, p: Promise[Int]): Unit = {
    if (p.isCompleted) return

    val future = Future[Int] {
      try {
        println(s"n = $n")

        if (n == N_TICKS / 2) {
          if (R == Result.Failure) { throw (new RuntimeException("Interrupted with a runtime error")) }
          if (R == Result.Doom) { dooms_day("The end of the game") }
        }
        if (n == N_TICKS) { p.success(n) }

        blocking { Thread.sleep(TICK_MS) }

        n + 1
      } catch { case e: Exception => { p.failure(e); n } }
    }

    for n <- future do next(n, p)
  }

  @main def main() = {
    val p = Promise[Int]()
    val f = Future { next(0, p) }

    p.future.onComplete {
      case Success(v) => { println(s"Driver loop finished with num = $v") }
      case Failure(e: DoomsDay) => { println(s"Doom's day: ${e.getMessage()}") }
      case Failure(e) => { e.printStackTrace() }
    }

    Await.ready(awaitable = p.future, atMost = Duration.Inf)
  }
}