diff --git a/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/PureConcGenerators.scala b/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/PureConcGenerators.scala index e35a7e5e68..64e44512dd 100644 --- a/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/PureConcGenerators.scala +++ b/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/PureConcGenerators.scala @@ -39,9 +39,8 @@ object PureConcGenerators { override def recursiveGen[B: Arbitrary: Cogen](deeper: GenK[PureConc[E, *]]) = super .recursiveGen[B](deeper) - .filterNot( - _._1 == "racePair" - ) // remove the racePair generator since it reifies nondeterminism, which cannot be law-tested + .filterNot(gen => + gen._1 == "racePair" || gen._1 == "join") // remove generators which reify nondeterminism and cannot be law-tested } implicit def arbitraryPureConc[E: Arbitrary: Cogen, A: Arbitrary: Cogen] diff --git a/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/TimeT.scala b/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/TimeT.scala index baa42a033d..8068a639b4 100644 --- a/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/TimeT.scala +++ b/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/TimeT.scala @@ -18,7 +18,7 @@ package cats.effect package kernel package testkit -import cats.{~>, Group, Monad, Monoid, Order} +import cats.{~>, Eq, Group, Monad, Monoid, Order} import cats.data.Kleisli import cats.syntax.all._ @@ -90,6 +90,9 @@ private[effect] object TimeT { a.map(_.inverse()) } + implicit def eqTimeT[F[_], A](implicit FA: Eq[F[A]]): Eq[TimeT[F, A]] = + Eq.by(TimeT.run) + implicit def orderTimeT[F[_], A](implicit FA: Order[F[A]]): Order[TimeT[F, A]] = Order.by(TimeT.run) @@ -111,15 +114,69 @@ private[effect] object TimeT { val forkA = time.fork() val forkB = time.fork() - // TODO this doesn't work (yet) because we need to force the "faster" effect to win the race, which right now isn't happening - F.racePair(fa.run(forkA), fb.run(forkB)).map { + def liftOutcome[C](oc: Outcome[F, E, C]): Outcome[TimeT[F, *], E, C] = + oc.mapK(TimeT.liftK[F]) + + F.racePair(fa.run(forkA), fb.run(forkB)).flatMap { case Left((oca, delegate)) => - time.now = forkA.now - Left((oca.mapK(TimeT.liftK[F]), fiberize(forkB, delegate))) + F.onCancel(F.race(delegate.join, F.cede), delegate.cancel).map { + case Left(ocb) if forkB.now < forkA.now => + time.now = forkB.now + Right((completedFiber(forkA, liftOutcome(oca)), liftOutcome(ocb))) + + case _ => + time.now = forkA.now + Left((liftOutcome(oca), fiberize(forkB, delegate))) + } case Right((delegate, ocb)) => - time.now = forkB.now - Right((fiberize(forkA, delegate), ocb.mapK(TimeT.liftK[F]))) + F.onCancel(F.race(delegate.join, F.cede), delegate.cancel).map { + case Left(oca) if forkA.now < forkB.now => + time.now = forkA.now + Left((liftOutcome(oca), completedFiber(forkB, liftOutcome(ocb)))) + + case _ => + time.now = forkB.now + Right((fiberize(forkA, delegate), liftOutcome(ocb))) + } + } + } + + override def race[A, B](fa: TimeT[F, A], fb: TimeT[F, B]): TimeT[F, Either[A, B]] = + uncancelable { poll => + poll(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 *> poll(f.join) flatMap { + case Outcome.Succeeded(fb) => fb.map(Right(_)) + case Outcome.Errored(eb) => raiseError(eb) + case Outcome.Canceled() => poll(canceled) *> never + } + } + + case Right((f, oc)) => + oc match { + case Outcome.Succeeded(fb) => f.cancel *> fb.map(Right(_)) + case Outcome.Errored(eb) => f.cancel *> raiseError(eb) + case Outcome.Canceled() => + 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 + } + } + } + } + + override def raceOutcome[A, B](fa: TimeT[F, A], fb: TimeT[F, B]) + : TimeT[F, Either[Outcome[TimeT[F, *], E, A], Outcome[TimeT[F, *], E, B]]] = + uncancelable { poll => + poll(racePair(fa, fb)).flatMap { + case Left((oc, f)) => f.cancel.as(Left(oc)) + case Right((f, oc)) => f.cancel.as(Right(oc)) } } @@ -156,5 +213,20 @@ private[effect] object TimeT { } } } + + private[this] def completedFiber[A]( + forked: Time, + outcome: Outcome[TimeT[F, *], E, A]): Fiber[TimeT[F, *], E, A] = + new Fiber[TimeT[F, *], E, A] { + + val cancel = + unit + + val join = + Kleisli { outerTime => + outerTime.now = outerTime.now.max(forked.now) + F.pure(outcome) + } + } } } diff --git a/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/pure.scala b/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/pure.scala index 52578252d7..bce5877de4 100644 --- a/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/pure.scala +++ b/kernel-testkit/shared/src/main/scala/cats/effect/kernel/testkit/pure.scala @@ -41,10 +41,56 @@ object pure { implicit val eq: Eq[MaskId] = Eq.fromUniversalEquals[MaskId] } - final case class FiberCtx[E]( + private[pure] final case class MaskFrame(id: MaskId) + + private[pure] final class CancelationListenerId + + private[pure] object CancelationListenerId { + implicit val eq: Eq[CancelationListenerId] = + Eq.fromUniversalEquals[CancelationListenerId] + } + + private[pure] final case class CancelationListener[E]( + id: CancelationListenerId, + action: PureConc[E, Unit]) + + private sealed trait MaskUpdate + + private object MaskUpdate { + case object Removed extends MaskUpdate + case object Shadowed extends MaskUpdate + case object Absent extends MaskUpdate + } + + sealed case class FiberCtx[E]( self: PureFiber[E, ?], masks: List[MaskId] = Nil, - finalizers: List[PureConc[E, Unit]] = Nil) + finalizers: List[PureConc[E, Unit]] = Nil) { + + private[pure] def finalizing: Boolean = false + + private[pure] def withFinalizers(value: List[PureConc[E, Unit]]): FiberCtx[E] = + FiberCtx.internal(self, masks, value, finalizing) + + private[pure] def withFinalizing(value: Boolean): FiberCtx[E] = + FiberCtx.internal(self, masks, finalizers, value) + } + + object FiberCtx { + private final class Internal[E]( + self: PureFiber[E, ?], + masks: List[MaskId], + finalizers: List[PureConc[E, Unit]], + override private[pure] val finalizing: Boolean) + extends FiberCtx[E](self, masks, finalizers) + + private[pure] def internal[E]( + self: PureFiber[E, ?], + masks: List[MaskId], + finalizers: List[PureConc[E, Unit]], + finalizing: Boolean): FiberCtx[E] = + new Internal(self, masks, finalizers, finalizing) + } type ResolvedPC[E, A] = ThreadT[IdOC[E, *], A] @@ -69,8 +115,19 @@ object pure { val back = Kleisli.ask[IdOC[E, *], FiberCtx[E]] map { ctx => val checker = ctx .self - .realizeCancelation - .ifM(ApplicativeThread[PureConc[E, *]].done, ().pure[PureConc[E, *]]) + .hasActivePoll + .ifM( + ().pure[PureConc[E, *]], + ctx + .self + .isFinalizing + .ifM( + ().pure[PureConc[E, *]], + ctx + .self + .realizeCancelationWith(ctx) + .ifM(ApplicativeThread[PureConc[E, *]].done[Unit], ().pure[PureConc[E, *]])) + ) checker >> mvarLiftF(ThreadT.liftF(ka)) } @@ -97,45 +154,71 @@ object pure { type Main[X] = MVarR[ResolvedPC[E, *], X] MVar.empty[Main, Outcome[PureConc[E, *], E, A]].flatMap { state0 => - val state = state0[Main] - - val fiber = new PureFiber[E, A](state0) - - val identified = canceled mapF { ta => - val fk = new (FiberR[E, *] ~> IdOC[E, *]) { - def apply[a](ke: FiberR[E, a]) = - ke.run(FiberCtx(fiber)) - } - - ta.mapK(fk) - } - - import Outcome._ - - val body = identified flatMap { a => - state.tryPut(Succeeded(a.pure[PureConc[E, *]])) - } handleErrorWith { e => state.tryPut(Errored(e)) } - - val results = state.read.flatMap { - case Canceled() => (Outcome.Canceled(): IdOC[E, A]).pure[Main] - case Errored(e) => (Outcome.Errored(e): IdOC[E, A]).pure[Main] - - case Succeeded(fa) => - val identifiedCompletion = fa.mapF { ta => - val fk = new (FiberR[E, *] ~> IdOC[E, *]) { - def apply[a](ke: FiberR[E, a]) = - ke.run(FiberCtx(fiber)) + MVar.empty[Main, Unit] flatMap { canceled0 => + MVar[Main, List[MaskFrame]](Nil) flatMap { masks => + MVar[Main, List[CancelationListener[E]]](Nil) flatMap { cancelationListeners => + MVar[Main, Boolean](false) flatMap { finalizing => + MVar[Main, List[List[Finalizer[E]]]](Nil) flatMap { activePolls => + val state = state0[Main] + val fiber = + new PureFiber[E, A]( + state0, + canceled0, + masks, + cancelationListeners, + finalizing, + activePolls) + + val completed = canceled.flatMap { a => + withCtx { ctx => + ctx + .self + .realizeCancelationWith(ctx) + .ifM(ApplicativeThread[PureConc[E, *]].done[A], a.pure[PureConc[E, *]]) + } + } + + val identified = completed mapF { ta => + val fk = new (FiberR[E, *] ~> IdOC[E, *]) { + def apply[a](ke: FiberR[E, a]) = + ke.run(FiberCtx(fiber)) + } + + ta.mapK(fk) + } + + import Outcome._ + + val body = identified flatMap { a => + state.tryPut(Succeeded(a.pure[PureConc[E, *]])) + } handleErrorWith { e => state.tryPut(Errored(e)) } + + val results = state.read.flatMap { + case Canceled() => (Outcome.Canceled(): IdOC[E, A]).pure[Main] + case Errored(e) => (Outcome.Errored(e): IdOC[E, A]).pure[Main] + + case Succeeded(fa) => + val identifiedCompletion = fa.mapF { ta => + val fk = new (FiberR[E, *] ~> IdOC[E, *]) { + def apply[a](ke: FiberR[E, a]) = + ke.run(FiberCtx(fiber)) + } + + ta.mapK(fk) + } + + identifiedCompletion.map(a => + Succeeded[Id, E, A](a): IdOC[E, A]) handleError { e => Errored(e) } + } + + Kleisli.ask[ResolvedPC[E, *], MVar.Universe].map { u => + ApplicativeThread[ResolvedPC[E, *]].start(body.run(u)) >> results.run(u) + } + } } - - ta.mapK(fk) - } - - identifiedCompletion.map(a => Succeeded[Id, E, A](a): IdOC[E, A]) handleError { e => - Errored(e) } + } } - - Kleisli.ask[ResolvedPC[E, *], MVar.Universe].map { u => body.run(u) >> results.run(u) } } } @@ -164,8 +247,9 @@ object pure { case (List(results), _) => results.mapK(optLift) case (_, false) => Outcome.Succeeded(None) - // we could make a writer that only receives one object, but that seems meh. just pretend we deadlocked - case _ => Outcome.Succeeded(None) + // in the case of never and such, we are awaiting the async cancel monitor + // this scenario only arises if the main fiber self cancels + case _ => Outcome.Canceled() } } @@ -191,26 +275,27 @@ object pure { def onCancel[A](fa: PureConc[E, A], fin: PureConc[E, Unit]): PureConc[E, A] = Thread.annotate("onCancel", true) { withCtx[E, A] { ctx => - val ctx2 = ctx.copy(finalizers = fin.attempt.void :: ctx.finalizers) + val ctx2 = ctx.withFinalizers(fin.attempt.void :: ctx.finalizers) localCtx(ctx2, fa) } } def canceled: PureConc[E, Unit] = - Thread.annotate("canceled") { - withCtx { ctx => - if (ctx.masks.isEmpty) - uncancelable(_ => ctx.self.cancel >> ctx.finalizers.sequence_ >> Thread.done) - else - ctx.self.cancel - } - } + Thread.annotate("canceled")(withCtx { ctx => + ctx.self.cancelAndRealizeWith(ctx).ifM(Thread.done, unit) + }) def cede: PureConc[E, Unit] = - Thread.cede + withCtx { ctx => + Thread.cede *> + ctx.self.realizeCancelationWith(ctx).ifM(Thread.done, unit) + } def never[A]: PureConc[E, A] = - Thread.annotate("never")(Thread.done[A]) + withCtx[E, A] { ctx => + // we monitor for asynchronous cancelation. if we're masked, this won't cancel and we hang + Thread.annotate("never")(ctx.self.awaitCancelationWith(ctx) *> Thread.done) + } def ref[A](a: A): PureConc[E, Ref[PureConc[E, *], A]] = MVar[PureConc[E, *], A](a).flatMap(mVar => Kleisli.pure(unsafeRef(mVar))) @@ -218,6 +303,9 @@ object pure { def deferred[A]: PureConc[E, Deferred[PureConc[E, *], A]] = MVar.empty[PureConc[E, *], A].flatMap(mVar => Kleisli.pure(unsafeDeferred(mVar))) + private[this] def interruptible[A](ctx: FiberCtx[E], fa: PureConc[E, A]): PureConc[E, A] = + ctx.self.interruptible(ctx)(fa) + private def unsafeRef[A](mVar: MVar[A]): Ref[PureConc[E, *], A] = new Ref[PureConc[E, *], A] { override def get: PureConc[E, A] = mVar.read[PureConc[E, *]] @@ -273,7 +361,8 @@ object pure { private def unsafeDeferred[A](mVar: MVar[A]): Deferred[PureConc[E, *], A] = new Deferred[PureConc[E, *], A] { - override def get: PureConc[E, A] = mVar.read[PureConc[E, *]] + override def get: PureConc[E, A] = + withCtx { ctx => interruptible(ctx, mVar.read[PureConc[E, *]]) } override def complete(a: A): PureConc[E, Boolean] = mVar.tryPut[PureConc[E, *]](a) @@ -283,30 +372,161 @@ object pure { def start[A](fa: PureConc[E, A]): PureConc[E, Fiber[PureConc[E, *], E, A]] = Thread.annotate("start", true) { MVar.empty[PureConc[E, *], Outcome[PureConc[E, *], E, A]].flatMap { state => - val fiber = new PureFiber[E, A](state) + MVar.empty[PureConc[E, *], Unit] flatMap { canceled => + MVar[PureConc[E, *], List[MaskFrame]](Nil) flatMap { masks => + MVar[PureConc[E, *], List[CancelationListener[E]]](Nil) flatMap { + cancelationListeners => + MVar[PureConc[E, *], Boolean](false) flatMap { finalizing => + MVar[PureConc[E, *], List[List[Finalizer[E]]]](Nil) flatMap { + activePolls => + val fiber = + new PureFiber[E, A]( + state, + canceled, + masks, + cancelationListeners, + finalizing, + activePolls) + + // the tryPut here is interesting: it encodes first-wins semantics on cancelation/completion + val body = guaranteeCase(fa)(state.tryPut[PureConc[E, *]](_).void) + val identified = localCtx(FiberCtx(fiber), body) + Thread.start(identified.attempt.void).as(fiber) + } + } + } + } + } + } + } - // the tryPut here is interesting: it encodes first-wins semantics on cancelation/completion - val body = guaranteeCase(fa)(state.tryPut[PureConc[E, *]](_).void) - val identified = localCtx(FiberCtx(fiber), body) - Thread.start(identified.attempt.void).as(fiber) + override def racePair[A, B](fa: PureConc[E, A], fb: PureConc[E, B]): PureConc[ + E, + Either[ + (Outcome[PureConc[E, *], E, A], Fiber[PureConc[E, *], E, B]), + (Fiber[PureConc[E, *], E, A], Outcome[PureConc[E, *], E, B])]] = + uncancelable { poll => + for { + result <- deferred[ + Either[Outcome[PureConc[E, *], E, A], Outcome[PureConc[E, *], E, B]]] + + fibA <- start(fa) + fibB <- start(fb) + + _ <- start( + fibA + .join + .flatMap(oc => + result + .complete(Left(oc): Either[ + Outcome[PureConc[E, *], E, A], + Outcome[PureConc[E, *], E, B]]) + .void)) + _ <- start( + fibB + .join + .flatMap(oc => + result + .complete(Right(oc): Either[ + Outcome[PureConc[E, *], E, A], + Outcome[PureConc[E, *], E, B]]) + .void)) + + back <- onCancel( + poll(result.get), + for { + canA <- start(fibA.cancel) + canB <- start(fibB.cancel) + + _ <- canA.join + _ <- canB.join + } yield ()) + } yield back match { + case Left(oc) => Left((oc, fibB)) + case Right(oc) => Right((fibA, oc)) } } def uncancelable[A](body: Poll[PureConc[E, *]] => PureConc[E, A]): PureConc[E, A] = Thread.annotate("uncancelable", true) { - val mask = new MaskId + withCtx { ctx => + val mask = new MaskId - val poll = new Poll[PureConc[E, *]] { - def apply[a](fa: PureConc[E, a]) = - withCtx { ctx => - val ctx2 = ctx.copy(masks = ctx.masks.dropWhile(mask === _)) - localCtx(ctx2, fa.attempt <* ctx.self.realizeCancelation).rethrow + val self = ctx.self + def updateMasks[B](f: List[MaskFrame] => (List[MaskFrame], B)): PureConc[E, B] = + self.masks.read[PureConc[E, *]].flatMap { ms => + val (updated, b) = f(ms) + self.masks.swap[PureConc[E, *]](updated).as(b) } - } - withCtx { ctx => - val ctx2 = ctx.copy(masks = mask :: ctx.masks) - localCtx(ctx2, body(poll)) + val addF = updateMasks(ms => (MaskFrame(mask) :: ms, ())) + val removeF = updateMasks { + case MaskFrame(`mask`) :: ms => (ms, MaskUpdate.Removed) + case ms if ms.exists(_.id === mask) => (ms, MaskUpdate.Shadowed) + case ms => (ms, MaskUpdate.Absent) + } + + def restore(update: MaskUpdate) = + update match { + case MaskUpdate.Removed => self.exitPoll *> addF + case MaskUpdate.Shadowed | MaskUpdate.Absent => unit + } + + val poll = new Poll[PureConc[E, *]] { + def apply[a](fa: PureConc[E, a]) = + withCtx { callCtx => + if (callCtx.self eq self) + removeF.flatMap { update => + val restoreF = restore(update) + + val enterF = update match { + case MaskUpdate.Removed => self.enterPoll(callCtx.finalizers) + case MaskUpdate.Shadowed | MaskUpdate.Absent => unit + } + + enterF *> + localCtx( + callCtx, + onCancel( + self + .realizeCancelationWith(callCtx) + .ifM( + Thread.done, + fa.attempt.flatMap { result => + self + .realizeCancelationWith(callCtx) + .ifM( + Thread.done, + restoreF *> result.pure[PureConc[E, *]].rethrow) + }), + restoreF + ) + ) + } + else fa + } + } + + val runBody = + addF *> body(poll).attempt.flatMap { result => + removeF.flatMap { + case MaskUpdate.Removed => + val back = result.pure[PureConc[E, *]].rethrow + + self.currentPollFinalizers.flatMap { + case Some(finalizers) => + self + .realizeCancelationWith(ctx.withFinalizers(finalizers)) + .ifM(Thread.done, back) + case None => back + } + + case MaskUpdate.Shadowed | MaskUpdate.Absent => + result.pure[PureConc[E, *]].rethrow + } + } + + onCancel(runBody, removeF.void) } } @@ -315,7 +535,7 @@ object pure { Defer[PureConc[E, *]].defer(pure(new Unique.Token())) def forceR[A, B](fa: PureConc[E, A])(fb: PureConc[E, B]): PureConc[E, B] = - Thread.annotate("forceR")(productR(attempt(fa))(fb)) + Thread.annotate("forceR")(productR(handleError(fa.void)(_ => ()))(fb)) def flatMap[A, B](fa: PureConc[E, A])(f: A => PureConc[E, B]): PureConc[E, B] = M.flatMap(fa)(f) @@ -363,34 +583,219 @@ object pure { } // todo: MVar is not Serializable, release then update here - final class PureFiber[E, A](val state0: MVar[Outcome[PureConc[E, *], E, A]]) + final class PureFiber[E, A]( + val state0: MVar[Outcome[PureConc[E, *], E, A]], + private[this] val canceled0: MVar[Unit], + private[pure] val masks: MVar[List[MaskFrame]], + private[this] val cancelationListeners: MVar[List[CancelationListener[E]]], + private[this] val finalizing: MVar[Boolean], + private[this] val activePolls: MVar[List[List[Finalizer[E]]]]) extends Fiber[PureConc[E, *], E, A] with Serializable { + def this(state0: MVar[Outcome[PureConc[E, *], E, A]]) = + this(state0, null, null, null, null, null) + private[this] val state = state0[PureConc[E, *]] - private[pure] val canceled: PureConc[E, Boolean] = - state.tryRead.map(_.map(_.fold(true, _ => false, _ => false)).getOrElse(false)) - - private[pure] val realizeCancelation: PureConc[E, Boolean] = - withCtx { ctx => - val checkM = ctx.masks.isEmpty.pure[PureConc[E, *]] - - checkM.ifM( - canceled.ifM( - // if unmasked and canceled, finalize - allocateForPureConc[E].uncancelable(_ => ctx.finalizers.sequence_.as(true)), - // if unmasked but not canceled, ignore - false.pure[PureConc[E, *]] - ), - // if masked, ignore cancelation state but retain until unmasked - false.pure[PureConc[E, *]] - ) + private[pure] val currentMasks: PureConc[E, List[MaskFrame]] = + if (masks eq null) List.empty[MaskFrame].pure[PureConc[E, *]] + else masks.read[PureConc[E, *]] + + private[pure] val hasActivePoll: PureConc[E, Boolean] = + if (activePolls eq null) false.pure[PureConc[E, *]] + else activePolls.read[PureConc[E, *]].map(_.nonEmpty) + + private[pure] def enterPoll(finalizers: List[Finalizer[E]]): PureConc[E, Unit] = + if (activePolls eq null) ().pure[PureConc[E, *]] + else + activePolls.read[PureConc[E, *]].flatMap { polls => + activePolls.swap[PureConc[E, *]](finalizers :: polls).void + } + + private[pure] val exitPoll: PureConc[E, Unit] = + if (activePolls eq null) ().pure[PureConc[E, *]] + else + activePolls.read[PureConc[E, *]].flatMap { + case _ :: polls => activePolls.swap[PureConc[E, *]](polls).void + case Nil => ().pure[PureConc[E, *]] + } + + private[pure] val currentPollFinalizers: PureConc[E, Option[List[Finalizer[E]]]] = + if (activePolls eq null) none[List[Finalizer[E]]].pure[PureConc[E, *]] + else activePolls.read[PureConc[E, *]].map(_.headOption) + + private[pure] def registerCancelationListener( + notify: PureConc[E, Unit]): PureConc[E, CancelationListenerId] = { + val id = new CancelationListenerId + + if (cancelationListeners eq null) id.pure[PureConc[E, *]] + else + cancelationListeners.read[PureConc[E, *]].flatMap { listeners => + cancelationListeners + .swap[PureConc[E, *]](CancelationListener(id, notify) :: listeners) + .as(id) + } + } + + private[pure] def removeCancelationListener(id: CancelationListenerId): PureConc[E, Unit] = + if (cancelationListeners eq null) ().pure[PureConc[E, *]] + else + cancelationListeners.read[PureConc[E, *]].flatMap { listeners => + cancelationListeners.swap[PureConc[E, *]](listeners.filterNot(_.id === id)).void + } + + private[this] def notifyCancelationListeners: PureConc[E, Unit] = + if (cancelationListeners eq null) ().pure[PureConc[E, *]] + else cancelationListeners.swap[PureConc[E, *]](Nil).flatMap(_.traverse_(_.action)) + + private[pure] def interruptible[B](ctx: FiberCtx[E])(fb: PureConc[E, B]): PureConc[E, B] = { + val Thread = ApplicativeThread[PureConc[E, *]] + + ctx.self.currentMasks.flatMap { + case Nil => + MVar.empty[PureConc[E, *], Option[B]].flatMap { signal => + val notifyCancelation = signal.tryPut[PureConc[E, *]](None).void + + ctx.self.registerCancelationListener(notifyCancelation).flatMap { listener => + val awaitCompletion = + Thread.start(fb.flatMap(b => signal.tryPut[PureConc[E, *]](Some(b)).void)) + + val checkCancelation = + signal.tryRead[PureConc[E, *]].flatMap { + case Some(_) => ().pure[PureConc[E, *]] + case None => + ctx + .self + .realizeCancelationWith(ctx) + .ifM(notifyCancelation, ().pure[PureConc[E, *]]) + } + + awaitCompletion *> + checkCancelation *> + signal.read[PureConc[E, *]].flatMap { + case Some(b) => + ctx.self.removeCancelationListener(listener).as(b) + + case None => + ctx.self.removeCancelationListener(listener) *> + ctx.self.realizeCancelationWith(ctx) *> + Thread.done + } + } + } + + case _ => + fb + } + } + + private[pure] val isFinalizing: PureConc[E, Boolean] = + if (finalizing eq null) false.pure[PureConc[E, *]] + else finalizing.read[PureConc[E, *]] + + private[this] def setFinalizing(value: Boolean): PureConc[E, Unit] = + if (finalizing eq null) ().pure[PureConc[E, *]] + else finalizing.swap[PureConc[E, *]](value).void + + private[this] def finalizeWith( + ctx: FiberCtx[E], + finalizers: List[PureConc[E, Unit]]): PureConc[E, Boolean] = + localCtx( + ctx.withFinalizers(Nil).withFinalizing(true), + allocateForPureConc[E].uncancelable(_ => finalizers.sequence_) *> + (state0.tryPut[PureConc[E, *]](Outcome.Canceled()).flatMap { + case true => true.pure[PureConc[E, *]] + case false => + state.read.map { + case Outcome.Canceled() => true + case _ => false + } + } <* setFinalizing(false)) + ) + + private[this] def whileFinalizing[B](ctx: FiberCtx[E])(fb: PureConc[E, B]): PureConc[E, B] = + localCtx(ctx.withFinalizers(Nil).withFinalizing(true), setFinalizing(true) *> fb) + + private[this] def finalizationOutcome: PureConc[E, Boolean] = + state.read.map { + case Outcome.Canceled() => true + case _ => false } - val cancel: PureConc[E, Unit] = state.tryPut(Outcome.Canceled()).void + private[pure] def realizeCancelationWith(ctx: FiberCtx[E]): PureConc[E, Boolean] = + if (ctx.finalizing) false.pure[PureConc[E, *]] + else + isFinalizing.ifM( + finalizationOutcome, + canceled0.tryRead[PureConc[E, *]].flatMap { + case Some(_) => + ctx + .self + .currentMasks + .map(_.isEmpty) + .ifM( + whileFinalizing(ctx)(finalizeWith(ctx, ctx.finalizers)), + false.pure[PureConc[E, *]] + ) + case None => false.pure[PureConc[E, *]] + } + ) + + private[pure] def awaitCancelationWith(ctx: FiberCtx[E]): PureConc[E, Boolean] = { + def blocked = + MVar.empty[PureConc[E, *], Unit].flatMap(_.read[PureConc[E, *]]).as(false) + + if (ctx.finalizing) blocked + else + ctx.self.currentMasks.flatMap { + case Nil => + isFinalizing.ifM( + canceled0.tryRead[PureConc[E, *]].map(_.isEmpty), + canceled0.read[PureConc[E, *]] *> realizeCancelationWith(ctx)) + + case _ => + isFinalizing.ifM(canceled0.tryRead[PureConc[E, *]].map(_.isEmpty), blocked) + } + } + + private[pure] def cancelAndRealizeWith(ctx: FiberCtx[E]): PureConc[E, Boolean] = + if (ctx.finalizing) false.pure[PureConc[E, *]] + else + isFinalizing.ifM( + ctx.self.currentMasks.map(_.isEmpty), + ctx.self.currentMasks.flatMap { + case Nil => + whileFinalizing(ctx) { + requestCancelation *> finalizeWith(ctx, ctx.finalizers) + } + + case _ => + requestCancelation.as(false) + } + ) + + private[this] def requestCancelation: PureConc[E, Unit] = + canceled0 + .tryPut[PureConc[E, *]](()) + .flatMap(inserted => + if (inserted) notifyCancelationListeners else ().pure[PureConc[E, *]]) val join: PureConc[E, Outcome[PureConc[E, *], E, A]] = - state.read + if (canceled0 eq null) state.read + else { + withCtx { ctx => ctx.self.interruptible(ctx)(state.read) } + } + + val cancel: PureConc[E, Unit] = + if (canceled0 eq null) state.tryPut(Outcome.Canceled()).void + else + allocateForPureConc[E].uncancelable { _ => + state.tryRead.flatMap { + case Some(_) => ().pure[PureConc[E, *]] + case None => + requestCancelation *> state.read.void + } + } } } diff --git a/laws/shared/src/test/scala/cats/effect/laws/PureConcSuite.scala b/laws/shared/src/test/scala/cats/effect/laws/PureConcSuite.scala index 5da3183f1e..487f5612d7 100644 --- a/laws/shared/src/test/scala/cats/effect/laws/PureConcSuite.scala +++ b/laws/shared/src/test/scala/cats/effect/laws/PureConcSuite.scala @@ -17,8 +17,9 @@ package cats.effect package laws +import cats.{Eq, Order} import cats.effect.kernel.testkit.{pure, OutcomeGenerators, PureConcGenerators, TimeT} -import cats.effect.kernel.testkit.TimeT._ +import cats.effect.kernel.testkit.TimeT.{eqTimeT => _, orderTimeT => _, _} import cats.effect.kernel.testkit.pure._ import cats.laws.discipline.arbitrary._ @@ -28,13 +29,27 @@ import scala.concurrent.duration._ import munit.DisciplineSuite -class PureConcSuite extends DisciplineSuite with BaseSuite { +private[laws] trait PureConcSuiteLowPriorityTimeTInstances { + implicit def orderTimeTPureConcFiniteDuration( + implicit FA: Order[PureConc[Int, FiniteDuration]]) + : Order[TimeT[PureConc[Int, *], FiniteDuration]] = + TimeT.orderTimeT +} + +class PureConcSuite + extends DisciplineSuite + with BaseSuite + with PureConcSuiteLowPriorityTimeTInstances { import PureConcGenerators._ import OutcomeGenerators._ implicit def exec(fb: TimeT[PureConc[Int, *], Boolean]): Prop = Prop(pure.run(TimeT.run(fb)).fold(false, _ => false, _.getOrElse(false))) + implicit def eqTimeTPureConc[A]( + implicit FA: Eq[PureConc[Int, A]]): Eq[TimeT[PureConc[Int, *], A]] = + TimeT.eqTimeT + { import cats.effect.kernel.{GenConcurrent, Outcome} import cats.effect.kernel.implicits._ @@ -52,6 +67,10 @@ class PureConcSuite extends DisciplineSuite with BaseSuite { } test("short-circuit on canceled") { + assertEquals(pure.run(F.canceled), Outcome.Canceled[Option, Int, Unit]()) + assertEquals( + pure.run((F.never[Unit], F.canceled).parTupled), + Outcome.Canceled[Option, Int, (Unit, Unit)]()) assert( pure.run((F.never[Unit], F.canceled).parTupled.start.flatMap(_.join)) === Outcome .Succeeded(Some(Outcome.canceled[F, Int, (Unit, Unit)]))) @@ -78,6 +97,500 @@ class PureConcSuite extends DisciplineSuite with BaseSuite { } } + { + import cats.effect.kernel.{GenConcurrent, GenTemporal, Outcome} + import cats.effect.kernel.implicits._ + import cats.syntax.all._ + + type F[A] = PureConc[Int, A] + val F = GenConcurrent[F] + + test("run finalizers when canceling never") { + val t = for { + c <- F.ref(0) + latch <- F.deferred[Unit] + fib <- F.start((latch.complete(()) *> F.never[Unit]).onCancel(c.update(_ + 1))) + _ <- latch.get + _ <- fib.cancel + v <- c.get + } yield v + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(1))) + } + + test("run finalizers when canceling Deferred#get") { + val t = for { + c <- F.ref(0) + latch <- F.deferred[Unit] + hang <- F.deferred[Unit] + fib <- F.start((latch.complete(()) *> hang.get).onCancel(c.update(_ + 1))) + _ <- latch.get + _ <- fib.cancel + v <- c.get + } yield v + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(1))) + } + + test("run finalizers when canceling Fiber#join") { + val t = for { + c <- F.ref(0) + latch <- F.deferred[Unit] + hang <- F.start(F.never[Unit]) + fib <- F.start((latch.complete(()) *> hang.join).onCancel(c.update(_ + 1))) + _ <- latch.get + _ <- fib.cancel + v <- c.get + } yield v + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(1))) + } + + test("hang when canceling uncancelable never") { + val t = for { + latch <- F.deferred[Unit] + f <- F.start((latch.complete(()) *> F.never[Unit]).uncancelable) + _ <- latch.get + _ <- f.cancel + } yield () + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Unit](None)) + } + + test("hang when canceling uncancelable Deferred#get") { + val t = for { + latch <- F.deferred[Unit] + hang <- F.deferred[Unit] + f <- F.start((latch.complete(()) *> hang.get).uncancelable) + _ <- latch.get + _ <- f.cancel + } yield () + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Unit](None)) + } + + test("hang when canceling uncancelable Fiber#join") { + val t = for { + latch <- F.deferred[Unit] + hang <- F.start(F.never[Unit]) + f <- F.start((latch.complete(()) *> hang.join).uncancelable) + _ <- latch.get + _ <- f.cancel + } yield () + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Unit](None)) + } + + test("hang when canceling fiber blocked on cancel finalization") { + val t = for { + targetStarted <- F.deferred[Unit] + finalizerStarted <- F.deferred[Unit] + target <- F.start( + (targetStarted.complete(()) *> F.never[Unit]) + .onCancel(finalizerStarted.complete(()) *> F.never[Unit])) + _ <- targetStarted.get + canceler <- F.start(target.cancel) + _ <- finalizerStarted.get + _ <- canceler.cancel + } yield () + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Unit](None)) + } + + test("run finalizers in order") { + val t = for { + results <- F.ref[String]("") + f <- F start { + F.canceled.onCancel(results.update(_ + "A")).onCancel(results.update(_ + "B")) + } + _ <- f.join + back <- results.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, String](Some("AB"))) + } + + test("ignore cancelation of a fiber after racePair has completed") { + val t = for { + finalized <- F.ref(0) + fiber <- F.start { + F.racePair(F.unit, F.never[Unit]).void.onCancel(finalized.update(_ + 1)) + } + _ <- fiber.join + _ <- fiber.cancel + _ <- F.cede + back <- finalized.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(0))) + } + + test("ignore cancelation of a fiber after race has completed") { + val t = for { + finalized <- F.ref(0) + fiber <- F.start { + F.race(F.unit, F.never[Unit]).void.onCancel(finalized.update(_ + 1)) + } + _ <- fiber.join + _ <- fiber.cancel + _ <- F.cede + back <- finalized.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(0))) + } + + test("correctly interpret uncancelable cancelation followed by suspension") { + val t = F.uncancelable(_ => F.canceled *> F.never[Unit]) + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Unit](None)) + + val forked = pure.run(F.start(t).flatMap(_.joinWith(F.canceled *> F.never[Unit]))) + assertEquals(forked, Outcome.Succeeded[Option, Int, Unit](None)) + } + + test("ignore poll from another fiber") { + val t = for { + started <- F.deferred[Unit] + polled <- F.deferred[Unit] + + parent <- F.start { + F.uncancelable { poll => + started.complete(()) *> + F.start(poll(polled.complete(()) *> F.never[Unit])).void *> + polled.get *> + F.never[Unit] + } + } + + _ <- started.get + _ <- polled.get + _ <- parent.cancel + } yield () + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Unit](None)) + } + + test("observe external cancelation while blocked inside poll") { + val t = for { + started <- F.deferred[Unit] + polled <- F.deferred[Unit] + gate <- F.deferred[Unit] + ran <- F.ref(false) + fiber <- F.start { + F.uncancelable { poll => + started.complete(()) *> + poll(polled.complete(()) *> gate.get *> ran.set(true)) + } + } + canceler <- F.start(polled.get *> fiber.cancel) + _ <- started.get + _ <- canceler.join + back <- ran.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Boolean](Some(false))) + } + + test("select current finalizers for external cancelation inside poll") { + val t = for { + polled <- F.deferred[Unit] + gate <- F.deferred[Unit] + finalized <- F.ref(0) + fiber <- F.start { + F.uncancelable { poll => + poll(F.onCancel(polled.complete(()) *> gate.get, finalized.update(_ + 1))) + } + } + _ <- polled.get + _ <- fiber.cancel + back <- finalized.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(1))) + } + + test("unregister finalizers before observing masked external cancelation") { + val t = for { + masked <- F.deferred[Unit] + gate <- F.deferred[Unit] + finalized <- F.ref(0) + fiber <- F.start { + F.onCancel( + F.uncancelable(_ => masked.complete(()) *> gate.get), + finalized.update(_ + 1)) + } + _ <- masked.get + releaser <- F.start(F.cede *> gate.complete(())) + _ <- fiber.cancel + _ <- releaser.join + outcome <- fiber.join + back <- finalized.get + } yield (outcome === Outcome.canceled[F, Int, Unit], back) + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, (Boolean, Int)](Some((true, 0)))) + } + + test("run finalizers around a self-canceling polled region") { + val t = for { + finalized <- F.ref(0) + fiber <- F.start { + F.uncancelable { poll => F.onCancel(poll(F.canceled), finalized.update(_ + 1)) } + } + _ <- fiber.join + back <- finalized.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(1))) + } + + test("run outer finalizers around a self-canceling polled region") { + val t = for { + finalized <- F.ref(0) + fiber <- F.start { + F.onCancel(F.uncancelable { poll => poll(F.canceled) }, finalized.update(_ + 1)) + } + _ <- fiber.join + back <- finalized.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(1))) + } + + test("run outer finalizers when a masked self-cancel is observed inside poll") { + val t = for { + finalized <- F.ref(0) + fiber <- F.start { + F.onCancel( + F.uncancelable { poll => poll(F.uncancelable(_ => F.canceled)) }, + finalized.update(_ + 1)) + } + _ <- fiber.join + back <- finalized.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Int](Some(1))) + } + + test("observe pending self-cancel before running a polled region") { + val t = for { + finalized <- F.ref(0) + ran <- F.ref(false) + fiber <- F.start { + F.uncancelable { poll => + F.canceled *> F.onCancel(poll(ran.set(true)), finalized.update(_ + 1)) + } + } + _ <- fiber.join + fin <- finalized.get + body <- ran.get + } yield (fin, body) + + assertEquals( + pure.run(t), + Outcome.Succeeded[Option, Int, (Int, Boolean)](Some((1, false)))) + } + + test("run only finalizers installed after a masked self-cancel") { + val t = for { + before <- F.ref(0) + after <- F.ref(0) + ran <- F.ref(false) + fiber <- F.start { + F.uncancelable { poll => + F.onCancel(F.canceled, before.update(_ + 1)) *> + F.onCancel(poll(ran.set(true)), after.update(_ + 1)) + } + } + _ <- fiber.join + beforeCount <- before.get + afterCount <- after.get + body <- ran.get + } yield (beforeCount, afterCount, body) + + assertEquals( + pure.run(t), + Outcome.Succeeded[Option, Int, (Int, Int, Boolean)](Some((0, 1, false)))) + } + + test("select the innermost active poll finalizers") { + val t = for { + finalized <- F.ref("") + fiber <- F.start { + F.uncancelable { outerPoll => + F.onCancel( + outerPoll { + F.uncancelable { innerPoll => + F.onCancel( + innerPoll(F.uncancelable(_ => F.canceled)), + finalized.update(_ + "B")) + } + }, + finalized.update(_ + "A")) + } + } + _ <- fiber.join + back <- finalized.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, String](Some("BA"))) + } + + test("restore outer poll finalizers after an inner poll completes") { + val t = for { + outerFinalized <- F.ref(0) + innerFinalized <- F.ref(0) + fiber <- F.start { + F.uncancelable { outerPoll => + F.onCancel( + outerPoll { + F.uncancelable { innerPoll => + F.onCancel(innerPoll(F.unit), innerFinalized.update(_ + 1)) + } *> F.uncancelable(_ => F.canceled) + }, + outerFinalized.update(_ + 1) + ) + } + } + _ <- fiber.join + outer <- outerFinalized.get + inner <- innerFinalized.get + } yield (outer, inner) + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, (Int, Int)](Some((1, 0)))) + } + + test("observe nested self-cancel inside a polled region before continuing") { + val t = for { + ran <- F.ref(false) + fiber <- F.start { + F.uncancelable { poll => + poll { + F.uncancelable(_ => F.canceled) *> ran.set(true) + } + } + } + _ <- fiber.join + back <- ran.get + } yield back + + assertEquals(pure.run(t), Outcome.Succeeded[Option, Int, Boolean](Some(false))) + } + + test("preserve masked self-cancel through poll") { + val maskedCancel = F.uncancelable(_ => F.canceled) + val fa = F.onCancel(maskedCancel, F.never[Unit]) + + assertEquals(pure.run(maskedCancel), Outcome.Canceled[Option, Int, Unit]()) + assertEquals( + pure.run(F.uncancelable(poll => poll(maskedCancel))), + Outcome.Canceled[Option, Int, Unit]()) + assertEquals(pure.run(fa), Outcome.Canceled[Option, Int, Unit]()) + assertEquals( + pure.run(F.uncancelable(poll => poll(fa))), + Outcome.Canceled[Option, Int, Unit]()) + } + + test("run a guarantee finalizer around a masked self-cancel") { + val fa = F.guarantee(F.uncancelable(_ => F.canceled), F.never[Unit]) + + assertEquals(pure.run(fa), Outcome.Succeeded[Option, Int, Unit](None)) + } + + test("associate finalizers across an uncancelable boundary") { + val left = F.uncancelable(_ => F.onCancel(F.canceled, F.never[Unit])) + val right = F.onCancel(F.uncancelable(_ => F.canceled), F.never[Unit]) + + assertEquals(pure.run(left), Outcome.Canceled[Option, Int, Unit]()) + assertEquals(pure.run(right), Outcome.Canceled[Option, Int, Unit]()) + } + + test("implement locals via Kleisli and FreeT") { + import cats.{~>, Eval, Id} + import cats.data.Kleisli + import cats.free.FreeT + import cats.syntax.all._ + + type F[A] = FreeT[Id, Kleisli[Eval, Int, *], A] + + def read[A](f: Int => F[A]): F[A] = + FreeT.liftT(Kleisli.ask[Eval, Int]).flatMap(f) + + def withLocal[A](i: Int)(fa: F[A]): F[A] = + fa.mapK(new (Kleisli[Eval, Int, *] ~> Kleisli[Eval, Int, *]) { + def apply[a](kea: Kleisli[Eval, Int, a]) = + Kleisli((_: Int) => kea(i)) + }) + + def run[A](i: Int)(fa: F[A]): A = + fa.runM(fta => Kleisli.liftF(Eval.now(fta))).apply(i).value + + val _ = run(1) { + withLocal(42) { + read { i => + FreeT + .liftT[Id, Kleisli[Eval, Int, *], Unit]( + Kleisli.liftF[Eval, Int, Unit](Eval.later(assertEquals(i, 42)))) + .flatMap(_ => + read { i2 => FreeT.liftT(Kleisli.liftF(Eval.later(assertEquals(i2, 42)))) }) + } + } *> read { i => FreeT.liftT(Kleisli.liftF(Eval.later(assertEquals(i, 1)))) } + } + } + + test("race TimeT values against never") { + type T[A] = TimeT[F, A] + val T = GenTemporal[T, Int] + + assertEquals( + pure.run(TimeT.run(T.race(T.pure(1), T.never[Unit]))), + Outcome.Succeeded[Option, Int, Either[Int, Unit]](Some(Left(1)))) + assertEquals( + pure.run(TimeT.run(T.race(T.never[Unit], T.pure(1)))), + Outcome.Succeeded[Option, Int, Either[Unit, Int]](Some(Right(1)))) + assertEquals( + pure.run(TimeT.run(T.race(T.sleep(1.second).as(1), T.never[Unit]))), + Outcome.Succeeded[Option, Int, Either[Int, Unit]](Some(Left(1)))) + assertEquals( + pure.run(TimeT.run(T.race(T.never[Unit], T.sleep(1.second).as(1)))), + Outcome.Succeeded[Option, Int, Either[Unit, Int]](Some(Right(1)))) + assertEquals( + pure.run( + TimeT.run(T.race(T.sleep(2.seconds).as("slow"), T.sleep(1.second).as("fast")))), + Outcome.Succeeded[Option, Int, Either[String, String]](Some(Right("fast"))) + ) + assertEquals( + pure.run( + TimeT.run(T.race(T.sleep(1.second).as("fast"), T.sleep(2.seconds).as("slow")))), + Outcome.Succeeded[Option, Int, Either[String, String]](Some(Left("fast"))) + ) + assertEquals( + pure.run(TimeT.run(T.race(T.canceled, T.never[Unit]).void)), + Outcome.Canceled[Option, Int, Unit]()) + assertEquals( + pure.run(TimeT.run(T.race(T.never[Unit], T.canceled).void)), + Outcome.Canceled[Option, Int, Unit]()) + assertEquals( + pure.run( + TimeT.run(T.race(TimeT.liftF(F.uncancelable(_ => F.canceled.as(1))), T.never[Unit]))), + Outcome.Canceled[Option, Int, Either[Int, Unit]]()) + assertEquals( + pure.run( + TimeT.run(T.race(T.never[Unit], TimeT.liftF(F.uncancelable(_ => F.canceled.as(1)))))), + Outcome.Canceled[Option, Int, Either[Unit, Int]]()) + assertEquals( + pure.run( + TimeT.run(T.race(TimeT.liftF(F.start(F.unit).flatMap(_.join).as(1)), T.never[Unit]))), + Outcome.Succeeded[Option, Int, Either[Int, Unit]](Some(Left(1))) + ) + assertEquals( + pure.run( + TimeT.run(T.race(T.never[Unit], TimeT.liftF(F.start(F.unit).flatMap(_.join).as(1))))), + Outcome.Succeeded[Option, Int, Either[Unit, Int]](Some(Right(1))) + ) + } + + } + checkAll( "TimeT[PureConc]", GenTemporalTests[TimeT[PureConc[Int, *], *], Int].temporal[Int, Int, Int](10.millis)