Variant of suspension based on fibers

Author: odersky

tests/run/suspend-strawman-2.check

@@ -0,0 +1,3 @@
+test async:
+33
+(22,11)

tests/run/suspend-strawman-2/Async.scala

@@ -1,8 +1,8 @@
 package concurrent
 import java.util.concurrent.atomic.AtomicBoolean
 import scala.collection.mutable
-import runtime.suspend
-import scala.util.boundary
+import fiberRuntime.suspend
+import fiberRuntime.boundary
 
 /** A context that allows to suspend waiting for asynchronous data sources
  */
@@ -53,23 +53,48 @@ object Async:
       checkCancellation()
       src.poll().getOrElse:
         try
+          var result: Option[T] = None
           suspend[T, Unit]: k =>
             src.onComplete: x =>
               scheduler.schedule: () =>
-                k.resume(x)
+                result = Some(x)
+                k.resume()
               true // signals to `src` that result `x` was consumed
+          result.get
         finally checkCancellation()
 
   end Impl
 
+  private class Blocking(val scheduler: Scheduler = Scheduler) extends Async:
+
+    def root = Cancellable.empty
+
+    protected def checkCancellation(): Unit = ()
+
+    private var hasResumed = false
+
+    def await[T](src: Source[T]): T = synchronized:
+      src.poll() match
+        case Some(x) => x
+        case None =>
+          var result: Option[T] = None
+          src.onComplete: x =>
+            synchronized:
+              result = Some(x)
+              notify()
+            true
+          while result.isEmpty do wait()
+          result.get
+
+  def blocking[T](body: Async ?=> T, scheduler: Scheduler = Scheduler): T =
+    body(using Blocking())
+
   /** The currently executing Async context */
   inline def current(using async: Async): Async = async
 
   /** Await source result in currently executing Async context */
   inline def await[T](src: Source[T])(using async: Async): T = async.await(src)
 
-
-
   /** A function `T => Boolean` whose lineage is recorded by its implementing
    *  classes. The Listener function accepts values of type `T` and returns
    *  `true` iff the value was consumed by an async block.

tests/run/suspend-strawman-2/Test.scala

@@ -1,4 +1,24 @@
 import concurrent.*
+import fiberRuntime.boundary.setName
+
+@main def Test =
+  given Scheduler = Scheduler
+  val x = Future:
+    setName("x")
+    val a = Future{ setName("xa"); 22 }
+    val b = Future{ setName("xb"); 11 }
+    val c = Future { setName("xc"); assert(false); 1 }
+    c.alt(Future{ setName("alt1"); a.value + b.value }).alt(c).value
+  val y = Future:
+    setName("y")
+    val a = Future{ setName("ya"); 22 }
+    val b = Future{ setName("yb"); 11 }
+    a.zip(b).value
+  println("test async:")
+  Async.blocking:
+    println(x.value)
+    println(y.value)
+  //println("test choices:")
+  //println(TestChoices)
 
-@main def Test = ()
 

tests/run/suspend-strawman-2/channels.scala

@@ -1,7 +1,7 @@
 package concurrent
 import scala.collection.mutable, mutable.ListBuffer
-import scala.util.boundary, boundary.Label
-import runtime.suspend
+import fiberRuntime.boundary, boundary.Label
+import fiberRuntime.suspend
 import java.util.concurrent.CancellationException
 import Async.{Listener, await}
 

tests/run/suspend-strawman-2/fiberRuntime.scala

@@ -0,0 +1,29 @@
+package fiberRuntime
+
+/** A delimited contination, which can be invoked with `resume` */
+class Suspension:
+  private var hasResumed = false
+  def resume(): Unit = synchronized:
+    hasResumed = true
+    notify()
+  def suspend(): Unit = synchronized:
+    if !hasResumed then
+      wait()
+
+def suspend[T, R](body: Suspension => Unit): Unit =
+  val susp = Suspension()
+  body(susp)
+  susp.suspend()
+
+object boundary:
+  final class Label[-T]()
+
+  def setName(name: String) = ()
+
+  def apply[T](body: Label[T] ?=> Unit): Unit =
+    new Thread:
+      override def run() =
+        body(using Label[T]())
+    .start()
+
+

tests/run/suspend-strawman-2/futures.scala

@@ -1,7 +1,7 @@
 package concurrent
 
 import scala.collection.mutable, mutable.ListBuffer
-import scala.util.boundary
+import fiberRuntime.boundary
 import scala.compiletime.uninitialized
 import scala.util.{Try, Success, Failure}
 import scala.annotation.unchecked.uncheckedVariance
@@ -16,11 +16,6 @@ trait Future[+T] extends Async.OriginalSource[Try[T]], Cancellable:
    */
   def value(using async: Async): T
 
-  /** Block thread until future is completed and return result
-   *  N.B. This should be parameterized with a timeout.
-   */
-  def force(): T
-
   /** Eventually stop computation of this future and fail with
    *  a `Cancellation` exception. Also cancel all children.
    */
@@ -81,10 +76,6 @@ object Future:
     def value(using async: Async): T =
       async.await(this).get
 
-    def force(): T = synchronized:
-      while !hasCompleted do wait()
-      result.get
-
     /** Complete future with result. If future was cancelled in the meantime,
      *  return a CancellationException failure instead.
      *  Note: @uncheckedVariance is safe here since `complete` is called from
@@ -99,8 +90,6 @@ object Future:
         this.result = result
         hasCompleted = true
       for listener <- extract(waiting) do listener(result)
-      synchronized:
-        notifyAll()
 
   end CoreFuture
 
@@ -151,7 +140,8 @@ object Future:
      *  If either task succeeds, succeed with the success that was returned first
      *  and cancel the other. Otherwise, fail with the failure that was returned last.
      */
-    def alt[T2 >: T1](f2: Future[T2])(using Async.Config): Future[T2] = Future:
+    def alt[T2 >: T1](f2: Future[T2], name: String = "alt")(using Async.Config): Future[T2] = Future:
+      boundary.setName(name)
       Async.await(Async.either(f1, f2)) match
         case Left(Success(x1))    => f2.cancel(); x1
         case Right(Success(x2))   => f1.cancel(); x2
@@ -210,5 +200,5 @@ def add(x: Future[Int], xs: List[Future[Int]])(using Scheduler): Future[Int] =
 end add
 
 def Main(x: Future[Int], xs: List[Future[Int]])(using Scheduler): Int =
-  add(x, xs).force()
+  Async.blocking(add(x, xs).value)