Provide an async iterator version of as_completed

@hniksic

BPO	33533
Nosy	@hniksic, @asvetlov, @Fak3, @1st1, @mivade, @JustinTArthur, @carlodri
PRs	gh-77714: Implement as_completed as an asynchronous generator #10251 gh-77714: Provide an async iterator version of as_completed #22491

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = None
closed_at = None
created_at = <Date 2018-05-16.06:48:52.976>
labels = ['type-feature', '3.10', 'expert-asyncio']
title = 'Provide an async iterator version of as_completed'
updated_at = <Date 2020-10-16.06:44:21.716>
user = 'https://github.com/hniksic'

bugs.python.org fields:

activity = <Date 2020-10-16.06:44:21.716>
actor = 'carlodri'
assignee = 'none'
closed = False
closed_date = None
closer = None
components = ['asyncio']
creation = <Date 2018-05-16.06:48:52.976>
creator = 'hniksic'
dependencies = []
files = []
hgrepos = []
issue_num = 33533
keywords = ['patch']
message_count = 11.0
messages = ['316773', '316774', '316963', '317247', '326748', '328454', '328527', '377797', '377803', '377840', '377927']
nosy_count = 7.0
nosy_names = ['hniksic', 'asvetlov', 'Roman.Evstifeev', 'yselivanov', 'mivade', 'JustinTArthur', 'carlodri']
pr_nums = ['10251', '22491']
priority = 'normal'
resolution = None
stage = 'patch review'
status = 'open'
superseder = None
type = 'enhancement'
url = 'https://bugs.python.org/issue33533'
versions = ['Python 3.10']

Judging by questions on the StackOverflow python-asyncio tag[1][2], it seems that users find it hard to understand how to use as_completed correctly. I have identified three issues:

It's somewhat sparingly documented.

A StackOverflow user ([2]) didn't find it obvious that it runs the futures in parallel. Unless one is already aware of the meaning, the term "as completed" could suggest that they are executed and completed sequentially.

Unlike its concurrent.futures counter-part, it's non-blocking.

This sounds like a good idea because it's usable from synchronous code, but it means that the futures it yields aren't completed, you have to await them first. This is confusing for a function with "completed" in the name, and is not the case with concurrent.futures.as_completed, nor with other waiting functions in asyncio (gather, wait, wait_for).

It yields futures other than those that were passed in.

This prevents some usual patterns from working, e.g. associating the results with context data, such as Python docs itself uses for concurrent.futures.as_completed in https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor-example . See SO question [1] for a similar request in asyncio.

Here is my proposal to address the issues.

I believe the usage problems stem from as_completed predating the concept of async iterators. If we had async iterators back then, as_completed would have been an obvious candidate to be one. In that case it could be both "blocking" (but not for the event loop) and return the original futures. For example:

async def as_completed2(fs):
    pending = set(map(asyncio.ensure_future(fs)))
    while pending:
        done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
        yield from done

(It is straightforward to add support for a timeout argument.)

I propose to deprecate asyncio.as_completed and advertise the async-iterator version like the one presented here - under a nicer name, such as as_done(), or as_completed_async().

[1] https://stackoverflow.com/questions/50028465/python-get-reference-to-original-task-after-ordering-tasks-by-completion
[2] https://stackoverflow.com/questions/50355944/yield-from-async-generator-in-python-asyncio

Of course, yield from done would actually have to be for future in done: yield future, since async generators don't support yield from.

I like the idea. Let's revisit it after Python 3.7 is released.

Another option occurred to me: as_completed could return an object that implements both synchronous and asynchronous iteration protocol:

class as_completed:
    def __init__(fs, *, loop=None, timeout=None):
        self.__fs = fs
        self.__loop = loop
        self.__timeout = timeout

    def __iter__(self):
        # current implementation here
        ...

    async def __aiter__(self):
        # new async implementation here
        ...

    def __next__(self):
        # defined for backward compatibility with code that expects
        # as_completed() to return an iterator rather than an iterable
        if self._iter is None:
            self._iter = iter(self)
        return next(self._iter)

With that design there wouldn't need to be a new function under a different name; instead, as_completed could just be documented as an asynchronous iterable, with the old synchronous iteration supported for backward compatibility.

If there is interest in this, I'd like to attempt a PR for a sync/async variant of as_completed.

Note that the new docs are *much* clearer, so the first (documentation) problem from the description is now fixed. Although the documentation is still brief, it now contains the key pieces of information: 1) that the futures are actually run in parallel, and 2) that each yielded future produces the next result that becomes available. Neither was actually stated in the old docs (!), so this is a marked improvement.

Is there any progress on this? I was thinking the exact same thing regarding the backwards-compatible approach and would like to work on it if no one else is.

