Python detect and properly handle multiprocessing Manager shutdown

Question

I am receiving IOError: [Errno 32] Broken pipe exceptions when I use a Manager().list() object in several child processes. I understand that a Manager class will:

Manager processes will be shutdown as soon as they are garbage collected or their parent process exits.

https://docs.python.org/2/library/multiprocessing.html#managers

I assume that the Manager shutting down is what's causing this exception:

Traceback (most recent call last):
  File "/opt/django/mdp/bin/mdp.py", line 83, in run
a_tags, iframe_tags = self.get_page_links()
  File "/opt/django/mdp/bin/mdp.py", line 135, in get_page_links
if (stuff not in self.work_tracker) and (stuff not in a_tags):
  File "<string>", line 2, in __contains__
  File "/usr/lib64/python2.7/multiprocessing/managers.py", line 758, in _callmethod
    conn.send((self._id, methodname, args, kwds))
IOError: [Errno 32] Broken pipe

But for the sake of completeness, here's my main function:

if __name__ == "__main__":
    manager = Manager()
    work_queue = Queue()
    work_tracker = manager.list()
    work_results = manager.list()

    work_queue.put('work fed in by a loop or something')

    workers = 4
    processes = []

    for i in range(workers):
        mdp = MDP(work_queue, work_tracker, work_results)
        mdp.start()

    for proc in processes:
        proc.join()

    printable_results = {}
    for each in work_results:
        printable_results[each['stuff']] = each

    print json.dumps(printable_results, indent=4)

And here's the much shorter version of the "MDP" process:

class MDP(Process):
    def __init__(self, work_queue, work_tracker, work_results):
        Process.__init__(self)
        self.exit = Event()
        self.work_queue = work_queue
        self.work_tracker = work_tracker
        self.results = work_results

    def run(self):        
        # Main loop
        while not self.exit.is_set():
            if self.work_queue.empty():
                print '[!] %s sees that the work queue is empty' % self.name
                self.shutdown()
            else:
                try: 
                    job = self.work_queue.get(timeout=3)

                    results = do_something_on_another_thing_with_the_thing(job)

                    self.results.append(results)

                except KeyboardInterrupt:
                    print '[!] %s got Ctrl-C!  Stopping!' % self.name
                    self.shutdown()

                except IOError as ex:
                    print traceback.format_exc()
                    pass

                except Exception as ex:
                    print traceback.format_exc()
                    pass

    def shutdown(self):
        print "[!] Shutting down %s" % self.name
        self.exit.set()

I don't want the Manager to shut down as soon as one child process detects that the queue is empty and shuts itself down. How do I keep the Manager lists open long enough to finish?

Answer 1

It's always comical to me how staring at the same code for a day can lead to becoming a complete idiot. On a whim, I consulted with a co-worker who mentioned that if the processes list is empty, how can the processes join?

This:

for i in range(workers):
    mdp = MDP(work_queue, work_tracker, work_results)
    mdp.start()

Should actually be this:

for i in range(workers):
    mdp = MDP(work_queue, work_tracker, work_results)
    mdp.start()
    processes.append(mdp) # HURRRRRRRRRRRR

I no longer receive the broken pipe error. The pipe is remaining unbroken. Whole. In one piece.