What kind of problems (if any) would there be combining asyncio with multiprocessing?

by

You should be able to safely combine asyncio and multiprocessing without too much trouble, though you shouldn’t be using multiprocessing directly. The cardinal sin of asyncio (and any other event-loop based asynchronous framework) is blocking the event loop. If you try to use multiprocessing directly, any time you block to wait for a child process, you’re going to block the event loop. Obviously, this is bad.

The simplest way to avoid this is to use BaseEventLoop.run_in_executor to execute a function in a concurrent.futures.ProcessPoolExecutor. ProcessPoolExecutor is a process pool implemented using multiprocessing.Process, but asyncio has built-in support for executing a function in it without blocking the event loop. Here’s a simple example:

import time
import asyncio
from concurrent.futures import ProcessPoolExecutor

def blocking_func(x):
   time.sleep(x) # Pretend this is expensive calculations
   return x * 5

@asyncio.coroutine
def main():
    #pool = multiprocessing.Pool()
    #out = pool.apply(blocking_func, args=(10,)) # This blocks the event loop.
    executor = ProcessPoolExecutor()
    out = yield from loop.run_in_executor(executor, blocking_func, 10)  # This does not
    print(out)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

For the majority of cases, this is function alone is good enough. If you find yourself needing other constructs from multiprocessing, like Queue, Event, Manager, etc., there is a third-party library called aioprocessing (full disclosure: I wrote it), that provides asyncio-compatible versions of all the multiprocessing data structures. Here’s an example demoing that:

import time
import asyncio
import aioprocessing
import multiprocessing

def func(queue, event, lock, items):
    with lock:
        event.set()
        for item in items:
            time.sleep(3)
            queue.put(item+5)
    queue.close()

@asyncio.coroutine
def example(queue, event, lock):
    l = [1,2,3,4,5]
    p = aioprocessing.AioProcess(target=func, args=(queue, event, lock, l)) 
    p.start()
    while True:
        result = yield from queue.coro_get()
        if result is None:
            break
        print("Got result {}".format(result))
    yield from p.coro_join()

@asyncio.coroutine
def example2(queue, event, lock):
    yield from event.coro_wait()
    with (yield from lock):
        yield from queue.coro_put(78)
        yield from queue.coro_put(None) # Shut down the worker

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    queue = aioprocessing.AioQueue()
    lock = aioprocessing.AioLock()
    event = aioprocessing.AioEvent()
    tasks = [ 
        asyncio.async(example(queue, event, lock)),
        asyncio.async(example2(queue, event, lock)),
    ]   
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

More Related Contents:

  1. multiprocessing vs multithreading vs asyncio in Python 3
  2. multiprocessing vs multithreading vs asyncio
  3. How to use asyncio with threading in Python is Method threading.Timer
  4. Multiprocessing vs Threading Python [duplicate]
  5. Can’t pickle when using multiprocessing Pool.map()
  6. How to run functions in parallel?
  7. multiprocessing.Pool: When to use apply, apply_async or map?
  8. What are the differences between the threading and multiprocessing modules?
  9. Asynchronous method call in Python?
  10. How to limit concurrency with Python asyncio?
  11. How to use multiprocessing queue in Python?
  12. FastAPI runs api-calls in serial instead of parallel fashion
  13. Processing single file from multiple processes
  14. How to combine python asyncio with threads?
  15. Asyncio.gather vs asyncio.wait
  16. Sharing python objects across multiple workers
  17. Simplest async/await example possible in Python
  18. How can I wrap a synchronous function in an async coroutine?
  19. Python requests with multithreading
  20. asyncio.sleep() vs time.sleep() in Python
  21. Use asyncio and Tkinter (or another GUI lib) together without freezing the GUI
  22. How to do multiprocessing in FastAPI
  23. Threads vs. Async
  24. Using multiprocessing.Process with a maximum number of simultaneous processes
  25. how to add a coroutine to a running asyncio loop?
  26. How to combine Celery with asyncio?
  27. How to get the return value from a thread in Python?
  28. Updating GUI elements in MultiThreaded PyQT
  29. How does asyncio actually work?
  30. How to close a thread when multithreading? [duplicate]

Leave a Comment