I didn't start working on the PR, so please go ahead if you're interested.

One small suggestion: If you're implementing this, please note that the proof-of-concept implementation shown in the description is not very efficient because each call to wait has to iterate over all the futures (which can be potentially large in number) to set up and tear down the done callbacks on each one. A more efficient implementation would set up the callbacks only once - see https://stackoverflow.com/a/51403277/1600898 for an example.

I've added a new PR, PR 22491. This one has as_completed returning an iterator and includes tests for both the old and new style.

I see a trivial amount of latency added in extra call stack over Guido's original implementation. Should we decide to deprecate the plain-iterator-of-awaitables form, we can possibly regain that latency in refactoring this as an asynchronous generator.

Hi, thanks for providing a PR. One thing I noticed is that the implementation in the PR yields results of the futures from the generator. This issue proposes a generator that instead yields the futures passed to as_completed. This is needed not just for consistency with concurrent.futures.as_completed, but also to allow associating results with the requests that produced them, which is an important use case for as_completed.

An example of how this is useful is the snippet at https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor-example which you cannot implement with an as_completed that doesn't yield the original futures. The StackOverflow question at [1] inquires about that issue, and would be easily resolved by yielding the futures as concurrent.futures.as_completed does.

As far as I can tell, the needed change is simple: just yield f instead of f.result() from _wait_for_one when invoked from anext. (It must still return f.result() when called from next for backward compatibility.)

[1] https://stackoverflow.com/questions/50028465/python-get-reference-to-original-task-after-ordering-tasks-by-completion

Thanks, Hrvoje. I've updated the patch to match this bug's suggested format and have updated the documentation and What's New.

The one quirk that comes with the benefit of getting the same futures back is that we still allow both coroutines and futures to be passed in. Coroutines will not be yielded back in their original form, but instead a new task.

Justin, thanks for updating the PR! I'll take another look at the code.

@serhiy-storchaka

@serhiy-storchaka Could you review the PRs from this issue? There seem two PRs by different authors.

While it is documented that as_completed() returns an iterator, actually it returns a generator, which has a close() method, so it can be explicitly closed or used with the closing() context manager. For asynchronous generators deterministic closing is even more important, because they always create reference loops. So I think that it is important to implement both close() and aclose() methods in new hybrid iterator. It is less likely that methods send() and throw() or attributes like gi_running are used in real code, so I think we can omit them.

There seem to be two PRs for this one issue. #10251 and #22491. Which one should we prefer?

The latter is more compatible (it returns an iterator, not simply an iterable) and contains many tests (duplicate tests for synchronous iterator and add new tests for details which we want to preserve).

I am planning to have a look at the PR. As always, with this type of project, I am (perhaps unnecessarily) worried that this improvement, however much of an improvement it is for new code (which can rely on the Futures' identities, for example), it may break corner cases where users have production systems that somehow, rarely, depends on some clever abuse of a detail of the old version (maybe someone assumes that they never get the same Future back, and they store input and output Futures in a dict expecting to have both available).

I have some other catching up to do before I get to this, but it is in my queue.

They only get the same Future back when they use an asynchronous iterator. I.e. if they change their code.

gh-77714: Provide an async iterator version of as_completed (GH-22491)

Squashed commit of the following:

pythongh-77714: Provide an async iterator version of as_completed (py…

Great to see this to land in asyncio!
Here's my version just FYI: https://github.com/achimnol/aiotools/blob/6d8a93ed0418d61e7c40c7f5c06b061af4581c06/src/aiotools/utils.py#L33
Here is a race() implementation on top of it: https://github.com/achimnol/aiotools/blob/6d8a93ed0418d61e7c40c7f5c06b061af4581c06/src/aiotools/utils.py#L108

Navigation Menu

Repository navigation

Provide an async iterator version of as_completed #77714

6 remaining items

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Provide an async iterator version of as_completed #77714

Description

Activity

hniksic commented on May 16, 2018

hniksic commented on May 16, 2018

1st1 commented on May 17, 2018

hniksic commented on May 21, 2018

hniksic commented on Sep 30, 2018

mivade commented on Oct 25, 2018

hniksic commented on Oct 26, 2018

JustinTArthur commented on Oct 2, 2020

hniksic commented on Oct 2, 2020

JustinTArthur commented on Oct 2, 2020

hniksic commented on Oct 4, 2020

6 remaining items

gvanrossum commented on Dec 16, 2023

serhiy-storchaka commented on Dec 18, 2023

gvanrossum commented on Dec 24, 2023

serhiy-storchaka commented on Dec 24, 2023

gvanrossum commented on Jan 10, 2024

serhiy-storchaka commented on Jan 10, 2024

achimnol commented on May 16, 2024

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions