FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

gh-131466: `concurrent.futures.Executor.map`: avoid temporarily exceeding `buffersize` while collecting the next result by ebonnal · Pull Request #131467 · python/cpython · GitHub

Repository navigation

Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (2) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
36 changes: 18 additions & 18 deletions Lib/concurrent/futures/_base.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -304,18 +304,6 @@ def wait(fs, timeout=None, return_when=ALL_COMPLETED):
done.update(waiter.finished_futures)
return DoneAndNotDoneFutures(done, fs - done)


def _result_or_cancel(fut, timeout=None):
try:
try:
return fut.result(timeout)
finally:
fut.cancel()
finally:
# Break a reference cycle with the exception in self._exception
del fut
Comment thread
ebonnal marked this conversation as resolved.


class Future(object):
"""Represents the result of an asynchronous computation."""

Expand Down Expand Up @@ -625,23 +613,35 @@ def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):
# before the first iterator value is required.
def result_iterator():
try:
# reverse to keep finishing order
# reverse so that the next (FIFO) future is on the right
fs.reverse()
# careful not to keep references to futures or results
while fs:
# wait for the next result
if timeout is None:
fs[-1].result()
else:
fs[-1].result(end_time - time.monotonic())

# buffer next task
if (
buffersize
and (executor := executor_weakref())
and (args := next(zipped_iterables, None))
):
fs.appendleft(executor.submit(fn, *args))
# Careful not to keep a reference to the popped future
if timeout is None:
yield _result_or_cancel(fs.pop())
else:
yield _result_or_cancel(fs.pop(), end_time - time.monotonic())

# yield the awaited result
yield fs.pop()._result

finally:
# break the reference cycle with fs[-1]._exception's traceback
if fs:
fs.pop().cancel()

for future in fs:
future.cancel()

return result_iterator()

def shutdown(self, wait=True, *, cancel_futures=False):
Expand Down
27 changes: 26 additions & 1 deletion Lib/test/test_concurrent_futures/executor.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@
import time
import weakref
from concurrent import futures
from operator import add
from contextlib import suppress
from functools import partial
from operator import add, truediv
from test import support
from test.support import Py_GIL_DISABLED

Expand Down Expand Up @@ -143,6 +145,29 @@ def test_map_buffersize_when_buffer_is_full(self):
msg="should have fetched only `buffersize` elements from `ints`.",
)

def test_map_buffersize_when_error(self):
ints = [1, 2, 3, 0, 4, 5, 6]
index_of_zero = ints.index(0)
ints_iter = iter(ints)
buffersize = 2
reciprocal = partial(truediv, 1)
results = []
with suppress(ZeroDivisionError):
for result in self.executor.map(
reciprocal, ints_iter, buffersize=buffersize
):
results.append(result)
self.assertEqual(
len(results),
index_of_zero,
msg="should have mapped until reaching the zero.",
)
self.assertEqual(
len(results) + buffersize + len(list(ints_iter)),
len(ints),
msg="ints should be either processed, or buffered, or not fetched.",
)

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
Expand Down

Back | FazBrowse Home | New Git URL