Path: csiph.com!v102.xanadu-bbs.net!xanadu-bbs.net!feeder.erje.net!eu.feeder.erje.net!newsfeed.xs4all.nl!newsfeed2.news.xs4all.nl!xs4all!newsgate.cistron.nl!newsgate.news.xs4all.nl!post.news.xs4all.nl!not-for-mail
MIME-Version: 1.0
In-Reply-To: <CABTXsnfPZtBy+s7HjoOk2S6QbqHF1ObRNY3fvqMOOixzRVucKA@mail.gmail.com>
References: <CABTXsnfPZtBy+s7HjoOk2S6QbqHF1ObRNY3fvqMOOixzRVucKA@mail.gmail.com>
From: Maxime Steisel <maximesteisel@gmail.com>
Date: Thu, 17 Jul 2014 01:09:02 +0200
Subject: Re: Anything better than asyncio.as_completed() and asyncio.wait() to manage execution of large amount of tasks?
To: Valery Khamenya <khamenya@gmail.com>
Content-Type: text/plain; charset=UTF-8
Cc: python-list@python.org
Precedence: list
Newsgroups: comp.lang.python
Message-ID: <mailman.11904.1405552164.18130.python-list@python.org>
Lines: 57
NNTP-Posting-Host: 2001:888:2000:d::a6
Xref: csiph.com comp.lang.python:74595

2014-07-15 14:20 GMT+02:00 Valery Khamenya <khamenya@gmail.com>:
> Hi,
>
> both asyncio.as_completed() and asyncio.wait() work with lists only. No
> generators are accepted. Are there anything similar to those functions that
> pulls Tasks/Futures/coroutines one-by-one and processes them in a limited
> task pool?


Something like this (adapted from as_completed) should do the work:

import asyncio
from concurrent import futures

def parallelize(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)
    exhausted = False

    @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()) for i in range(max_workers)]

    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()

    timeout_handle.cancel()