multiprocessing.Connection does not communicate pipe closure between parent and child · Issue #56697 · python/cpython · GitHub

BPO	12488
Nosy	@pitrou

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 = None
created_at = <Date 2011-07-04.11:43:29.105>
labels = ['type-bug', '3.8', '3.9', '3.10', 'docs']
title = 'multiprocessing.Connection does not communicate pipe closure between parent and child'
updated_at = <Date 2020-11-14.11:11:23.404>
user = 'https://bugs.python.org/lcampagn'

bugs.python.org fields:

activity = <Date 2020-11-14.11:11:23.404>
actor = 'neologix'
assignee = 'docs@python'
closed = False
closed_date = None
closer = None
components = ['Documentation']
creation = <Date 2011-07-04.11:43:29.105>
creator = 'lcampagn'
dependencies = []
files = []
hgrepos = []
issue_num = 12488
keywords = []
message_count = 8.0
messages = ['139754', '139765', '139769', '139788', '139789', '139791', '139792', '159282']
nosy_count = 4.0
nosy_names = ['syeberman', 'pitrou', 'docs@python', 'lcampagn']
pr_nums = []
priority = 'normal'
resolution = None
stage = 'needs patch'
status = 'open'
superseder = None
type = 'behavior'
url = 'https://bugs.python.org/issue12488'
versions = ['Python 3.8', 'Python 3.9', 'Python 3.10']

I have found that when using multiprocessing.Connection objects to pass data between two processes, closing one end of the pipe is not properly communicated to the other end. My expectation was that when calling recv() on the remote end, it should raise EOFError if the pipe has been closed. Instead, the remote recv() blocks indefinitely. This behavior exists on Linux and Cygwin, but NOT on native Windows.

Example:

    import multiprocessing as m

    def fn(pipe):
        print "recv:", pipe.recv()
        print "recv:", pipe.recv()

    if __name__ == '__main__':
        p1, p2 = m.Pipe()
        pr = m.Process(target=fn, args=(p2,))
        pr.start()
        p1.send(1)
        p1.close()  ## should generate EOFError in remote process

That's because the other end of the pipe (p1) is open in the child process (FDs are inherited on fork()).
Just add
p1.close()

at the beginning of fn() and you'll get EOF.
Closing as invalid.

That's interesting, thanks for your response.

It is also a bit awkward..
Might I recommend adding a note to the documentation? It is not really
intuitive that each child should need to close the end of the pipe it isn't
using (especially since it is possible to create a child that has no
explicit access to that end of the pipe, even though it has inherited the
file descriptor).

2011/7/4 Charles-François Natali <report@bugs.python.org>

Charles-François Natali <neologix@free.fr> added the comment:

That's because the other end of the pipe (p1) is open in the child process
(FDs are inherited on fork()).
Just add
p1.close()

at the beginning of fn() and you'll get EOF.
Closing as invalid.

----------
nosy: +neologix
resolution: -> invalid
stage: -> committed/rejected
status: open -> closed

Python tracker <report@bugs.python.org>
<http://bugs.python.org/issue12488\>

