Lossless cancelation and joinOrCancel - #4641
Conversation
The sys.error call is a side-effect and should be suspended in IO
- add polling cancelable. This is needed to safely start the join in `joinOrCancel` without introducing a cancelation boundary that could drop an already started fiber - replace unsafe usages of `join.onCancel(cancel)` construct with `joinOrCancel` - replace fromCompletableFuture with an implementation that uses cancelable
In IO, cancelable can be implemented without hoisting the operation to a separate thread, by invoking the callback when cancelation is requested. Partly based on Arman's previous attempt in typelevel#3491 Co-authored-by: Arman Bilge <armanbilge@gmail.com>
IO.racePair has to be uncancelable because cancelable introduces a cancelation boundary when used unmasked
2596350 to
2a9e3b6
Compare
|
I have tested this on the http4s test suite. Getting timeouts in |
|
Narrowed issue down to fs2-io. Unix socket tests in there fail with the changes. |
- cancelable needs to be on a separate fiber since it is for blocking operations which do not suspend the fiber, so we cannot start the acks - onCancelRequested becomes a synonym for onCancel by default. In IO, onCancelRequested masks. This lets the implementation keep the `poll(fa).onCancelRequested(ack)`, which will at least cancel in other Fs, but lose data, while working correctly in IO. - cancelable now uses onCancelRequested instead of onCancel so that cancelable operations get a chance to terminate.
The current onCancelRequested doesn't work with it. It looks like it should be fixable without it, by waiting to transition to Unevaluated until the fiber completes.
Needed to preserve bin-compat.
|
Revised things, and updated the main description. I've broken back out a separate |
|
To make |
- fix up naming issues - restore cancelable tests - adjust scaladoc
I don't really understand this. Why does this mean the old behavior is required? |
My solution avoids this by requiring control of the runloop to start the finalizers, and completes them before running any It's probably possible to write an algorithm to safely perform step 3, but that's going to have to be very carefully designed to avoid races. IMO, that's not worthwhile for the relatively few pieces of code that need |
|
|
||
| case 9 => succeeded(Left(error), depth) // attemptK | ||
|
|
||
| case 10 => // onCancelRequestedFailureK |
There was a problem hiding this comment.
-1 is lost? error inside an onCancelRequested region permanently leaks a mask
| succeeded(Right(result), depth) | ||
|
|
||
| case 10 => // onCancelRequestedSuccessK | ||
| masks -= 1 |
There was a problem hiding this comment.
decrement before acks.pop()
at the outermost mask level the ack-join is thrown away by shouldFinalize() at the top of runLoop and fiber.cancel returns before the acknowledgement finishes.
There was a problem hiding this comment.
decrement before acks.pop()
I'm not following the concern here. masks should only be used by runloop code, so why does the order matter? This code is called if the inner operation completed, at which point the ack should no longer matter.
There was a problem hiding this comment.
right, no race, and the body's result doesn't care.
But the next runloop iteration reads masks before running the IO we just returned: startedAcks implies canceled, so if the decrement takes masks to 0 then shouldFinalize() is true and the ack join gets dropped as _cur0. The ack fiber keeps running, nobody waits for it, and with no finalizers fiber.cancel returns while it's still going. The acks-before-finalizers drain can't save it either, the ack is already off the stack.
Nothing in the lib hits this??, all call sites are inside uncancelable.
Public onCancelRequested is. Pushing UncancelableK before acks.pop().as(result) would hold the mask across the join, and the failed twin needs the same ordering.
I think I can try to test spec this if it helps.
There was a problem hiding this comment.
Ahh, I see. I didn't consider that scenario. Well, if the onCancelRequested operation succeeded, the ack completion doesn't matter any more, so I doubt any CE code hitting it would encounter an issue. Outside of some async tests, it's all fiber cancellation, and those cancellations are going to no-op if we got this far.
It could be an issue though, since a finalizer could then clean up a resource the ack depends on (this is the main reason tracking acks is important).
I think I can try to test spec this if it helps.
Sure, I'm not going to have a chance to fix this today.
There was a problem hiding this comment.
Here are 2 small tests, should fail, but please recheck - I'm bit lost a track trying to repro
reardonj#1
|
|
||
| // otherwise it is too late to request cancelation | ||
| if (!finalizing) { | ||
| masks += 1 |
There was a problem hiding this comment.
"UnmaskRunLoop" only unmasks when masks == cur.id
any enclosing poll used inside an "onCancelRequested" body silently does nothing.
There was a problem hiding this comment.
This is intentional. It doesn't really make sense to poll inside an onCancelRequested body, since the construct exists to wrap code that can't safely be canceled, and instead needs to use its own cancelation protocol.
To make this whole thing kind of function without a new major release, the default implementation of onCancelRequested in Spawn just uses regular cancellation. So, the user-land code does wrap the part than can be canceled in poll, so that it can be canceled in Spawn, but for the better version of onCancelRequested in IO, the poll has to be ignored so the ack can actually run.
Yes, this is awful. I don't see any other way out that doesn't involve a breaking binary compatibility change though.
There was a problem hiding this comment.
Then it makes sense,
IMHO but worth documenting - the behaviour differs by instance: uncancelable(poll => poll(never).onCancelRequested(fin)) hangs on cancel under IO (I've seen in your tests assertation), while under Kleisli[IO, R, *] it hits the default fa.onCancel(ack), poll stays live, and it cancels.
same joinOrCancel -> takes a Poll that does nothing at IO.
There was a problem hiding this comment.
while under Kleisli[IO, R, *] it hits the default fa.onCancel(ack), poll stays live, and it cancels.
same joinOrCancel -> takes a Poll that does nothing at IO.
Oh, well that's bad. I guess it would need explicit delegation set up?
I have been holding off on expanding on documentation until we decide this is even the way we want to fix the underlying problem with cancellation, and if the awful default is acceptable to start with.
| } | ||
| } | ||
|
|
||
| onCancelRequested(poll(wait), void(delay(cf.cancel(true)))) |
There was a problem hiding this comment.
Canceling fromCompletableFuture/fromCompletionStag will yieldOutcome.Errored(CancellationException) instead of Canceled(). IMHO - contract change not sure is it a real issue.
| } | ||
| def fromCompletableFuture[A](fut: F[CompletableFuture[A]]): F[A] = | ||
| uncancelable { poll => | ||
| flatMap(fut) { cf => |
There was a problem hiding this comment.
Thats a change, the poll around the acquisition was dropped (G.flatMap(poll(lift(fut))) → flatMap(fut)), so producing the CompletableFuture is uncancelable?
| /** | ||
| * Suspend a `java.util.concurrent.CompletableFuture` into the `F[_]` context. | ||
| * | ||
| * @note |
There was a problem hiding this comment.
I think it should be at least changesd,
AsyncPlatform.scala:55 - Before ifM on cf.cancel(true)'s return value is gone, so instances that don't override onCancelRequested get fire-and-forget on a CF that refuses cancelation.
There was a problem hiding this comment.
I have been holding off on expanding on documentation until we decide this is even the way we want to fix
I 've seen the comment after I already reporte, but anyway is it desirable change?
| 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")) |
There was a problem hiding this comment.
I think this statement is not true anymore and reachable.
survives only because shouldFinalize() discards the suspended delay first
There was a problem hiding this comment.
If needed I can spend time crafting the test - I may be wrong.
There was a problem hiding this comment.
It's been a while since I looked at this, but iirc, it ends up being canceled before the sys.error executes.
|
So I had a think about this, and I think it's still possible to make a primitive This is definitely all sorts of weird when you think about exclusivity, too. For example, what happens if we start running the action and then the body completes? Do we interrupt it? Error out? Do we prevent the body from completing? Do we block the cancelee? In all these cases I think we can just mimic what the default implementation does and avoid getting too fancy: no exclusivity, no blocking, it's a best effort fire-and-forget. |
|
Or alternative Give cancelable one atomic with three states: Unclaimed | ClaimedByCanceler | CompletedByBody.
The canceler only schedules fin, it never runs user code inline. |
My concern with fire-and-forget is the interaction with resource cleanup. If the canceled fiber can continue running while the action is running, the canceled fiber could clean up a resource that the action depends on. Irrelevant for |
|
@stasimus , I think the coordination is a bit more complicated.
I'm a little iffy on trying to acquire 2 separate CASes in sequence, but maybe it doesn't matter if the runloop gets suspended in between.
In this case, the runloop would CAS into CompletedByBody, take
If it loses, the runloop first has to wait for the right
|
They don't have to be acquired together.
My fault for writing "one atomic", I meant one per
That's better than what I had, and it removes the "wait for the right One thing I'm unsure about: |
Not that they're together, but in general multi-threaded code that is working with multiple locks in any way raises alarm bells in my head that we are entering dangerous waters. @stasimus , @djspiewak , I worry about the level of complexity we're getting into and more generally the oddness of trying to jam this into the existing API (in particular, needing Acknowledgement is needed to safely terminate asynchronous operations (i.e. cancel or complete a fiber before cancelation is effective so we don't lose data). We are really trying to force in a new behavior [ ie. (1) ] into the typeclass hierarchy that properly should be a binary breaking change. re: @djspiewak 's comment from @armanbilge 's attempt at this, maybe it's time to talk about CE4 if we really want to do this right. Realistically, any other |
|
My company just migrated (I hope all repos) CE->CE3 in 2025, CE4 please wait) "Multiple locks" isn't the shape of the CAS way/proposal. Nothing waits while holding anything, each participant does at most one CAS on a monotonic per-node word, so no deadlock cycle. Deadlock needs someone holding one lock while waiting on another. A CAS isn't held, so that can't happen here? Complexity (maintenance) yes could be a huge. =========== IMHO The Poll[F] bothers more than the CAS stuff. It's inert in IO but load-bearing in other Fs, and unlike the coordination question it lands in the API. You said in July you weren't convinced joinOrCancel pulls its weight. Lets just cut it, are those 3 reports from header need it? |
A breaking change to add this little bit would be much closer to the 3.5.0 breaks than all the semantic changes of CE2 -> CE3. That release also required downstreams to update, which smells awfully like a major release 😜
💯
I don't follow, cutting |
|
Opps, I conflated two things. Cutting joinOrCancel doesn't touch the polls, they come from onCancelRequested itself. Though that does narrow the version question. The polls only exist to make the default implementation work: with fa.onCancel(ack) the ack fires only if cancelation actually gets observed, so call sites have to unmask. IO doesn't need it, because its override starts acks under the mask. Make onCancelRequested abstract with IO's semantics and no call site needs a poll, and there's one behavior instead of two. So it's either keep the default, and pay for it with polls that are inert in IO plus a guarantee that depends on which instance you picked, or drop the default and break bincompat? In this way I changed opinion towards your position... |
Context
There are a number of open bugs relating to data loss during races:
Fiber#joinOrCancel#4620Fundamentally, these all must occur because there is no way to do all of the following together:
In particular, the fiber that started the other fiber doesn't know it's getting canceled until it observes cancelation (ie.
onCancelruns). At this point it is no longer possible for the fiber to complete, it must cancel. So all it can do is terminate its child fibers and carry on1.I had previously tried to solve this problem with a new
onCancelRequestedcombinator (#4633 ), but that would break otherFs. This implementation instead follows @djspiewak's suggestion to base the solution on an @armanbilge'scancelablefix in #3491 which would givecancelabledifferent behavior inIO, but also leave a working (but not ideal) implementation for otherFs already.New Behaviour
Unfortunately, we still need the old
cancelablebehavior, as it is needed to cancel blocking operations on a fiber (as in literallyF.blocking). This design only works on a suspended fiber. What we can do, while letting otherFs remain no more broken than they are today, is implementonCancelRequestedasonCancelby default, then inIOimplement it by masking the operation, so it can only be canceled by theonCancelRequestedeffect.By adding this behavior, we can run some finalizers before the fiber observes cancelation. Now we can give the operation a chance to return a result before the fiber is canceled and unable to return a result, but also try canceling the operation to make sure cancelation does actually happen if we can't complete.
Changes to Use This Behaviour
Unsafe usages of
join.onCancel(cancel)are replaced with usages ofjoinOrCancelso that they can become lossless inIO. In IO,joinOrCancelwill result in exactly one of getting outcome of the fiber or the fiber being canceled2.onCancelRequestedremains lossy by default, but this is unfixable without new semantics which would break compatibility.The implementation of
cancelableis updated to useonCancelRequested.The implementation of
fromCompletableFutureis also updated to useonCancelRequestedinstead of it's bespokecontimplementation. This implementation is no better by default, but will no longer lose data inIO.Implementation
IOFiber
onCancelRequestedinIOis now handled as a new primitiveIO.OnCancelRequestedclass. This operation introduces a second finalizer (referred to as acks to differentiation from the actual finalizers) stack toIOFiber. When the fiber receives a cancelation request, all acks are immediately run in parallel and any acks that get added after this point are also immediately started. This behavior is intended to drive the fiber towards cancelation as quickly as possible, with the expectation that acks are safe to run at any time, unlike finalizers, which clean up resources. The fiber waits for all acks to complete before running finalizers, since the acks could depend on a resource that will be disposed by a finalizer. If theIO.OnCancelRequestedwould come off the stack, the fiber will also wait on the ack to complete3IO.racePairis also updated to useonCancelRequestedinstead of async cancelation so it does not lose data during a race, following @armanbilge's earlier implementation.Footnotes
You can do a little better if you know the fiber is returning a resource, and clean up that resource, but you still can't get data back out. ↩
or non-termination, of course. ↩
The ack should already be completed, since the ack cancelable action should ↩