Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]

Groups > comp.lang.python > #74879 > unrolled thread

Re: Anything better than asyncio.as_completed() and asyncio.wait() to manage execution of large amount of tasks?

Back to article view | Back to comp.lang.python

This discussion starts older than the indexed window; earlier articles aren't shown. The article labeled Started by below is the oldest one visible, not the original post.

Contents

  Re: Anything better than asyncio.as_completed() and asyncio.wait() to manage execution of large amount of tasks? Valery Khamenya <khamenya@gmail.com> - 2014-07-20 14:16 +0200

#74879 — Re: Anything better than asyncio.as_completed() and asyncio.wait() to manage execution of large amount of tasks?

[Multipart message — attachments visible in raw view] — view raw

Hi Maxime,

many thanks for your great solution. It would be so great to have it in
stock asyncio and use it out-of-the-box...
I've made 4 fixes to it that are rather of "cosmetic" nature. Here is the
final version:

import asyncio
from concurrent import futures


def as_completed_with_max_workers(tasks, *, loop=None, max_workers=5,
timeout=None):
    loop = loop if loop is not None else asyncio.get_event_loop()
    workers = []
    pending = set()
    done = asyncio.Queue(maxsize=max_workers, loop=loop) # Valery: respect
the "loop" parameter
    exhausted = False
    timeout_handle = None # Valery: added to see, if we indeed have to call
timeout_handle.cancel()

    @asyncio.coroutine
    def _worker():
        nonlocal exhausted
        while not exhausted:
            try:
                t = next(tasks)
                pending.add(t)
                yield from t
                yield from done.put(t)
                pending.remove(t)
            except StopIteration:
                exhausted = True

    def _on_timeout():
        for f in workers:
            f.cancel()
        workers.clear()
        # Wake up _wait_for_one()
        done.put_nowait(None)

    @asyncio.coroutine
    def _wait_for_one():
        f = yield from done.get()
        if f is None:
            raise futures.TimeoutError()
        return f.result()

    workers = [asyncio.async(_worker(), loop=loop) for _ in
range(max_workers)] # Valery: respect the "loop" parameter

    if workers and timeout is not None:
        timeout_handle = loop.call_later(timeout, _on_timeout)

    while not exhausted or pending or not done.empty():
        yield _wait_for_one()

    if timeout_handle: # Valery: call timeout_handle.cancel() only if it is
needed
        timeout_handle.cancel()


best regards
--
Valery A.Khamenya

[toc] | [standalone]

Back to top | Article view | comp.lang.python

csiph-web