Akka Streams with Scala 3: mapAsync fails

Johann Heinzelreiter 0
Viewed 50

I made some first experiments with Akka Streams and Scala 3. Although I know that Akka has only experimental support for Scala 3, I am a little bit surprised that the following simple program does not work:

object HelloAkka extends App {

  implicit val actorSystem: ActorSystem = ActorSystem();
  implicit val executionContext: ExecutionContext = actorSystem.dispatcher

  val done = Source(1 to 5)
    .mapAsync(2)(n => Future {
      println(s"processing item $n in thread ${Thread.currentThread().getId}")
    })
    .runWith(Sink.ignore)

  Await.result(done, Duration.Inf)
  println("stream processing finished.")
  Await.result(actorSystem.terminate(), Duration.Inf)
  println("main thread terminated.")
}

When I build this program with Scala 3.2.0 using the following build.sbt the program hangs when mapAsync is invoked (no future is ever executed):

ThisBuild / version := "0.1.0-SNAPSHOT"

ThisBuild / scalaVersion := "3.2.0"  // fails
// ThisBuild / scalaVersion := "2.13.8" // works

lazy val root = (project in file("."))
  .settings(
    name := "hello-akka",
    libraryDependencies += "com.typesafe.akka" %% "akka-stream" % "2.6.20"
  )

When changing the Scala version to 2.13.8 the program behaves as expected. Also version 2.7.0-M1 of the Akka Streams library shows the same behavior.

Since mapAsync is elementary to Akka Streams programming, I suspect that I miss an obvious error.

Full running example: https://scastie.scala-lang.org/jheinzel/PHzMfGNDSmaSbssTX5YhWw.

akka-stream scala-3 scala
1 Answers

I suspect that you're encountering weirdness with extends App which relies on DelayedInit being treated specially by the compiler. Scala 3 no longer gives DelayedInit special treatment, instead the code in the body of HelloAkka becomes part of the static initializer of the object, which means it executes before the HelloAkka object is considered to be a real object by the JVM. This means that any thread outside the thread initializing HelloAkka (i.e. the equivalent of the "main thread") will block until such time as the initializing thread completes, which in this case entails a deadlock thanks to the Awaits in the body of HelloAkka.

Moving your code into a main method (which is not typically necessary with Scastie) causes your program to run as expected.

Removing the Awaits to allow the static initializer to return causes the program to run as expected

Awaiting in an App which is intended to be compiled for Scala 3 will deadlock if some asynchronous task (in this case the Akka Stream's mapAsync stage) reads a val/var from the App.

The Scala 3 docs recommend either @main annotations or an explicit def main.

Levi Ramsey0
Related