wip

Author: ebonnal

Lib/concurrent/futures/_base.py

@@ -562,7 +562,7 @@ def submit(self, fn, /, *args, **kwargs):
         """
         raise NotImplementedError()
 
-    def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):
+    def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None, as_completed=False):
         """Returns an iterator equivalent to map(fn, iter).
 
         Args:
@@ -579,6 +579,7 @@ def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):
                 iterables pauses until a result is yielded from the buffer.
                 If None, all input elements are eagerly collected, and a task is
                 submitted for each.
+            as_completed: Set to yield the results as they become available.
 
         Returns:
             An iterator equivalent to: map(func, *iterables) but the calls may

Lib/concurrent/futures/process.py

@@ -813,7 +813,7 @@ def submit(self, fn, /, *args, **kwargs):
             return f
     submit.__doc__ = _base.Executor.submit.__doc__
 
-    def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):
+    def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None, as_completed=False):
         """Returns an iterator equivalent to map(fn, iter).
 
         Args:
@@ -829,6 +829,7 @@ def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):
                 iterables pauses until a result is yielded from the buffer.
                 If None, all input elements are eagerly collected, and a task is
                 submitted for each.
+            as_completed: Set to yield the results as they become available.
 
         Returns:
             An iterator equivalent to: map(func, *iterables) but the calls may
@@ -845,7 +846,8 @@ def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):
         results = super().map(partial(_process_chunk, fn),
                               itertools.batched(zip(*iterables), chunksize),
                               timeout=timeout,
-                              buffersize=buffersize)
+                              buffersize=buffersize
+                              as_completed=as_completed)
         return _chain_from_iterable_of_lists(results)
 
     def shutdown(self, wait=True, *, cancel_futures=False):

Lib/test/test_concurrent_futures/executor.py

@@ -26,6 +26,11 @@ def make_dummy_object(_):
     return MyObject()
 
 
+def sleep_and_return(seconds: int):
+    time.sleep(seconds)
+    return seconds
+
+
 class ExecutorTest:
 
     # Executor.shutdown() and context manager usage is tested by
@@ -168,6 +173,22 @@ def test_map_buffersize_when_error(self):
             msg="ints should be either processed, or buffered, or not fetched.",
         )
 
+    def test_map_as_completed(self):
+        ints = [4, 2, 1, 0]
+        self.assertListEqual(
+            list(self.executor.map(sleep_and_return, ints, as_completed=True)),
+            [0, 1, 2, 4],
+            msg="should yield in First Done First Out if `as_completed=True`.",
+        )
+
+    def test_map_buffersize_as_completed(self):
+        ints = [4, 2, 1, 0]
+        self.assertListEqual(
+            list(self.executor.map(sleep_and_return, ints, buffersize=2, as_completed=True)),
+            [2, 1, 0, 4],
+            msg="should yield in First Done First Out within buffer if `buffersize` is set and `as_completed=True`.",
+        )
+
     def test_shutdown_race_issue12456(self):
         # Issue #12456: race condition at shutdown where trying to post a
         # sentinel in the call queue blocks (the queue is full while processes