[Feature Request] Support raising cancellation in sync multiprocessed activities

[gregbrowndev]

Is your feature request related to a problem? Please describe.

Currently, activities running on a multi-processed worker can not be cancelled, either voluntarily by the user/workflow or involuntarily e.g. by a missed heartbeat (or activity timeout, I think?).

This leads to wasted resources and potential pool starvation, as the activity will run to completion anyway just to get the error below and subsequently retried:

Activity not found on completion. This may happen if the activity has already been cancelled but completed anyway.

Additionally, in the case of missed heartbeats / timeout, the server will show the activity has failed and will retry the activity. Users might not expect or protect against this behaviour, leading to possible data corruption (e.g. if the activity was trying to write to a specific file) or duplicate side effects (e.g. if the activity did something at the end, like send an email).

Reproduction: hello_cancellation.py shows a sync, multi-processed activity running forever in a while loop, preventing the worker from shutting down after the workflow was cancelled.

Describe the solution you'd like

• Cancellation should be best effort in sync, multi-processed workers, like it is for async and multithreaded workers, with the known expectation that a cancellation cannot be received unless the activity is instrumented with heartbeats.

• Provide clearer documentation in Interrupt a Workflow Execution and/or Python SDK sync-vs-async that sync, multiprocess workers do not currently support cancellation

• Provide guidance to avoid problematic side effects in such activities, e.g.

  • ensure output is written to a unique location or some kind of resource lock is acquired by the activity,
  • trigger "only-once" side effects in subsequent activities in the workflow, e.g. sending an email (and that this requires a separate async or multithreaded worker / task queue)
  • etc.

Additional context

Other relevant work/issues:

• Support raising cancellation in sync multithreaded activities #217
• [Feature Request] Set OpenTelemetry span status to failed for heartbeat timeout cancellations #1047

Root cause:

• Activity cancellation upon a missed heartbeat or timeout happens here in _ActivityWorker
• This sets various cancellation properties/events in _RunningActivity
• For sync, multiprocessed activities, the cancelled_event is initialised with a multiprocessing Manager event that can communicate across the process boundary.
• Multi-threaded activities initialise the _ThreadExceptionRaiser in _execute_sync_activity with the thread ID, but this isn't done for the multi-processed case (I imagine it doesn't work the same way).
• So it looks like the cancelled_event is set and can be observed within the child process via _Context, but there is currently no built-in way for the user activity implementation to react to this event.

I also assume this affects the "cancelled due to activity pause", at least after the activity has already started?

Out of interest, how does the the _handle_start_activity_task work in this case? Does the _ActivityWorker continue to poll for tasks for an activity it has already picked up, as seems to be suggested by the cancel task? Does the server somehow ensure these cancel tasks are sticky, so the correct worker instance that picked up the start task can process them? I'm assuming this mechanism only works for newly picked up activities, so start was never triggered. If not though, why isn't this mechanism also used for heartbeat failures?

Cheers,

I hope this assessment is accurate 🙏🏻

[cretz]

For cancel, we leverage interruption. It is easy for us to cancel/interrupt asyncio and we use an advanced Python C function to interrupt a thread, but for non-thread-pool executors, there is no clear way to interrupt processing. But utilities like activity.is_cancelled() and activity.wait_for_cancelled_sync() should work. If you use multiprocess activities, you'll have to manually check for cancellation due to there being no clear way to interrupt non-thread-pool executor (of which process-pool is one type).

[gregbrowndev]

thanks @cretz for the explanation.

Just so I understand, the reason the _ThreadExceptionRaiser / underlying C function approach cannot be used for the multi-processed worker is because:

1. We cannot serialise _ThreadExceptionRaiser across the process pool, or
2. The C function only works in the current Python process (i.e. in the parent process where it is called)? or
3. The exception raised by _ThreadExceptionRaiser requires a kind of wait point, similar to the asyncio CancelledError. Since CPU bound work may take a long time before hitting a OS call, e.g. sleep, etc. we wouldn't able to rely on this to interrupt the thread in a timely manner?

The problem with manually checking if the task is cancelled is because it might not be viable to do so, e.g. in a tight loop (check every iteration?) or calling into some long-running library function outside of your control. For the same reason, I do the heartbeats in a separate thread following the async autoheartbeater example, so I can send one periodically. But then you still have the same issue, how to interrupt the main cpu-bound thread from the io-bound heartbeat/cancellation-checking thread?

One possible solution could be to use signals to interrupt the child process? This seems to work fairly well:

https://gist.github.com/gregbrowndev/ea3e699224185591c10c4f68fb0e65b9

[cretz]

Just so I understand, the reason the _ThreadExceptionRaiser / underlying C function approach cannot be used for the multi-processed worker is because

I think the reason is more simply - we don't know what executor is being used so we can't guess at how to interrupt code.

For the same reason, I do the heartbeats in a separate thread following the async autoheartbeater example, so I can send one periodically. But then you still have the same issue, how to interrupt the main cpu-bound thread from the io-bound heartbeat/cancellation-checking thread?

This is general purpose code at this point, and only you know how best to interrupt your Python code. How might you interrupt this Python code if it weren't a Temporal activity? Technically you maybe can reuse temporalio.bridge.runtime.Runtime._raise_in_thread, though it is untested in child process and we can't guarantee compatibility. You might be able to signal the child.

One possible solution could be to use signals to interrupt the child process?

We can't even be sure the user is using multiprocessing, this code path is just for the non-threadpool executor, not specifically process pool executor.

