Add OriginalSource abstract class

Author: odersky

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

@@ -95,7 +95,7 @@ object Async:
   trait Source[+T]:
 
     /** If data is available at present, pass it to function `k`
-     *  and return the result of this call.
+     *  and return the result of this call. Otherwise return false.
      *  `k` returns true iff the data was consumed in an async block.
      *  Calls to `poll` are always synchronous.
      */
@@ -122,6 +122,19 @@ object Async:
 
   end Source
 
+  /** An original source has a standard definition of `onCopmplete` in terms
+   *  of `poll` and `addListener`.
+   */
+  abstract class OriginalSource[+T] extends Source[T]:
+
+    /** Add `k` to the listener set of this source */
+    protected def addListener(k: Listener[T]): Unit
+
+    def onComplete(k: Listener[T]): Unit = synchronized:
+      if !poll(k) then addListener(k)
+
+  end OriginalSource
+
   /** A source that transforms an original source in some way */
   abstract class DerivedSource[T, U](val original: Source[T]) extends Source[U]:
 

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

@@ -8,7 +8,7 @@ import Async.{Listener, await}
 /** An unbounded asynchronous channel. Senders do not wait for matching
  *  readers.
  */
-class UnboundedChannel[T] extends Async.Source[T]:
+class UnboundedChannel[T] extends Async.OriginalSource[T]:
 
   private val pending = ListBuffer[T]()
   private val waiting = mutable.Set[Listener[T]]()
@@ -37,8 +37,8 @@ class UnboundedChannel[T] extends Async.Source[T]:
   def poll(k: Listener[T]): Boolean = synchronized:
     drainPending(k)
 
-  def onComplete(k: Listener[T]): Unit = synchronized:
-    if !drainPending(k) then waiting += k
+  def addListener(k: Listener[T]): Unit = synchronized:
+    waiting += k
 
   def dropListener(k: Listener[T]): Unit = synchronized:
     waiting -= k
@@ -79,20 +79,20 @@ object SyncChannel:
       var r: Option[T] = None
       if k2 { x => r = Some(x); true } then r else None
 
-    val canRead = new Async.Source[T]:
+    val canRead = new Async.OriginalSource[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 =
+      def addListener(k: Listener[T]) = synchronized:
+        pendingReads += k
+      def dropListener(k: Listener[T]): Unit = synchronized:
         pendingReads -= k
 
-    val canSend = new Async.Source[Listener[T]]:
+    val canSend = new Async.OriginalSource[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 =
+      def addListener(k: Listener[Listener[T]]) = synchronized:
+        pendingSends += k
+      def dropListener(k: Listener[Listener[T]]): Unit = synchronized:
         pendingSends -= k
 
 end SyncChannel

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

@@ -1,5 +1,5 @@
 import scala.util.boundary, boundary.Label
-import runtime.*
+import runtime.suspend
 
 trait Choice:
   def choose[A](choices: A*): A

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

@@ -9,7 +9,7 @@ import java.util.concurrent.CancellationException
 
 /** A cancellable future that can suspend waiting for other asynchronous sources
  */
-trait Future[+T] extends Async.Source[Try[T]], Cancellable:
+trait Future[+T] extends Async.OriginalSource[Try[T]], Cancellable:
 
   /** Wait for this future to be completed, return its value in case of success,
    *  or rethrow exception in case of failure.
@@ -56,10 +56,10 @@ object Future:
     def poll(k: Async.Listener[Try[T]]): Boolean =
       hasCompleted && k(result)
 
-    def onComplete(k: Async.Listener[Try[T]]): Unit = synchronized:
-      if !poll(k) then waiting += k
+    def addListener(k: Async.Listener[Try[T]]): Unit = synchronized:
+      waiting += k
 
-    def dropListener(k: Async.Listener[Try[T]]): Unit =
+    def dropListener(k: Async.Listener[Try[T]]): Unit = synchronized:
       waiting -= k
 
     // Cancellable method implementations

tests/run/suspend-strawman-2/monadic-reflect.scala

@@ -1,5 +1,5 @@
 import scala.util.boundary
-import runtime.*
+import runtime.suspend
 
 trait Monad[F[_]]:
 

tests/pos/suspend-strawman-2/scheduler.scala renamed to tests/run/suspend-strawman-2/scheduler.scala

@@ -2,7 +2,7 @@ package concurrent
 
 /** A hypothetical task scheduler trait */
 trait Scheduler:
-  def schedule(task: Runnable): Unit = ???
+  def schedule(task: Runnable): Unit = task.run()
 
 object Scheduler extends Scheduler:
   given fromAsyncConfig(using ac: Async.Config): Scheduler = ac.scheduler

tests/run/suspend-strawman-2/simple-futures.scala

@@ -2,7 +2,7 @@ package simpleFutures
 
 import scala.collection.mutable.ListBuffer
 import scala.util.boundary, boundary.Label
-import runtime.*
+import runtime.suspend
 import concurrent.Scheduler
 
 trait Async: