Move timing dependent test (of multiple callback failures) into separate

sub-test and use a barrier to ensure it is reliable. We must ensure all tasks
are submitted before any of them run and mark the pool as broken. Since
barriers can't be shared between processes, do not run that sub-test with
process-based parallelism.

Author: duaneg

Lib/test/_test_multiprocessing.py

@@ -3110,22 +3110,20 @@ def _map(pool, target, **kwargs):
 
                 # Fail upon trying to reuse a broken pool after error callback failures:
                 # - BrokenPoolError containing:
-                #   - 3x CallbackError each containing:
+                #   - CallbackError containing:
                 #     - Error thrown from the callback
                 #     - Original error
                 with self.assertRaises(BrokenPoolError) as pool_ctx:
-                    with self.Pool(3) as pool:
-                        res = [func(pool, raising2, error_callback=raising)
-                               for _ in range(3)]
-                        for r in res:
-                            with self.assertRaises(CallbackError) as res_ctx:
-                                r.get()
-                            self._check_subexceptions(res_ctx.exception,
-                                                      [KeyError, IndexError])
+                    with self.Pool(1) as pool:
+                        res = func(pool, raising2, error_callback=raising)
+                        with self.assertRaises(CallbackError) as res_ctx:
+                            res.get()
+                        self._check_subexceptions(res_ctx.exception,
+                                                  [KeyError, IndexError])
                     pool.apply_async(noop)
-                self._check_subexceptions(pool_ctx.exception, [CallbackError] * 3)
-                for se in pool_ctx.exception.exceptions:
-                    self._check_subexceptions(se, [KeyError, IndexError])
+                self._check_subexceptions(pool_ctx.exception, [CallbackError])
+                self._check_subexceptions(pool_ctx.exception.exceptions[0],
+                                          [KeyError, IndexError])
 
                 # Exiting the context manager with a "normal" error and a failed callback
                 # - BrokenPoolError containing:
@@ -3153,13 +3151,33 @@ def _map(pool, target, **kwargs):
                 self._check_subexceptions(pool_ctx.exception, [CallbackError])
                 self._check_subexceptions(pool_ctx.exception.exceptions[0], [KeyError])
 
+        # Skip this test for process-based parallelism as sharing the barrier will fail
+        if self.TYPE != 'processes':
+            with self.subTest(name="Multiple callback failures"):
+                # Fail with 3x callback failure:
+                # - BrokenPoolError containing:
+                #   - 3x CallbackError containing:
+                #     - Error thrown from the callback
+                with self.assertRaises(BrokenPoolError) as pool_ctx:
+                    kwds = {'barrier': self.Barrier(3)}
+                    with self.Pool(3) as pool:
+                        res = [pool.apply_async(noop, kwds=kwds, callback=raising)
+                               for _ in range(3)]
+                        for r in res:
+                            with self.assertRaises(CallbackError) as res_ctx:
+                                r.get()
+                self._check_subexceptions(pool_ctx.exception, [CallbackError] * 3)
+                for se in pool_ctx.exception.exceptions:
+                    self._check_subexceptions(se, [KeyError])
+
     def _check_subexceptions(self, group, sub_types):
         self.assertEqual(len(group.exceptions), len(sub_types))
         for sub_exc, sub_type in zip(group.exceptions, sub_types):
             self.assertIsInstance(sub_exc, sub_type)
 
-def noop(*args):
-    pass
+def noop(*args, barrier=None):
+    if barrier:
+        barrier.wait()
 
 def raising(*args):
     raise KeyError("key")