Add Async.Source, Async.Runner abstractions

Async.Source encapsulates the core functionality of allowing
to poll or wait for an asynchronous event.

Async.Runner encapsulates the core functionality of a scheduled
activity that can be canceled.

Also adds an unbounded channel that implements Async.Source for
demonstration purposes.

Author: odersky

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

@@ -12,103 +12,164 @@ trait Scheduler:
   def schedule(task: Runnable): Unit = ???
 
 object Scheduler extends Scheduler:
-  given fromAsync(using async: Async): Scheduler = async.client.scheduler
+  given fromAsync(using async: Async): Scheduler = async.runner.scheduler
 end Scheduler
 
+/** A context that allows one to suspend waiting for asynchronous data sources */
 trait Async:
 
-  /** Wait for completion of future `f`. This means:
-   *   - ensure that computing `f` has started
-   *   - wait for the completion and return the completed Try
-   */
-  def await[T](f: Future[T]): Try[T]
+  /** Wait for completion of async source `src` and return the result */
+  def await[T](src: Async.Source[T]): T
 
-  /** Wait for completion of the first of the futures `f1`, `f2`
-   *  @return  `Left(r1)`  if `f1` completed first with `r1`
-   *           `Right(r2)` if `f2` completed first with `r2`
+  /** Wait for completion of the first of the sources `src1`, `src2`
+   *  @return  `Left(r1)`  if `src1` completed first with `r1`
+   *           `Right(r2)` if `src2` completed first with `r2`
   */
-  def awaitEither[T1, T2](f1: Future[T1], f2: Future[T2]): Either[Try[T1], Try[T2]]
+  def awaitEither[T1, T2](src1: Async.Source[T1], src2: Async.Source[T2]): Either[T1, T2]
 
-  /** The future computed by this async computation. */
-  def client: Future[?]
+  /** The runner underlying this async computation. */
+  def runner: Async.Runner
 
 object Async:
+
+  /** The currently executing Async context */
   inline def current(using async: Async): Async = async
+
+  /** An asynchronous data source. Sources can be persistent or ephemeral.
+   *  A persistent source will always return the same data to calls to `poll`
+   *  and pass the same data to calls of `handle`. An ephemeral source might pass new
+   *  data in every call. An example of a persistent source is  `Future`. An
+   *  example of an ephemeral source is `Channel`.
+   */
+  trait Source[+T]:
+
+    /** Poll whether data is available
+     *  @return  The data or None in an option. Depending on the nature of the
+     *           source, data might be returned only once in a poll. E.g. if
+     *           the source is a channel, a Some result might skip to the next
+     *           entry.
+     */
+    def poll: Option[T]
+
+    /** When data is available, pass it to function `k`.
+     */
+    def handleWith(k: T => Unit): Unit
+
+  end Source
+
+  /** A thread-like entity that can be cancelled */
+  trait Runner:
+
+    /** The scheduler on which this computation is running */
+    def scheduler: Scheduler
+
+    /** Cancel computation for this runner and all its children */
+    def cancel(): Unit
+
+    /** Add a given child to this runner */
+    def addChild(child: Runner): Unit
+  end Runner
+
 end Async
 
-class Future[+T](body: Async ?=> T)(using val scheduler: Scheduler):
+
+class Future[+T](body: Async ?=> T)(using val scheduler: Scheduler)
+extends Async.Source[Try[T]], Async.Runner:
   import Future.{Status, Cancellation}, Status.*
 
   @volatile private var status: Status = Started
   private var result: Try[T] = uninitialized
   private var waiting: ListBuffer[Try[T] => Unit] = ListBuffer()
-  private var children: mutable.Set[Future[?]] = mutable.Set()
-
-  private def addWaiting(k: Try[T] => Unit): Unit = synchronized:
-    if status == Completed then k(result)
-    else waiting += k
+  private var children: mutable.Set[Async.Runner] = mutable.Set()
 
   private def currentWaiting(): List[Try[T] => Unit] = synchronized:
     val ws = waiting.toList
     waiting.clear()
     ws
 
-  private def currentChildren(): List[Future[?]] = synchronized:
+  private def currentChildren(): List[Async.Runner] = synchronized:
     val cs = children.toList
     children.clear()
     cs
 
-  private def checkCancellation(): Unit =
-    if status == Cancelled then throw Cancellation()
+  def poll: Option[Try[T]] = status match
+    case Started   =>  None
+    case Completed => Some(result)
+    case Cancelled => Some(Failure(Cancellation()))
+
+  def handleWith(k: Try[T] => Unit): Unit = synchronized:
+    if status == Completed then k(result)
+    else waiting += k
+
+  /** Eventually stop computation of this future and fail with
+   *  a `Cancellation` exception. Also cancel all linked children.
+   */
+  def cancel(): Unit = synchronized:
+    if status != Completed && status != Cancelled then
+      status = Cancelled
+      for f <- currentChildren() do f.cancel()
+
+  def addChild(child: Async.Runner): Unit = synchronized:
+    if status == Completed then child.cancel()
+    else children += this
+
+  /** Links the future as a child to the current async client.
+   *  This means the future will be cancelled when the async client
+   *  completes.
+   */
+  def linked(using async: Async): this.type = synchronized:
+    if status != Completed then async.runner.addChild(this)
+    this
 
   /** Wait for this future to be completed, return its value in case of success,
    *  or rethrow exception in case of failure.
    */
   def value(using async: Async): T = async.await(this).get
 
+  /** Block thread until future is completed and return result
+   *  N.B. This should be parameterized with a timeout.
+   */
+  def force(): T =
+    while status != Completed do wait()
+    result.get
+
   // a handler for Async
   private def async(body: Async ?=> Unit): Unit =
     boundary [Unit]:
       given Async with
-
-        private def resultOption[T](f: Future[T]): Option[Try[T]] = f.status match
-          case Started =>
-            None
-          case Completed =>
-            Some(f.result)
-          case Cancelled =>
-            Some(Failure(Cancellation()))
+        private def checkCancellation(): Unit =
+          if status == Cancelled then throw Cancellation()
 
         private inline def cancelChecked[T](op: => T): T =
           checkCancellation()
           val res = op
           checkCancellation()
           res
 
-        def await[T](f: Future[T]): Try[T] =
+        def await[T](src: Async.Source[T]): T =
           cancelChecked:
-            resultOption(f).getOrElse:
-              suspend[Try[T], Unit]: s =>
-                f.addWaiting: result =>
+            src.poll.getOrElse:
+              suspend[T, Unit]: k =>
+                src.handleWith: result =>
                   scheduler.schedule: () =>
-                    s.resume(result)
+                    k.resume(result)
 
-        def awaitEither[T1, T2](f1: Future[T1], f2: Future[T2]): Either[Try[T1], Try[T2]] =
+        def awaitEither[T1, T2](src1: Async.Source[T1], src2: Async.Source[T2]): Either[T1, T2] =
           cancelChecked:
-            resultOption(f1).map(Left(_)).getOrElse:
-              resultOption(f2).map(Right(_)).getOrElse:
-                suspend[Either[Try[T1], Try[T2]], Unit]: s =>
+            src1.poll.map(Left(_)).getOrElse:
+              src2.poll.map(Right(_)).getOrElse:
+                suspend[Either[T1, T2], Unit]: k =>
                   var found = AtomicBoolean()
-                  f1.addWaiting: result =>
+                  src1.handleWith: result =>
                     if !found.getAndSet(true) then
                       scheduler.schedule: () =>
-                        s.resume(Left(result))
-                  f2.addWaiting: result =>
+                        k.resume(Left(result))
+                  src2.handleWith: result =>
                     if !found.getAndSet(true) then
                       scheduler.schedule: () =>
-                        s.resume(Right(result))
+                        k.resume(Right(result))
 
-        def client = Future.this
+        def runner = Future.this
       end given
 
       body
@@ -121,36 +182,9 @@ class Future[+T](body: Async ?=> T)(using val scheduler: Scheduler):
         catch case ex: Exception => Failure(ex)
       status = Completed
     for task <- currentWaiting() do task(result)
-    cancelChildren()
+    cancel()
     notifyAll()
 
-  /** Links the future as a child to the current async client.
-   *  This means the future will be cancelled when the async client
-   *  completes.
-   */
-  def linked(using async: Async): this.type = synchronized:
-    if status != Completed then
-      async.client.children += this
-    this
-
-  private def cancelChildren(): Unit =
-    for f <- currentChildren() do f.cancel()
-
-  /** Eventually stop computation of this future and fail with
-   *  a `Cancellation` exception. Also cancel all linked children.
-   */
-  def cancel(): Unit = synchronized:
-    if status != Completed && status != Cancelled then
-      status = Cancelled
-      cancelChildren()
-
-  /** Block thread until future is completed and return result
-   *  N.B. This should be parameterized with a timeout.
-   */
-  def force(): T =
-    while status != Completed do wait()
-    result.get
-
   scheduler.schedule(() => complete())
 end Future
 
@@ -187,6 +221,30 @@ class Task[+T](val body: Async ?=> T):
         case Right(_: Failure[?]) => f1.value
 end Task
 
+/** An unbounded channel */
+class Channel[T] extends Async.Source[T]:
+  private val pending = ListBuffer[T]()
+  private val waiting = ListBuffer[T => Unit]()
+  def send(x: T): Unit = synchronized:
+    if waiting.isEmpty then pending += x
+    else
+      val k = waiting.head
+      waiting.dropInPlace(1)
+      k(x)
+  def poll: Option[T] = synchronized:
+    if pending.isEmpty then None
+    else
+      val x = pending.head
+      pending.dropInPlace(1)
+      Some(x)
+  def handleWith(k: T => Unit): Unit = synchronized:
+    if pending.isEmpty then waiting += k
+    else
+      val x = pending.head
+      pending.dropInPlace(1)
+      k(x)
+end Channel
+
 def Test(x: Future[Int], xs: List[Future[Int]])(using Scheduler): Future[Int] =
   Future:
     x.value + xs.map(_.value).sum