Timer and AsyncOperations

Author: natsukagami

jvm/src/main/scala/async/JvmAsyncOperations.scala

@@ -1,5 +1,7 @@
 package gears.async
 
+import language.experimental.captureChecking
+
 object JvmAsyncOperations extends AsyncOperations:
 
   private def jvmInterruptible[T](fn: => T)(using Async): T =

jvm/src/main/scala/async/VThreadSupport.scala

@@ -1,5 +1,7 @@
 package gears.async
 
+import language.experimental.captureChecking
+
 import java.lang.invoke.{MethodHandles, VarHandle}
 import java.util.concurrent.locks.ReentrantLock
 import scala.annotation.unchecked.uncheckedVariance
@@ -11,13 +13,15 @@ object VThreadScheduler extends Scheduler:
     .name("gears.async.VThread-", 0L)
     .factory()
 
-  override def execute(body: Runnable): Unit =
+  override def execute(body: Runnable^): Unit =
     val th = VTFactory.newThread(body)
     th.start()
 
-  override def schedule(delay: FiniteDuration, body: Runnable): Cancellable = ScheduledRunnable(delay, body)
+  override def schedule(delay: FiniteDuration, body: Runnable^): Cancellable =
+    val sr = ScheduledRunnable(delay, body)
+    () => sr.cancel()
 
