Skip to content

Commit d745d7e

Browse files
neko-kaineko-kai
authored
Fix #503 implement MonadCancel#canceled by sending an external interrupt to current fiber via Fiber.unsafeCurrentFiber (#544)
* Workaround zio/zio#6911 Only consider the *current* fiber being interrupted as `Outcome.Canceled`, consider internal interruption & Die to be `Outcome.Errored` Trigger `MonadCancel#onCancel` only if the *current* fiber is being interrupted Implement `MonadCancel#bracketFull` & `bracketCase` * Fix #503 Find a way to send interrupt to self via `Fiber.unsafeCurrentFiber`, implement external-only Cancel semantic for `start` and `raceWith` * make `MonadError#onError` consistent with `guaranteeCase` starting from `MonadCancel` instance * Add tests for zio/zio#6911 (require #543) * Fix js implementation of `async` * arbitrary instance: convert from cats-effect * rebase * Use throwable instead of generic instance in `ZManaged#toResource[F]` * Remove exceptions for inner interruption in generic error instances & generate Interrupted Causes for generic tests * In tests, add more generators + cats conversion generators - disabled by default, because they surface further law failures still * Fix `calls finalizers when using resource is canceled` test, fix unreliability of `canceled` when used with `unsafeRunToFuture` * Change `toEffect` implementation to avoid getting a `BoxedException` when the underlying fiber is interrupted. * Relax condition for signalling non-interruption, because ZIO sometimes creates invalid Causes without Interrupt node when interrupted * Add `genNever` to random generators * Change the implementation of `Async#async` to follow `async left is uncancelable sequenced raiseError` law. Consider the law code: ```scala // 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 ``` if `fu` is `F.never`, then an implmentation based on ZIO.effectAsyncM can never satisfy this law, because it runs the register code on a separate fiber and ignores its effect. The law clearly states that register effect must run first and must run on the same fiber as the callback listener, so as to supercede it. * 2.12 build * Remove incorrect `onError` definition. (should not be uncancelable) * Remove redundant `resetForkScope` * Acknowledge `onError`/`Outcome.Errored` incoherence in tests * Restore no-op F.onError in `parTraverse + ZIO.die` test
1 parent 9992c56 commit d745d7e

File tree

10 files changed

+530
-139
lines changed

10 files changed

+530
-139
lines changed

‎zio-interop-cats-tests/jvm/src/test/scala/zio/interop/CatsInteropSpec.scala

+93-2
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
11
package zio.interop
22

3-
import cats.effect.{ Async, IO as CIO, LiftIO }
3+
import cats.effect.{ Async, IO as CIO, LiftIO, Outcome }
44
import cats.effect.kernel.{ Concurrent, Resource }
55
import zio.interop.catz.*
66
import zio.test.*
7-
import zio.{ Promise, Task }
7+
import zio.{ Promise, Task, ZIO }
88

99
object CatsInteropSpec extends CatsRunnableSpec {
1010
def spec = suite("Cats interop")(
@@ -54,6 +54,97 @@ object CatsInteropSpec extends CatsRunnableSpec {
5454
sanityCheckCIO <- fromEffect(test[CIO])
5555
zioResult <- test[Task]
5656
} yield zioResult && sanityCheckCIO
57+
},
58+
testM("onCancel is not triggered by ZIO.parTraverse + ZIO.fail https://github.com/zio/zio/issues/6911") {
59+
val F = Concurrent[Task]
60+
61+
for {
62+
counter <- F.ref("")
63+
_ <- F.guaranteeCase(
64+
F.onError(
65+
F.onCancel(
66+
ZIO.collectAllPar(
67+
List(
68+
ZIO.unit.forever,
69+
counter.update(_ + "A") *> ZIO.fail(new RuntimeException("x")).unit
70+
)
71+
),
72+
counter.update(_ + "1")
73+
)
74+
) { case _ => counter.update(_ + "B") }
75+
) {
76+
case Outcome.Errored(_) => counter.update(_ + "C")
77+
case Outcome.Canceled() => counter.update(_ + "2")
78+
case Outcome.Succeeded(_) => counter.update(_ + "3")
79+
}.run
80+
res <- counter.get
81+
} yield assertTrue(!res.contains("1")) && assertTrue(res == "ABC")
82+
},
83+
testM("onCancel is not triggered by ZIO.parTraverse + ZIO.die https://github.com/zio/zio/issues/6911") {
84+
val F = Concurrent[Task]
85+
86+
for {
87+
counter <- F.ref("")
88+
_ <- F.guaranteeCase(
89+
F.onError(
90+
F.onCancel(
91+
ZIO.collectAllPar(
92+
List(
93+
ZIO.unit.forever,
94+
counter.update(_ + "A") *> ZIO.die(new RuntimeException("x")).unit
95+
)
96+
),
97+
counter.update(_ + "1")
98+
)
99+
) { case _ => counter.update(_ + "B") }
100+
) {
101+
case Outcome.Errored(_) => counter.update(_ + "C")
102+
case Outcome.Canceled() => counter.update(_ + "2")
103+
case Outcome.Succeeded(_) => counter.update(_ + "3")
104+
}.run
105+
res <- counter.get
106+
} yield assertTrue(!res.contains("1")) && assertTrue(res == "AC")
107+
},
108+
testM("onCancel is not triggered by ZIO.parTraverse + ZIO.interrupt https://github.com/zio/zio/issues/6911") {
109+
val F = Concurrent[Task]
110+
111+
for {
112+
counter <- F.ref("")
113+
_ <- F.guaranteeCase(
114+
F.onError(
115+
F.onCancel(
116+
ZIO.collectAllPar(
117+
List(
118+
ZIO.unit.forever,
119+
counter.update(_ + "A") *> ZIO.interrupt.unit
120+
)
121+
),
122+
counter.update(_ + "1")
123+
)
124+
) { case _ => counter.update(_ + "B") }
125+
) {
126+
case Outcome.Errored(_) => counter.update(_ + "C")
127+
case Outcome.Canceled() => counter.update(_ + "2")
128+
case Outcome.Succeeded(_) => counter.update(_ + "3")
129+
}.run
130+
res <- counter.get
131+
} yield assertTrue(!res.contains("1")) && assertTrue(res == "AC")
132+
},
133+
test("F.canceled.toEffect results in CancellationException, not BoxedException") {
134+
val F = Concurrent[Task]
135+
136+
val exception: Option[Throwable] =
137+
try {
138+
F.canceled.toEffect[cats.effect.IO].unsafeRunSync()
139+
None
140+
} catch {
141+
case t: Throwable => Some(t)
142+
}
143+
144+
assertTrue(
145+
!exception.get.getMessage.contains("Boxed Exception") &&
146+
exception.get.getMessage.contains("The fiber was canceled")
147+
)
57148
}
58149
)
59150
}

‎zio-interop-cats-tests/jvm/src/test/scala/zio/interop/CatsZManagedSyntaxSpec.scala

+69-7
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package zio.interop
22

3-
import cats.effect.kernel.Resource
3+
import cats.effect.kernel.{ Concurrent, Resource }
44
import cats.effect.IO as CIO
55
import zio.*
66
import zio.interop.catz.*
@@ -17,13 +17,37 @@ object CatsZManagedSyntaxSpec extends CatsRunnableSpec {
1717
def spec =
1818
suite("CatsZManagedSyntaxSpec")(
1919
suite("toManaged")(
20-
test("calls finalizers correctly when use is interrupted") {
20+
test("calls finalizers correctly when use is externally interrupted") {
2121
val effects = new mutable.ListBuffer[Int]
2222
def res(x: Int): Resource[CIO, Unit] =
2323
Resource.makeCase(CIO.delay(effects += x).void) {
2424
case (_, Resource.ExitCase.Canceled) =>
2525
CIO.delay(effects += x + 1).void
26-
case _ => CIO.unit
26+
case (_, _) =>
27+
CIO.unit
28+
}
29+
30+
val testCase = {
31+
val managed: ZManaged[Any, Throwable, Unit] = res(1).toManaged
32+
Promise.make[Nothing, Unit].flatMap { latch =>
33+
managed
34+
.use(_ => latch.succeed(()) *> ZIO.never)
35+
.forkDaemon
36+
.flatMap(latch.await *> _.interrupt)
37+
}
38+
}
39+
40+
unsafeRun(testCase)
41+
assert(effects.toList)(equalTo(List(1, 2)))
42+
},
43+
test("calls finalizers correctly when use is internally interrupted") {
44+
val effects = new mutable.ListBuffer[Int]
45+
def res(x: Int): Resource[CIO, Unit] =
46+
Resource.makeCase(CIO.delay(effects += x).void) {
47+
case (_, Resource.ExitCase.Errored(_)) =>
48+
CIO.delay(effects += x + 1).void
49+
case (_, _) =>
50+
CIO.unit
2751
}
2852

2953
val testCase = {
@@ -118,7 +142,7 @@ object CatsZManagedSyntaxSpec extends CatsRunnableSpec {
118142
}
119143
),
120144
suite("toManagedZIO")(
121-
test("calls finalizers correctly when use is interrupted") {
145+
test("calls finalizers correctly when use is externally interrupted") {
122146
val effects = new mutable.ListBuffer[Int]
123147
def res(x: Int): Resource[Task, Unit] =
124148
Resource.makeCase(Task(effects += x).unit) {
@@ -127,6 +151,28 @@ object CatsZManagedSyntaxSpec extends CatsRunnableSpec {
127151
case _ => Task.unit
128152
}
129153

154+
val testCase = {
155+
val managed: ZManaged[Any, Throwable, Unit] = res(1).toManagedZIO
156+
Promise.make[Nothing, Unit].flatMap { latch =>
157+
managed
158+
.use(_ => latch.succeed(()) *> ZIO.never)
159+
.forkDaemon
160+
.flatMap(latch.await *> _.interrupt)
161+
}
162+
}
163+
164+
unsafeRun(testCase)
165+
assert(effects.toList)(equalTo(List(1, 2)))
166+
},
167+
test("calls finalizers correctly when use is internally interrupted") {
168+
val effects = new mutable.ListBuffer[Int]
169+
def res(x: Int): Resource[Task, Unit] =
170+
Resource.makeCase(Task(effects += x).unit) {
171+
case (_, Resource.ExitCase.Errored(_)) =>
172+
Task(effects += x + 1).unit
173+
case _ => Task.unit
174+
}
175+
130176
val testCase = {
131177
val managed: ZManaged[Any, Throwable, Unit] = res(1).toManagedZIO
132178
managed.use(_ => ZIO.interrupt.unit)
@@ -242,7 +288,22 @@ object CatsZManagedSyntaxSpec extends CatsRunnableSpec {
242288
unsafeRun(testCase.orElse(ZIO.unit))
243289
assert(effects.toList)(equalTo(List(1, 2)))
244290
},
245-
test("calls finalizers when using resource is canceled") {
291+
test("calls finalizers when using resource is internally interrupted") {
292+
val effects = new mutable.ListBuffer[Int]
293+
def man(x: Int): ZManaged[Any, Throwable, Unit] =
294+
ZManaged.makeExit(ZIO.effectTotal(effects += x).unit) {
295+
case (_, Exit.Failure(c)) if !c.interrupted && c.failureOption.nonEmpty =>
296+
ZIO.effectTotal(effects += x + 1)
297+
case _ =>
298+
ZIO.unit
299+
}
300+
301+
val testCase = man(1).toResource[RIO[ZEnv, _]].use(_ => ZIO.interrupt)
302+
try unsafeRun(testCase)
303+
catch { case _: Throwable => }
304+
assert(effects.toList)(equalTo(List(1, 2)))
305+
},
306+
test("calls finalizers when using resource is externally interrupted") {
246307
val effects = new mutable.ListBuffer[Int]
247308
def man(x: Int): ZManaged[Any, Throwable, Unit] =
248309
ZManaged.makeExit(ZIO.effectTotal(effects += x).unit) {
@@ -252,8 +313,9 @@ object CatsZManagedSyntaxSpec extends CatsRunnableSpec {
252313
ZIO.unit
253314
}
254315

255-
val testCase = man(1).toResource[RIO[ZEnv, _]].use(_ => ZIO.interrupt)
256-
unsafeRun(testCase.orElse(ZIO.unit))
316+
val testCase = man(1).toResource[RIO[ZEnv, _]].use(_ => Concurrent[RIO[ZEnv, _]].canceled)
317+
try unsafeRun(testCase)
318+
catch { case _: Throwable => }
257319
assert(effects.toList)(equalTo(List(1, 2)))
258320
},
259321
test("acquisition of Reservation preserves cancellability in new F") {

‎zio-interop-cats-tests/shared/src/test/scala/zio/interop/CatsSpecBase.scala

+3
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,8 @@ private[zio] trait CatsSpecBase
9696
implicit val eqForNothing: Eq[Nothing] =
9797
Eq.allEqual
9898

99+
// workaround for laws `evalOn local pure` & `executionContext commutativity`
100+
// (ZIO cannot implement them at all due to `.executor.asEC` losing the original executionContext)
99101
implicit val eqForExecutionContext: Eq[ExecutionContext] =
100102
Eq.allEqual
101103

@@ -114,6 +116,7 @@ private[zio] trait CatsSpecBase
114116
implicit def eqForUIO[A: Eq](implicit ticker: Ticker): Eq[UIO[A]] = { (uio1, uio2) =>
115117
val exit1 = unsafeRun(uio1)
116118
val exit2 = unsafeRun(uio2)
119+
// println(s"comparing $exit1 $exit2")
117120
(exit1 eqv exit2) || {
118121
println(s"$exit1 was not equal to $exit2")
119122
false

‎zio-interop-cats-tests/shared/src/test/scala/zio/interop/GenIOInteropCats.scala

+43-18
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,22 @@
11
package zio.interop
22

3+
import cats.effect.GenConcurrent
34
import org.scalacheck.*
45
import zio.*
56

6-
/**
7-
* Temporary fork of zio.GenIO that overrides `genParallel` with ZManaged-based code
8-
* instead of `io.zipPar(parIo).map(_._1)`
9-
* because ZIP-PAR IS NON-DETERMINISTIC IN ITS SPAWNED EC TASKS (required for TestContext equality)
10-
*/
117
trait GenIOInteropCats {
128

9+
// FIXME generating anything but success (even genFail)
10+
// surfaces multiple further unaddressed law failures
11+
def betterGenerators: Boolean = false
12+
13+
// FIXME cats conversion surfaces failures in the following laws:
14+
// `async left is uncancelable sequenced raiseError`
15+
// `async right is uncancelable sequenced pure`
16+
// `applicativeError onError raise`
17+
// `canceled sequences onCanceled in order`
18+
def catsConversionGenerator: Boolean = false
19+
1320
/**
1421
* Given a generator for `A`, produces a generator for `IO[E, A]` using the `IO.point` constructor.
1522
*/
@@ -26,8 +33,35 @@ trait GenIOInteropCats {
2633
*/
2734
def genSuccess[E, A: Arbitrary]: Gen[IO[E, A]] = Gen.oneOf(genSyncSuccess[E, A], genAsyncSuccess[E, A])
2835

29-
def genIO[E, A: Arbitrary]: Gen[IO[E, A]] =
30-
genSuccess[E, A]
36+
def genFail[E: Arbitrary, A]: Gen[IO[E, A]] = Arbitrary.arbitrary[E].map(IO.fail[E](_))
37+
38+
def genDie(implicit arbThrowable: Arbitrary[Throwable]): Gen[UIO[Nothing]] = arbThrowable.arbitrary.map(IO.die(_))
39+
40+
def genInternalInterrupt: Gen[UIO[Nothing]] = ZIO.interrupt
41+
42+
def genCancel[E, A: Arbitrary](implicit F: GenConcurrent[IO[E, _], ?]): Gen[IO[E, A]] =
43+
Arbitrary.arbitrary[A].map(F.canceled.as(_))
44+
45+
def genNever: Gen[UIO[Nothing]] = ZIO.never
46+
47+
def genIO[E: Arbitrary, A: Arbitrary](implicit
48+
arbThrowable: Arbitrary[Throwable],
49+
F: GenConcurrent[IO[E, _], ?]
50+
): Gen[IO[E, A]] =
51+
if (betterGenerators)
52+
Gen.oneOf(
53+
genSuccess[E, A],
54+
genFail[E, A],
55+
genDie,
56+
genInternalInterrupt,
57+
genCancel[E, A],
58+
genNever
59+
)
60+
else
61+
Gen.oneOf(
62+
genSuccess[E, A],
63+
genNever
64+
)
3165

3266
def genUIO[A: Arbitrary]: Gen[UIO[A]] =
3367
Gen.oneOf(genSuccess[Nothing, A], genIdentityTrans(genSuccess[Nothing, A]))
@@ -98,17 +132,8 @@ trait GenIOInteropCats {
98132
Gen.const(io.flatMap(a => IO.succeed(a)))
99133

100134
private def genOfRace[E, A](io: IO[E, A]): Gen[IO[E, A]] =
101-
Gen.const(io.raceFirst(ZIO.never.interruptible))
135+
Gen.const(io.interruptible.raceFirst(ZIO.never.interruptible))
102136

103137
private def genOfParallel[E, A](io: IO[E, A])(gen: Gen[IO[E, A]]): Gen[IO[E, A]] =
104-
gen.map { parIo =>
105-
// this should work, but generates more random failures on CI
106-
// io.interruptible.zipPar(parIo.interruptible).map(_._1)
107-
Promise.make[Nothing, Unit].flatMap { p =>
108-
ZManaged
109-
.fromEffect(parIo *> p.succeed(()))
110-
.fork
111-
.use_(p.await *> io)
112-
}
113-
}
138+
gen.map(parIo => io.interruptible.zipPar(parIo.interruptible).map(_._1))
114139
}

‎zio-interop-cats-tests/shared/src/test/scala/zio/interop/ZioSpecBase.scala

+30-7
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,7 @@ private[interop] trait ZioSpecBase extends CatsSpecBase with ZioSpecBaseLowPrior
2929
Gen.oneOf(
3030
e.arbitrary.map(Cause.Fail(_)),
3131
Arbitrary.arbitrary[Throwable].map(Cause.Die(_)),
32-
// Generating interrupt failures causes law failures (`canceled`/`Outcome.Canceled` are ill-defined as of now https://github.com/zio/interop-cats/issues/503#issuecomment-1157101175=)
33-
// Gen.long.flatMap(l1 => Gen.long.map(l2 => Cause.Interrupt(Fiber.Id(l1, l2)))),
32+
Gen.long.flatMap(l1 => Gen.long.map(l2 => Cause.Interrupt(Fiber.Id(l1, l2)))),
3433
Gen.delay(self.map(Cause.Traced(_, ZTrace(Fiber.Id.None, Nil, Nil, None)))),
3534
Gen.delay(self.map(Cause.stackless)),
3635
Gen.delay(self.flatMap(e1 => self.map(e2 => Cause.Both(e1, e2)))),
@@ -54,17 +53,41 @@ private[interop] trait ZioSpecBaseLowPriority { self: ZioSpecBase =>
5453

5554
implicit def arbitraryIO[E: CanFail: Arbitrary: Cogen, A: Arbitrary: Cogen]: Arbitrary[IO[E, A]] = {
5655
implicitly[CanFail[E]]
57-
Arbitrary(Gen.oneOf(genIO[E, A], genLikeTrans(genIO[E, A]), genIdentityTrans(genIO[E, A])))
56+
import zio.interop.catz.generic.concurrentInstanceCause
57+
Arbitrary(
58+
Gen.oneOf(
59+
genIO[E, A],
60+
genLikeTrans(genIO[E, A]),
61+
genIdentityTrans(genIO[E, A])
62+
)
63+
)
5864
}
5965

6066
implicit def arbitraryZIO[R: Cogen, E: CanFail: Arbitrary: Cogen, A: Arbitrary: Cogen]: Arbitrary[ZIO[R, E, A]] =
6167
Arbitrary(Gen.function1[R, IO[E, A]](arbitraryIO[E, A].arbitrary).map(ZIO.environment[R].flatMap))
6268

63-
implicit def arbitraryRIO[R: Cogen, A: Arbitrary: Cogen]: Arbitrary[RIO[R, A]] =
64-
arbitraryZIO[R, Throwable, A]
69+
implicit def arbitraryTask[A: Arbitrary: Cogen](implicit ticker: Ticker): Arbitrary[Task[A]] = {
70+
val arbIO = arbitraryIO[Throwable, A]
71+
if (catsConversionGenerator)
72+
Arbitrary(Gen.oneOf(arbIO.arbitrary, genCatsConversionTask[A]))
73+
else
74+
arbIO
75+
}
6576

66-
implicit def arbitraryTask[A: Arbitrary: Cogen]: Arbitrary[Task[A]] =
67-
arbitraryIO[Throwable, A]
77+
def genCatsConversionTask[A: Arbitrary: Cogen](implicit ticker: Ticker): Gen[Task[A]] =
78+
arbitraryIO[A].arbitrary.map(liftIO(_))
79+
80+
def liftIO[A](io: cats.effect.IO[A])(implicit ticker: Ticker): zio.Task[A] =
81+
ZIO.effectAsyncInterrupt { k =>
82+
val (result, cancel) = io.unsafeToFutureCancelable()
83+
k(ZIO.fromFuture(_ => result).tapError {
84+
case c: scala.concurrent.CancellationException if c.getMessage == "The fiber was canceled" =>
85+
zio.interop.catz.concurrentInstance.canceled *> ZIO.interrupt
86+
case _ =>
87+
ZIO.unit
88+
})
89+
Left(ZIO.fromFuture(_ => cancel()).orDie)
90+
}
6891

6992
def zManagedArbitrary[R, E, A](implicit zio: Arbitrary[ZIO[R, E, A]]): Arbitrary[ZManaged[R, E, A]] =
7093
Arbitrary(zio.arbitrary.map(ZManaged.fromEffect))

0 commit comments

Comments
 (0)