Visitar URL original
gh-66587: Fix deadlock from pool worker death without communication by applio · Pull Request #16103 · python/cpython · GitHub
Skip to content

gh-66587: Fix deadlock from pool worker death without communication #16103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
applio wants to merge 7 commits into from
Closed

gh-66587: Fix deadlock from pool worker death without communication #16103

Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c8f4896
Adds tracking of which process in the pool takes which job from the q…
applio Sep 13, 2019
315ec3d
Added blurb.
applio Sep 13, 2019
bcbd7d3
Fix for missing checks on resources still being available during tear…
applio Sep 15, 2019
e1a9eb5
Remove spurious space.
applio Sep 22, 2019
6459284
Fix result position for killed workers, add Steve suggested change to…
applio Sep 23, 2019
fa54afb
Merge branch 'main' into fix_multiprocessing_worker_died_indefinite_hang
gpshead May 22, 2025
5bc6db5
Merge branch 'main' into fix_multiprocessing_worker_died_indefinite_hang
gpshead Jun 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Fix result position for killed workers, add Steve suggested change to… 
… use dict.pop().
  • Loading branch information
@applio
applio committed Sep 23, 2019
commit 64592844258585099b517b2f3ef8caa73d8b4464
13 changes: 7 additions & 6 deletions Lib/multiprocessing/pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,12 +299,13 @@ def _join_exited_workers(pool, outqueue, job_assignments):
pid = worker.ident
worker.join()
cleaned = True
if pid in job_assignments:
job_info = job_assignments.pop(pid, None)

@pablogsal pablogsal Jun 29, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This depends on _handle_results() recording the assignment before _join_exited_workers() runs. If the worker dies first, we miss the job and the result can still hang.

@gpshead gpshead Jun 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think there's still a race condition regardless.

if job_info is not None:
# If the worker process died without communicating back
# while running a job, add a default result for it.
job = job_assignments[pid]
outqueue.put((job, i, (False, RuntimeError("Worker died"))))
del job_assignments[pid]
outqueue.put(
(*job_info, (False, RuntimeError("Worker died")))
)
del pool[i]
return cleaned

Expand Down Expand Up @@ -602,7 +603,7 @@ def _handle_results(outqueue, get, cache, job_assignments):
# task_info is True or False when a task has completed but
# None indicates information about which process has
# accepted a job from the queue.
job_assignments[value] = job
job_assignments[value] = (job, i)

@pablogsal pablogsal Jun 29, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This entry is never cleared when the worker sends the real result, so maxtasksperchild can report Worker died for a completed map chunk. Can we clear it when handling the result?

else:
try:
cache[job]._set(i, (task_info, value))

@tomMoral tomMoral Sep 20, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why don't you remove the job from job_assignement here? It would avoid unecessary operation when a worker died gracefully.

Expand All @@ -623,7 +624,7 @@ def _handle_results(outqueue, get, cache, job_assignments):

job, i, (task_info, value) = task
if task_info is None:
job_assignments[value] = job
job_assignments[value] = (job, i)
else:
try:
cache[job]._set(i, (task_info, value))
Expand Down