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..c543e0e83e 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 CancelableK = 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..4872f55cf2 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 CancelableK = 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..ac1cfcfb5c 100644 --- a/core/shared/src/main/scala/cats/effect/IO.scala +++ b/core/shared/src/main/scala/cats/effect/IO.scala @@ -447,7 +447,7 @@ sealed abstract class IO[+A] private () extends IOPlatform[A] { * [[onCancel]] */ def cancelable(fin: IO[Unit]): IO[A] = - Spawn[IO].cancelable(this, fin) + IO.Cancelable(this, fin) def forceR[B](that: IO[B]): IO[B] = // cast is needed here to trick the compiler into avoiding the IO[Any] @@ -2059,6 +2059,12 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara def onCancel[A](ioa: IO[A], fin: IO[Unit]): IO[A] = ioa.onCancel(fin) + override def cancelable[A](poll: Poll[IO], ioa: IO[A], ack: IO[Unit]): IO[A] = + ioa.cancelable(ack) + + override def cancelable[A](ioa: IO[A], ack: IO[Unit]): IO[A] = + ioa.cancelable(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 +2334,10 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara def tag = 24 } + private[effect] final case class Cancelable[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..d3c323a683 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) } @@ -908,46 +918,56 @@ private final class IOFiber[A]( case 18 => 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) + 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 + ) + + val cancel = + for { + cancelA <- fiberA.cancel.start + cancelB <- fiberB.cancel.start + _ <- cancelA.join + _ <- cancelB.join + } yield () + + type RacePairResult = + Either[(OutcomeIO[Any], FiberIO[Any]), (FiberIO[Any], OutcomeIO[Any])] + + val callback: ((Either[Throwable, RacePairResult] => Unit) => Unit) = cb => { + fiberA.setCallback(oc => cb(Right(Left((oc, fiberB))))) + fiberB.setCallback(oc => cb(Right(Right((fiberA, oc))))) + + scheduleFiber(ec, fiberA) + scheduleFiber(ec, fiberB) + } + + // inline and specialize `async_` so the `G.uncancelable` call be be removed, since + // the entire operation must be uncancelable. + val next = IO + .cont { + new Cont[IO, RacePairResult, RacePairResult] { + def apply[G[_]](implicit G: MonadCancel[G, Throwable]) = { + (resume, get, lift) => G.flatMap(lift(IO.delay(callback(resume))))(_ => get) } + } } + .cancelable(cancel) + .uncancelable runLoop(next, nextCancelation, nextAutoCede) @@ -1068,6 +1088,25 @@ private final class IOFiber[A]( /* ReadRT */ case 24 => runLoop(succeeded(runtime, 0), nextCancelation, nextAutoCede) + + /* Cancelable */ + case 25 => + val cur = cur0.asInstanceOf[Cancelable[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, CancelableK) + } + + runLoop(cur.f, nextCancelation, nextAutoCede) + } } } @@ -1118,6 +1157,7 @@ private final class IOFiber[A]( conts = null objectState.invalidate() finalizers.invalidate() + acks.invalidate() currentCtx = null if (isStackTracing) { @@ -1130,7 +1170,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 +1186,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 +1204,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 +1249,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 +1342,14 @@ private final class IOFiber[A]( case 9 => // attemptK succeeded(Right(result), depth) + + case 10 => // cancelableSuccessK + 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 +1412,14 @@ private final class IOFiber[A]( failed(error, depth + 1) case 9 => succeeded(Left(error), depth) // attemptK + + case 10 => // cancelableFailureK + if (startedAcks) { + acks.pop() >> failed(error, depth + 1) + } else { + val _ = acks.pop() + failed(error, depth + 1) + } } } @@ -1393,6 +1486,7 @@ private final class IOFiber[A]( objectState.init(16) finalizers.init(16) + acks.init(16) val io = resumeIO resumeIO = null @@ -1411,12 +1505,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 +1559,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 +1613,7 @@ private final class IOFiber[A]( scheduleOnForeignEC(ec, this) IO.EndFiber } else { + acknowledgeCancelation() prepareFiberForCancelation(null) } } @@ -1531,6 +1632,7 @@ private final class IOFiber[A]( scheduleOnForeignEC(ec, this) IO.EndFiber } else { + acknowledgeCancelation() prepareFiberForCancelation(null) } } diff --git a/kernel/jvm/src/main/scala/cats/effect/kernel/AsyncPlatform.scala b/kernel/jvm/src/main/scala/cats/effect/kernel/AsyncPlatform.scala index b0e6492770..2d5e0f18d3 100644 --- a/kernel/jvm/src/main/scala/cats/effect/kernel/AsyncPlatform.scala +++ b/kernel/jvm/src/main/scala/cats/effect/kernel/AsyncPlatform.scala @@ -38,36 +38,21 @@ private[kernel] trait AsyncPlatform[F[_]] extends Serializable { this: Async[F] * @param fut * The `java.util.concurrent.CompletableFuture` to suspend in `F[_]` */ - def fromCompletableFuture[A](fut: F[CompletableFuture[A]]): F[A] = cont { - new Cont[F, A, A] { - def apply[G[_]]( - implicit - G: MonadCancelThrow[G]): (Either[Throwable, A] => Unit, G[A], F ~> G) => G[A] = { - (resume, get, lift) => - G.uncancelable { poll => - G.flatMap(poll(lift(fut))) { cf => - val go = delay { - cf.handle[Unit] { - case (a, null) => resume(Right(a)) - case (_, t) => - resume(Left(t match { - case e: CompletionException if e.getCause ne null => e.getCause - case _ => t - })) - } - } - - val await = G.onCancel( - poll(get), - // if cannot cancel, fallback to get - G.ifM(lift(delay(cf.cancel(true))))(G.unit, G.void(get)) - ) - - G.productR(lift(go))(await) - } + def fromCompletableFuture[A](fut: F[CompletableFuture[A]]): F[A] = + uncancelable { poll => + flatMap(fut) { cf => + val wait = async_[A] { cb => + val _ = cf.handle[Unit] { + case (a, null) => cb(Right(a)) + case (_, t) => + cb(Left(t match { + case e: CompletionException if e.getCause ne null => e.getCause + case _ => t + })) } + } + + cancelable(poll, wait, void(delay(cf.cancel(true)))) } } - } - } 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..8d80163926 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,19 @@ 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 safer version of `join.onCancel(cancel)` for [[GenSpawn]] + * implementations where + * [[cats.effect.kernel.GenSpawn.cancelable[A](poll:cats\.effect\.kernel\.Poll[F],fa:F[A],fin:F[Unit]):* the polling cancelable]] + * has a data-loss safe implementation. + */ + def joinOrCancel(poll: Poll[F])(implicit F: GenSpawn[F, E]): F[Outcome[F, E, A]] = + F.cancelable(poll, F.uncancelable(_ => 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..144f6516c0 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala @@ -77,16 +77,24 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { val eval = go.start.flatMap { fiber => deferredFiber.complete(fiber) *> - poll(fiber.join.flatMap(_.embed(productR(canceled)(never)))) - .onCancel(unsubscribe(deferredFiber)) + poll( + fiber + .join + .flatMap(_.embed(productR(canceled)(never))) + .cancelable(unsubscribe(deferredFiber))) + } Evaluating(deferredFiber, 1) -> eval case (poll, Evaluating(fiber, subscribers)) => Evaluating(fiber, subscribers + 1) -> - poll(fiber.get.flatMap(_.join).flatMap(_.embed(productR(canceled)(never)))) - .onCancel(unsubscribe(fiber)) + poll( + fiber + .get + .flatMap(_.join) + .flatMap(_.embed(productR(canceled)(never))) + .cancelable(unsubscribe(fiber))) case (_, finished @ Finished(result)) => finished -> fromEither(result).flatten @@ -167,8 +175,9 @@ 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), + back <- cancelable( + poll, + result.get, for { canA <- start(fibA.cancel) canB <- start(fibB.cancel) 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..60ee55bc01 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala @@ -252,6 +252,13 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { * be equal to `never` (similar to [[race]]). Under normal circumstances, if `fa` * self-cancels, that cancelation will be propagated to the calling context. * + * @note + * The default implementation of `cancelable` ensures that `fa` is completed before + * cancelation continues, but cannot ensure that `fa` gets canceled before `fa` completes + * normally. When this race condition occurs, the result of `fa` is lost. Implementations of + * [[GenSpawn]] should override `cancelable` with an implementation that returns normally if + * `fa` wins the race between it and `fin`. + * * @param fa * the effect to be canceled * @param fin @@ -264,12 +271,44 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { def cancelable[A](fa: F[A], fin: F[Unit]): F[A] = uncancelable { poll => start(fa) flatMap { fiber => + // Note: cannot be replaced with joinOrCancel, as this is used to implement joinOrCancel poll(fiber.join) .onCancel(fin.guarantee(fiber.cancel)) .flatMap(_.embed(poll(canceled *> never))) } } + /** + * An override of [[cancelable[A](fa:F[A],fin:F[Unit]):* cancelable]] that can be safely used + * when `fa` and `fin` use a resource-like construct that must be used without allowing + * cancelation. + * + * @note + * The default implementation of `cancelable` ensures that `fa` is completed before + * cancelation continues, but cannot ensure that `fa` gets canceled before `fa` completes + * normally. When this race condition occurs, the result of `fa` is lost. Implementations of + * [[GenSpawn]] should override `cancelable` with an implementation that returns normally if + * `fa` wins the race between it and `fin`. + * + * @param poll + * the poller for the uncancelable context the cancelable finalizer is constructed in. + * @param fa + * the effect to be canceled + * @param fin + * an effect which orchestrates some external state which terminates `fa` + * @see + * [[uncancelable]] + * @see + * [[onCancel]] + */ + def cancelable[A](poll: Poll[F], fa: F[A], fin: F[Unit]): F[A] = + start(fa) flatMap { fiber => + // Note: cannot be replaced with joinOrCancel, as this is used to implement joinOrCancel. + poll(fiber.join) + .onCancel(fin.guarantee(fiber.cancel)) + .flatMap(_.embed(poll(canceled *> never))) + } + /** * A non-terminating effect that never completes, which causes a fiber to semantically block * indefinitely. This is the purely functional, asynchronous equivalent of an infinite while @@ -442,8 +481,8 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { 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) + case Left((oc, f)) => f.joinOrCancel(poll)(this).tupleLeft(oc) + case Right((f, oc)) => f.joinOrCancel(poll)(this).tupleRight(oc) } } @@ -477,7 +516,7 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] { case Left((oc, f)) => oc match { case Outcome.Succeeded(fa) => - poll(f.join).onCancel(f.cancel).flatMap { + f.joinOrCancel(poll)(this).flatMap { case Outcome.Succeeded(fb) => fa.product(fb) case Outcome.Errored(eb) => raiseError(eb) case Outcome.Canceled() => poll(canceled) *> never @@ -488,7 +527,7 @@ 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 { + f.joinOrCancel(poll)(this).flatMap { case Outcome.Succeeded(fa) => fa.product(fb) case Outcome.Errored(ea) => raiseError(ea) case Outcome.Canceled() => poll(canceled) *> never diff --git a/tests/shared/src/test/scala/cats/effect/IOSuite.scala b/tests/shared/src/test/scala/cats/effect/IOSuite.scala index 73dedf6129..56e16117ea 100644 --- a/tests/shared/src/test/scala/cats/effect/IOSuite.scala +++ b/tests/shared/src/test/scala/cats/effect/IOSuite.scala @@ -1096,6 +1096,41 @@ class IOSuite extends BaseScalaCheckSuite with DisciplineSuite with IOPlatformSu Outcome.succeeded[IO, Throwable, Int](IO.pure(42))) } + real("racePair completes when canceled") { + for { + started <- IO.deferred[Unit] + racingFiber <- IO + .racePair( + IO.deferred[Unit] + .flatMap(complete => + started.complete(()) *> complete.get.cancelable(complete.complete(()).void)) + .uncancelable, + IO.never.as(()) + ) + .start + _ <- started.get + _ <- racingFiber.cancel + outcome <- racingFiber.join + } yield assert(outcome.isSuccess, s"racing fiber was unable to complete, was $outcome") + } + + real("cancelable callback handled before inner onCancel") { + for { + started <- IO.deferred[Unit] + cancelable <- IO.deferred[Boolean] + callbacks <- { + started.complete(()) *> cancelable + .get + // with the default cancelable implementation, the onCancel would run first + .onCancel(cancelable.complete(false).void) + .cancelable(cancelable.complete(true).void) + }.uncancelable.start + _ <- started.get + _ <- callbacks.cancel + isCancelable <- cancelable.get + } yield assert(isCancelable, s"cancelable finalizer not called before cancelation observed") + } + real("async - race - immediately cancel inner race when outer unit") { for { start <- IO.monotonic 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(())