Instantly share code, notes, and snippets.

What would you like to do?
Combining Python 3 asyncio coroutines with thread pool and process pool executors
#!/usr/bin/env python3
# Combining coroutines running in an asyncio event loop with
# blocking tasks in thread pool and process pool executors.
# Based on, but this version runs both
# threads and processes at the same time and interleaves them with asyncio coroutines.
# All appears to be working.
import asyncio
import concurrent.futures
import logging
import sys
import time
def block_task(prefix, n):
"""A blocking task to be executed in a thread or process executor"""
log = logging.getLogger(f'{prefix}_blocks({n})')'running')
return f'b{n ** 2}'
async def async_task(prefix, n):
"""A coroutine intended to run in the asyncio event loop to verify that it
works concurrently with the blocking tasks"""
log = logging.getLogger(f'{prefix}_asyncio({n})')
for i in range(5):'running {i}')
await asyncio.sleep(0.5)'done')
return f'a{n ** 2}'
async def run_tasks(prefix, executor):
"""Runs blocking tasks in the executor and spawns off a few coroutines to run
concurrently with the blocking tasks."""
log = logging.getLogger(f'{prefix}_run_blocking_tasks')'starting')'creating executor tasks')
loop = asyncio.get_event_loop()
blocking_tasks = [
loop.run_in_executor(executor, block_task, prefix, i)
for i in range(6)
] + [async_task(prefix, i) for i in range(3)]'waiting for executor tasks')
completed, pending = await asyncio.wait(blocking_tasks)
results = [t.result() for t in completed]'results: {!r}'.format(results))'exiting')
if __name__ == '__main__':
format='PID %(process)5s %(threadName)-25s %(name)-25s: %(message)s',
th_executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
pr_executor = concurrent.futures.ProcessPoolExecutor(max_workers=3)
event_loop = asyncio.get_event_loop()
w = asyncio.wait([run_tasks('th', th_executor),
run_tasks('pr', pr_executor)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment