Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]
Groups > comp.lang.python > #74879
| References | <CABTXsnfPZtBy+s7HjoOk2S6QbqHF1ObRNY3fvqMOOixzRVucKA@mail.gmail.com> <CAGqiJR_5RzzoD2zNVd3HjmQpncY6ed5VCRKXn2NrBLKsMg-rbw@mail.gmail.com> |
|---|---|
| From | Valery Khamenya <khamenya@gmail.com> |
| Date | 2014-07-20 14:16 +0200 |
| Subject | Re: Anything better than asyncio.as_completed() and asyncio.wait() to manage execution of large amount of tasks? |
| Newsgroups | comp.lang.python |
| Message-ID | <mailman.12096.1405858609.18130.python-list@python.org> (permalink) |
[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
Back to comp.lang.python | Previous | Next | Find similar | Unroll thread
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
csiph-web