Return an iterator whose values are coroutines. When waiting for the yielded coroutines you'll get the results (or exceptions!) of the original Futures (or coroutines), in the order in which and as soon as they complete. This differs from PEP 3148; the proper way to use this
(fs, *, timeout=None)
| 566 | |
| 567 | # This is *not* a @coroutine! It is just an iterator (yielding Futures). |
| 568 | def as_completed(fs, *, timeout=None): |
| 569 | """Return an iterator whose values are coroutines. |
| 570 | |
| 571 | When waiting for the yielded coroutines you'll get the results (or |
| 572 | exceptions!) of the original Futures (or coroutines), in the order |
| 573 | in which and as soon as they complete. |
| 574 | |
| 575 | This differs from PEP 3148; the proper way to use this is: |
| 576 | |
| 577 | for f in as_completed(fs): |
| 578 | result = await f # The 'await' may raise. |
| 579 | # Use result. |
| 580 | |
| 581 | If a timeout is specified, the 'await' will raise |
| 582 | TimeoutError when the timeout occurs before all Futures are done. |
| 583 | |
| 584 | Note: The futures 'f' are not necessarily members of fs. |
| 585 | """ |
| 586 | if futures.isfuture(fs) or coroutines.iscoroutine(fs): |
| 587 | raise TypeError(f"expect an iterable of futures, not {type(fs).__name__}") |
| 588 | |
| 589 | from .queues import Queue # Import here to avoid circular import problem. |
| 590 | done = Queue() |
| 591 | |
| 592 | loop = events._get_event_loop() |
| 593 | todo = {ensure_future(f, loop=loop) for f in set(fs)} |
| 594 | timeout_handle = None |
| 595 | |
| 596 | def _on_timeout(): |
| 597 | for f in todo: |
| 598 | f.remove_done_callback(_on_completion) |
| 599 | done.put_nowait(None) # Queue a dummy value for _wait_for_one(). |
| 600 | todo.clear() # Can't do todo.remove(f) in the loop. |
| 601 | |
| 602 | def _on_completion(f): |
| 603 | if not todo: |
| 604 | return # _on_timeout() was here first. |
| 605 | todo.remove(f) |
| 606 | done.put_nowait(f) |
| 607 | if not todo and timeout_handle is not None: |
| 608 | timeout_handle.cancel() |
| 609 | |
| 610 | async def _wait_for_one(): |
| 611 | f = await done.get() |
| 612 | if f is None: |
| 613 | # Dummy value from _on_timeout(). |
| 614 | raise exceptions.TimeoutError |
| 615 | return f.result() # May raise f.exception(). |
| 616 | |
| 617 | for f in todo: |
| 618 | f.add_done_callback(_on_completion) |
| 619 | if todo and timeout is not None: |
| 620 | timeout_handle = loop.call_later(timeout, _on_timeout) |
| 621 | for _ in range(len(todo)): |
| 622 | yield _wait_for_one() |
| 623 | |
| 624 | |
| 625 | @types.coroutine |
nothing calls this directly
no test coverage detected