How to combine python asyncio with threads?

It's pretty simple to delegate a method to a thread or sub-process using BaseEventLoop.run_in_executor:

import asyncioimport timefrom concurrent.futures import ProcessPoolExecutordef cpu_bound_operation(x):    time.sleep(x) # This is some operation that is CPU-bound@asyncio.coroutinedef main():    # Run cpu_bound_operation in the ProcessPoolExecutor    # This will make your coroutine block, but won't block    # the event loop; other coroutines can run in meantime.    yield from loop.run_in_executor(p, cpu_bound_operation, 5)loop = asyncio.get_event_loop()p = ProcessPoolExecutor(2) # Create a ProcessPool with 2 processesloop.run_until_complete(main())

As for whether to use a ProcessPoolExecutor or ThreadPoolExecutor, that's kind of hard to say; pickling a large object will definitely eat some CPU cycles, which initially would you make think ProcessPoolExecutor is the way to go. However, passing your 100MB object to a Process in the pool would require pickling the instance in your main process, sending the bytes to the child process via IPC, unpickling it in the child, and then pickling it again so you can write it to disk. Given that, my guess is the pickling/unpickling overhead will be large enough that you're better off using a ThreadPoolExecutor, even though you're going to take a performance hit because of the GIL.

That said, it's very simple to test both ways and find out for sure, so you might as well do that.

I also used run_in_executor, but I found this function kinda gross under most circumstances, since it requires partial() for keyword args and I'm never calling it with anything other than a single executor and the default event loop. So I made a convenience wrapper around it with sensible defaults and automatic keyword argument handling.

from time import sleepimport asyncio as aioloop = aio.get_event_loop()class Executor:    """In most cases, you can just use the 'execute' instance as a    function, i.e. y = await execute(f, a, b, k=c) => run f(a, b, k=c) in    the executor, assign result to y. The defaults can be changed, though,    with your own instantiation of Executor, i.e. execute =    Executor(nthreads=4)"""    def __init__(self, loop=loop, nthreads=1):        from concurrent.futures import ThreadPoolExecutor        self._ex = ThreadPoolExecutor(nthreads)        self._loop = loop    def __call__(self, f, *args, **kw):        from functools import partial        return self._loop.run_in_executor(self._ex, partial(f, *args, **kw))execute = Executor()...def cpu_bound_operation(t, alpha=30):    sleep(t)    return 20*alphaasync def main():    y = await execute(cpu_bound_operation, 5, alpha=-2)loop.run_until_complete(main())

Another alternative is to use loop.call_soon_threadsafe along with an asyncio.Queue as the intermediate channel of communication.

The current documentation for Python 3 also has a section on Developing with asyncio - Concurrency and Multithreading:

import asyncio# This method represents your blocking codedef blocking(loop, queue):    import time    while True:        loop.call_soon_threadsafe(queue.put_nowait, 'Blocking A')        time.sleep(2)        loop.call_soon_threadsafe(queue.put_nowait, 'Blocking B')        time.sleep(2)# This method represents your async codeasync def nonblocking(queue):    await asyncio.sleep(1)    while True:        queue.put_nowait('Non-blocking A')        await asyncio.sleep(2)        queue.put_nowait('Non-blocking B')        await asyncio.sleep(2)# The main sets up the queue as the communication channel and synchronizes themasync def main():    queue = asyncio.Queue()    loop = asyncio.get_running_loop()    blocking_fut = loop.run_in_executor(None, blocking, loop, queue)    nonblocking_task = loop.create_task(nonblocking(queue))    running = True  # use whatever exit condition    while running:        # Get messages from both blocking and non-blocking in parallel        message = await queue.get()        # You could send any messages, and do anything you want with them        print(message)asyncio.run(main())

How to send asyncio tasks to loop running in other thread may also help you.