Use CanFilter type to distinguish FilterableChannels

Author: odersky

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

@@ -32,6 +32,9 @@ trait Async extends AsyncConfig:
 
 object Async:
 
+  /** A marker type for Source#CanFilter */
+  opaque type Yes = Unit
+
   abstract class Impl(val root: Cancellable, val scheduler: Scheduler)
      (using boundary.Label[Unit]) extends Async:
 
@@ -83,6 +86,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.
      *  `k` returns true iff the data was consumed in an async block.
@@ -105,27 +110,9 @@ object Async:
 
   end Source
 
-   /** A source that can be mapped, filtered, or raced. Only ComposableSources
-    *  can pass `false` to the `Listener` in `poll` or `onComplete`. They do
-    *  that if the data is rejected by a filter or did not come first in a race.
-    */
-  trait ComposableSource[+T] extends Source[T]:
-
-    /** Pass on data transformed by `f` */
-    def map[U](f: T => U): ComposableSource[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): ComposableSource[T] =
-      new DerivedSource[T, T](this):
-        def listen(x: T, k: Listener[T]) = p(x) && k(x)
-
-  end ComposableSource
-
   /** As source that transforms an original source in some way */
 
-  abstract class DerivedSource[T, U](src: Source[T]) extends ComposableSource[U]:
+  abstract class DerivedSource[T, U](val original: Source[T]) extends Source[U]:
 
     /** Handle a value `x` passed to the original source by possibly
      *  invokiong the continuation for this source.
@@ -137,16 +124,34 @@ object Async:
         def apply(x: T): Boolean = listen(x, k)
 
     def poll(k: Listener[U]): Boolean =
-      src.poll(transform(k))
+      original.poll(transform(k))
     def onComplete(k: Listener[U]): Unit =
-      src.onComplete(transform(k))
+      original.onComplete(transform(k))
     def dropListener(k: Listener[U]): Unit =
-      src.dropListener(transform(k))
+      original.dropListener(transform(k))
   end DerivedSource
 
+  extension [T](src: Source[T])
+
+    /** Pass on data transformed by `f` */
+    def map[U](f: T => U): Source[U] { type CanFilter = src.CanFilter } =
+      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 } =
+      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](sources: ComposableSource[T]*): ComposableSource[T] =
-    new ComposableSource:
+  def race[T, CF](sources: Source[T] { type CanFilter <: CF} *): Source[T] { type CanFilter <: CF } =
+    new Source[T]:
+      type CanFilter <: CF
 
       def poll(k: Listener[T]): Boolean =
         val it = sources.iterator
@@ -180,8 +185,11 @@ object Async:
   /** If left (respectively, right) source succeeds with `x`, pass `Left(x)`,
    *  (respectively, Right(x)) on to the continuation.
    */
-  def either[T, U](src1: ComposableSource[T], src2: ComposableSource[U]): ComposableSource[Either[T, U]] =
-    race[Either[T, U]](src1.map(Left(_)), src2.map(Right(_)))
+  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(_)))
 
 end Async
 

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

@@ -2,10 +2,14 @@ package concurrent
 import scala.collection.mutable, mutable.ListBuffer
 import scala.util.boundary.Label
 import runtime.suspend
-import Async.{Listener, await}
+import Async.{Listener, await, Yes}
+
+/** An unbounded channel
+ *  Unbounded channels are composable async sources.
+ */
+class UnboundedChannel[T] extends Async.Source[T]:
+  type CanFilter = Yes
 
-/** An unbounded channel */
-class UnboundedChannel[T] extends Async.ComposableSource[T]:
   private val pending = ListBuffer[T]()
   private val waiting = mutable.Set[Listener[T]]()
 
@@ -45,69 +49,108 @@ class UnboundedChannel[T] extends Async.ComposableSource[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.
+ */
 trait SyncChannel[T]:
-  def canRead: Async.Source[T]
-  def canSend: Async.Source[Listener[T]]
+  thisCannel =>
+
+  type CanFilter
+
+  val canRead: Async.Source[T]           { type CanFilter = thisCannel.CanFilter }
+  val canSend: Async.Source[Listener[T]] { type CanFilter = thisCannel.CanFilter }
 
   def send(x: T)(using Async): Unit = await(canSend)(x)
 
   def read()(using Async): T = await(canRead)
 
 object SyncChannel:
-  def apply[T](): SyncChannel[T] = new Impl[T]:
-    val canRead = new ReadSource
-    val canSend = new SendSource
+  def apply[T](): SyncChannel[T] = Impl[T]()
+
+  class Impl[T] extends SyncChannel[T]:
 
-  abstract class Impl[T] extends SyncChannel[T]:
-    protected val pendingReads = mutable.Set[Listener[T]]()
-    protected val pendingSends = mutable.Set[Listener[Listener[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) => op(elem); true
+        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
         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
 
-    protected class ReadSource extends Async.Source[T]:
+    private class ReadSource extends Async.Source[T]:
+      type CanFilter = Impl.this.CanFilter
       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
 
-    protected class SendSource extends Async.Source[Listener[T]]:
+    private class SendSource extends Async.Source[Listener[T]]:
+      type CanFilter = Impl.this.CanFilter
       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
-  end Impl
 
+    val canRead = new ReadSource
+    val canSend = new SendSource
+  end Impl
 end SyncChannel
 
-trait ComposableSyncChannel[T] extends SyncChannel[T]:
-  def canRead: Async.ComposableSource[T]
-  def canSend: Async.ComposableSource[Listener[T]]
-
-object ComposableSyncChannel:
-  def apply[T](): ComposableSyncChannel[T] = new Impl[T]:
-    val canRead = new ComposableReadSource
-    val canSend = new ComposableSendSource
-
-  abstract class Impl[T] extends SyncChannel.Impl[T], ComposableSyncChannel[T]:
+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
 
-    class ComposableReadSource extends ReadSource, Async.ComposableSource[T]
-    class ComposableSendSource extends SendSource, Async.ComposableSource[Listener[T]]
-  end Impl
-
-end ComposableSyncChannel
+end FilterableSyncChannel
+
+def TestRace =
+  val c1, c2 = FilterableSyncChannel[Int]()
+  val s = c1.canSend
+  val c3 = Async.race(c1.canRead, c2.canRead)
+  val c4 = c3.filter(_ >= 0)
+  val d0 = SyncChannel[Int]()
+  val d1 = Async.race(c1.canRead, c2.canRead, d0.canRead)
+  val d2 = d1.map(_ + 1)
+  val c5 = Async.either(c1.canRead, c2.canRead)
+    .map:
+      case Left(x) => -x
+      case Right(x) => x
+    .filter(_ >= 0)
+
+  //val d3bad = d1.filter(_ >= 0)
+  val d5 = Async.either(c1.canRead, d2)
+    .map:
+      case Left(x) => -x
+      case Right(x) => x
+  //val d6bad = d5.filter(_ >= 0)

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

@@ -10,7 +10,9 @@ import runtime.suspend
 
 /** A cancellable future that can suspend waiting for other synchronous sources
  */
-trait Future[+T] extends Async.ComposableSource[Try[T]], Cancellable:
+trait Future[+T] extends Async.Source[Try[T]], Cancellable:
+
+  type CanFilter = Async.Yes
 
   /** Wait for this future to be completed, return its value in case of success,
    *  or rethrow exception in case of failure.