Parent process hanging in multiprocessing if children terminate unexpectedly

Lib/test/test_concurrent_futures.py

 import test.support
 
 # Skip tests if _multiprocessing wasn't built.
 test.support.import_module('_multiprocessing')
 # Skip tests if sem_open implementation is broken.
 test.support.import_module('multiprocessing.synchronize')
 # import threading after _multiprocessing to raise a more revelant error
 # message: "No module named _multiprocessing". _multiprocessing is not compiled
 # without thread support.
 test.support.import_module('threading')
 
 from test.script_helper import assert_python_ok
 
 import sys
 import threading
 import time
 import unittest
 
 from concurrent import futures
 from concurrent.futures._base import (
     PENDING, RUNNING, CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED, Future)
-import concurrent.futures.process
+from concurrent.futures.process import BrokenProcessPool
 
 
 def create_future(state=PENDING, exception=None, result=None):
     f = Future()
     f._state = state
     f._exception = exception
     f._result = result
     return f
 
 
@@ skipping 114 matching lines @@
         pass
 
     def test_processes_terminate(self):
         self.executor.submit(mul, 21, 2)
         self.executor.submit(mul, 6, 7)
         self.executor.submit(mul, 3, 14)
         self.assertEqual(len(self.executor._processes), 5)
         processes = self.executor._processes
         self.executor.shutdown()
 
-        for p in processes:
+        for p in processes.values():
             p.join()
 
     def test_context_manager_shutdown(self):
         with futures.ProcessPoolExecutor(max_workers=5) as e:
             processes = e._processes
             self.assertEqual(list(e.map(abs, range(-5, 5))),
                              [5, 4, 3, 2, 1, 0, 1, 2, 3, 4])
 
-        for p in processes:
+        for p in processes.values():
             p.join()
 
     def test_del_shutdown(self):
         executor = futures.ProcessPoolExecutor(max_workers=5)
         list(executor.map(abs, range(-5, 5)))
         queue_management_thread = executor._queue_management_thread
         processes = executor._processes
         del executor
 
         queue_management_thread.join()
-        for p in processes:
+        for p in processes.values():
             p.join()
 
 class WaitTests(unittest.TestCase):
     def test_first_completed(self):
         future1 = self.executor.submit(mul, 21, 2)
         future2 = self.executor.submit(time.sleep, 1.5)
 
         done, not_done = futures.wait(
                 [CANCELLED_FUTURE, future1, future2],
                  return_when=futures.FIRST_COMPLETED)
@@ skipping 186 matching lines @@
         finished = []
         def record_finished(n):
             finished.append(n)
 
         self.executor.map(record_finished, range(10))
         self.executor.shutdown(wait=True)
         self.assertCountEqual(finished, range(10))
 
 
 class ProcessPoolExecutorTest(ProcessPoolMixin, ExecutorTest):
-    pass
+    def test_killed_child(self):
+        # When a child process is abruptly terminated, the whole pool gets
+        # "broken".
+        futures = [self.executor.submit(time.sleep, 3)]
+        # Get one of the processes, and terminate (kill) it
+        p = next(iter(self.executor._processes.values()))
+        p.terminate()
+        for fut in futures:

[bquinlan, 2011/06/08 12:31:32]

fut => f (for consistency)

+            self.assertRaises(BrokenProcessPool, fut.result)
+        # Submitting other jobs fails as well.
+        self.assertRaises(BrokenProcessPool, self.executor.submit, pow, 2, 8)
 
 
 class FutureTests(unittest.TestCase):
     def test_done_callback_with_result(self):
         callback_result = None
         def fn(callback_future):
             nonlocal callback_result
             callback_result = callback_future.result()
 
         f = Future()
@@ skipping 217 matching lines @@
                               ProcessPoolWaitTests,
                               ThreadPoolWaitTests,
                               ProcessPoolAsCompletedTests,
                               ThreadPoolAsCompletedTests,
                               FutureTests,
                               ProcessPoolShutdownTest,
                               ThreadPoolShutdownTest)
 
 if __name__ == "__main__":
     test_main()