gh-155852: Do not cancel remaining Executor.map calls on a callable's TimeoutError (GH-155853)

A TimeoutError raised by the mapped callable was re-raised like the
map(timeout=...) wait timeout, aborting the iteration and cancelling the
remaining calls, unlike every other exception since gh-108518. A wait timeout
only occurs while the future is still running, so a TimeoutError from an
already-finished future is treated as the callable's own result.

Author: tonghuaroot

Lib/concurrent/futures/_base.py

@@ -309,10 +309,13 @@ def wait(fs, timeout=None, return_when=ALL_COMPLETED):
 def _result_or_cancel(fut, timeout=None):
     try:
         try:
-            return (fut.result(timeout), None)
-        except TimeoutError:
-            raise
-        except BaseException as exc:
+            # fut.exception() returns the call's own error but raises
+            # TimeoutError only for a map() timeout.
+            exc = fut.exception(timeout)
+            if exc is not None:
+                return (None, exc)
+            return (fut.result(), None)
+        except CancelledError as exc:
             return (None, exc)
         finally:
             fut.cancel()

Lib/test/test_concurrent_futures/executor.py

@@ -29,6 +29,12 @@ def raiser(exception, msg='std'):
     raise exception(msg)
 
 
+def timeout_on_one(x):
+    if x == 1:
+        raise TimeoutError
+    return x
+
+
 class FalseyBoolException(Exception):
     def __bool__(self):
         return False
@@ -87,6 +93,20 @@ def test_map_exception(self):
         self.assertRaises(StopIteration, next, i)
         self.assertRaises(StopIteration, next, i)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_map_timeout_from_callable(self):
+        # A TimeoutError from the callable is not the map() timeout, whether
+        # or not a map() timeout is set.
+        for timeout in (None, support.SHORT_TIMEOUT):
+            with self.subTest(timeout=timeout):
+                i = self.executor.map(timeout_on_one, [0, 1, 2, 3],
+                                      timeout=timeout)
+                self.assertEqual(next(i), 0)
+                self.assertRaises(TimeoutError, next, i)
+                self.assertEqual(next(i), 2)
+                self.assertEqual(next(i), 3)
+                self.assertRaises(StopIteration, next, i)
+
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @support.requires_resource('walltime')
     def test_map_timeout(self):