Drop distinction between filterable and normal sync channels

Filterable sync channels don't require significantly more complexity than
normal channels. The only thing a normal channel could do that a filterable
channel could not, was to discard a reader and sender immediately if there
was no match. If there are no filters on the source, then the only way a
source could not match is by losing a race. But then the source will be
removed anyway by the `race` method. So the only improvement would be had
if there was a data race condition where the removal by race happens after
a second channel was triggered.

Author: odersky

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

@@ -42,10 +42,3 @@ object Future:
 def Test(x: Future[Int], xs: List[Future[Int]]) =
   Future:
     x.await + xs.map(_.await).sum
-
-
-
-
-
-
-

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

@@ -4,43 +4,52 @@ 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 extends AsyncConfig:
+/** A context that allows to suspend waiting for asynchronous data sources
+ */
+trait Async extends Async.Config:
 
   /** Wait for completion of async source `src` and return the result */
   def await[T](src: Async.Source[T]): T
 
 object Async:
 
-  /** A marker type for Source#CanFilter */
-  opaque type Yes = Unit
+  /** The underlying configuration of an async block */
+  trait Config:
+
+    /** The cancellable async source underlying this async computation */
+    def root: Cancellable
+
+    /** The scheduler for runnables defined in this async computation */
+    def scheduler: Scheduler
 
+  object Config:
+
+    /** 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 Config
+
+  /** A possible implementation of Async. Defines an `await` method based
+   *  on a method to check for cancellation that needs to be implemented by
+   *  subclasses.
+   *
+   *   @param   root      the root of the Async's config
+   *   @param   scheduler the scheduler of the Async's config
+   *   @param   label     the label of the boundary that defines the representedd async block
+   */
   abstract class Impl(val root: Cancellable, val scheduler: Scheduler)
-     (using boundary.Label[Unit]) extends Async:
+     (using label: boundary.Label[Unit]) extends Async:
 
     protected def checkCancellation(): Unit
 
-    def await[T](src: Async.Source[T]): T =
+    /** Await a source first by polling it, and, if that fails, by suspending
+     *  in a onComplete call.
+     */
+    def await[T](src: Source[T]): T =
       checkCancellation()
       var resultOpt: Option[T] = None
       src.poll: x =>
@@ -76,7 +85,7 @@ object Async:
   /** 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]
+  abstract case class FinalListener[T](apply: T => Boolean) extends Listener[T]
 
   /** An asynchronous data source. Sources can be persistent or ephemeral.
    *  A persistent source will always pass same data to calls of `poll and `onComplete`.
@@ -86,10 +95,8 @@ object Async:
    */
   trait Source[+T]:
 
-    type CanFilter
-
     /** If data is available at present, pass it to function `k`
-     *  and return the result if this call.
+     *  and return the result of this call.
      *  `k` returns true iff the data was consumed in an async block.
      *  Calls to `poll` are always synchronous.
      */
@@ -104,14 +111,13 @@ object Async:
 
     /** 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.
+     *  to drop the  listener from their waiting sets.
      */
     def dropListener(k: Listener[T]): Unit
 
   end Source
 
-  /** As source that transforms an original source in some way */
-
+  /** A source that transforms an original source in some way */
   abstract class DerivedSource[T, U](val original: Source[T]) extends Source[U]:
 
     /** Handle a value `x` passed to the original source by possibly
@@ -134,24 +140,18 @@ object Async:
   extension [T](src: Source[T])
 
     /** Pass on data transformed by `f` */
-    def map[U](f: T => U): Source[U] { type CanFilter = src.CanFilter } =
+    def map[U](f: T => U): Source[U]  =
       new DerivedSource[T, U](src):
-        type CanFilter = src.CanFilter
         def listen(x: T, k: Listener[U]) = k(f(x))
 
-  extension [T](src: Source[T] { type CanFilter = Yes })
-
     /** Pass on only data matching the predicate `p` */
-    def filter(p: T => Boolean): Source[T] { type CanFilter = src.CanFilter } =
+    def filter(p: T => Boolean): Source[T] =
       new DerivedSource[T, T](src):
-        type CanFilter = src.CanFilter
         def listen(x: T, k: Listener[T]) = p(x) && k(x)
 
-
   /** Pass first result from any of `sources` to the continuation */
-  def race[T, CF](sources: Source[T] { type CanFilter <: CF} *): Source[T] { type CanFilter <: CF } =
+  def race[T](sources: Source[T]*): Source[T] =
     new Source[T]:
-      type CanFilter <: CF
 
       def poll(k: Listener[T]): Boolean =
         val it = sources.iterator
@@ -185,11 +185,8 @@ object Async:
   /** If left (respectively, right) source succeeds with `x`, pass `Left(x)`,
    *  (respectively, Right(x)) on to the continuation.
    */
-  def either[T, U, CF](
-      src1: Source[T] { type CanFilter <: CF },
-      src2: Source[U] { type CanFilter <: CF })
-    : Source[Either[T, U]] { type CanFilter <: CF } =
-    race[Either[T, U], CF](src1.map(Left(_)), src2.map(Right(_)))
+  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
 

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