-  private class ScheduledRunnable(val delay: FiniteDuration, val body: Runnable) extends Cancellable {
+  private class ScheduledRunnable(val delay: FiniteDuration, val body: Runnable^) extends Cancellable {
     @volatile var interruptGuard = true // to avoid interrupting the body
 
     val th = VTFactory.newThread: () =>

shared/src/main/scala/async/Async.scala

@@ -372,21 +372,21 @@ object Async:
     * @see
     *   [[Async$.select Async.select]] where [[SelectCase]] is used.
     */
-  opaque type SelectCase[T] = (Source[?], Nothing => T)
-  //                           ^ unsafe types, but we only construct SelectCase from `handle` which is safe
+  case class SelectCase[+T] private[Async] (src: Source[Any]^, f: Nothing => T)
+  //                                        ^ unsafe types, but we only construct SelectCase from `handle` which is safe
 
   extension [T](src: Source[T]^)
     /** Attach a handler to `src`, creating a [[SelectCase]].
       * @see
       *   [[Async$.select Async.select]] where [[SelectCase]] is used.
       */
-    inline def handle[U](f: T => U): SelectCase[U]^{src, f} = (src, f)
+    def handle[U](f: T => U): SelectCase[U]^{src, f} = SelectCase(src, f)
 
     /** Alias for [[handle]]
       * @see
       *   [[Async$.select Async.select]] where [[SelectCase]] is used.
       */
-    inline def ~~>[U](f: T => U): SelectCase[U]^{src, f} = src.handle(f)
+    def ~~>[U](f: T => U): SelectCase[U]^{src, f} = src.handle(f)
 
   /** Race a list of sources with the corresponding handler functions, once an item has come back. Like [[race]],
     * [[select]] guarantees exactly one of the sources are polled. Unlike [[transformValuesWith]], the handler in
@@ -410,7 +410,7 @@ object Async:
     */
   def select[T](cases: (SelectCase[T]^)*)(using Async) =
     val (input, which) = raceWithOrigin(cases.map(_._1)*).awaitResult
-    val (_, handler) = cases.find(_._1.symbol == which).get
+    val SelectCase(_, handler) = cases.find(_._1.symbol == which).get
     handler.asInstanceOf[input.type => T](input)
 
   /** Race two sources, wrapping them respectively in [[Left]] and [[Right]] cases.

shared/src/main/scala/async/AsyncOperations.scala

@@ -1,6 +1,6 @@
 package gears.async
 
-import gears.async.AsyncOperations.sleep
+import language.experimental.captureChecking
 
 import java.util.concurrent.TimeoutException
 import scala.concurrent.duration.FiniteDuration
@@ -19,14 +19,14 @@ object AsyncOperations:
     * @param millis
     *   The duration to suspend, in milliseconds. Must be a positive integer.
     */
-  inline def sleep(millis: Long)(using AsyncOperations, Async): Unit =
+  def sleep(millis: Long)(using AsyncOperations, Async): Unit =
     summon[AsyncOperations].sleep(millis)
 
   /** Suspends the current [[Async]] context for `duration`.
     * @param duration
     *   The duration to suspend. Must be positive.
     */
-  inline def sleep(duration: FiniteDuration)(using AsyncOperations, Async): Unit =
+  def sleep(duration: FiniteDuration)(using AsyncOperations, Async): Unit =
     sleep(duration.toMillis)
 
 /** Runs [[op]] with a timeout. When the timeout occurs, [[op]] is cancelled through the given [[Async]] context, and
@@ -36,7 +36,7 @@ def withTimeout[T](timeout: FiniteDuration)(op: Async ?=> T)(using AsyncOperatio
   Async.group:
     Async.select(
       Future(op).handle(_.get),
-      Future(sleep(timeout)).handle: _ =>
+      Future(AsyncOperations.sleep(timeout)).handle: _ =>
         throw TimeoutException()
     )
 
@@ -47,5 +47,5 @@ def withTimeoutOption[T](timeout: FiniteDuration)(op: Async ?=> T)(using AsyncOp
   Async.group:
     Async.select(
       Future(op).handle(v => Some(v.get)),
-      Future(sleep(timeout)).handle(_ => None)
+      Future(AsyncOperations.sleep(timeout)).handle(_ => None)
     )

shared/src/main/scala/async/AsyncSupport.scala

@@ -1,7 +1,9 @@
 package gears.async
 
 import language.experimental.captureChecking
+
 import scala.concurrent.duration._
+import scala.annotation.capability
 
 /** The delimited continuation suspension interface. Represents a suspended computation asking for a value of type `T`
   * to continue (and eventually returning a value of type `R`).

shared/src/main/scala/async/Timer.scala

@@ -1,5 +1,7 @@
 package gears.async
 
+import language.experimental.captureChecking
+
 import gears.async.Listener
 
 import java.util.concurrent.CancellationException
@@ -9,10 +11,12 @@ import scala.collection.mutable
 import scala.concurrent.TimeoutException
 import scala.concurrent.duration._
 import scala.util.{Failure, Success, Try}
+import scala.annotation.unchecked.uncheckedCaptures
 
 import AsyncOperations.sleep
 import Future.Promise
 
+
 /** Timer exposes a steady [[Async.Source]] of ticks that happens every `tickDuration` milliseconds. Note that the timer
   * does not start ticking until `start` is called (which is a blocking operation, until the timer is cancelled).
   *
@@ -27,23 +31,32 @@ class Timer(tickDuration: Duration) extends Cancellable {
   private var isCancelled = false
 
   private object Source extends Async.OriginalSource[this.TimerEvent] {
-    val listeners = mutable.Set[Listener[TimerEvent]]()
-    def tick() = synchronized {
+    private val listeners : mutable.Set[(Listener[TimerEvent]^) @uncheckedCaptures] =
+      mutable.Set[(Listener[TimerEvent]^) @uncheckedCaptures]()
+
+    def tick(): Unit = synchronized {
       listeners.filterInPlace(l =>
-        l.completeNow(TimerEvent.Tick, this)
+        l.completeNow(TimerEvent.Tick, src)
         false
       )
     }
-    override def poll(k: Listener[TimerEvent]): Boolean =
+    override def poll(k: Listener[TimerEvent]^): Boolean =
       if isCancelled then k.completeNow(TimerEvent.Cancelled, this)
       else false // subscribing to a timer always takes you to the next tick
-    override def dropListener(k: Listener[TimerEvent]): Unit = listeners -= k
-    override protected def addListener(k: Listener[TimerEvent]): Unit =
+    override def dropListener(k: Listener[TimerEvent]^): Unit = listeners -= k
+    override protected def addListener(k: Listener[TimerEvent]^): Unit =
       if isCancelled then k.completeNow(TimerEvent.Cancelled, this)
       else
         Timer.this.synchronized:
           if isCancelled then k.completeNow(TimerEvent.Cancelled, this)
           else listeners += k
+
+    def cancel(): Unit =
+      synchronized { isCancelled = true }
+      src.synchronized {
+        Source.listeners.foreach(_.completeNow(TimerEvent.Cancelled, src))
+        Source.listeners.clear()
+      }
   }
 
   /** Ticks of the timer are delivered through this source. Note that ticks are ephemeral. */
@@ -62,10 +75,6 @@ class Timer(tickDuration: Duration) extends Cancellable {
       Source.tick()
       loop()
 
-  override def cancel(): Unit =
-    synchronized { isCancelled = true }
-    src.synchronized {
-      Source.listeners.foreach(_.completeNow(TimerEvent.Cancelled, src))
-      Source.listeners.clear()
-    }
+  override def cancel(): Unit = Source.cancel()
 }
+

shared/src/main/scala/async/futures.scala

@@ -108,9 +108,11 @@ object Future:
   end CoreFuture
 
   private class CancelSuspension[U](val src: Async.Source[U]^)(val ac: Async, val suspension: ac.support.Suspension[Try[U], Unit]) extends Cancellable:
+   self: CancelSuspension[U]^{src, ac} =>
     var listener: Listener[U]^{ac} = Listener.acceptingListener[U]: (x, _) =>
       val completedBefore = complete()
-      if !completedBefore then ac.support.resumeAsync(suspension)(Success(x))
+      if !completedBefore then
+        ac.support.resumeAsync(suspension)(Success(x))
       unlink()
     var completed = false
 

shared/src/test/scala/TimerBehavior.scala

@@ -1,3 +1,5 @@
+import language.experimental.captureChecking
+
 import gears.async.AsyncOperations._
 import gears.async._
 
@@ -25,7 +27,7 @@ class TimerBehavior extends munit.FunSuite {
       assert(timer.src.awaitResult == timer.TimerEvent.Tick)
   }
 
-  def `cancel future after timeout`[T](d: Duration, f: Future[T])(using Async, AsyncOperations): Try[T] =
+  def `cancel future after timeout`[T](d: Duration, f: Future[T]^)(using Async, AsyncOperations): Try[T] =
     Async.group:
       f.link()
       val t = Future { sleep(d.toMillis) }