Add explanation document and align implementation with it

Author: odersky

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

@@ -19,15 +19,15 @@ trait Async:
 object Async:
 
   /** The underlying configuration of an async block */
-  case class Config(scheduler: ExecutionContext, group: Cancellable.Group)
+  case class Config(scheduler: ExecutionContext, group: CancellationGroup)
 
   trait LowPrioConfig:
 
     /** A toplevel async group with given scheduler and a synthetic root that
      *  ignores cancellation requests
      */
     given fromExecutionContext(using scheduler: ExecutionContext): Config =
-      Config(scheduler, Cancellable.Unlinked)
+      Config(scheduler, CancellationGroup.Unlinked)
 
   end LowPrioConfig
 
@@ -69,14 +69,14 @@ object Async:
   inline def await[T](src: Source[T])(using async: Async): T = async.await(src)
 
   def group[T](body: Async ?=> T)(using async: Async): T =
-    val newGroup = Cancellable.Group().link()
+    val newGroup = CancellationGroup().link()
     body(using async.withConfig(async.config.copy(group = newGroup)))
 
   /** 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 Listener[-T] extends Function[T, Boolean]
+  trait Listener[-T] extends (T => Boolean)
 
   /** A listener for values that are processed by the given source `src` and
    *  that are demanded by the continuation listener `continue`.

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

@@ -4,15 +4,15 @@ import scala.collection.mutable
 /** A trait for cancellable entities that can be grouped */
 trait Cancellable:
 
-  private var group: Cancellable.Group = Cancellable.Unlinked
+  private var group: CancellationGroup = CancellationGroup.Unlinked
 
   /** Issue a cancel request */
   def cancel(): Unit
 
   /** Add this cancellable to the given group after removing
    *  it from the previous group in which it was.
    */
-  def link(group: Cancellable.Group): this.type =
+  def link(group: CancellationGroup): this.type =
     this.group.drop(this)
     this.group = group
     this.group.add(this)
@@ -26,37 +26,37 @@ trait Cancellable:
 
   /** Unlink this cancellable from its group. */
   def unlink(): this.type =
-    link(Cancellable.Unlinked)
+    link(CancellationGroup.Unlinked)
 
-object Cancellable:
+end Cancellable
+
+class CancellationGroup extends Cancellable:
+  private var members: mutable.Set[Cancellable] = mutable.Set()
 
-  /** A group of cancellable members */
-  class Group extends Cancellable:
-    private var members: mutable.Set[Cancellable] = mutable.Set()
+  /** Cancel all members and clear the members set */
+  def cancel() =
+    members.toArray.foreach(_.cancel())
+    members.clear()
 
-    /** Cancel all members and clear the members set */
-    def cancel() =
-      members.toArray.foreach(_.cancel())
-      members.clear()
+  /** Add given member to the members set */
+  def add(member: Cancellable): Unit = synchronized:
+    members += member
 
-    /** Add given member to the members set */
-    def add(member: Cancellable): Unit = synchronized:
-      members += member
+  /** Remove given member from the members set if it is an element */
+  def drop(member: Cancellable): Unit = synchronized:
+    members -= member
 
-    /** Remove given member from the members set if it is an element */
-    def drop(member: Cancellable): Unit = synchronized:
-      members -= member
-  end Group
+object CancellationGroup:
 
   /** A sentinal group of cancellables that are in fact not linked
    *  to any real group. `cancel`, `add`, and `drop` do nothing when
    *  called on this group.
    */
-  object Unlinked extends Group:
+  object Unlinked extends CancellationGroup:
     override def cancel() = ()
     override def add(member: Cancellable): Unit = ()
     override def drop(member: Cancellable): Unit = ()
   end Unlinked
 
-end Cancellable
+end CancellationGroup
 

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

@@ -5,13 +5,25 @@ import fiberRuntime.suspend
 import scala.concurrent.ExecutionContext
 import Async.{Listener, await}
 
+/** A common interface for channels */
+trait Channel[T]:
+  def read()(using Async): T
+  def send(x: T)(using Async): Unit
+  def close(): Unit
+
+class ChannelClosedException extends Exception
+
 /** An unbounded asynchronous channel. Senders do not wait for matching
  *  readers.
  */
-class UnboundedChannel[T] extends Async.OriginalSource[T]:
+class AsyncChannel[T] extends Async.OriginalSource[T], Channel[T]:
 
   private val pending = ListBuffer[T]()
   private val waiting = mutable.Set[Listener[T]]()
+  private var isClosed = false
+
+  private def ensureOpen() =
+    if isClosed then throw ChannelClosedException()
 
   private def drainWaiting(x: T): Boolean =
     waiting.iterator.find(_(x)) match
@@ -30,11 +42,13 @@ class UnboundedChannel[T] extends Async.OriginalSource[T]:
   def read()(using Async): T = synchronized:
     await(this)
 
-  def send(x: T): Unit = synchronized:
+  def send(x: T)(using Async): Unit = synchronized:
+    ensureOpen()
     val sent = pending.isEmpty && drainWaiting(x)
     if !sent then pending += x
 
   def poll(k: Listener[T]): Boolean = synchronized:
+    ensureOpen()
     drainPending(k)
 
   def addListener(k: Listener[T]): Unit = synchronized:
@@ -43,7 +57,10 @@ class UnboundedChannel[T] extends Async.OriginalSource[T]:
   def dropListener(k: Listener[T]): Unit = synchronized:
     waiting -= k
 
-end UnboundedChannel
+  def close() =
+    isClosed = true
+
+end AsyncChannel
 
 /** An unbuffered, synchronous channel. Senders and readers both block
  *  until a communication between them happens. The channel provides two
@@ -52,7 +69,7 @@ end UnboundedChannel
  *  waiting sender the data is transmitted directly. Otherwise we add
  *  the operation to the corresponding pending set.
  */
-trait SyncChannel[T]:
+trait SyncChannel[T] extends Channel[T]:
 
   val canRead: Async.Source[T]
   val canSend: Async.Source[Listener[T]]
@@ -67,8 +84,13 @@ object SyncChannel:
 
     private val pendingReads = mutable.Set[Listener[T]]()
     private val pendingSends = mutable.Set[Listener[Listener[T]]]()
+    private var isClosed = false
+
+    private def ensureOpen() =
+      if isClosed then throw ChannelClosedException()
 
     private def link[T](pending: mutable.Set[T], op: T => Boolean): Boolean =
+      ensureOpen()
       // Since sources are filterable, we have to match all pending readers or writers
       // against the incoming request
       pending.iterator.find(op) match
@@ -95,6 +117,9 @@ object SyncChannel:
       def dropListener(k: Listener[Listener[T]]): Unit = synchronized:
         pendingSends -= k
 
+    def close() =
+      isClosed = true
+
 end SyncChannel
 
 def TestChannel(using ExecutionContext) =