diff --git a/core/js-native/src/main/scala/cats/effect/IOFiberConstants.scala b/core/js-native/src/main/scala/cats/effect/IOFiberConstants.scala index b93f462ef3..796171d90a 100644 --- a/core/js-native/src/main/scala/cats/effect/IOFiberConstants.scala +++ b/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 diff --git a/core/js/src/main/scala/cats/effect/IOApp.scala b/core/js/src/main/scala/cats/effect/IOApp.scala index 8133b54fee..1bd9235c59 100644 --- a/core/js/src/main/scala/cats/effect/IOApp.scala +++ b/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), diff --git a/core/jvm/src/main/java/cats/effect/IOFiberConstants.java b/core/jvm/src/main/java/cats/effect/IOFiberConstants.java index c4310aea05..1eeb285449 100644 --- a/core/jvm/src/main/java/cats/effect/IOFiberConstants.java +++ b/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; diff --git a/core/shared/src/main/scala/cats/effect/IO.scala b/core/shared/src/main/scala/cats/effect/IO.scala index 41fb79db63..5e25e72208 100644 --- a/core/shared/src/main/scala/cats/effect/IO.scala +++ b/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 diff --git a/core/shared/src/main/scala/cats/effect/IOFiber.scala b/core/shared/src/main/scala/cats/effect/IOFiber.scala index e34f3d6586..03fba20271 100644 --- a/core/shared/src/main/scala/cats/effect/IOFiber.scala +++ b/core/shared/src/main/scala/cats/effect/IOFiber.scala @@ -103,6 +103,10 @@ private final class IOFiber[A]( private[this] var masks: Int = 0 private[this] var finalizing: Boolean = false + // async cancelation handling + private[this] val acks: ArrayStack[IO[Unit]] = ArrayStack() + private[this] var startedAcks: Boolean = false + @volatile private[this] var outcome: OutcomeIO[A] = _ @@ -142,14 +146,14 @@ private final class IOFiber[A]( private[this] var _cancel: IO[Unit] = IO uncancelable { _ => canceled = true - // println(s"${name}: attempting cancelation") + // println(s"${this}: attempting cancelation") /* check to see if the target fiber is suspended */ if (resume()) { /* ...it was! was it masked? */ if (isUnmasked()) { /* ...nope! take over the target fiber's runloop and run the finalizers */ - // println(s"<$name> running cancelation (finalizers.length = ${finalizers.unsafeIndex()})") + // println(s"$this: running cancelation (finalizers.length = ${finalizers.unsafeIndex()})") /* if we have async finalizers, runLoop may return early */ IO.async_[Unit] { fin => @@ -160,15 +164,17 @@ private final class IOFiber[A]( scheduleFiber(ec, this) } } else { + // println(s"$this: masked, it will cancel) /* * it was masked, so we need to wait for it to finish whatever * it was doing and cancel itself */ + acknowledgeCancelation() suspend() /* allow someone else to take the runloop */ join.void } } else { - // println(s"${name}: had to join") + // println(s"$this: had to join") /* it's already being run somewhere; await the finalizers */ join.void } @@ -239,6 +245,7 @@ private final class IOFiber[A]( } } + acknowledgeCancelation() if (shouldFinalize()) { val fin = prepareFiberForCancelation(null) runLoop(fin, nextCancelation, nextAutoCede) @@ -531,6 +538,7 @@ private final class IOFiber[A]( /* Canceled */ case 10 => canceled = true + acknowledgeCancelation() if (isUnmasked()) { /* run finalizers immediately */ val fin = prepareFiberForCancelation(null) @@ -822,7 +830,7 @@ private final class IOFiber[A]( * race condition check: we may have been canceled * after setting the state but before we suspended */ - if (shouldFinalize()) { + if (canceled && (isUnmasked() || !startedAcks)) { /* * if we can re-acquire the run-loop, we can finalize, * otherwise somebody else acquired it and will eventually finalize. @@ -832,6 +840,7 @@ private final class IOFiber[A]( * finalisers. */ if (resume()) { + acknowledgeCancelation() if (shouldFinalize()) { val fin = prepareFiberForCancelation(null) runLoop(fin, nextCancelation, nextAutoCede) @@ -877,6 +886,7 @@ private final class IOFiber[A]( * we were canceled, but `cancel` cannot run the finalisers * because the runloop was not suspended, so we have to run them */ + acknowledgeCancelation() val fin = prepareFiberForCancelation(null) runLoop(fin, nextCancelation, nextAutoCede) } @@ -909,44 +919,39 @@ private final class IOFiber[A]( val cur = cur0.asInstanceOf[RacePair[Any, Any]] val next = - IO.async[Either[(OutcomeIO[Any], FiberIO[Any]), (FiberIO[Any], OutcomeIO[Any])]] { - cb => - IO { - val ec = currentCtx - val rt = runtime - - val fiberA = new IOFiber[Any]( - localState, - null, - cur.ioa, - ec, - rt - ) - - val fiberB = new IOFiber[Any]( - localState, - null, - cur.iob, - ec, - rt - ) - - fiberA.setCallback(oc => cb(Right(Left((oc, fiberB))))) - fiberB.setCallback(oc => cb(Right(Right((fiberA, oc))))) - - scheduleFiber(ec, fiberA) - scheduleFiber(ec, fiberB) - - val cancel = - for { - cancelA <- fiberA.cancel.start - cancelB <- fiberB.cancel.start - _ <- cancelA.join - _ <- cancelB.join - } yield () - - Some(cancel) - } + IO.asyncParCancelable[ + Either[(OutcomeIO[Any], FiberIO[Any]), (FiberIO[Any], OutcomeIO[Any])]] { cb => + IO { + val ec = currentCtx + val rt = runtime + + val fiberA = new IOFiber[Any]( + localState, + null, + cur.ioa, + ec, + rt + ) + + val fiberB = new IOFiber[Any]( + localState, + null, + cur.iob, + ec, + rt + ) + + fiberA.setCallback(oc => cb(Right(Left((oc, fiberB))))) + fiberB.setCallback(oc => cb(Right(Right((fiberA, oc))))) + + scheduleFiber(ec, fiberA) + scheduleFiber(ec, fiberB) + + val cancel = + fiberA.cancel.start.flatMap(cancelA => fiberB.cancel *> cancelA.join.void) + + Some(cancel) + } } runLoop(next, nextCancelation, nextAutoCede) @@ -1068,6 +1073,30 @@ private final class IOFiber[A]( /* ReadRT */ case 24 => runLoop(succeeded(runtime, 0), nextCancelation, nextAutoCede) + + /* OnCancelRequested */ + case 25 => + val cur = cur0.asInstanceOf[OnCancelRequested[Any]] + val ack = EvalOn(cur.ack, currentCtx) + + // otherwise it is too late to request cancelation + if (!finalizing) { + val push = + if (startedAcks) + runAcknowledgement(ack) // already started, run immediately + else + ack + acks.push(push) + conts = ByteStack.push(conts, OnCancelRequestedK) + } + + runLoop(cur.f, nextCancelation, nextAutoCede) + case 26 => + if (!startedAcks) { + val _ = acks.pop() + } + + runLoop(IO.unit, nextCancelation, nextAutoCede) } } } @@ -1118,6 +1147,7 @@ private final class IOFiber[A]( conts = null objectState.invalidate() finalizers.invalidate() + acks.invalidate() currentCtx = null if (isStackTracing) { @@ -1130,7 +1160,7 @@ private final class IOFiber[A]( * because cancelation has been triggered. */ private[this] def prepareFiberForCancelation(cb: Either[Throwable, Unit] => Unit): IO[Any] = { - if (!finalizers.isEmpty()) { + if (!finalizers.isEmpty() || !acks.isEmpty()) { if (!finalizing) { // Do not nuke the fiber execution state repeatedly. finalizing = true @@ -1146,7 +1176,8 @@ private final class IOFiber[A]( } // Return the first finalizer for execution. - finalizers.pop() + if (!acks.isEmpty()) acks.pop() + else finalizers.pop() } else { // There are no finalizers to execute. @@ -1163,6 +1194,39 @@ private final class IOFiber[A]( } } + private[this] def acknowledgeCancelation(): Unit = { + if (shouldAcknowledgeCancelation()) { + startedAcks = true + + // Replace all of the pending acks with running acks unsafely to minimize allocations + var i = acks.unsafeIndex() + val ackBuffer = acks.unsafeBuffer() + + while (i > 0) { + i -= 1 + val acknowledgement = ackBuffer(i).asInstanceOf[IO[Unit]] + ackBuffer(i) = runAcknowledgement(acknowledgement) + } + + } + } + + private[this] def runAcknowledgement(acknowledgement: IO[Unit]): IO[Unit] = { + // println(s"$this: starting cancelation acknowledgement in thread ${Thread.currentThread()}") + val ec = currentCtx + val rt = runtime + + val runningAcknowledgement = new IOFiber[Unit]( + localState, + null, + acknowledgement, + ec, + rt + ) + scheduleFiber(ec, runningAcknowledgement) + runningAcknowledgement.join.flatMap(_.embed(IO.unit)) + } + /* * We should attempt finalization if all of the following are true: * 1) We own the runloop @@ -1175,6 +1239,9 @@ private final class IOFiber[A]( private[this] def isUnmasked(): Boolean = masks == 0 + private[this] def shouldAcknowledgeCancelation(): Boolean = + canceled && !finalizing && !startedAcks + /* * You should probably just read this as `suspended.compareAndSet(true, false)`. * This implementation has the same semantics as the above, except that it guarantees @@ -1265,6 +1332,14 @@ private final class IOFiber[A]( case 9 => // attemptK succeeded(Right(result), depth) + + case 10 => // onCancelRequestedSuccessK + if (startedAcks) { + acks.pop().as(result) + } else { + val _ = acks.pop() + succeeded(result, depth + 1) + } } private[this] def failed(error: Throwable, depth: Int): IO[Any] = { @@ -1327,6 +1402,14 @@ private final class IOFiber[A]( failed(error, depth + 1) case 9 => succeeded(Left(error), depth) // attemptK + + case 10 => // onCancelRequestedFailureK + if (startedAcks) { + acks.pop() >> failed(error, depth + 1) + } else { + val _ = acks.pop() + failed(error, depth + 1) + } } } @@ -1393,6 +1476,7 @@ private final class IOFiber[A]( objectState.init(16) finalizers.init(16) + acks.init(16) val io = resumeIO resumeIO = null @@ -1411,12 +1495,14 @@ private final class IOFiber[A]( } private[this] def asyncContinueCanceledR(): Unit = { + acknowledgeCancelation() val fin = prepareFiberForCancelation(null) runLoop(fin, runtime.cancelationCheckThreshold, runtime.autoYieldThreshold) } private[this] def asyncContinueCanceledWithFinalizerR(): Unit = { val cb = objectState.pop().asInstanceOf[Either[Throwable, Unit] => Unit] + acknowledgeCancelation() val fin = prepareFiberForCancelation(cb) runLoop(fin, runtime.cancelationCheckThreshold, runtime.autoYieldThreshold) } @@ -1463,7 +1549,11 @@ private final class IOFiber[A]( /* Implementations of continuations */ private[this] def cancelationLoopSuccessK(): IO[Any] = { - if (!finalizers.isEmpty()) { + if (!acks.isEmpty()) { + // There are still remaining finalizers to execute. Continue. + conts = ByteStack.push(conts, CancelationLoopK) + acks.pop() + } else if (!finalizers.isEmpty()) { // There are still remaining finalizers to execute. Continue. conts = ByteStack.push(conts, CancelationLoopK) finalizers.pop() @@ -1513,6 +1603,7 @@ private final class IOFiber[A]( scheduleOnForeignEC(ec, this) IO.EndFiber } else { + acknowledgeCancelation() prepareFiberForCancelation(null) } } @@ -1531,6 +1622,7 @@ private final class IOFiber[A]( scheduleOnForeignEC(ec, this) IO.EndFiber } else { + acknowledgeCancelation() prepareFiberForCancelation(null) } } diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/Async.scala b/kernel/shared/src/main/scala/cats/effect/kernel/Async.scala index f9e20e1c4e..f147af9c50 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/Async.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/Async.scala @@ -105,6 +105,51 @@ trait Async[F[_]] extends AsyncPlatform[F] with Sync[F] with Temporal[F] { cont(body) } + /** + * Suspends an asynchronous side effect with optional immediate result in `F` and asynchronous + * cancelation. + * + * The given function `k` will be invoked during evaluation of `F` 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 `F` (the outer `F` of + * `F[Either[Option[F[Unit]], A]]`). + * + * The effect returns `Either[Option[F[Unit]], A]` where: + * - right side `A` is an immediate result of computation (callback invocation will be + * dropped); + * - left side `Option[F[Unit]]` is an optional acknowledgement to be run in the event that + * cancelation of the fiber running `asyncCheckAttemptParCancelable(k)` is requested. + * + * Also, note that `asyncCheckAttemptParCancelable` is uncancelable during its registration. + * + * @see + * [[async]] for a simplified variant without an option for immediate result + * @see + * [[async_]] for a simplified variant without an option for immediate result or cancelation + * support + */ + def asyncCheckAttemptParCancelable[A]( + k: (Either[Throwable, A] => Unit) => F[Either[Option[F[Unit]], A]] + ): F[A] = { + val body = new Cont[F, 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 + } + } + } + } + + cont(body) + } + /** * Suspends an asynchronous side effect in `F`. * @@ -135,6 +180,36 @@ trait Async[F[_]] extends AsyncPlatform[F] with Sync[F] with Temporal[F] { def async[A](k: (Either[Throwable, A] => Unit) => F[Option[F[Unit]]]): F[A] = asyncCheckAttempt[A](cb => map(k(cb))(Left(_))) + /** + * Suspends an asynchronous side effect in `F` with asynchronous cancelation. + * + * The given function `k` will be invoked during evaluation of the `F` 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 `F` (the outer `F` of + * `F[Option[F[Unit]]]`). + * + * The effect returns `Option[F[Unit]]` which is an optional acknowledgement to be run in the + * event that the 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 + * finalizer (i.e. evaluates to `None`). If you need the created task to be cancelable, + * return a finalizer effect upon the registration. In a rare case when there's nothing to + * finalize, you can return `Some(F.unit)` for that. + * + * @see + * [[async_]] for a simplified variant without cancelation support + * @see + * [[asyncCheckAttemptParCancelable]] for more generic version with option of providing + * immediate result of computation + */ + def asyncParCancelable[A](k: (Either[Throwable, A] => Unit) => F[Option[F[Unit]]]): F[A] = + asyncCheckAttemptParCancelable[A](cb => map(k(cb))(Left(_))) + /** * Suspends an asynchronous side effect in `F`. * @@ -151,10 +226,10 @@ trait Async[F[_]] extends AsyncPlatform[F] with Sync[F] with Temporal[F] { * asyncronous effect to be cancelable, consider using `async` instead. * * @see - * [[async]] for more generic version providing a finalizer + * [[Async!.async]] for more generic version providing a finalizer * @see - * [[asyncCheckAttempt]] for more generic version with option of providing immediate result - * of computation and finalizer + * [[Async!.asyncCheckAttempt]] for more generic version with option of providing immediate + * result of computation and finalizer */ def async_[A](k: (Either[Throwable, A] => Unit) => Unit): F[A] = async[A](cb => as(delay(k(cb)), None)) @@ -272,6 +347,10 @@ trait Async[F[_]] extends AsyncPlatform[F] with Sync[F] with Temporal[F] { /** * Like [[fromFuture]], but is cancelable via the provided finalizer. + * + * @see + * [[onCancelRequested]] for a safer alternative. This method can lose data if the future + * completes before the finalizaer can stop it. */ def fromFutureCancelable[A](futCancel: F[(Future[A], F[Unit])]): F[A] = flatMap(executionContext) { implicit ec => @@ -285,6 +364,20 @@ trait Async[F[_]] extends AsyncPlatform[F] with Sync[F] with Temporal[F] { } } + /** + * Like [[fromFuture]], but is cancelable via asynchronous cancelation. + */ + def fromFutureParCancelable[A](futCancel: F[(Future[A], F[Unit])]): F[A] = + flatMap(executionContext) { implicit ec => + uncancelable { poll => + flatMap(poll(futCancel)) { + case (fut, fin) => + asyncParCancelable[A](cb => + as(delay(fut.onComplete(t => cb(t.toEither))), Some(fin))) + } + } + } + /** * Translates this `F[A]` into a `G` value which, when evaluated, runs the original `F` to its * completion, the `limit` number of stages, or until the first stage that cannot be expressed diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/Fiber.scala b/kernel/shared/src/main/scala/cats/effect/kernel/Fiber.scala index 1588a18ee7..34182a5d64 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/Fiber.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/Fiber.scala @@ -53,6 +53,17 @@ trait Fiber[F[_], E, A] extends Serializable { */ def join: F[Outcome[F, E, A]] + /** + * Awaits the completion of the fiber bound to this [[Fiber]] and returns its [[Outcome]] once + * it completes and cancels the fiber if cancelation is requested. + * + * @note + * This method provides a race free version of `join.onCancel(cancel)`, which can lose data + * if the fiber completes after cancelation is observed. + */ + def joinOrCancel(implicit F: MonadCancel[F, E]): F[Outcome[F, E, A]] = + F.onCancelRequested(join, cancel) + /** * Awaits the completion of the bound fiber and returns its result once it completes. * diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala index 024f640197..563fcf92c3 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala @@ -159,7 +159,7 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { : F[Either[(Outcome[F, E, A], Fiber[F, E, B]), (Fiber[F, E, A], Outcome[F, E, B])]] = { implicit val F: GenConcurrent[F, E] = this - uncancelable { poll => + uncancelable { _ => for { result <- deferred[Either[Outcome[F, E, A], Outcome[F, E, B]]] @@ -167,15 +167,10 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { fibA <- start(guaranteeCase(fa)(oc => result.complete(Left(oc)).void)) fibB <- start(guaranteeCase(fb)(oc => result.complete(Right(oc)).void)) - back <- onCancel( - poll(result.get), - for { - canA <- start(fibA.cancel) - canB <- start(fibB.cancel) - - _ <- canA.join - _ <- canB.join - } yield ()) + back <- onCancelRequested( + result.get, + start(fibA.cancel).flatMap(canA => fibB.cancel *> canA.join.void) + ) } yield back match { case Left(oc) => Left((oc, fibB)) case Right(oc) => Right((fibA, oc)) diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala index bfbdb55687..68a6d02f5a 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala @@ -260,6 +260,9 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { * [[uncancelable]] * @see * [[onCancel]] + * @note + * this method can lose data if `fa` completes before `fin` can stop it. See + * [[onCancelRequested]] for a safer alternative. */ def cancelable[A](fa: F[A], fin: F[Unit]): F[A] = uncancelable { poll => @@ -368,8 +371,8 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { * [[race]] for a simpler variant that returns the successful outcome. */ def raceOutcome[A, B](fa: F[A], fb: F[B]): F[Either[Outcome[F, E, A], Outcome[F, E, B]]] = - uncancelable { poll => - poll(racePair(fa, fb)).flatMap { + uncancelable { _ => + racePair(fa, fb).flatMap { case Left((oc, f)) => f.cancel.as(Left(oc)) case Right((f, oc)) => f.cancel.as(Right(oc)) } @@ -400,13 +403,13 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { */ def race[A, B](fa: F[A], fb: F[B]): F[Either[A, B]] = uncancelable { poll => - poll(racePair(fa, fb)).flatMap { + racePair(fa, fb).flatMap { case Left((oc, f)) => oc match { case Outcome.Succeeded(fa) => f.cancel *> fa.map(Left(_)) case Outcome.Errored(ea) => f.cancel *> raiseError(ea) case Outcome.Canceled() => - f.cancel *> f.join flatMap { + f.cancel *> poll(f.join) flatMap { case Outcome.Succeeded(fb) => fb.map(Right(_)) case Outcome.Errored(eb) => raiseError(eb) case Outcome.Canceled() => poll(canceled) *> never @@ -417,7 +420,7 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { case Outcome.Succeeded(fb) => f.cancel *> fb.map(Right(_)) case Outcome.Errored(eb) => f.cancel *> raiseError(eb) case Outcome.Canceled() => - f.cancel *> f.join flatMap { + f.cancel *> poll(f.join) flatMap { case Outcome.Succeeded(fa) => fa.map(Left(_)) case Outcome.Errored(ea) => raiseError(ea) case Outcome.Canceled() => poll(canceled) *> never @@ -440,10 +443,10 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { * [[both]] for a simpler variant that returns the results of both fibers. */ def bothOutcome[A, B](fa: F[A], fb: F[B]): F[(Outcome[F, E, A], Outcome[F, E, B])] = - uncancelable { poll => - poll(racePair(fa, fb)).flatMap { - case Left((oc, f)) => poll(f.join).onCancel(f.cancel).tupleLeft(oc) - case Right((f, oc)) => poll(f.join).onCancel(f.cancel).tupleRight(oc) + uncancelable { _ => + racePair(fa, fb).flatMap { + case Left((oc, f)) => f.joinOrCancel.tupleLeft(oc) + case Right((f, oc)) => f.joinOrCancel.tupleRight(oc) } } @@ -473,11 +476,12 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { */ def both[A, B](fa: F[A], fb: F[B]): F[(A, B)] = uncancelable { poll => - poll(racePair(fa, fb)).flatMap { + racePair(fa, fb).flatMap { case Left((oc, f)) => oc match { case Outcome.Succeeded(fa) => - poll(f.join).onCancel(f.cancel).flatMap { + // We can only cancel `f` if cancelation is observed, otherwise, the canceled case below never returns + f.joinOrCancel.flatMap { case Outcome.Succeeded(fb) => fa.product(fb) case Outcome.Errored(eb) => raiseError(eb) case Outcome.Canceled() => poll(canceled) *> never @@ -488,7 +492,8 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { case Right((f, oc)) => oc match { case Outcome.Succeeded(fb) => - poll(f.join).onCancel(f.cancel).flatMap { + // We can only cancel `f` if cancelation is observed, otherwise, the canceled case below never returns + f.joinOrCancel.flatMap { case Outcome.Succeeded(fa) => fa.product(fb) case Outcome.Errored(ea) => raiseError(ea) case Outcome.Canceled() => poll(canceled) *> never diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenTemporal.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenTemporal.scala index 9527bc5992..26fd498e84 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenTemporal.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenTemporal.scala @@ -100,7 +100,7 @@ trait GenTemporal[F[_], E] extends GenConcurrent[F, E] with Clock[F] { uncancelable { poll => implicit val F: GenTemporal[F, E] = this - poll(racePair(fa, sleep(duration))) flatMap { + racePair(fa, sleep(duration)) flatMap { case Left((oc, f)) => f.cancel *> oc.embed(poll(F.canceled) *> F.never) case Right((f, _)) => f.cancel *> f.join.flatMap { oc => oc.embed(fallback) } } @@ -137,7 +137,7 @@ trait GenTemporal[F[_], E] extends GenConcurrent[F, E] with Clock[F] { uncancelable { poll => implicit val F: GenTemporal[F, E] = this - poll(racePair(fa, sleep(duration))) flatMap { + racePair(fa, sleep(duration)) flatMap { case Left((oc, f)) => f.cancel *> oc.embed(poll(F.canceled) *> F.never) case Right((f, _)) => f.cancel *> f.join.flatMap { oc => @@ -178,7 +178,7 @@ trait GenTemporal[F[_], E] extends GenConcurrent[F, E] with Clock[F] { uncancelable { poll => implicit val F: GenTemporal[F, E] = this - poll(racePair(fa, sleep(duration))) flatMap { + racePair(fa, sleep(duration)) flatMap { case Left((oc, f)) => poll(f.cancel *> oc.embed(poll(F.canceled) *> F.never)) diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/MonadCancel.scala b/kernel/shared/src/main/scala/cats/effect/kernel/MonadCancel.scala index fbe6dc312f..b826461edf 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/MonadCancel.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/MonadCancel.scala @@ -335,6 +335,26 @@ trait MonadCancel[F[_], E] extends MonadError[F, E] { */ def onCancel[A](fa: F[A], fin: F[Unit]): F[A] + /** + * Registers a cancelation acknowledgment that is invoked if cancelation is requested. + * + * If the evaluation of `fa` completes without the cancelation being requested, the + * acknowledgement is unregistered before proceeding. If a cancelation request is observed + * while `fa` is running, then `ack` is started concurrently. + * + * After `ack` has been started, if `fa` completes, the fiber is blocked until the cancelation + * request completes. If the fiber is canceled before `fa` completes, then the fiber will + * block on the completion of `ack` before running finalizers. After finalizers have started, + * [[onCancelRequested]] has no effect. + * + * @note + * If asynchronous cancelation is not supported by `F`, this function equivalent to `fa`. + */ + def onCancelRequested[A](fa: F[A], ack: F[Unit]): F[A] = { + val _ = ack + fa + } + /** * Specifies an effect that is always invoked after evaluation of `fa` completes, regardless * of the outcome. @@ -648,6 +668,9 @@ object MonadCancel { def onCancel[A](fa: OptionT[F, A], fin: OptionT[F, Unit]): OptionT[F, A] = OptionT(F.onCancel(fa.value, fin.value.void)) + override def onCancelRequested[A](fa: OptionT[F, A], ack: OptionT[F, Unit]): OptionT[F, A] = + OptionT(F.onCancelRequested(fa.value, ack.value.void)) + def forceR[A, B](fa: OptionT[F, A])(fb: OptionT[F, B]): OptionT[F, B] = OptionT( F.forceR(fa.value)(fb.value) @@ -702,6 +725,11 @@ object MonadCancel { def onCancel[A](fa: EitherT[F, E0, A], fin: EitherT[F, E0, Unit]): EitherT[F, E0, A] = EitherT(F.onCancel(fa.value, fin.value.void)) + override def onCancelRequested[A]( + fa: EitherT[F, E0, A], + ack: EitherT[F, E0, Unit]): EitherT[F, E0, A] = + EitherT(F.onCancelRequested(fa.value, ack.value.void)) + def forceR[A, B](fa: EitherT[F, E0, A])(fb: EitherT[F, E0, B]): EitherT[F, E0, B] = EitherT( F.forceR(fa.value)(fb.value) @@ -756,6 +784,9 @@ object MonadCancel { def onCancel[A](fa: IorT[F, L, A], fin: IorT[F, L, Unit]): IorT[F, L, A] = IorT(F.onCancel(fa.value, fin.value.void)) + override def onCancelRequested[A](fa: IorT[F, L, A], ack: IorT[F, L, Unit]): IorT[F, L, A] = + IorT(F.onCancelRequested(fa.value, ack.value.void)) + def forceR[A, B](fa: IorT[F, L, A])(fb: IorT[F, L, B]): IorT[F, L, B] = IorT( F.forceR(fa.value)(fb.value) @@ -810,6 +841,11 @@ object MonadCancel { def onCancel[A](fa: Kleisli[F, R, A], fin: Kleisli[F, R, Unit]): Kleisli[F, R, A] = Kleisli { r => F.onCancel(fa.run(r), fin.run(r)) } + override def onCancelRequested[A]( + fa: Kleisli[F, R, A], + ack: Kleisli[F, R, Unit]): Kleisli[F, R, A] = + Kleisli { r => F.onCancelRequested(fa.run(r), ack.run(r)) } + def forceR[A, B](fa: Kleisli[F, R, A])(fb: Kleisli[F, R, B]): Kleisli[F, R, B] = Kleisli(r => F.forceR(fa.run(r))(fb.run(r))) @@ -864,6 +900,11 @@ object MonadCancel { def onCancel[A](fa: WriterT[F, L, A], fin: WriterT[F, L, Unit]): WriterT[F, L, A] = WriterT(F.onCancel(fa.run, fin.value.void)) + override def onCancelRequested[A]( + fa: WriterT[F, L, A], + ack: WriterT[F, L, Unit]): WriterT[F, L, A] = + WriterT(F.onCancelRequested(fa.run, ack.value.void)) + def forceR[A, B](fa: WriterT[F, L, A])(fb: WriterT[F, L, B]): WriterT[F, L, B] = WriterT( F.forceR(fa.run)(fb.run) @@ -926,6 +967,11 @@ object MonadCancel { def onCancel[A](fa: StateT[F, S, A], fin: StateT[F, S, Unit]): StateT[F, S, A] = StateT[F, S, A](s => F.onCancel(fa.run(s), fin.runA(s))) + override def onCancelRequested[A]( + fa: StateT[F, S, A], + ack: StateT[F, S, Unit]): StateT[F, S, A] = + StateT[F, S, A](s => F.onCancelRequested(fa.run(s), ack.runA(s))) + def uncancelable[A](body: Poll[StateT[F, S, *]] => StateT[F, S, A]): StateT[F, S, A] = StateT[F, S, A] { s => F uncancelable { poll => @@ -991,6 +1037,12 @@ object MonadCancel { fin: ReaderWriterStateT[F, E0, L, S, Unit]): ReaderWriterStateT[F, E0, L, S, A] = ReaderWriterStateT[F, E0, L, S, A]((e, s) => F.onCancel(fa.run(e, s), fin.runA(e, s))) + override def onCancelRequested[A]( + fa: ReaderWriterStateT[F, E0, L, S, A], + ack: ReaderWriterStateT[F, E0, L, S, Unit]): ReaderWriterStateT[F, E0, L, S, A] = + ReaderWriterStateT[F, E0, L, S, A]((e, s) => + F.onCancelRequested(fa.run(e, s), ack.runA(e, s))) + def uncancelable[A]( body: Poll[ReaderWriterStateT[F, E0, L, S, *]] => ReaderWriterStateT[F, E0, L, S, A]) : ReaderWriterStateT[F, E0, L, S, A] = diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/syntax/MonadCancelSyntax.scala b/kernel/shared/src/main/scala/cats/effect/kernel/syntax/MonadCancelSyntax.scala index 946aedd0fc..bc524ce305 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/syntax/MonadCancelSyntax.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/syntax/MonadCancelSyntax.scala @@ -48,6 +48,9 @@ final class MonadCancelOps_[F[_], A] private[syntax] (private val wrapped: F[A]) def onCancel(fin: F[Unit])(implicit F: MonadCancel[F, ?]): F[A] = F.onCancel(wrapped, fin) + def onCancelRequested(ack: F[Unit])(implicit F: MonadCancel[F, ?]): F[A] = + F.onCancelRequested(wrapped, ack) + def guarantee(fin: F[Unit])(implicit F: MonadCancel[F, ?]): F[A] = F.guarantee(wrapped, fin) diff --git a/laws/shared/src/main/scala/cats/effect/laws/AsyncLaws.scala b/laws/shared/src/main/scala/cats/effect/laws/AsyncLaws.scala index 0c0319e081..c58e39b762 100644 --- a/laws/shared/src/main/scala/cats/effect/laws/AsyncLaws.scala +++ b/laws/shared/src/main/scala/cats/effect/laws/AsyncLaws.scala @@ -31,35 +31,72 @@ trait AsyncLaws[F[_]] extends GenTemporalLaws[F, Throwable] with SyncLaws[F] { (F.asyncCheckAttempt[A](_ => F.pure(Right(a))) <* F.unit) <-> (F.pure(a)) // format: on + // format: off + def asyncCheckAttemptParCancelableImmediateIsPure[A](a: A) = + (F.asyncCheckAttemptParCancelable[A](_ => F.pure(Right(a))) <* F.unit) <-> (F.pure(a)) + // format: on + // format: off def asyncCheckAttemptSuspendedRightIsAsyncRight[A](a: A, fu: F[Unit]) = (F.asyncCheckAttempt[A](k => F.delay(k(Right(a))) >> fu.as(Left(None))) <* F.unit) <-> (F.async[A](k => F.delay(k(Right(a))) >> fu.as(None)) <* F.unit) // format: on + // format: off + def asyncCheckAttemptParCancelableSuspendedRightIsAsyncRight[A](a: A, fu: F[Unit]) = + (F.asyncCheckAttemptParCancelable[A](k => F.delay(k(Right(a))) >> fu.as(Left(None))) <* F.unit) <-> (F.async[A](k => F.delay(k(Right(a))) >> fu.as(None)) <* F.unit) + // format: on + // format: off def asyncCheckAttemptSuspendedLeftIsAsyncLeft[A](e: Throwable, fu: F[Unit]) = (F.asyncCheckAttempt[A](k => F.delay(k(Left(e))) >> fu.as(Left(None))) <* F.unit) <-> (F.async[A](k => F.delay(k(Left(e))) >> fu.as(None)) <* F.unit) // format: on + // format: off + def asyncCheckAttemptParCancelableSuspendedLeftIsAsyncLeft[A](e: Throwable, fu: F[Unit]) = + (F.asyncCheckAttemptParCancelable[A](k => F.delay(k(Left(e))) >> fu.as(Left(None))) <* F.unit) <-> (F.async[A](k => F.delay(k(Left(e))) >> fu.as(None)) <* F.unit) + // format: on + // format: off def asyncRightIsUncancelableSequencedPure[A](a: A, fu: F[Unit]) = (F.async[A](k => F.delay(k(Right(a))) >> fu.as(None)) <* F.unit) <-> (F.uncancelable(_ => fu) >> F.pure(a)) // format: on + // format: off + def asyncParCancelableRightIsUncancelableSequencedPure[A](a: A, fu: F[Unit]) = + (F.asyncParCancelable[A](k => F.delay(k(Right(a))) >> fu.as(None)) <* F.unit) <-> (F.uncancelable(_ => fu) >> F.pure(a)) + // format: on + // format: off def asyncLeftIsUncancelableSequencedRaiseError[A](e: Throwable, fu: F[Unit]) = (F.async[A](k => F.delay(k(Left(e))) >> fu.as(None)) <* F.unit) <-> (F.uncancelable(_ => fu) >> F.raiseError(e)) // format: on + // format: off + def asyncParCancelableLeftIsUncancelableSequencedRaiseError[A](e: Throwable, fu: F[Unit]) = + (F.asyncParCancelable[A](k => F.delay(k(Left(e))) >> fu.as(None)) <* F.unit) <-> (F.uncancelable(_ => fu) >> F.raiseError(e)) + // format: on + def asyncRepeatedCallbackIgnored[A](a: A) = F.async[A](k => F.delay(k(Right(a))) >> F.delay(k(Right(a))).as(None)) <-> F.pure(a) + // format: off + def asyncParCancelableRepeatedCallbackIgnored[A](a: A) = + F.asyncParCancelable[A](k => F.delay(k(Right(a))) >> F.delay(k(Right(a))).as(None)) <-> F + .pure(a) + // format: on + def asyncCancelTokenIsUnsequencedOnCompletion[A](a: A, fu: F[Unit]) = F.async[A](k => F.delay(k(Right(a))) >> F.pure(Some(fu))) <-> F.pure(a) + def asyncParCancelableCancelTokenIsUnsequencedOnCompletion[A](a: A, fu: F[Unit]) = + F.asyncParCancelable[A](k => F.delay(k(Right(a))) >> F.pure(Some(fu))) <-> F.pure(a) + def asyncCancelTokenIsUnsequencedOnError[A](e: Throwable, fu: F[Unit]) = F.async[A](k => F.delay(k(Left(e))) >> F.pure(Some(fu))) <-> F.raiseError(e) + def asyncParCancelableCancelTokenIsUnsequencedOnError[A](e: Throwable, fu: F[Unit]) = + F.asyncParCancelable[A](k => F.delay(k(Left(e))) >> F.pure(Some(fu))) <-> F.raiseError(e) + def neverIsDerivedFromAsync[A] = F.never[A] <-> F.async[A](_ => F.pure(Some(F.unit))) diff --git a/tests/shared/src/test/scala/cats/effect/IOSuite.scala b/tests/shared/src/test/scala/cats/effect/IOSuite.scala index 73dedf6129..50b0c6dfc7 100644 --- a/tests/shared/src/test/scala/cats/effect/IOSuite.scala +++ b/tests/shared/src/test/scala/cats/effect/IOSuite.scala @@ -2209,6 +2209,81 @@ class IOSuite extends BaseScalaCheckSuite with DisciplineSuite with IOPlatformSu } yield () } + real("onCancelRequested - not invoked when not canceled") { + for { + requested <- Deferred[IO, Unit] + _ <- IO.unit.onCancelRequested(requested.complete(()).void) + _ <- requested + .complete(()) + .flatMap(completed => IO(assert(completed, "onCancelRequested executed"))) + } yield () + } + + real("onCancelRequested - invoked when canceled") { + for { + ready <- Deferred[IO, Unit] + requested <- Deferred[IO, Unit] + fiber <- (ready.complete(()) *> IO.never) + .onCancelRequested(requested.complete(()).void) + .start + _ <- ready.get + _ <- fiber.cancel + _ <- requested.get + } yield () + } + + real("onCancelRequested - invoked once per ack when canceled") { + for { + ready <- Deferred[IO, Unit] + requests <- Ref.of[IO, Int](0) + canceled <- Deferred[IO, Unit] + done <- Deferred[IO, Unit] + fiber <- (ready.complete(()) *> canceled.get.void) + .onCancelRequested(requests.update(_ + 1) *> canceled.complete(()).void) + .onCancelRequested(requests.update(_ + 1) *> done.complete(()).void) + .uncancelable + .start + _ <- ready.get + _ <- fiber.cancel + _ <- done.get + times <- requests.get + } yield assertEquals(times, 2) + } + + real("onCancelRequested - invoked when canceled while masked") { + for { + ready <- Deferred[IO, Unit] + requested <- Deferred[IO, Unit] + fiber <- (ready.complete(()) *> requested.get) + .onCancelRequested(requested.complete(()).void) + .uncancelable + .start + _ <- ready.get + _ <- fiber.cancel + } yield () + } + + real("joinOrCancel - gets result") { + for { + ioaReady <- Deferred[IO, Unit] + fiberReady <- Deferred[IO, Unit] + requested <- Deferred[IO, Boolean] + ioa <- (ioaReady.complete(()) *> requested.get) + .onCancelRequested(requested.complete(true).void) + .onCancel(requested.complete(false).void) + .uncancelable + .start + fiber <- (fiberReady.complete(()) *> ioa + .joinOrCancel + .flatMap(_.fold(IO.pure(false), _ => IO.pure(false), identity))).uncancelable.start + _ <- ioaReady.get + _ <- fiberReady.get + _ <- fiber.cancel + fiberResult <- fiber.join.flatMap(_.fold(IO.pure(false), _ => IO.pure(false), identity)) + requestedResult <- requested.get + } yield assert(fiberResult == requestedResult) + } + property("serialize") { forAll { (io: IO[Int]) => serializable(io) }( implicitly, diff --git a/tests/shared/src/test/scala/cats/effect/kernel/AsyncSuite.scala b/tests/shared/src/test/scala/cats/effect/kernel/AsyncSuite.scala index e1e982346b..e563fcce45 100644 --- a/tests/shared/src/test/scala/cats/effect/kernel/AsyncSuite.scala +++ b/tests/shared/src/test/scala/cats/effect/kernel/AsyncSuite.scala @@ -29,6 +29,7 @@ import org.scalacheck.Arbitrary.arbitrary import scala.concurrent.{ExecutionContext, Promise} import scala.concurrent.duration._ +import scala.util.Failure import java.util.concurrent.atomic.AtomicBoolean @@ -111,6 +112,51 @@ class AsyncSuite extends BaseSuite with DisciplineSuite { .map(r => assert(r.forall(identity(_)))) } + real("fromFutureParCancelable should cancel on fiber cancelation") { + val smallDelay: IO[Unit] = IO.sleep(10.millis) + def mkf() = Promise[Unit]() + + val go = for { + canceled <- IO(new AtomicBoolean) + fiber <- IO.fromFutureParCancelable { + IO(mkf()).map(f => + f.future -> IO { + canceled.set(true); f.complete(Failure(new InterruptedException())) + }.void) + }.start + _ <- smallDelay + _ <- fiber.cancel + res <- IO(assert(canceled.get())) + } yield res + + TestControl.executeEmbed(go, IORuntimeConfig(1, 2)).replicateA_(1000) + + } + + real("fromFutureParCancelable should backpressure on cancelation") { + // a non-cancelable, never-completing Future + def mkf() = Promise[Unit]().future + + val go = for { + started <- IO(new AtomicBoolean) + fiber <- IO.fromFutureParCancelable { + IO { + started.set(true) + mkf() + }.map(f => f -> IO.never) + }.start + _ <- IO.cede.whileM_(IO(!started.get)) + _ <- fiber.cancel + } yield () + + TestControl + .executeEmbed(go, IORuntimeConfig(1, 2)) + .as(false) + .recover { case _: TestControl.NonTerminationException => true } + .replicateA(1000) + .map(r => assert(r.forall(identity(_)))) + } + final class AsyncIO[A](val io: IO[A]) implicit def asyncForAsyncIO: Async[AsyncIO] = new Async[AsyncIO] diff --git a/tests/shared/src/test/scala/cats/effect/std/MutexSuite.scala b/tests/shared/src/test/scala/cats/effect/std/MutexSuite.scala index c742d0b910..93d0adcd54 100644 --- a/tests/shared/src/test/scala/cats/effect/std/MutexSuite.scala +++ b/tests/shared/src/test/scala/cats/effect/std/MutexSuite.scala @@ -155,7 +155,7 @@ final class MutexSuite extends BaseSuite with DetectPlatform { m.lock.use_ } - tsk.replicateA_(if (isJVM) 3000 else 5) + tsk.replicateA_(if (isJVM) 1000 else 5) } p.mustEqual(())