Fix multiprocessing deadlock after using prefect_test_harness on Linux #19116
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
CodSpeed Performance ReportMerging #19116 will not alter performanceComparing Summary
|
This PR fixes a deadlock that occurs on Linux when using multiprocessing after `prefect_test_harness()`. The issue was that background threads (WorkerThread, EventLoopThread, and QueueServices like APILogWorker and EventsWorker) with locks remain alive when the test harness exits. On Linux, `multiprocessing.Process()` defaults to `fork()`, which copies these lock states into the child process, causing deadlocks. The fix registers `os.register_at_fork()` handlers that reset thread and service state in child processes after fork(), ensuring clean state without inherited locks. Changes: - Add fork handlers to `prefect._internal.concurrency.threads` module - Add fork handlers to `prefect._internal.concurrency.services` module - Add smoke tests to verify fork handlers are registered Closes #19112 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
- Move fork reset logic into class methods to avoid accessing protected members - Consolidate thread tracking into single WeakSet - Remove broad exception handling in favor of cleaner class methods - All type errors fixed, all tests still pass
Remove underscore prefix from reset_for_fork methods to avoid reportPrivateUsage errors when calling from module-level fork handlers. Verified with: uv run pyright -p pyrightconfig-ci.json --level error
zzstoatzz
force-pushed
the
fix/multiprocessing-after-test-harness
branch
from
9a04c62 to
6ac8315
Compare
…eadlock Moved the test from internal concurrency tests to testing utilities where it belongs, and replaced the useless smoke tests with a real test that reproduces the actual issue: using multiprocessing.Process() after prefect_test_harness(). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
On Linux with fork(), the child process inherits Prefect's logging and event state. When the worker function returns normally, Python's cleanup tries to flush logs/close connections, which can fail since those resources are in an inconsistent state after fork. Using os._exit(0) bypasses cleanup entirely. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
zzstoatzz
commented
Oct 7, 2025
tests/testing/test_utilites.py
Outdated
Comment on lines
19
to
36
| def _multiprocessing_worker(): | ||
| """ | ||
| Worker function for multiprocessing test. Must be at module level for pickling. | ||
| Explicitly exits with code 0 to avoid any Python cleanup issues in the forked process. | ||
| """ | ||
| import os | ||
| import sys | ||
|
|
||
| try: | ||
| # Explicitly exit with success code instead of relying on return value cleanup | ||
| os._exit(0) | ||
| except Exception as e: | ||
| print(f"Worker error: {e}", file=sys.stderr) | ||
| import traceback | ||
|
|
||
| traceback.print_exc() | ||
| os._exit(1) |
Collaborator
Author
There was a problem hiding this comment.
i don't know if this is a good thing ™️ to do
Collaborator
Author
|
this does appear to solve the issue.... but feels rather hacky. which might be fine because its only a testing util. but it might be better interpreted as signal that we should make |
multiprocessing's _bootstrap() explicitly handles SystemExit raised by sys.exit(), making it the public API for setting exit codes. This avoids using the private os._exit() function. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
On Linux with fork(), the child process inherits Prefect's logging/event state. Normal exit (return or sys.exit) triggers Python cleanup that fails with this inherited state, causing exitcode=1. os._exit() bypasses cleanup and is documented for use "in the child process after os.fork()". Despite the underscore prefix, this is the intended API for this scenario. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
Contributor
|
This pull request is stale because it has been open 14 days with no activity. To keep this pull request open remove stale label or comment. |
desertaxle
approved these changes
Oct 30, 2025
Comment on lines
+49
to
+50
| # Might fail in certain contexts (e.g., if already in a child process) | ||
| pass |
Member
There was a problem hiding this comment.
Can we add a debug log here for breadcrumbs in case this is ever an issue?
Collaborator
Author
There was a problem hiding this comment.
oops, i missed this, will add in a follow up PR
| T = TypeVar("T", infer_variance=True) | ||
|
|
||
| # Track all active instances for fork handling | ||
| _active_instances: weakref.WeakSet[WorkerThread | EventLoopThread] = weakref.WeakSet() |
zzstoatzz
added a commit
that referenced
this pull request
Oct 31, 2025
closes #19116 this PR adds a debug log when fork handler registration fails, as requested in review feedback from the original PR. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
This was referenced Nov 10, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
Fixes a deadlock that occurs on Linux when using multiprocessing after
prefect_test_harness().Issue
On Linux,
multiprocessing.Process()defaults tofork(), which copies all process state including locks from background threads. Whenprefect_test_harness()exits, background threads (WorkerThread,EventLoopThread, and services likeAPILogWorkerandEventsWorker) remain alive with their locks. When fork() copies these locks into the child process, the locks remain in inconsistent states (held by threads that don't exist in the child), causing deadlocks.Solution
Registers
os.register_at_fork()handlers that reset thread and service state in child processes after fork(), ensuring clean state without inherited locks from dead threads.Changes
prefect._internal.concurrency.threadsmodule to resetWorkerThreadandEventLoopThreadstateprefect._internal.concurrency.servicesmodule to resetQueueServiceinstancestests/testing/test_utilites.pythat reproduces the reported issueTest Notes
The regression test uses
os._exit(0)in the worker function despite the underscore prefix. This is necessary because:returnorsys.exit()) triggers Python cleanup that fails with inherited state, causing exitcode=1os._exit()bypasses cleanup and is documented for use "in the child process after os.fork()" - which is exactly this scenarioTest Results
Closes #19112
🤖 Generated with Claude Code