PoC onCancelRequested

core/js-native/src/main/scala/cats/effect/IOFiberConstants.scala

@@ -34,6 +34,7 @@ private object IOFiberConstants {
   final val UncancelableK = 7
   final val UnmaskK = 8
   final val AttemptK = 9
+  final val OnCancelRequestedK = 10
 
   // resume ids
   final val ExecR = 0

core/js/src/main/scala/cats/effect/IOApp.scala

@@ -270,7 +270,7 @@ trait IOApp {
         case Left(Outcome.Errored(t)) => IO.raiseError(t)
         case Left(Outcome.Succeeded(code)) => code
         case Right(Outcome.Errored(t)) => IO.raiseError(t)
-        case Right(_) => sys.error("impossible")
+        case Right(_) => IO.delay(sys.error("impossible"))
       }
       .unsafeRunFiber(
         hardExit(cancelCode),

core/jvm/src/main/java/cats/effect/IOFiberConstants.java

@@ -36,6 +36,7 @@ final class IOFiberConstants {
   static final byte UncancelableK = 7;
   static final byte UnmaskK = 8;
   static final byte AttemptK = 9;
+  static final byte OnCancelRequestedK = 10;
 
   // resume ids
   static final byte ExecR = 0;

core/shared/src/main/scala/cats/effect/IO.scala

@@ -445,6 +445,9 @@ sealed abstract class IO[+A] private () extends IOPlatform[A] {
    *   [[uncancelable]]
    * @see
    *   [[onCancel]]
+   * @note
+   *   this method can lose data if `fa` completes before `fin` can stop it. See
+   *   [[IO!.onCancelRequested]] for a safer alternative.
    */
   def cancelable(fin: IO[Unit]): IO[A] =
     Spawn[IO].cancelable(this, fin)
@@ -590,6 +593,10 @@ sealed abstract class IO[+A] private () extends IOPlatform[A] {
   def onCancel(fin: IO[Unit]): IO[A] =
     IO.OnCancel(this, fin)
 
+  def onCancelRequested(ack: IO[Unit]): IO[A] = {
+    IO.OnCancelRequested(this, ack)
+  }
+
   @deprecated("Use onError with PartialFunction argument", "3.6.0")
   private[effect] def onError(f: Throwable => IO[Unit]): IO[A] = {
     val pf: PartialFunction[Throwable, IO[Unit]] = { case t => f(t).reportError }
@@ -850,7 +857,7 @@ sealed abstract class IO[+A] private () extends IOPlatform[A] {
   def timeoutTo[A2 >: A](duration: Duration, fallback: IO[A2]): IO[A2] = {
     handleDuration[IO[A2]](duration, this) { finiteDuration =>
       IO.uncancelable { poll =>
-        poll(racePair(IO.sleep(finiteDuration))) flatMap {
+        racePair(IO.sleep(finiteDuration)) flatMap {
           case Left((oc, f)) => f.cancel *> oc.embed(poll(IO.canceled) *> IO.never)
           case Right((f, _)) =>
             f.cancel *> f.join.flatMap { oc => oc.fold(fallback, IO.raiseError, identity) }
@@ -1367,6 +1374,46 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara
     IOCont(body, Tracing.calculateTracingEvent(k))
   }
 
+  /**
+   * Suspends an asynchronous side effect with optional immediate result in `IO`.
+   *
+   * The given function `k` will be invoked during evaluation of the `IO` to:
+   *   - check if result is already available;
+   *   - "schedule" the asynchronous callback, where the callback of type `Either[Throwable, A]
+   *     \=> Unit` is the parameter passed to that function. Only the ''first'' invocation of
+   *     the callback will be effective! All subsequent invocations will be silently dropped.
+   *
+   * The process of registering the callback itself is suspended in `IO` (the outer `IO` of
+   * `IO[Either[Option[IO[Unit]], A]]`).
+   *
+   * The effect returns `Either[Option[IO[Unit]], A]` where:
+   *   - right side `A` is an immediate result of computation (callback invocation will be
+   *     dropped);
+   *   - left side `Option[IO[Unit]]` is an optional acknowledgement to be run in the event that
+   *     cancelation of the fiber running `asyncCheckAttemptParCancelable(k)` is requested.
+   *
+   * Note that `asyncCheckAttemptParCancelable` is uncancelable during its registration.
+   *
+   * @see
+   *   [[async]] for a simplified variant without an option for immediate result
+   */
+  def asyncCheckAttemptParCancelable[A](
+      k: (Either[Throwable, A] => Unit) => IO[Either[Option[IO[Unit]], A]]): IO[A] = {
+    val body = new Cont[IO, A, A] {
+      def apply[G[_]](implicit G: MonadCancel[G, Throwable]) = { (resume, get, lift) =>
+        G.uncancelable { _ =>
+          lift(k(resume)) flatMap {
+            case Right(a) => G.pure(a)
+            case Left(Some(fin)) => G.onCancelRequested(get, lift(fin))
+            case Left(None) => get
+          }
+        }
+      }
+    }
+
+    IOCont(body, Tracing.calculateTracingEvent(k))
+  }
+
   /**
    * Suspends an asynchronous side effect in `IO`.
    *
@@ -1430,6 +1477,50 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara
     IOCont(body, Tracing.calculateTracingEvent(k))
   }
 
+  /**
+   * Suspends an asynchronous side effect in `IO`.
+   *
+   * The given function `k` will be invoked during evaluation of the `IO` to "schedule" the
+   * asynchronous callback, where the callback of type `Either[Throwable, A] => Unit` is the
+   * parameter passed to that function. Only the ''first'' invocation of the callback will be
+   * effective! All subsequent invocations will be silently dropped.
+   *
+   * The process of registering the callback itself is suspended in `IO` (the outer `IO` of
+   * `IO[Option[IO[Unit]]]`).
+   *
+   * The effect returns `Option[IO[Unit]]` which is an optional cancelation acknowledgement to
+   * be run in the event that cancelation of the fiber running `asyncParCancelable(k)` is
+   * requested.
+   *
+   * @note
+   *   `asyncParCancelable` is always uncancelable during its registration. The created effect
+   *   will be uncancelable during its execution if the registration callback provides no
+   *   cancelation acknowledgement (i.e. evaluates to `None`). If you need the created task to
+   *   be cancelable, return a acknowledgement effect upon the registration. In a rare case when
+   *   there's nothing to finalize, you can return `Some(IO.unit)` for that.
+   *
+   * @see
+   *   [[async_]] for a simplified uncancelable variant
+   * @see
+   *   [[asyncCheckAttemptParCancelable]] for more generic version providing an optional
+   *   immediate result of computation
+   */
+  def asyncParCancelable[A](
+      k: (Either[Throwable, A] => Unit) => IO[Option[IO[Unit]]]): IO[A] = {
+    val body = new Cont[IO, A, A] {
+      def apply[G[_]](implicit G: MonadCancel[G, Throwable]) = { (resume, get, lift) =>
+        G.uncancelable { _ =>
+          lift(k(resume)) flatMap {
+            case Some(fin) => G.onCancelRequested(get, lift(fin))
+            case None => get
+          }
+        }
+      }
+    }
+
+    IOCont(body, Tracing.calculateTracingEvent(k))
+  }
+
   /**
    * Suspends an asynchronous side effect in `IO`.
    *
@@ -1765,10 +1856,20 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara
 
   /**
    * Like [[fromFuture]], but is cancelable via the provided finalizer.
+   *
+   * @see
+   *   [[IO!.onCancelRequested]] for a safer alternative. This method can lose data if the
+   *   future completes before the finalizaer can stop it.
    */
   def fromFutureCancelable[A](fut: IO[(Future[A], IO[Unit])]): IO[A] =
     asyncForIO.fromFutureCancelable(fut)
 
+  /**
+   * Like [[fromFuture]], but is cancelable asynchronously.
+   */
+  def fromFutureParCancelable[A](fut: IO[(Future[A], IO[Unit])]): IO[A] =
+    asyncForIO.fromFutureParCancelable(fut)
+
   /**
    * Run two IO tasks concurrently, and return the first to finish, either in success or error.
    * The loser of the race is canceled.
@@ -2059,6 +2160,9 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara
     def onCancel[A](ioa: IO[A], fin: IO[Unit]): IO[A] =
       ioa.onCancel(fin)
 
+    override def onCancelRequested[A](ioa: IO[A], ack: IO[Unit]): IO[A] =
+      ioa.onCancelRequested(ack)
+
     override def bracketFull[A, B](acquire: Poll[IO] => IO[A])(use: A => IO[B])(
         release: (A, OutcomeIO[B]) => IO[Unit]): IO[B] =
       IO.bracketFull(acquire)(use)(release)
@@ -2328,6 +2432,10 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara
     def tag = 24
   }
 
+  private[effect] final case class OnCancelRequested[A](f: IO[A], ack: IO[Unit]) extends IO[A] {
+    def tag = 25
+  }
+
   // INTERNAL, only created by the runloop itself as the terminal state of several operations
   private[effect] case object EndFiber extends IO[Nothing] {
     def tag = -1