こんにちは。
ウェブクルー AdventCalendar 22日目の記事です。
昨日は @wcsakurai さんの「 NuxtユーザがSapper(svelte)に入門してみた」でした。

今日は Scala での非同期処理をデバッグしやすくする方法について書いていきます。

課題点

Scala に限らずですが、JVM での非同期処理はデバッグが困難になりがちです。
その要因は、スタックトレースから例外発生時のコールフローが読みとれない為です。

例として、次のソースを実行してみましょう。

object Example extends App {
  implicit val ec: ExecutionContext = ExecutionContext.global
  Await.result(ProcessA.exec, Duration.Inf)
}

object ProcessA {
  def exec(implicit ec: ExecutionContext): Future[Unit] =
    Future(()).flatMap(_ => ProcessB.exec)
}

object ProcessB {
  def exec(implicit ec: ExecutionContext): Future[Nothing] =
    Future(throw new RuntimeException("error"))
}

これを実行すると、スタックトレースは以下のように表示されます。

java.lang.RuntimeException: error
        at ProcessB$.$anonfun$exec$3(Example.scala:16)
        at scala.concurrent.Future$.$anonfun$apply$1(Future.scala:672)
        at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:431)
        at java.base/java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1426)
        at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
        at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
        at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
        at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
        at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)

ProcessB#exec で RuntimeException が発生している事は確認できます。
しかし、ProcessB#exec が何処から呼び出されているかは、後続のコールフローからは判断する事ができません。

今回の例では、ProcessB#exec は ProcessA#exec からしか呼び出されていない事が明白なので、あまり問題に感じられないかもしれませんが・・・。
もし ProcessB#exec が他のメソッドから呼びされる事がある状況であった場合、どのメソッドから呼び出されてエラーになったかスタックトレースからは判断できなくなってしまいます。

対応策

この問題への対応策として、非同期処理ライブラリを利用する方法があります。

ここでは Monix を利用します。
Monix では 3.3.0 以降のバージョンで非同期スタックトレースに対応しています。

object ExampleTask extends App {
  implicit val s: Scheduler = Scheduler.global
  ProcessTaskA.exec.runSyncUnsafe()
}

object ProcessTaskA {
  def exec(implicit ec: ExecutionContext): Task[Unit] =
    Task(()).flatMap(_ => ProcessTaskB.exec)
}

object ProcessTaskB {
  def exec(implicit ec: ExecutionContext): Task[Nothing] =
    Task(throw new RuntimeException("error"))
}

これを実行すると...

java.lang.RuntimeException: error
        at ProcessTaskB$.$anonfun$exec$3(ExampleTask.scala:18)
        at flatMap @ ProcessTaskA$.exec(ExampleTask.scala:13)

このようなスタックトレースになります。
これで ProcessTaskB#exec は ProcessTaskA#exec から呼び出されている事が読み取れるようになりました。

トレースモード

Monix では、トレースモードを3種類用意しています。
システムプロパティ -Dmonix.eval.stackTracingMode=<モード名> で設定します。

モード	内容
NONE	トレースを行わない。パフォーマンスへの影響はほぼ無し。
CACHED	map や flatMap 等、特定の操作のみトレースを行い、キャッシュする。
FULL	トレースを詳細に行う。パフォーマンスが大幅に低下する為、開発環境用。

デフォルトでは CACHED が有効になっています。本番環境での推奨設定のようです。
ただし、公式のベンチマーク結果では、10〜30% ほど余分に負荷がかかっているそうです。 (実稼働上ではそれほどでは無いかも、と添えられつつ)

パフォーマンスを最優先にするのであれば、テスト環境のみ CACHED や FULL にしておいて、本番環境では NONE にすると良いと思います。

FULL では CACHED よりも詳細にスタックトレースを得ることができます。
前述のサンプルソースを FULL で実行すると、下記のようなスタックトレースになります。

java.lang.RuntimeException: error
        at ProcessTaskB$.$anonfun$exec$3(ExampleTask.scala:18)
        at apply @ ProcessTaskB$.exec(ExampleTask.scala:18)
        at apply @ ProcessTaskA$.exec(ExampleTask.scala:13)
        at flatMap @ ProcessTaskA$.exec(ExampleTask.scala:13)

apply の操作についてもスタックトレース上に出力されるようになりました。

利用例

Playframework での利用例です。

Controller の実装に Action.async を利用する場合、ブロック内の戻り値は Future[Result] にする必要があります。
ここではブロック内の戻り値として Task[Result] を受け取れるよう ActionBuilder を拡張しています。

@Singleton
class HomeController @Inject()(
  val controllerComponents: ControllerComponents, ec: ExecutionContext
) extends BaseController with TaskAction {
  implicit val scheduler: Scheduler = Scheduler(ec)

  def index(): Action[AnyContent] = Action.task { implicit request =>
    Usecase.execute.map(_ => Ok(""))
  }
}

trait TaskAction {
  implicit class TaskActionBuilder[R[_], A](ab: ActionBuilder[R, A]) {
    def task(block: R[A] => Task[Result])(implicit s: Scheduler): Action[A] = {
      ab.async(block.andThen(_.runToFuture))
    }
  }
}

object Usecase {
  def execute: Task[Unit] = Repository.fetch.map(_ => ())
}

object Repository {
  def fetch: Task[Nothing] = Task(throw new RuntimeException("error"))
}

これで HomeController#index に対応するエントリーポイントを実行すると...

play.api.http.HttpErrorHandlerExceptions$$anon$1: Execution exception[[RuntimeException: error]]
        at play.api.http.HttpErrorHandlerExceptions$.throwableToUsefulException(HttpErrorHandler.scala:359)
        at play.api.http.DefaultHttpErrorHandler.onServerError(HttpErrorHandler.scala:261)
        at play.core.server.AkkaHttpServer$$anonfun$2.applyOrElse(AkkaHttpServer.scala:430)
        at play.core.server.AkkaHttpServer$$anonfun$2.applyOrElse(AkkaHttpServer.scala:422)
        at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:453)
        at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:56)
        // ...中略...
Caused by: java.lang.RuntimeException: error
        at repositories.Repository$.$anonfun$fetch$1(Repository.scala:5)
        at map @ usecases.Usecase$.execute(Usecase.scala:7)
        at map @ controllers.HomeController.$anonfun$index$1(HomeController.scala:19)

このようなスタックトレースになります。
これなら Repository がどの Usecase から呼び出されたか読み取れますね。

その他の非同期処理ライブラリ

今回は Monix を利用しましたが、非同期スタックトレースに対応したライブラリは他にもあります。

cats-effect
https://typelevel.org/cats-effect/
ZIO
https://github.com/zio/zio

まとめ

非同期処理のデバッグ問題を、ライブラリで解決できるお話でした。
実装の効率化だけでなく、トレーサビリティを確保できるという点においても、非同期処理ライブラリを導入するメリットがありますね。

明日は @wc-keisuke_tokunaga さんです。
よろしくお願いします！

参考

https://monix.io/docs/current/eval/stacktraces.html