Parent process hanging in multiprocessing if children terminate unexpectedly · Issue #53451 · python/cpython · GitHub

BPO	9205
Nosy	@jcea, @brianquinlan, @pitrou, @vstinner, @vlasovskikh
Dependencies	bpo-11743: Rewrite PipeConnection and Connection in pure Python bpo-12040: Expose a Process.sentinel property (and fix polling loop in Process.join())
Files	termination.patch: Patch to deal with unexpected termination of children shutdown.patch: Patch to catch exceptions raised upon interpeter shutdown pickling_error.patch: Inform an application that is has failed if a task is unpickleable multiprocessing-trunk@82502-termination3.patch multiprocessing-trunk@82502-termination-trackjobs.patch assign-tasks.patch: Record the assignment of tasks to workers multiprocessing-trunk@82502-termination-trackjobs2.patch multiprocessing-trunk@82502-termination-trackjobs3.patch sentinels.patch sentinels2.patch sentinels3.patch sentinels4.patch sentinels5.patch multiproc_broken_pool.diff: Raise BrokenProcessPool if a process unexecptedly dies in a multiprocessing.Pool

Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.

Show more details

GitHub fields:

assignee = None
closed_at = <Date 2011-06-08.15:25:30.134>
created_at = <Date 2010-07-08.20:00:44.723>
labels = ['type-bug', 'library']
title = 'Parent process hanging in multiprocessing if children terminate unexpectedly'
updated_at = <Date 2014-09-11.22:34:12.788>
user = 'https://bugs.python.org/gdb'

bugs.python.org fields:

activity = <Date 2014-09-11.22:34:12.788>
actor = 'dan.oreilly'
assignee = 'jnoller'
closed = True
closed_date = <Date 2011-06-08.15:25:30.134>
closer = 'pitrou'
components = ['Library (Lib)']
creation = <Date 2010-07-08.20:00:44.723>
creator = 'gdb'
dependencies = ['11743', '12040']
files = ['17905', '17934', '17987', '18015', '18026', '18513', '18657', '18664', '21865', '21923', '21928', '21937', '22266', '36454']
hgrepos = []
issue_num = 9205
keywords = ['patch']
message_count = 78.0
messages = ['109585', '109867', '109885', '109910', '109922', '109936', '110129', '110136', '110139', '110142', '110152', '110169', '110174', '110197', '110207', '110256', '110283', '110285', '110288', '110353', '110366', '110369', '110370', '110386', '110387', '110399', '110428', '110979', '111025', '111028', '111124', '111559', '111690', '111696', '111706', '111915', '113828', '114423', '114449', '115065', '115107', '115114', '115118', '115125', '115128', '132646', '132660', '132661', '132664', '132665', '132666', '135013', '135021', '135025', '135026', '135027', '135028', '135461', '135493', '135502', '135513', '135544', '135624', '135895', '135896', '135901', '135904', '135968', '136090', '137767', '137855', '137912', '137913', '138065', '225833', '226699', '226794', '226806']
nosy_count = 16.0
nosy_names = ['jcea', 'bquinlan', 'pitrou', 'vstinner', 'jnoller', 'hongqn', 'asksol', 'vlasovskikh', 'neologix', 'gdb', 'Albert.Strasheim', 'aljungberg', 'python-dev', 'sbt', 'gkcn', 'dan.oreilly']
pr_nums = []
priority = 'normal'
resolution = 'fixed'
stage = 'resolved'
status = 'closed'
superseder = None
type = 'behavior'
url = 'https://bugs.python.org/issue9205'
versions = ['Python 3.3']