Well, in this regard it behaves like a Unix pipe/socket (in the duplex case it's implemented with a Unix domain socket), so I find it quite natural (of course, you have to know about FD inheritance upon fork()).
I'm not convinced it's necessary, Antoine any thought on that?

Well, I think it deserves a comment in the documentation that behaviour of Pipes and Queues when one of the process terminates is undefined and implementation-dependent.

By the way, there's internal support in 3.3 to reliably detect killed children, and it's used by concurrent.futures: http://docs.python.org/dev/library/concurrent.futures.html#concurrent.futures.BrokenProcessPool. However, I'm not sure there's an easy way to detect a killed master process from one of the worker processes.

Alright.
Luke, if you're motivated, feel free to provide a patch.
The relevant file is Doc/library/multiprocessing.rst.

By the way, if you don't want children processes to continue running when the master exits, just make them daemonic processes (by adding "daemon=True" to the Process() constructor call).

This issue _does_ exist on Windows, and is not limited to the case where the master process exits before its children. The following code, which is almost exactly that from the 2.7.3 documentation, deadlocks on Win7 (Py3.2 and 2.7) and WinXP (Py3.2 and 2.6):

    from multiprocessing import Process, Pipe
    import sys

    def f(conn):
        #conn.send([42, None, 'hello'])  # uncomment second
        conn.close()

    if __name__ == "__main__":
        parent_conn, child_conn = Pipe()
        p = Process(target=f, args=(child_conn,))
        p.start()
        #child_conn.close()  # uncomment first
        sys.stdout.write( "about to receive\n" )
        sys.stdout.write( "%s\n"%parent_conn.recv() )
        sys.stdout.write( "received\n" )
        p.join()

If you "uncomment first", recv raises an EOFError; if you also "uncomment second", recv succeeds.

If this behaviour is the same on other platforms, then it seems all that is required is to update the documentation.

The problem is still out there under Linux. At least python3.10 and python3.11 are affected.
It works perfectly under Windows, but under Linux child copy of pipe object never knows that the parent process has closed the pipe.
See the code and output for both Windows and Linux below.

import multiprocessing
import multiprocessing.connection
import logging
import random
import time

logging.basicConfig(level=logging.DEBUG)
MESSAGE_SIZE = 1920 * 1080 * 4


def main_parent(pipe_main: multiprocessing.connection.Connection):
    logger = logging.getLogger("MAIN")
    message = random.randbytes(MESSAGE_SIZE)
    for i in range(5):
        t1 = time.time()
        pipe_main.send_bytes(message)
        t2 = time.time()
        logger.info(f"[SEND] frame {i} DeltaT={1000 * (t2 - t1):.1f}")
    pipe_main.close()
    logger.info(f"Main thread is about to stop")


def main_child(pipe_another: multiprocessing.connection.Connection):
    logger = logging.getLogger("NTHR")
    logger.info(f"Hello! The thread has started")
    i = 0
    t1 = time.time()
    try:
        while True:
            t1 = time.time()
            if not pipe_another.poll(timeout=1.0):
                logger.debug(f"No data! {pipe_another.closed=} {pipe_another.fileno()=}")
                continue
            _ = pipe_another.recv_bytes(MESSAGE_SIZE)
            t2 = time.time()
            logger.info(f"[GET] frame {i=} DeltaT={1000 * (t2 - t1):.1f}")
            i += 1
    except Exception as exc:
        t2 = time.time()
        logger.info(f"[GET] Stopped after {exc=} {i=} Delta={1000 * (t2 - t1):.1f}")


def main():
    pipe_reader, pipe_sender = multiprocessing.Pipe(duplex=False)
    proc_child = multiprocessing.Process(target=main_child, args=(pipe_reader,))
    proc_child.start()
    main_parent(pipe_sender)
    proc_child.join()


if __name__ == "__main__":
    main()

Output under Windows (python3.10)

INFO:NTHR:Hello! The thread has started
INFO:MAIN:[SEND] frame 0 DeltaT=65.0
INFO:NTHR:[GET] frame i=0 DeltaT=5.0
INFO:MAIN:[SEND] frame 1 DeltaT=6.0
INFO:NTHR:[GET] frame i=1 DeltaT=7.0
INFO:MAIN:[SEND] frame 2 DeltaT=8.0
INFO:NTHR:[GET] frame i=2 DeltaT=8.0
INFO:MAIN:[SEND] frame 3 DeltaT=8.0
INFO:NTHR:[GET] frame i=3 DeltaT=9.0
INFO:MAIN:[SEND] frame 4 DeltaT=9.0
INFO:MAIN:Main thread is about to stop
INFO:NTHR:[GET] frame i=4 DeltaT=7.0
INFO:NTHR:[GET] Stopped after exc=BrokenPipeError(32, 'Pipe has been closed', None, 109, None) i=5 Delta=1.0

Output under Linux (Ubuntu 22.04LTS, python3.10 and python3.11)

INFO:NTHR:Hello! The thread has started
INFO:NTHR:[GET] frame i=0 DeltaT=24.2
INFO:MAIN:[SEND] frame 0 DeltaT=5.5
INFO:MAIN:[SEND] frame 1 DeltaT=5.4
INFO:NTHR:[GET] frame i=1 DeltaT=6.0
INFO:MAIN:[SEND] frame 2 DeltaT=5.9
INFO:NTHR:[GET] frame i=2 DeltaT=5.9
INFO:MAIN:[SEND] frame 3 DeltaT=4.8
INFO:NTHR:[GET] frame i=3 DeltaT=4.3
INFO:MAIN:[SEND] frame 4 DeltaT=5.2
INFO:MAIN:Main thread is about to stop
INFO:NTHR:[GET] frame i=4 DeltaT=6.3
DEBUG:NTHR:No data! pipe_another.closed=False pipe_another.fileno()=3
DEBUG:NTHR:No data! pipe_another.closed=False pipe_another.fileno()=3
DEBUG:NTHR:No data! pipe_another.closed=False pipe_another.fileno()=3
...
The program never ends

Could you guide me through the problem? Maybe there is a workaround to that problem?

There is also a similar problem in python3.12.4 (linux, x86, NixOS) and python3.11.5 (MacOS, x86) with .recv() on a Connection:

import time
from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe(duplex=False)  # same problem no matter if duplex or not
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, 'hello']"
    time.sleep(2)
    print("reading...")
    parent_conn.recv()  # this should do the broken Pipe thing, but it just blocks indefinitely
    p.join()

I also tried to close both ends in the child process (in duplex mode) but it didn't help at all.
But the docs promise an EOFError.

multiprocessing.Connection does not communicate pipe closure between parent and child · Issue #56697 · python/cpython · GitHub

Repository navigation

multiprocessing.Connection does not communicate pipe closure between parent and child #56697

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

multiprocessing.Connection does not communicate pipe closure between parent and child · Issue #56697 · python/cpython · GitHub

Repository navigation

multiprocessing.Connection does not communicate pipe closure between parent and child #56697

Description

Activity

lcampagn commented on Jul 4, 2011

neologix commented on Jul 4, 2011

lcampagn commented on Jul 4, 2011

neologix commented on Jul 4, 2011

pitrou commented on Jul 4, 2011

neologix commented on Jul 4, 2011

pitrou commented on Jul 4, 2011

syeberman commented on Apr 25, 2012

darksidds commented on May 26, 2023

nonchris commented on Sep 11, 2024

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions