Add derived sources and channels

odersky · odersky · commit 4cf756c64403 · 2023-02-11T15:40:41.000+01:00
diff --git a/tests/pos/suspend-strawman-2/Async.scala b/tests/pos/suspend-strawman-2/Async.scala
@@ -1,53 +1,189 @@
 package concurrent
+import java.util.concurrent.atomic.AtomicBoolean
+import scala.collection.mutable
+import runtime.suspend
+import scala.util.boundary
+
+/** The underlying configuration of an async block */
+trait AsyncConfig:
+
+  /** The cancellable async source underlying this async computation */
+  def root: Cancellable
+
+  /** The scheduler for runnables defined in this async computation */
+  def scheduler: Scheduler
+
+object AsyncConfig:
+
+  /** A toplevel async group with given scheduler and a synthetic root that
+   *  ignores cancellation requests
+   */
+  given fromScheduler(using s: Scheduler): AsyncConfig with
+    def root = Cancellable.empty
+    def scheduler = s
+
+end AsyncConfig
 
 /** A context that allows to suspend waiting for asynchronous data sources */
-trait Async:
+trait Async extends AsyncConfig:
 
   /** 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 sources `src1`, `src2`
-   *  @return  `Left(r1)`  if `src1` completed first with `r1`
-   *           `Right(r2)` if `src2` completed first with `r2`
-  */
-  def awaitEither[T1, T2](src1: Async.Source[T1], src2: Async.Source[T2]): Either[T1, T2]
+object Async:
 
-  /** The cancellable runner underlying this async computation. */
-  def runner: Cancellable
+  abstract class AsyncImpl(val root: Cancellable, val scheduler: Scheduler)
+     (using boundary.Label[Unit]) extends Async:
 
-  /** The scheduler for runnables defined in this async computation */
-  def scheduler: Scheduler
+    protected def checkCancellation(): Unit
 
-object Async:
+    private var result: T
+
+    def await[T](src: Async.Source[T]): T =
+      checkCancellation()
+      var resultOpt: Option[T] = None
+      if src.poll: x =>
+        result = x
+        true
+      then result
+      else
+        try suspend[T, Unit]: k =>
+          src.onComplete: x =>
+            scheduler.schedule: () =>
+              k.resume(x)
+            true // signals to `src` that result `x` was consumed
+        finally checkCancellation()
+
+  end AsyncImpl
 
   /** 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`.
+  /** 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.
    */
-  trait Source[+T]:
-    thisSource =>
+  trait Listener[-T] extends Function[T, Boolean]
 
-    /** 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
+  /** A listener for values that are processed by the given source `src` and
+   *  that are demanded by the continuation listener `continue`.
+   */
+  abstract case class ForwardingListener[T](src: Source[?], continue: Listener[?]) extends Listener[T]
+
+  /** A listener for values that are processed directly in an async block.
+   *  Closures of type `T => Boolean` can be SAM converted to this type.
+   */
+  abstract case class FinalListener[T]() extends Listener[T]
+
+  /** A source that cannot be mapped, filtered, or raced. In other words,
+   *  an item coming from a direct source must be immediately consumed in
+   *  another async computation; no rejection of this item is possible.
+   */
+  trait DirectSource[+T]:
+
+    /** If data is available at present, pass it to function `k`
+     *  and return the result if this call.
+     *  `k` returns true iff the data was consumed in an async block.
+     *  Calls to `poll` are always synchronous.
+     */
+    def poll(k: Listener[T]): Boolean
+
+    /** Once data is available, pass it to function `k`.
+     *  `k` returns true iff the data was consumed in an async block.
+     *  Calls to `onComplete` are usually asynchronous, meaning that
+     *  the passed continuation `k` is a suspension.
      */
-    def poll: Option[T]
+    def onComplete(k: Listener[T]): Unit
 
-    /** When data is available, pass it to function `k`.
+    /** Signal that listener `k` is dead (i.e. will always return `false` from now on).
+     *  This permits original, (i.e. non-derived) sources like futures or channels
+     *  to drop the  listener from their `waiting` sets.
      */
-    def handleWith(k: T => Unit): Unit
+    def dropListener(k: Listener[T]): Unit
 
-    def map[U](f: T => U): Source[U] = new Source:
-      def poll = thisSource.poll.map(f)
-      def handleWith(k: U => Unit): Unit = thisSource.handleWith(f.andThen(k))
+  end DirectSource
+
+  /** An asynchronous data source. Sources can be persistent or ephemeral.
+   *  A persistent source will always pass same data to calls of `poll and `onComplete`.
+   *  An ephememral source can 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] extends DirectSource[T]:
+
+    /** Pass on data transformed by `f` */
+    def map[U](f: T => U): Source[U] =
+      new DerivedSource[T, U](this):
+        def listen(x: T, k: Listener[U]) = k(f(x))
+
+    /** Pass on only data matching the predicate `p` */
+    def filter(p: T => Boolean): Source[T] =
+      new DerivedSource[T, T](this):
+        def listen(x: T, k: Listener[T]) = p(x) && k(x)
 
   end Source
 
+  /** As source that transforms an original source in some way */
+
+  abstract class DerivedSource[T, U](src: Source[T]) extends Source[U]:
+
+    /** Handle a value `x` passed to the original source by possibly
+     *  invokiong the continuation for this source.
+     */
+    protected def listen(x: T, k: Listener[U]): Boolean
+
+    private def transform(k: Listener[U]): Listener[T] =
+      new ForwardingListener[T](this, k):
+        def apply(x: T): Boolean = listen(x, k)
+
+    def poll(k: Listener[U]): Boolean =
+      src.poll(transform(k))
+    def onComplete(k: Listener[U]): Unit =
+      src.onComplete(transform(k))
+    def dropListener(k: Listener[U]): Unit =
+      src.dropListener(transform(k))
+  end DerivedSource
+
+  /** Pass first result from any of `sources` to the continuation */
+  def race[T](sources: Source[T]*): Source[T] = new Source:
+
+    def poll(k: Listener[T]): Boolean =
+      val it = sources.iterator
+      var found = false
+      while it.hasNext && !found do
+        it.next.poll: x =>
+          found = k(x)
+          found
+      found
+
+    def onComplete(k: Listener[T]): Unit =
+      val listener = new ForwardingListener[T](this, k):
+        var foundBefore = false
+        def continueIfFirst(x: T): Boolean = synchronized:
+          if foundBefore then false else { foundBefore = k(x); foundBefore }
+        def apply(x: T): Boolean =
+          val found = continueIfFirst(x)
+          if found then sources.foreach(_.dropListener(this))
+          found
+      sources.foreach(_.onComplete(listener))
+
+    def dropListener(k: Listener[T]): Unit =
+      val listener = new ForwardingListener[T](this, k):
+        def apply(x: T): Boolean = ???
+          // not to be called, we need the listener only for its
+          // hashcode and equality test.
+      sources.foreach(_.dropListener(listener))
+
+  end race
+
+  /** If left (respectively, right) source succeeds with `x`, pass `Left(x)`,
+   *  (respectively, Right(x)) on to the continuation.
+   */
+  def either[T, U](src1: Source[T], src2: Source[U]): Source[Either[T, U]] =
+    race[Either[T, U]](src1.map(Left(_)), src2.map(Right(_)))
+
 end Async
 
diff --git a/tests/pos/suspend-strawman-2/Cancellable.scala b/tests/pos/suspend-strawman-2/Cancellable.scala
@@ -1,15 +1,21 @@
 package concurrent
 
-/** A trait for cancellable entiries that can be grouped */
+/** A trait for cancellable entities that can be grouped */
 trait Cancellable:
 
+  /** Issue a cancel request */
   def cancel(): Unit
 
   /** Add a given child to this Cancellable, so that the child will be cancelled
    *  when the Cancellable itself is cancelled.
    */
   def addChild(child: Cancellable): Unit
 
-  def isCancelled: Boolean
+object Cancellable:
+
+  /** A cancelled entity that ignores all `cancel` and `addChild` requests */
+  object empty extends Cancellable:
+    def cancel() = ()
+    def addChild(child: Cancellable) = ()
 
 end Cancellable
diff --git a/tests/pos/suspend-strawman-2/channels.scala b/tests/pos/suspend-strawman-2/channels.scala
@@ -0,0 +1,122 @@
+package concurrent
+import scala.collection.mutable, mutable.ListBuffer
+import scala.util.boundary.Label
+import runtime.suspend
+import Async.{Listener, await}
+
+/** An unbounded channel */
+class UnboundedChannel[T] extends Async.Source[T]:
+  private val pending = ListBuffer[T]()
+  private val waiting = mutable.Set[Listener[T]]()
+
+  private def drainWaiting(x: T): Boolean =
+    val it = waiting.iterator
+    var sent = false
+    while it.hasNext && !sent do
+      val k = it.next()
+      sent = k(x)
+      if sent then waiting -= k
+    sent
+
+  private def drainPending(k: Listener[T]): Boolean =
+    val sent = pending.nonEmpty && k(pending.head)
+    if sent then
+      while
+        pending.dropInPlace(1)
+        pending.nonEmpty && drainWaiting(pending.head)
+      do ()
+    sent
+
+  def read()(using Async): T = synchronized:
+    await(this)
+
+  def send(x: T): Unit = synchronized:
+    val sent = pending.isEmpty && drainWaiting(x)
+    if !sent then pending += x
+
+  def poll(k: Listener[T]): Boolean = synchronized:
+    drainPending(k)
+
+  def onComplete(k: Listener[T]): Unit = synchronized:
+    if !drainPending(k) then waiting += k
+
+  def dropListener(k: Listener[T]): Unit = synchronized:
+    waiting -= k
+
+end UnboundedChannel
+
+class SyncChannel[T]:
+
+  private val pendingReads = mutable.Set[Listener[T]]()
+  private val pendingSends = mutable.Set[Listener[Listener[T]]]()
+
+  private def collapse[T](k2: Listener[Listener[T]]): Option[T] =
+    var r: Option[T] = None
+    if k2 { x => r = Some(x); true } then r else None
+
+  protected def link[T](pending: mutable.Set[T], op: T => Boolean): Boolean =
+    pending.iterator.find(op) match
+      case Some(elem) => pending -= elem; true
+      case None => false
+
+  val canRead = new Async.Source[T]:
+    def poll(k: Listener[T]): Boolean =
+      link(pendingSends, sender => collapse(sender).map(k) == Some(true))
+    def onComplete(k: Listener[T]): Unit =
+      if !poll(k) then pendingReads += k
+    def dropListener(k: Listener[T]): Unit =
+      pendingReads -= k
+
+  val canSend = new Async.Source[Listener[T]]:
+    def poll(k: Listener[Listener[T]]): Boolean =
+      link(pendingReads, k)
+    def onComplete(k: Listener[Listener[T]]): Unit =
+      if !poll(k) then pendingSends += k
+    def dropListener(k: Listener[Listener[T]]): Unit =
+      pendingSends -= k
+
+  def send(x: T)(using Async): Unit =
+    await(canSend)(x)
+
+  def read()(using Async): T =
+    await(canRead)
+
+end SyncChannel
+
+class DirectSyncChannel[T]:
+
+  private val pendingReads = mutable.Set[Listener[T]]()
+  private val pendingSends = mutable.Set[Listener[Listener[T]]]()
+
+  private def collapse[T](k2: Listener[Listener[T]]): Option[T] =
+    var r: Option[T] = None
+    if k2 { x => r = Some(x); true } then r else None
+
+  private def link[T](pending: mutable.Set[T], op: T => Unit): Boolean =
+    pending.headOption match
+      case Some(elem) => op(elem); true
+      case None => false
+
+  val canRead = new Async.Source[T]:
+    def poll(k: Listener[T]): Boolean =
+      link(pendingSends, sender => collapse(sender).map(k))
+    def onComplete(k: Listener[T]): Unit =
+      if !poll(k) then pendingReads += k
+    def dropListener(k: Listener[T]): Unit =
+      pendingReads -= k
+
+  val canSend = new Async.Source[Listener[T]]:
+    def poll(k: Listener[Listener[T]]): Boolean =
+      link(pendingReads, k(_))
+    def onComplete(k: Listener[Listener[T]]): Unit =
+      if !poll(k) then pendingSends += k
+    def dropListener(k: Listener[Listener[T]]): Unit =
+      pendingSends -= k
+
+  def send(x: T)(using Async): Unit =
+    await(canSend)(x)
+
+  def read()(using Async): T =
+    await(canRead)
+
+end DirectSyncChannel
diff --git a/tests/pos/suspend-strawman-2/futures.scala b/tests/pos/suspend-strawman-2/futures.scala