I have recently begun using multiprocessing for a variety of batch
jobs. It's a great library, and it's been quite useful. However, I have been bitten several times by situations where a worker process in a Pool will unexpectedly die, leaving multiprocessing hanging in a wait. A simple example of this is produced by the following:
"""
#!/usr/bin/env python
import multiprocessing, sys
def foo(x):
sys.exit(1)
multiprocessing.Pool(1).apply(foo, [1])
"""
The child will exit and the parent will hang forever. A similar occurrence happens if one pushes C-c while a child process is running (this special case is noted in http://bugs.python.org/issue8296) or killed by a signal.

Attached is a patch to handle unexpected terminations of children
processes and prevent the parent process from hanging. A test case is included. (Developed and tested on 64-bit Ubuntu.) Please let me know what you think. Thanks!

thanks greg; I'm going to take a look and think about this. I'd like to resolve bug 9207 first though

Cool, thanks.  I'll note that with this patch applied, using the test program from 9207 I consistently get the following exception:
"""
Exception in thread Thread-1 (most likely raised during interpreter shutdown):
Traceback (most recent call last):
  File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner
  File "/usr/lib/python2.6/threading.py", line 484, in run
  File "/home/gdb/repositories/multiprocessing/pool.py", line 312, in _handle_workers
  File "/home/gdb/repositories/multiprocessing/pool.py", line 190, in _maintain_pool
  File "/home/gdb/repositories/multiprocessing/pool.py", line 158, in _join_exited_workers
<type 'exceptions.TypeError'>: 'NoneType' object is not callable
"""

This is line 148 in the unpatched source, namely the 'reversed(range(len(self._pool)))' line of _join_exited_workers. Looks like the same issue, where instead reversed/range/len have been set to None.

So I think by changing how much time the worker_handler spends in various functions, I've made it possible (or just more likely?) that if we lose the race with interpreter shutdown the worker_handler will be in the middle of _join_exited_workers. This may mean that someone should keep around a local reference to reversed/range/len... not sure if there's a better solution.

Ugh. I'm going to have to think about the cleanest way of handling this case of functions vanishing from us since this is going to be more widespread inside the code. Suggestions welcome.

What about just catching the exception? See e.g. the attached patch. (Disclaimer: not heavily tested).

A+ for creativity; I wouldn't have thought of that ;)

termination.patch, in the result handler you've added:

while cache and thread._state != TERMINATE and not failed

why are you terminating the second pass after finding a failed process?

Unpickleable errors and other errors occurring in the worker body are not exceptional cases, at least not now that the pool is supervised by _handle_workers. I think the result should be set also in this case, so the user can inspect the exception after the fact.

I have some other suggestions too, so I will review this patch tomorrow.

For shutdown.patch, I thought this only happened in the worker handler, but you've enabled this for the result handler too? I don't care about the worker handler, but with the result handler I'm worried that I don't know what ignoring these exceptions actually means. For example, is there a possibility that we may lose results at shutdown?

Thanks much for taking a look at this!

why are you terminating the second pass after finding a failed
process?
Unfortunately, if you've lost a worker, you are no longer guaranteed that cache will eventually be empty. In particular, you may have lost a task, which could result in an ApplyResult waiting forever for a _set call.

More generally, my chief assumption that went into this is that the unexpected death of a worker process is unrecoverable. It would be nice to have a better workaround than just aborting everything, but I couldn't see a way to do that.

Unpickleable errors and other errors occurring in the worker body are
not exceptional cases, at least not now that the pool is supervised
by _handle_workers.
I could be wrong, but that's not what my experiments were indicating. In particular, if an unpickleable error occurs, then a task has been lost, which means that the relevant map, apply, etc. will wait forever for completion of the lost task.

I think the result should be set also in this case, so the user can
inspect the exception after the fact.
That does sound useful. Although, how can you determine the job (and the value of i) if it's an unpickleable error? It would be nice to be able to retrieve job/i without having to unpickle the rest.

For shutdown.patch, I thought this only happened in the worker
handler, but you've enabled this for the result handler too? I don't
care about the worker handler, but with the result handler I'm
worried that I don't know what ignoring these exceptions actually
means.
You have a good point. I didn't think about the patch very hard. I've only seen these exceptions from the worker handler, but AFAICT there's no guarantee that bad luck with the scheduler wouldn't result in the same problem in the result handler. One option would be to narrow the breadth of the exceptions caught by _make_shutdown_safe (do we need to catch anything but TypeErrors?). Another option would be to enable only for the worker handler. I don't have a particularly great sense of what the Right Thing to do here is.

Greg - I asked Ask to take a look - his celery package is a huge consumer of multiprocessing, and so I tend to run things past him as well.

That said - to both of you - the fundamental problem the shutdown patch is trying to scratch is located in bpo-9207 - greg's termination patch just exposes the problem in 9207 a lot more.

Focusing specifically on the shutdown patch; our issue is that during interpreter shutdown, sys.modules is iterated, and entries are set to None - for threads which "live on" well into that cycle can end up losing imported functions/modules/etc. The multiple daemon threads in the Pool code are exposing this as code which executed imported functions (such as the debug() statement in handle_workers) which will fire after the pool has exited and the interpreter is shut down.

We can work around the shutdown issue (really, bug 9207) by ignoring the exception such as shutdown.patch does, or passing in references/adding references to the functions those methods need. Or (as Brett suggested) converting them to class methods and adding references to the class. Or passing them in via the signature like this _handle_workers(arg, _debug=debug), etc.

Unfortunately, if you've lost a worker, you are no
longer guaranteed that cache will eventually be empty.
In particular, you may have lost a task, which could
result in an ApplyResult waiting forever for a _set call.

More generally, my chief assumption that went into this
is that the unexpected death of a worker process is
unrecoverable. It would be nice to have a better workaround
than just aborting everything, but I couldn't see a way
to do that.

It would be a problem if the process simply disappeared,
But in this case you have the ability to put a result on the queue,
so it doesn't have to wait forever.

For processes disappearing (if that can at all happen), we could solve
that by storing the jobs a process has accepted (started working on),
so if a worker process is lost, we can mark them as failed too.

I could be wrong, but that's not what my experiments
were indicating. In particular, if an unpickleable error occurs,
then a task has been lost, which means that the relevant map,
apply, etc. will wait forever for completion of the lost task.

It's lost now, but not if we handle the error...
For a single map operation this behavior may make sense, but what about
someone running the pool as s long-running service for users to submit map operations to? Errors in this context are expected to happen, even unpickleable errors.

I guess that the worker handler works as a supervisor is a side effect,
as it was made for the maxtasksperchild feature, but for me it's a welcome one. With the supervisor in place, multiprocessing.pool is already fairly stable to be used for this use case, and there's not much to be done to make it solid (Celery is already running for months without issue, unless there's a pickling error...)

That does sound useful. Although, how can you determine the
job (and the value of i) if it's an unpickleable error?
It would be nice to be able to retrieve job/i without having
to unpickle the rest.

I was already working on this issue last week actually, and I managed
to do that in a way that works well enough (at least for me):
http://github.com/ask/celery/commit/eaa4d5ddc06b000576a21264f11e6004b418bda1#diff-1

For processes disappearing (if that can at all happen), we could solve
that by storing the jobs a process has accepted (started working on),
so if a worker process is lost, we can mark them as failed too.
Sure, this would be reasonable behavior. I had considered it but decided it as a larger change than I wanted to make without consulting the devs.

I was already working on this issue last week actually, and I managed
to do that in a way that works well enough (at least for me):
If I'm reading this right, you catch the exception upon pickling the result (at which point you have the job/i information already; totally reasonable). I'm worried about the case of unpickling the task failing. (Namely, the "task = get()" line of the "worker" method.) Try running the following:
"""
#!/usr/bin/env python
import multiprocessing
p = multiprocessing.Pool(1)
def foo(x):
pass
p.apply(foo, [1])
"""
And if "task = get()" fails, then the worker doesn't know what the relevant job/i values are.

Anyway, so I guess the question that is forming in my mind is, what sorts of errors do we want to handle, and how do we want to handle them? My answer is I'd like to handle all possible errors with some behavior that is not "hang forever". This includes handling children processes dying by signals or os._exit, raising unpickling errors, etc.

I believe my patch provides this functionality. By adding the extra mechanism that you've written/proposed, we can improve the error handling in specific recoverable cases (which probably constitute the vast majority of real-world cases).

I think I misunderstood the purpose of the patch. This is about handling errors on get(), not on put() like I was working on. So sorry for that confusion.

What kind of errors are you having that makes the get() call fail?

If the queue is not working, then I guess the only sensible approach is to shutdown the pool like suggested. I'll open up another issue for unpickleable errors then.

Not exactly. The select is done on the queue's pipe and on the workers'
fds *at the same time*. Thus there's no race condition.

You're right, I missed this part, it's perfectly safe.

But I think there's a problem with the new implementation in Python.
Writes to a pipe are guaranteed to be atomic if you write less than
PIPE_BUF (4K on Linux, 512 by POSIX) at a time. Writes to a datagram
Unix domain socket are also atomic.
But Lib/multiprocessing/connection.py does:

    def _send_bytes(self, buf):
        # For wire compatibility with 3.2 and lower
        n = len(buf)
        self._send(struct.pack("=i", len(buf)))
        # The condition is necessary to avoid "broken pipe" errors
        # when sending a 0-length buffer if the other end closed the pipe.
        if n > 0:
            self._send(buf)

This is definitely not atomic. If two processes write objects of
different size at the same time, it can probably lead to trouble.
Also, Pipe(duplex=True) should probably return a SOCK_DGRAM Unix
socket for the same reason.
If I missed something here, I promise to shut up ;-)

But Lib/multiprocessing/connection.py does:

def \_send_bytes(self, buf):
    # For wire compatibility with 3.2 and lower
    n = len(buf)
    self.\_send(struct.pack("=i", len(buf)))
    # The condition is necessary to avoid "broken pipe" errors
    # when sending a 0-length buffer if the other end closed the pipe.
    if n \> 0:
        self.\_send(buf)

This is definitely not atomic.

Indeed, it isn't, Pipe objects are not meant to be safe against multiple
access. Queue objects (in multiprocessing/queues.py) use locks so they
are safe.

Indeed, it isn't, Pipe objects are not meant to be safe against multiple
access. Queue objects (in multiprocessing/queues.py) use locks so they
are safe.

But if the write to the Pipe is not atomic, then the select isn't safe.
select will return as soon as some data is available for reading. So
let's say this happens:
parent process waiting inside Connection.recv() on poll():
def recv(self, sentinels=None):
if sentinels:
self._poll(-1.0, sentinels)
buf = self._recv_bytes()
return pickle.loads(buf.getbuffer())

worker process, inside Connection.send():
def send_bytes(self, buf):
n = len(buf)
self._send(struct.pack("=i", len(buf)))
[crash]
# The condition is necessary to avoid "broken pipe" errors
# when sending a 0-length buffer if the other end closed the pipe.
if n > 0:
self._send(buf)

In the parent process, _poll() will return self._handle as readable as
soon as len(buf) has been sent by the worker process. Thus,
Connection.recv_bytes() will be called:

    def _recv_bytes(self, maxsize=None):
        buf = self._recv(4)
        size, = struct.unpack("=i", buf.getvalue())
        if maxsize is not None and size > maxsize:
            return None
        return self._recv(size)

The first _recv() will succeed, since the length is in the Pipe.
The second one, however, will remain stuck on the read from the
pipe/unix socket, because there's no more data.
This can be reproduced easily by adding a short sleep right after the
sending of the length of the buffer inside send_bytes(), and then
sending a SIGKILL to a worker process.
(Also, I now remember why I made the comment about the handle being
read after the select, I spotted this earlier but completely forgot
about it afterwards...).

Thus,
Connection.recv_bytes() will be called:
def \_recv_bytes(self, maxsize=None):
    buf = self.\_recv(4)
    size, = struct.unpack("=i", buf.getvalue())
    if maxsize is not None and size \> maxsize:
        return None
    return self.\_recv(size)
The first _recv() will succeed, since the length is in the Pipe.
The second one, however, will remain stuck on the read from the
pipe/unix socket, because there's no more data.
This can be reproduced easily by adding a short sleep right after the
sending of the length of the buffer inside send_bytes(), and then
sending a SIGKILL to a worker process.

That's true. We would need to insert a select() before each raw read().

Ok, the dependencies are now committed. Here is a new patch addressing Charles-François' comments: select() is now called before each call to read() when sentinels are given, to avoid race conditions.

Ok, the dependencies are now committed. Here is a new patch addressing Charles-François' comments: select() is now called before each call to read() when sentinels are given, to avoid race conditions.

The patch looks fine to me (but I didn't look at win32-specific code).

New changeset 6d6099f7fe89 by Antoine Pitrou in branch 'default':
Issue bpo-9205: concurrent.futures.ProcessPoolExecutor now detects killed
http://hg.python.org/cpython/rev/6d6099f7fe89

So, concurrent.futures is fixed now. Unless someone wants to patch multiprocessing.Pool, I am closing this issue.

test_multiprocessing crashs ~700 times on Mac OS X Tiger, regression likely introduced by this issue (6d6099f7fe89): I created issue bpo-12310 for that.

> So, concurrent.futures is fixed now. Unless someone wants to patch multiprocessing.Pool, I am closing this issue.

I realize I'm 3 years late on this, but I've put together a patch for multiprocessing.Pool. Should a process in a Pool unexpectedly exit (meaning, *not* because of hitting the maxtasksperchild limit), the Pool will be closed/terminated and all cached/running tasks will return a BrokenProcessPool. These changes also prevent the Pool from going into a bad state if the "initializer" function raises an exception (previously, the pool would end up infinitely starting new processes, which would immediately die because of the exception).

One concern with the patch: The way timings are altered with these changes, the Pool seems to be particularly susceptible to bpo-6721 in certain cases. If processes in the Pool are being restarted due to maxtasksperchild just as the worker is being closed or joined, there is a chance the worker will be forked while some of the debug logging inside of Pool is running (and holding locks on either sys.stdout or sys.stderr). When this happens, the worker deadlocks on startup, which will hang the whole program. I believe the current implementation is susceptible to this as well, but I could reproduce it much more consistently with this patch. I think its rare enough in practice that it shouldn't prevent the patch from being accepted, but thought I should point it out.

(I do think bpo-6721 should be addressed, or at the very least internal I/O locks should always reset after forking.)

Is it possible to have this issue re-opened, so that the new patch is more likely to get attention? Or should I create a new issue for the multiprocessing patch?

You should certainly create a new issue!

Thanks, Antoine. I've opened bpo-22393.

Parent process hanging in multiprocessing if children terminate unexpectedly · Issue #53451 · python/cpython · GitHub

Repository navigation

Parent process hanging in multiprocessing if children terminate unexpectedly #53451

55 remaining items

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Parent process hanging in multiprocessing if children terminate unexpectedly · Issue #53451 · python/cpython · GitHub

Repository navigation

Parent process hanging in multiprocessing if children terminate unexpectedly #53451

Description

Activity

gdb commented on Jul 8, 2010

jnoller commented on Jul 10, 2010

gdb commented on Jul 10, 2010

jnoller commented on Jul 10, 2010

gdb commented on Jul 10, 2010

jnoller commented on Jul 10, 2010

asksol commented on Jul 12, 2010

gdb commented on Jul 12, 2010

jnoller commented on Jul 12, 2010

asksol commented on Jul 12, 2010

gdb commented on Jul 13, 2010

asksol commented on Jul 13, 2010

55 remaining items

neologix commented on May 13, 2011

pitrou commented on May 13, 2011

neologix commented on May 14, 2011

pitrou commented on May 16, 2011

pitrou commented on Jun 6, 2011

neologix commented on Jun 7, 2011

python-dev commented on Jun 8, 2011

pitrou commented on Jun 8, 2011

vstinner commented on Jun 10, 2011

danoreilly commented on Aug 24, 2014

danoreilly commented on Sep 10, 2014

pitrou commented on Sep 11, 2014

danoreilly commented on Sep 11, 2014

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions