Asyncio coroutine worker pool. No more juggling bounded semaphores and annoying timeouts, and allows you to run through millions of pieces of data efficiently.
Adapted from the awesome worker pool found at https://gist.github.com/thehesiod/7081ab165b9a0d4de2e07d321cc2391d
pip install asyncpool
importasyncpoolimportloggingimportasyncioasyncdefexample_coro(initial_number, result_queue):
print("Processing Value! -> {} * 2 = {}".format(initial_number, initial_number*2))
awaitasyncio.sleep(1)
awaitresult_queue.put(initial_number*2)
asyncdefresult_reader(queue):
whileTrue:
value=awaitqueue.get()
ifvalueisNone:
breakprint("Got value! -> {}".format(value))
asyncdefrun():
result_queue=asyncio.Queue()
reader_future=asyncio.ensure_future(result_reader(result_queue), loop=loop)
# Start a worker pool with 10 coroutines, invokes `example_coro` and waits for it to complete or 5 minutes to pass.asyncwithasyncpool.AsyncPool(loop, num_workers=10, name="ExamplePool",
logger=logging.getLogger("ExamplePool"),
worker_co=example_coro, max_task_time=300,
log_every_n=10) aspool:
foriinrange(50):
awaitpool.push(i, result_queue)
awaitresult_queue.put(None)
awaitreader_futureloop=asyncio.get_event_loop()
loop.run_until_complete(run())