Use multiprocessing.Manager for process based executors

Author: colesbury

Lib/test/test_concurrent_futures/test_wait.py

@@ -114,9 +114,9 @@ def test_all_completed(self):
 
     def test_timeout(self):
         short_timeout = 0.050
-        event = threading.Event()
+        self.event.clear()
 
-        future = self.executor.submit(event.wait)
+        future = self.executor.submit(self.event.wait)
 
         finished, pending = futures.wait(
                 [CANCELLED_AND_NOTIFIED_FUTURE,
@@ -126,7 +126,7 @@ def test_timeout(self):
                 timeout=short_timeout,
                 return_when=futures.ALL_COMPLETED)
 
-        event.set()
+        self.event.set()
         futures.wait([future], return_when=futures.ALL_COMPLETED)
 
         self.assertEqual(set([CANCELLED_AND_NOTIFIED_FUTURE,

Lib/test/test_concurrent_futures/util.py

@@ -1,5 +1,6 @@
 import multiprocessing
 import sys
+import threading
 import time
 import unittest
 from concurrent import futures
@@ -46,18 +47,24 @@ def setUp(self):
 
         self.t1 = time.monotonic()
         if hasattr(self, "ctx"):
+            self.manager = multiprocessing.Manager()
+            self.event = self.manager.Event()
             self.executor = self.executor_type(
                 max_workers=self.worker_count,
                 mp_context=self.get_context(),
                 **self.executor_kwargs)
         else:
+            self.event = threading.Event()
             self.executor = self.executor_type(
                 max_workers=self.worker_count,
                 **self.executor_kwargs)
 
     def tearDown(self):
         self.executor.shutdown(wait=True)
         self.executor = None
+        if hasattr(self, "ctx"):
+            self.manager.shutdown()
+            self.manager = None
 
         dt = time.monotonic() - self.t1
         if support.verbose: