How to use asyncio with existing blocking library?

by

There are (sort of) two questions here: first, how to run blocking code asynchronously, and second, how to run async code in parallel (asyncio is single-threaded, so the GIL still applies, so it isn’t truly parallel, but I digress).

Concurrent tasks can be created using asyncio.ensure_future, as documented here.

To run synchronous code, you will need to run the blocking code in an executor. Example:

import concurrent.futures
import asyncio
import time

def blocking(delay):
    time.sleep(delay)
    print('Completed.')

async def non_blocking(loop, executor):
    # Run three of the blocking tasks concurrently. asyncio.wait will
    # automatically wrap these in Tasks. If you want explicit access
    # to the tasks themselves, use asyncio.ensure_future, or add a
    # "done, pending = asyncio.wait..." assignment
    await asyncio.wait(
        fs={
            # Returns after delay=12 seconds
            loop.run_in_executor(executor, blocking, 12),
            
            # Returns after delay=14 seconds
            loop.run_in_executor(executor, blocking, 14),
            
            # Returns after delay=16 seconds
            loop.run_in_executor(executor, blocking, 16)
        },
        return_when=asyncio.ALL_COMPLETED
    )

loop = asyncio.get_event_loop()
executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
loop.run_until_complete(non_blocking(loop, executor))

If you want to schedule these tasks using a for loop (as in your example), you have several different strategies, but the underlying approach is to schedule the tasks using the for loop (or list comprehension, etc), await them with asyncio.wait, and then retrieve the results. Example:

done, pending = await asyncio.wait(
    fs=[loop.run_in_executor(executor, blocking_foo, *args) for args in inps],
    return_when=asyncio.ALL_COMPLETED
)

# Note that any errors raise during the above will be raised here; to
# handle errors you will need to call task.exception() and check if it
# is not None before calling task.result()
results = [task.result() for task in done]

More Related Contents:

  1. How can I periodically execute a function with asyncio?
  2. Simplest async/await example possible in Python
  3. Combine awaitables like Promise.all
  4. Does asyncio supports asynchronous I/O for file operations?
  5. How to call an async function without await?
  6. Using asyncio.Queue for producer-consumer flow
  7. asyncio.ensure_future vs. BaseEventLoop.create_task vs. simple coroutine?
  8. How can I await inside future-like object’s __await__?
  9. How does asyncio actually work?
  10. What are type hints in Python 3.5?
  11. Python 3.7 – asyncio.sleep() and time.sleep()
  12. asyncio.sleep() vs time.sleep()
  13. multiprocessing vs multithreading vs asyncio in Python 3
  14. Asyncio.gather vs asyncio.wait
  15. How can I send an HTTP request from my FastAPI app to another site (API)?
  16. “Asyncio Event Loop is Closed” when getting loop
  17. Self-reference or forward-reference of type annotations in Python [duplicate]
  18. How can I wrap a synchronous function in an async coroutine?
  19. What does “SSLError: [SSL] PEM lib (_ssl.c:2532)” mean using the Python ssl library?
  20. asyncio.sleep() vs time.sleep() in Python
  21. How can I run an async function using the schedule library?
  22. multiprocessing vs multithreading vs asyncio
  23. Why am I getting “invalid syntax” from an f-string? [duplicate]
  24. How to set Python3.5.2 as default Python version on CentOS?
  25. How to use await in a python lambda
  26. How to set class attribute with await in __init__
  27. Is there a way to use asyncio.Queue in multiple threads?
  28. Create generator that yields coroutine results as the coroutines finish
  29. how to add a coroutine to a running asyncio loop?
  30. How to combine Celery with asyncio?

Leave a Comment