[gregbrowndev]

Thanks again for explaining. I think the penny just dropped for me that you're talking about third-party, possibly distributed executors rather than just the standard library ones (I didn't know there were any others).

I do think it would be sensible and valuable for the SDK to support both standard library executors equally in time and add missing features such as this. Particularly now that I know its technically possible, its just the code path treats all the other executors as sync, non-threaded.

I can see this issue, like the other ProcessPool executor features I requested, will a bit further back on the priority list for the time being. So happy to leave it here. But let me know if contributions are welcome and I could pick it up.

[cretz]

I think the penny just dropped for me that you're talking about third-party, possibly distributed executors rather than just the standard library ones

It's not just that there are third party executors, even with process pool executor, there is no obvious, single interruption approach in Python I don't believe. How would you interrupt a process-pool executor based function/task in normal Python if Temporal weren't involved? Is the approach universally what people would want? It is also important that it can be traditionally caught, shielded from, etc.

If you need advanced logic beyond the default, what you may want is to use an async def activity and make your own run_in_executor call to your process pool executor, and wait for cancellation and apply it to your process-pool-based future how you prefer best. This is effectively what we do, with some added capabilities to send heartbeats and back to the parent process and cancel info to the child process.

[gregbrowndev]

I understand the concern, but I think there's a reasonable solution simply flipping your argument around:

Why not handle ProcessPoolExecutor out of the box in a somewhat opinionated, Temporal-aware way, and then if the user requires more advanced executor logic beyond the default, they can run a custom executor within an async/threaded worker?

---

I think all of the issues you raised are solvable without too much change:

How would you interrupt a process-pool executor based function/task in normal Python if Temporal weren't involved?

I think they should be interrupted in an analogous way as the async and threaded workers, i.e. the interruption is hands off, driven by the cancellation message received by the ActivityWorker, and doesn't require user code to explicitly check which may be infeasible.

Like I showed in the Gist above, using os.pkill(pid, SIGINT) looks as analogous to me as what the threaded worker's _ThreadExceptionRaiser low-level C function is doing. All it does it raise a KeyboardInterrupt in process specified by the pid. It doesn't destroy the actual process, it can be reused for further work. Its no different from any other uncaught exception in the task. We could go further and register a SIGINT signal handler in the process raise a temporalio.exceptions.CancelledError to be consistent with the sync, threaded approach.

Is the approach universally what people would want?

I think this would be a sensible default, consistent with the behaviour of the other workers. If its not the exact approach the user wants, they could run their own executor within a async/threaded worker as mentioned. I find it hard to imagine this behaviour wouldn't be the expectation of most users coming from the other two main worker types.

It is also important that it can be traditionally caught, shielded from, etc.

The exception raised by os.pkill(pid, SIGINT) in the child process would be a KeyboardInterrupt by default, but can be handled to raise a different exception. Its raised in the main thread of the process, and can be caught, ignored, reraised, etc. by normal Python code.

In the gist, I didn't handle shielding but this could be handled by backing the shield depth with the multiprocess.Manager.

[cretz]

Why not handle ProcessPoolExecutor out of the box in a somewhat opinionated, Temporal-aware way, and then if the user requires more advanced executor logic beyond the default, they can run a custom executor within an async/threaded worker?

There hasn't really been enough desire for explicit process pool executor support. It's a fairly significant challenge to maintain these abstractions if they are rarely used.

There is actually some regret in supporting non-thread-pool based executors because they are much less commonly used and it's usually best for a user to control how they want to handle multiprocessing instead of Temporal choose. No other Temporal SDK supports multiprocessing. By having Temporal just execute a function and a user choose how to delegate their logic (multiprocess, sidecars, microservices, etc), we prevent issues such as our existing opinionated abstraction not sharing the same opinions as different users (e.g. not choosing a specific interruption approach, or not choosing to explicitly support process pools, or not making OTel work natively across a process boundary, etc).

There is concern about heaping on additional logic/expectations onto this lesser used non-thread-pool abstraction compared to just documenting its limitations and encouraging users to exercise more explicit control over the delegation of work/interruption/contexts from their own functions.

[ikseek]

We use billiard.Pool adapted to ProcessPoolExecutor interface to run temporal sync activities that can segfault.
Is there a supported way I can hook into activity cancelation in such worker to send a SIGINT or in worst case SIGKILL to the subprocess executing the sync activity?
Thanks.

[cretz]

Inside the activity itself, you can wait on cancellation via activity.wait_for_cancellation_sync() in a background thread. It will return when the activity has been cancelled (worker shutdown or server-sent cancel assuming you are properly heartbeating) at which point the process can exit itself if it'd like. There is no way outside the activity to get the cancel, but you can use a regular async def activity and do https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor to your process pool executor yourself and on asyncio cancellation do something to that started execution (or just use def and process pool executor or subprocessing in asyncio or threaded directly).

[ikseek]

@cretz Thanks for the response!
Unfortunately wait_for_cancelled_sync raises RuntimeError("Not in activity context") when called not in the thread where activity executes.

[cretz]

Unfortunately wait_for_cancelled_sync raises RuntimeError("Not in activity context") when called not in the thread where activity executes.

This is touched on kinda at https://github.com/temporalio/sdk-python?tab=readme-ov-file#activity-context where it mentions the activity context uses https://docs.python.org/3/library/contextvars.html which do not span threads by default. You will have to use something like copy_context + .run to run code inside the same context.