CODEWITHSUNDEEP
pythonmultiprocessingconnectionpipepython-multiprocessing
Wasted Parameter
Wasted Parameter

Reputation: 1

Python multiprocessing, parent process hangs at recv after child process raises exception before send

Python is definitely one of the worst languages for parallel processing.

Just finna get the parent process to throw an error when the child process fails. Instead, the parent process hangs at the recv() method.

Some code in the child process:

try:
    condition.acquire()
    x = 0/0
    pipeConn2.send('test data')
    condition.notify_all()
    condition.release()
except:
    pipeConn2.close() # expect this to raise exception in parent process per docs
    condition.notify_all()
    condition.release()

Some code in the parent process:

try:
    condition0.acquire()
    condition0.wait()
    pipe0Conn1.recv() # hangs here instead of raising exception
except:
    # handle exception

Upvotes: 0

Views: 1070

Answers (2)

Ninjago77
Ninjago77

Reputation: 59

I had the exact same problem, that the Parent Pipe Connection was waiting indefinetly. Instead of using Lock, I used the Parent Pipe Connection . Poll function with returns a boolean value depending on if there is anything to recieve. Some Examples:-

Without Poll -

try:
    pipe0Conn1.recv() # Waits indefinetly (until it recieves)
except EOFError as e:
    # handling not recieved
except:
    # some other exception

Without Poll -

try:
    if pipe0Conn1.poll() is True:
        pipe0Conn1.recv() # Doesn't wait indefinetly (immediate recieve)
    else:
        # Some other code for handling nothing recieved
except EOFError as e:
    # handling not recieved
except:
    # some other exception

PS: Its my first answer on stackoverflow, and I am really sorry about any grammatical, format, code mistakes. Please comment any feedback if you wish.

Upvotes: 0

Booboo
Booboo

Reputation: 44108

First of all, you have a race condition. Even if you eliminate the divide-by-zero exception and never close the connection, the call to recv in the main process will hang anyway if the subprocess is first to acquire the implicit lock associated with the underlying condition instance (I am giving you the benefit of the doubt that condition and condition0 do reference the same instance.

The problem is that if your subprocess is the first to acquire the lock, it will send its data, do its notifying, release the lock and terminate. The main process will then be able to acquire the lock for the first time and then call wait. But when it calls wait it releases the lock and implicitly waits for a notification that has already occurred and will hang indefinitely on the wait. You can verify this by commenting out the statement x = 0/0 and inserting import time; time.sleep(.5) as the first statements within the main process's try block to ensure that the subprocess acquires the condition lock first.

Just to get past that problem I am using an Event instance to pause the subprocess until the main process signals to it that it has acquired the condition lock.

I have verified that there does indeed seem to be a problem with the main process hanging when the subprocess does a close on the connection. This, however, does work correctly when I substitute multithreading for multiprocessing. So there appears to be a bug in Python. I have tested this on Windows with Python 3.8.5 and Linux with Python 3.9.7.

This is what a minimal, reproducible example looks like:

USE_THREADING = False

from multiprocessing import Pipe
if USE_THREADING:
    from threading import Thread as SubTask, Condition, Event
else:
    from multiprocessing import Process as SubTask, Condition, Event

def foo(condition, pipeConn2, acquired_event):
    try:
        acquired_event.wait()
        condition.acquire()
        x = 0/0
        pipeConn2.send('test data')
    except:
        pipeConn2.close() # expect this to raise exception in parent process per docs
    finally:
        condition.notify_all()
        condition.release()

if __name__ == '__main__':
    condition = Condition()
    pipeConn1, pipeConn2 = Pipe()
    # Ensure that our main process/thread is first to acquire the condition:
    acquired_event = Event()
    p = SubTask(target=foo, args=(condition, pipeConn2, acquired_event))
    p.start()
    try:
        condition.acquire()
        # Now allow the subprocess/subthread to proceed to try to acquire the lock:
        acquired_event.set()
        condition.wait()
        print('Receiving data ...')
        data = pipeConn1.recv()
        print(f"Data received: '{data}'")
    except Exception as e:
        print('Exception:', type(e))
    p.join()

Prints:

Receiving data ...

Just as an aside, yours is not a particularly realistic example. A Pipe is essentially a single consumer/single producer communication vehicle unless you build on top of it a lot of "scaffolding" and then you might as well just use one of the Queue variants. Realistically then a process or thread would just be doing a blocking receive on the connection and your producer could write a special sentinel object (None often suffices) to signal that this is the "end of file". There is really no need to do an explicit close on a connection; it will be closed when it is garbage collected.

Upvotes: 1

Related Questions