Add `Fiber#joinOrCancel`

[djspiewak]

I think this may be the solution to all the resource leaks around nested fibers. For example, #4571

Here's the intuition:

ioa.start
  .flatMap { inner => 
    inner.join.onCancel(inner.cancel)
  }
  .start.flatMap { outer => 
    outer.cancel *> outer.join
  }

In this spaghetti, the really problematic bit is inner.join.onCancel(inner.cancel). This is problematic because there's a defiance of plain intent here in the semantics of cancelation. Specifically, what can happen is the inner.join can be canceled, which results in inner.cancel, but when the inner is canceled it may have already completed. If that happens, the value is lost regardless of what else we do, because inner.join is already yeeted and outer is committed to its cancelation. This pattern isn't universal when we use fibers, but it's fairly common. It's fundamental to racePair (and thus, timeout), and we also see it pop up in Resource quite a bit.

If this were all happening within a single fiber, the semantics of uncancelable would have saved us, since the inner.join expression would have either completed (in which case we have the value, so the onCancel doesn't run but the continuing flatMap does) or would have been canceled (in which case its finalizers would have cascaded), with none of this liminal half-and-half nonsense. joinOrCancel allows us to approximate this behavior across the fiber boundary.

The default implementation is obvious:

def joinOrCancel(implicit F: MonadCancel[F, _]): F[Outcome[F, E, A]] =
  join.onCancel(cancel)

IOFiber would override this to implement the same thing but atomically, matching the semantics of the analogous non-nested-fiber construction. For external observers, this is essentially semantic-preserving because the difference between the atomic and non-atomic version is unobservable up to race condition (so, same argument as our functor law breakage), with the only difference being that some races are now off the table.

Another much trickier option is we could detect join.onCancel and make it behave differently, but I think that's more brittle and not worth it (also requires the run loop pattern match to do something weird).

[reardonj]

If I understand correctly, the fixed code with joinOrCancel would be like this?

ioa.start
  .flatMap { inner => 
    inner.joinOrCancel // change here
  }
  .start.flatMap { outer => 
    outer.cancel *> outer.join
  }

The default Fiber implementation would not fix anything because it substitutes to the same code as the original. The IOFiber atomically either joining or cancelling, but never both, would. Is that correct?

[djspiewak]

Correct

[durban]

Intuitively, I quite like this. But I admit, I didn't play out all the known problems with this in mind. But intuitively, it should help solve them.

Further thoughts (warning: brain dump ahead):

---

A join is just an async, at least in IOFiber (well, an asyncCheckAttempt, but who's counting; a cont in any case). When I was thinking very hard about #3553, I became convinced, that what I really need is for a fiber to atomically either become masked (with a cb result) or become cancelled. This seems kind of like that. I wonder what's more fundamental: this join/cancel; or what I've wanted, the cb/cancel.

---

About how to implement this: I assume this should work for an arbitrary amount of nested fibers. Which means we will need a way for k fibers to change state atomically. That is suspiciously STM-like. (But I admit, I have a tendency to see STM in anything and everything.)

---

Going back to my previous remark about cont: what I concluded there, that currently there are 2 different memory locations for the ContState and the overall fiber state. Thus, to change them atomically is not possible with a CAS. We'd need a 2-CAS, which we don't have as a primitive, but can be built from CAS. Which is a first step towards building a fully general k-CAS. Which is practically halfway there for a full STM. (Yes.) However, to solve one specific problem, we definitely don't need a fully general solution, we might be able to get away with solving a special case.

---

To tie all these together somehow, I'm wondering, if we solved the "one fiber to atomically either become masked or become cancelled" problem, could this joinOrCancel fall out of that somehow? I really don't know.

(Sorry for closing/reopening, I've misclicked. I've also learnt a new english word today: liminal ;-)

[reardonj]

I've been trying to understand the code around this.

I didn't think there had to be something special to handle nested fibers. If there is atomic complete or cancel, the nesting should fall out of that? When this method is used, there is no longer a place for a value to 'fall into the gap' as it were, so normal fiber handling can take care of the value from there.

It would need to handle multiple joined fibers, so either all joined fibers cancel or complete. Because in principle, you can join multiple fibers to the ioa fiber (why? I don't know, but you can). Because of that, this looks like a 2-phase commit to me:

1. fiber is ready to complete
2. All joined fibers signal either canceled or become uncancelable.
3. If any joined fiber is cancelled, cancel everything, otherwise, all joined fibers are uncancelable and the fiber can complete.

---

I was also thinking if there is a simpler version of this method, like def linkCancelation: F[Unit] that just encodes that either both fibers complete or both cancel. Then joinOrCancel becomes F.uncancelable(poll => poll(ioa.linkCancelation) >> ioa.join). Although, without implementing it, I am not sure if that's actually really simpler than joinOrCancel, since the outcome might be available in the implementation of linkCancellation anyways, at which point we may as well return the outcome and it becomes joinOrCancel.

Which also made me wonder if the user method should be def join(orCancel: Boolean) and deprecate the old join method to force CE users to make the choice and help clean up all the places where it is being done wrong.

[durban]

Yes, it is very possible I'm overthinking this. As I've said, brain dump.

[reardonj]

I've been trying to implement joinOrCancel. I have it mostly working with an implementation like join, but I'm still hitting one case where inner completes but outer cancels.

I've uncommented a bunch of the printlns in IOFiber to try understanding the situations. With fiber tracing on full, the fiber toString I get when the outer cancellation happens is "canceled in runloop at cats.effect.IOFiber@55e823f RUNNING". My sense is that the outer fiber is being cancelled before it can begin executing at all, before it can even enter an uncancellable region to link up with the fiber, i.e:

ioa.start
  .flatMap { inner => 
   /* cancelation here */ inner.joinOrCancel
  }
  .start.flatMap { outer => 
    outer.cancel *> outer.join
  }

That seems to be allowed, correct? This makes me think that I need to implement the linking as a primitive IO op so the runloop can make a point of registering outer with inner before cancelling outer. Maybe that's what @djspiewak means by atomic?

[reardonj]

There are a lot of conversations floating around these cancelation issues, so it's quite possible I'm missing something, but I'm getting the sense that the core issue here is that:

1. a fiber can't do anything about cancelation until it has observed cancelation
2. at that point, the fiber can't do anything but cancel if it hasn't completed
3. after cancelation is observed, the 'parent' fiber could still observe a 'child' fiber complete before it can cancel them
4. the 'parent' fiber is no longer capable to returning this data
5. and now we have dropped data into nowhere

I don't think joinOrCancel can fix this. A fiber can spawn multiple fibers, but can only joinOrCancel one at a time. If the first joinOrCancel results in cancelling the parent, the one child is canceled, but the other children are still leaked, as they could have completed in the meantime. Maybe this is what lead @durban to think STM instead? It does look like we have to make a decision for all children before canceling.

More specifically, my understanding is that the solution ultimately needs to make a racePair implementation that never leaks, but I can't see how joinOrCancel will acheive that. Here is my reasoning:

racePair requires 3 fibers:

1. fa
2. fb
3. the race orchestrator

The orchestrator has to start 2 fibers. We can't just startfa and try to orchestrate in the fiber running fb, as without a fiber handle, we can't stop evaluating fb if fa finishes first.

The orchestrator can guarantee that one fiber joins or cancels when it is cancelled via joinOrCancel, but this still leaves a gap that the other fiber could complete after cancelation is observed, leaking that result.

---

Alternative 1: Two Phase Commit

I think we can fix this if we expose a two phase commit style mechanism. i.e. instead of racing on the outcomes, we race (allowing cancellation) on fa and fb fibers being ready to complete. After one becomes ready, we mask, commit, cancel the other fiber and join both. If the orchestator is canceled in the cancelable section, we cancel both racing fibers. Since neither of them has been committed, both can still be canceled (if they haven't been already).

IMO, this is dangerous, as we would have to expose a two phase commit mechanism and that is just asking for users to use incorrectly. It would be fairly easy to accidentally fail to confirm or abort during the commit phase, leading to a deadlock. Not to mention, it slows down all fibers as we always have to check before completing to perform the commit (even if we are able to fast-path it, there is still a check, and extra state). This is how I tried implementing joinOrCancel internally, and what I ended up with is pretty messy.

This also doesn't solve the chestnut, as there is still the cancellation gap between starting and registering the cancelation handler.

Alternative 2: A supervised start

I keep saying 'parent' up at the top, but that is a meaningless term in the current system. We keep wanting to pretend the starting fiber supervises the started fiber, but it doesn't. What we want feels somewhat analogous to an actor system actor spawning children then being shut down, where the parent can't actually be stopped until the children are stopped. But, in CE, started fibers are unattached from the parent, and the started fibers aren't canceled until the 'parent' has observed its own cancelation. i.e. the reverse of the actor supervision situation.

What if we provided a different mechanism (eg. spawn? startAttached? startChild?) that could spawn a fiber, but:

1. be attached to the parent fiber's lifetime from creation
2. when cancelation of the parent fiber is requested, the request is propagated to the child fibers and is not observed by the parent fiber until all the children have completed one way or another. Essentially, the parent adds a mask each time is startAttacheds a fiber, clears the mask when the child terminates, and propagates cancelation to children even if it is masked. Right now, we are trapped because we can't do anything about cancelation until we observe it, but by then its down late.

This does mean that the parent fiber inherits uncancelability (eg. IO.never) from any attached fibers. I don't think this is a fatal issue, as racePair can still complete the other side and return the cancelable fiber along with the other outcome. Racing 2 uncancelable, non-terminating fibers (eg. IO.never.uncancelable) would prevent the race from ever returning, but that's already the case with racePair.

This would be make fiber cancellation slower, since the fiber would have to notify its children, introducing an O(n) operation on number of children.

Note: this is very similar to ZIO's default fiber forking model, except their model additionally cancels ('interrupts' to use their terminology, but it looks like the same thing) the children when the parent completes. We could complete following this model by additionally: canceling and joining all children when the parent completes. With or without this, children that aren't terminated when the parent completes would still leak, but presumably that was acceptable if they weren't joined already. Ensuring the children are terminated does ensure all the fibers the parent fiber owns are cleaned up at at least.

Completing the ZIO model would also make regular fiber completion slower similar to the cancelation case, but the fast path in the event the fiber has no children should amount to one comparison if we wanted to (if the fiber is unmasked, there cannot be any children.

---

Solving the Chestnut

joinOrCancel doesn't quite solve the chestnut as we can't avoid a cancelation boundary between starting inner and calling joinOrCancel. joinOrCancel must be cancelable, so we will have a boundary at the start of a poll containing joinOrCancel.

We should be able to implement alternative 1 so that we can uncancelably register for cancelation before waiting on ready to complete unmasked, then complete masked. But this is tedious to implement.

Alternative 2 should solve the chestnut as even if outer is canceled, inner will be completed one way or another before the join can be canceled, as join blocks on completion of inner and outer cannot observe cancellation until it has no children ¹.

ioa.startAttached
  .flatMap { inner => 
    inner.join
  }
}
  .startAttached.flatMap { outer => 
    outer.cancel *> outer.join
  }

Footnotes

1. I do hate the parent/child terminology of supervision trees. You'd expect parents to want their children to survive them, but supervision demands the opposite. ↩

[reardonj]

Continuing to think on this, a supervised start is not primitive. I think the primitive is being able to observe a request to cancel while masked:

Old Version

trait MonadCancelRequested[F[_], E] extends GenSpawn[F[_], E] {
  /**
   * During the execution of `fa`, when cancelation is requested but masked, 
   * execute `ack` concurrently. When `onCancelRequested` is executed after
   * cancelation is requested but is masked, `ack` starts immediately. When
   * cancelation is observed, any running request acknowledgements are
   * canceled before running finalizers. When `onCancelRequested` is
   * executed in an unmasked scope, it has no effect.
  **/
  def onCancelRequested[A](fa: F[A], ack: F[Unit]): F[A]
}

This lets us asynchronously acknowledge cancelation in an uncancelable scope. It extends GenSpawn since these acknowledgements need to run concurrently to the main fiber. This allows the acknowledgement to interact with the running fiber. It's possible the acknowledgement should actually be canceled before we leave the scope of each onCancelRequested instead of just at the end. They are canceled at the end because they may depend on something the onCancel finalizers are about to clean up.

EDIT: chatted with @armanbilge and simplified the design. Now it is just a new method in MonadCancel

  def onCancelRequested[A](fa: F[A], ack: F[Unit]): F[A]

Same signature, simpler semantics:

1. The execution of ack is uncancelable
2. While in the scope of onCancelRequested , if a cancelation request would be observed¹, instead execute ack on the fiber (similar to how finalizers are run on the fiber)
3. After ack completes, continue evaluation fa

Executing everything in the same fiber simplifies the multithreading significantly, since we can just run the ack like we do onCancel finalizers. If we really want to run the ack on another fiber, we can always end the ack in a start, and could use a Deferred to await its completion later or cancel it (the fiber wouldn't be cancelable even though ack was) if we really wanted to.

---

With this, you can now write:

def joinOrCancel(fiber: Fiber[F, E, A]): F[Outcome[F, E, A]] = fiber.join.onCancelRequested(fiber.cancel).uncancelable

The join itself is uncancelable, we want to insist that we get the result of that. But, if someone tries to cancel the fiber, the fiber can respond by cancelling inner, then the join will complete in the usual fashion, which may give us a cancelled inner, but at least we actually got it

If we have two fibers, we do a parTupled variant that returns both outcomes:

{
  for {
    fibA <- ioa.start
    fibB <- iob.start
    a <- fibA.joinOrCancel
    b <- fibB.joinOrCancel
  } yield (a, b)
}.uncancelable

The whole thing has to be in uncancelable so we don't drop out between a & b. The same principle should be extendable to collections if we ever wanted to build a parTraverse that never loses anything.

Notes:

1. This does feel like a bit of abuse of uncancelable since you just don't ever poll (maybe the function should be pollAsync, as in, we can asynchronously acknowledge cancelation here)
2. I have defined this in a whole new typeclass. I doubt it has any laws though. It would probably actually get implemented in MonadCancel, since running ack is never guaranteed, since the uncancelable section could terminate before the request is observed. Since MonadCancel doesn't provide a way to do concurrency, ack simply never happens to run 🥲 . IOFiber would include a useful implementation of course, and utility methods could be added in appropriate typeclasses.
3. I haven't thought through how to implement this, but the actual operation happens on the fiber, so there shouldn't be any hairy synchronization issues. Acknowlgements could just be started where the fiber current checks if it should finalize to avoid any synchronization on the cancelation request.

Solving the Chestnut

I think these uncancelables were always needed, as otherwise there are cancelation boundaries between start and the fiber handling

ioa.start
  .flatMap { inner => 
    inner.joinOrCancel
  }.uncancelable 
  .start.flatMap { outer => 
    outer.cancel *> outer.join
  }.uncancelable

Footnotes

1. cancelation is observed when the fiber is actually canceled and will never continue executing. A "cancelation request" is observed if the fiber checks for cancelation and the fiber is canceled, regardless of masking, and the fiber is in a onCancelRequested scope. It would also be observed during a call to cancel if the fiber is suspended, at which point the calling fiber would take over the runloop as it does now and run the ack. If cancelation isn't masked, we just delay it briefly while the ack executes, then finalization will happen. There is probably no point in using onCancelRequested outside `uncancelable, but not having to think about it is nice. ↩

[durban]

@reardonj It seems to me you've reached a similar conclusion as @armanbilge here; see also #3491. What you call onCancelRequested is similar to cancelable, which already exists, but is somewhat problematic (see the linked discussion).

[reardonj]

@reardonj It seems to me you've reached a similar conclusion as @armanbilge here; see also #3491. What you call onCancelRequested is similar to cancelable, which already exists, but is somewhat problematic (see the linked discussion).

Ah, interesting. I think my proposal is actually the onCancelMasked suggestion from @djspiewak here: #3474 (comment)

Importantly, with cancelable, the fa is polled, so the finalizer is only able to try to clean up the mess. With onCancelRequested, there is no unmasking, so the fiber could still complete before observing cancellation.

[durban]

I'm not sure how is fa polled; with #3491 cancelable is a primitive. (The existing cancelable is of course different, that's the point of #3491.)

---

I think I finally understand how #3491 fixes racePair. (And what I wrote above about STM is probably wrong. With this model, essentially everything would be uncancelable. I've missed that previously.) With onCancelRequested, the fix for racePair would be essentially the same as in #3491, right?

---

The edited description is interesting. Am I right, that this way it would not help cancel a fiber blocked in blocking? (Because the ack would not be able to run.)