@@ -2,25 +2,20 @@ package concurrent
 import scala.collection.mutable, mutable.ListBuffer
 import scala.util.boundary.Label
 import runtime.suspend
-import Async.{Listener, await, Yes}
+import Async.{Listener, await}
 
-/** An unbounded channel
- *  Unbounded channels are composable async sources.
+/** An unbounded asynchronous channel. Senders do not wait for matching
+ *  readers.
  */
 class UnboundedChannel[T] extends Async.Source[T]:
-  type CanFilter = Yes
 
   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
+    waiting.iterator.find(_(x)) match
+      case Some(k) => waiting -= k; true
+      case None => false
 
   private def drainPending(k: Listener[T]): Boolean =
     val sent = pending.nonEmpty && k(pending.head)
@@ -50,92 +45,59 @@ class UnboundedChannel[T] extends Async.Source[T]:
 end UnboundedChannel
 
 /** An unbuffered, synchronous channel. Senders and readers both block
- *  until a communication between them happens.
- *  The channel provides two async sources, one for reading and one for
- *  sending. The two sources are not composable. This allows a simple
- *  implementation strategy where at each point either some senders
- *  are waiting for matching readers, or some readers are waiting for matching
- *  senders, or the channel is idle, i.e. there are no waiting readers or senders.
- *  If a send operation encounters some waiting readers, or a read operation
- *  encounters some waiting sender the data is transmitted directly. Otherwise
- *  we add the operation to the corresponding waiting pending set.
+ *  until a communication between them happens. The channel provides two
+ *  async sources, one for reading and one for sending. If a send operation
+ *  encounters some waiting readers, or a read operation encounters some
+ *  waiting sender the data is transmitted directly. Otherwise we add
+ *  the operation to the corresponding pending set.
  */
 trait SyncChannel[T]:
-  thisCannel =>
-
-  type CanFilter
 
-  val canRead: Async.Source[T]           { type CanFilter = thisCannel.CanFilter }
-  val canSend: Async.Source[Listener[T]] { type CanFilter = thisCannel.CanFilter }
+  val canRead: Async.Source[T]
+  val canSend: Async.Source[Listener[T]]
 
   def send(x: T)(using Async): Unit = await(canSend)(x)
 
   def read()(using Async): T = await(canRead)
 
 object SyncChannel:
-  def apply[T](): SyncChannel[T] = Impl[T]()
 
-  class Impl[T] extends SyncChannel[T]:
+  def apply[T](): SyncChannel[T] = new SyncChannel[T]:
 
     private val pendingReads = mutable.Set[Listener[T]]()
     private val pendingSends = mutable.Set[Listener[Listener[T]]]()
 
-    protected def link[T](pending: mutable.Set[T], op: T => Boolean): Boolean =
-      pending.headOption match
-        case Some(elem) =>
-          val ok = op(elem)
-          if !ok then
-            // Since sources are not filterable, we can be here only if a race
-            // was lost and the entry was not yet removed. In that case, remove
-            // it here.
-            pending -= pending.head
-            link(pending, op)
-          ok
+    private def link[T](pending: mutable.Set[T], op: T => Boolean): Boolean =
+      // Since sources are filterable, we have to match all pending readers or writers
+      // against the incoming request
+      pending.iterator.find(op) match
+        case Some(elem) => pending -= elem; true
         case None => false
 
     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 class ReadSource extends Async.Source[T]:
-      type CanFilter = Impl.this.CanFilter
+    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
 
-    private class SendSource extends Async.Source[Listener[T]]:
-      type CanFilter = Impl.this.CanFilter
+    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
 
-    val canRead = new ReadSource
-    val canSend = new SendSource
-  end Impl
 end SyncChannel
 
-object FilterableSyncChannel:
-  def apply[T](): SyncChannel[T] { type CanFilter = Yes } = Impl[T]()
-
-  class Impl[T] extends SyncChannel.Impl[T]:
-    type CanFilter = Yes
-    override protected def link[T](pending: mutable.Set[T], op: T => Boolean): Boolean =
-      // Since sources are filterable, we have to match all pending readers or writers
-      // against the incoming request
-      pending.iterator.find(op) match
-        case Some(elem) => pending -= elem; true
-        case None => false
-
-end FilterableSyncChannel
-
 def TestRace =
-  val c1, c2 = FilterableSyncChannel[Int]()
+  val c1, c2 = SyncChannel[Int]()
   val s = c1.canSend
   val c3 = Async.race(c1.canRead, c2.canRead)
   val c4 = c3.filter(_ >= 0)
@@ -148,9 +110,7 @@ def TestRace =
       case Right(x) => x
     .filter(_ >= 0)
 
-  //val d3bad = d1.filter(_ >= 0)
-  val d5 = Async.either(c1.canRead, d2)
+    val d5 = Async.either(c1.canRead, d2)
     .map:
       case Left(x) => -x
       case Right(x) => x
-  //val d6bad = d5.filter(_ >= 0)