Post from September 22, 2022, technically supplemented for the redesign. The original code examples are historical fragments, some deliberately broken or incomplete; don't adopt them without review. This documents an investigation, not a production-ready queue.
This post starts with asyncio patterns and problems in asynchronous programs. Then it turns to multiprocessing, multiple event loops and communication between processes and loops. It ends with a sketch and unresolved problems, not a finished solution.
My post on horizontal and vertical parallelism is another starting point. I assume you know the basics of coroutines, async and await. I learned from the Python docs and later used this book.
The original post credited a title image by r2hox / Wikimedia Commons, CC BY-SA 2.0; it is not embedded here.
Part I: asyncio
Coroutines are lighter than dedicated threads or processes. That's a reason to start several I/O tasks concurrently. They aren't free, however: tasks use memory and can overload a server. Bounded concurrency and backpressure are often useful.
The producer starts tasks and keeps references; we then wait for completed work. The original example is missing import random.
import asyncio
async def aio_producer(pending: set):
loop = asyncio.get_event_loop()
for _ in range(100):
work = random.randint(0,100) / 100
task = loop.create_task(consumer_coro(work))
pending.add(task)
# Wait until consumer has eaten all work
print("Producer done")
# this is a simple consumer coroutine now.
async def consumer_coro(work):
# simulates some i/o bound task
await asyncio.sleep(work)
return f"done {work}"
async def main():
pending = set() # we keep reference to the pending task. To wait for them and get the result
await aio_producer(pending)
while pending:
done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
for task in done:
print(task.result())
print(f"{len(done)=}, {len(pending)=}")
print("Bye")
asyncio.run(main(), debug=True)
As a counterexample I wanted to show that immediately awaiting each piece of work can serialize processing. Original mistake: the second block below is accidentally a copy of the first and does not demonstrate that counterexample. A queue is not inherently an anti-pattern either: multiple consumers and a bounded queue can be very useful.
async def aio_producer(pending: set):
loop = asyncio.get_event_loop()
for _ in range(100):
work = random.randint(0,100) / 100
task = loop.create_task(consumer_coro(work))
pending.add(task)
# Wait until consumer has eaten all work
print("Producer done")
# this is a simple consumer coroutine now.
async def consumer_coro(work):
# simultates some i/o bound task
await asyncio.sleep(work)
return f"done {work}"
async def main():
pending = set() # we keep reference to the pending task. To wait for them and get the result
await aio_producer(pending)
while pending:
done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
for task in done:
print(task.result())
print(f"{len(done)=}, {len(pending)=}")
print("Bye")
asyncio.run(main(), debug=True)Blocking the event loop
CPU-heavy code or a blocking function delays other tasks on the same event loop. To try it, replace asyncio.sleep with time.sleep.
Blocking I/O can be moved to a thread with loop.run_in_executor() or asyncio.to_thread(). Correction: the original loop.to_task does not exist; asyncio.to_thread is the relevant helper.
import asyncio
import time
def blocking_api_function():
print("working heavily")
time.sleep(1)
print("done")
return "42"
async def block_main():
# this will block and raise at least a warning. So it blocks the event loop for a second.
res = blocking_api_function()
print(res)
asyncio.run(block_main(), debug=True)
async def main():
# now we want to unblock the blocking api call
loop = asyncio.get_event_loop()
# we wrap that function into a future
fut = loop.run_in_executor(None, blocking_api_function)
# same as loop.to_task(...)
# we than can await the future if we need to
res = await fut # event loop could do other things now.
print(res)
asyncio.run(main(), debug=True)Many small computations can also add up to event-loop delays. My former blanket 30% CPU threshold wasn't a sound rule; latency, workload and measurement matter. Threads are mainly useful for blocking I/O; a process pool can be more appropriate for CPU-bound Python work. See the asyncio documentation.
Part II: multiprocessing
A ProcessPoolExecutor can handle larger CPU jobs. Multiple processes add startup, communication and serialization overhead. Not every object is easy to pass. The smaller the jobs, the more likely that overhead eats the benefit – measure rather than guess.
from concurrent.futures import ProcessPoolExecutor
def cpu_bound_func(work):
result = 0
for _ in range(work * 1000):
result = result + 1
return result
if __name__ == "__main__":
executor = ProcessPoolExecutor()
work = [1, 2, 3, 4]
results = executor.map(cpu_bound_func, work)
print(list(results))The event loop can also work with a process pool. This historical fragment requires imports and the function from the previous example; a real pool needs to be shut down properly.
async def main():
executor = ProcessPoolExecutor()
loop = asyncio.get_event_loop()
works = [1, 2, 3, 4]
pending = set()
for work in works:
pending.add(
loop.run_in_executor(executor, cpu_bound_func, work)
)
done, pending = await asyncio.wait(pending)
print([t.result() for t in done])
print("bye")
if __name__ == '__main__':
asyncio.run(main(), debug=True)Another idea is to run multiple processes, each with its own event loop. That doesn't distribute work automatically: responsibilities, communication and shutdown need explicit design.
Warning: The historical example below contains a faulty wait loop and no proper join(). It can run indefinitely and is not a template to execute.
async def main():
for _ in range(10):
cpu_bound_func(1) # should not take longer than 100 ms here
await asyncio.sleep(0)
print("async main gone at", multiprocessing.current_process())
def async_worker():
print("starting", multiprocessing.current_process())
asyncio.run(main(), debug=True)
print("bye from", multiprocessing.current_process())
def do_main():
processes = [multiprocessing.Process(target=async_worker, name=f"P<{count}>")
for count in range(multiprocessing.cpu_count())]
for p in processes:
p.start()
while not all(p.is_alive() for p in processes):
print(".", end="")
if __name__ == '__main__':
do_main()Part III: The missing link
Python offers pipes and multiprocessing.Queue, among other options, for communication between processes. Queues synchronize their operations and serialize objects. But a blocking get() called directly on an event loop blocks that loop. Polling may waste CPU time too.
import asyncio
import multiprocessing as mp
from queue import Empty
async def main():
q = mp.Queue()
q.put_nowait("stuff")
print(q.get()) # note that this would block if there is now stuff on the queue
try:
q.get(timeout=.31) # and now it will block for at least .31 seconds
except Empty:
print("queue is finally empty")
if __name__ == '__main__':
asyncio.run(main(), debug=True)
# prints:
# stuff
# queue is finally empty
# Executing <Task finished name='Task-1' coro=<main() done, defined at C:\ ... tasks.py:634> took 0.329 seconds
One can run get() in a thread, but many waiting calls can occupy the thread pool. The next example demonstrates a shutdown problem and should not be executed uncontrolled:
async def worker(q: mp.Queue, uid: str):
# this worker will not block the event loop.
# but it will consume one thread until it is unblocked
loop = asyncio.get_event_loop()
while True:
try:
item = await loop.run_in_executor(None, q.get)
except asyncio.CancelledError:
return
# .... do some useful stuff
print(f"{uid} did something with {item}")
async def main():
loop = asyncio.get_event_loop()
q = mp.Queue()
# create then workers waiting for work
# means consuming 10 threads
workers = [loop.create_task(worker(q, f"worker-{uid}")) for uid in range(10)]
await asyncio.sleep(0) # yield to the event loop so it can schedule above tasks
# feed them a bit
for key in range(5):
q.put_nowait(f"food-{key}")
await asyncio.sleep(0.5) # sleep here to let the worker consume that food
# enough we want to stop:
[worker.cancel() for worker in workers]
await asyncio.wait(workers)
if __name__ == '__main__':
asyncio.run(main(), debug=True)
# prints:
# worker-0 did something with food-0
# worker-1 did something with food-1
# worker-2 did something with food-2
# worker-3 did something with food-3
# worker-4 did something with food-4
#
# and then hangs forever or throws exception and complains about broken pipeCorrection of the original explanation: cancelling the asyncio task does not reliably stop a thread already blocked in the call. The waiting thread can prevent executor shutdown; other failures depend on process and queue handling. A thread automatically cancelled while holding a queue lock is not a generally valid explanation.
Next I wrapped the queue in an adapter. This makes the interface convenient but does not solve the shutdown and thread-pool problem. The fragment needs additional imports and helper functions and is not production-ready:
class Queue:
"""
An adapter to make multiprocessing.Queue available for async
"""
def __init__(self, maxsize: int = 0, *, ctx=None, queue: mp.Queue = None):
self.maxsize = maxsize
if not queue:
if not ctx:
self._mp_queue = mp.JoinableQueue(maxsize)
else:
self._mp_queue = mp.JoinableQueue(maxsize, ctx=ctx)
else:
self._mp_queue = queue
self._t_executor: None | concurrent.futures.ThreadPoolExecutor = None
self._loop = None
@property
def loop(self):
if not self._loop:
self._loop = asyncio.get_event_loop()
return self._loop
@property
def _executor(self):
if not self._t_executor:
self._t_executor = concurrent.futures.ThreadPoolExecutor()
return self._t_executor
async def get_async(self):
item = await self.loop.run_in_executor(self._executor, self._mp_queue.get)
return item
def get_sync(self):
return self._mp_queue.get()
def get_nowait(self):
return self._mp_queue.get_nowait()
def get(self):
if from_coroutine():
return self.get_async()
else:
return self.get_sync()
def put_sync(self, item):
self._mp_queue.put(item)
async def put_async(self, item):
item = await self.loop.run_in_executor(self._executor, self._mp_queue.put, item)
return item
def put_nowait(self, item):
return self._mp_queue.put_nowait(item)
def put(self, item):
if from_coroutine():
return self.put_async(item)
else:
return self.put_sync(item)
def task_done(self):
return self._mp_queue.task_done()
async def join(self):
return await asyncio.to_thread(self._mp_queue.join)
def sync_join(self):
return self._mp_queue.join()
def flush(self):
self._mp_queue.put_nowait(None)
def empty(self):
return self._mp_queue.empty()
def qsize(self):
return self._mp_queue.qsize()
def full(self):
return self._mp_queue.full()
def close(self):
return self._mp_queue.close()
def __str__(self):
return self._mp_queue.__str__()
def __repr__(self):
return f"<Queue at {id(self)} maxsize={self.maxsize}>"
Then I experimented with events and termination signals to coordinate startup and shutdown. A bounded thread pool can still stall when many calls depend on each other. Worker count and lifetime, and an explicit shutdown strategy, matter.
One inspiration was a talk by David Beazley about asyncio and threads. With processes, objects cannot simply be treated as shared Python references. Wrapping a Future with asyncio.wrap_future() doesn't make it complete: something still has to fulfil or cancel it safely and arrange inter-process communication.
For experiments I started a shared-memory queue. At the time of the original article it still lacked the required synchronization.
The following Future/deque sketch is also incomplete: the waiter lists are not connected between processes, and it has other bugs. Do not use it as a working queue.
class Queue3:
"""
An adapter to make multiprocessing.Queue available for async
using Futures instead of threads.
However, the communication of the putters and getters will not work,
as they are not connected. The use of deque will only work at that worker but fails
to communicate to other workers
"""
def __init__(self, maxsize: int = 0, *, ctx=None, queue: "mp.Queue[T]" = None):
self.maxsize = maxsize
if not queue:
if not ctx:
self._mp_queue: "mp.JoinableQueue[T]" = mp.JoinableQueue(maxsize)
else:
self._mp_queue: "mp.JoinableQueue[T]" = mp.JoinableQueue(maxsize, ctx=ctx)
else:
self._mp_queue = queue
self.putters: deque[Future] = deque()
self.getters: deque[Future] = deque()
def get_noblock(self) -> tuple[T | None, Future | None]:
item = None
fut = None
print("get_noblock", mp.current_process().name)
try:
item = self._mp_queue.get(block=False)
if self.putters:
self.putters.popleft().set_result(True)
except Empty:
print("empty", mp.current_process().name)
fut = Future()
self.getters.append(fut)
except Exception as e:
breakpoint()
finally:
return item, fut
async def get_async(self, timeout=None):
item, fut = self.get_noblock()
if fut:
print("wait for future at", mp.current_process())
item = await asyncio.wait_for(asyncio.wrap_future(fut), timeout)
return item
def get_sync(self, timeout=None):
item, fut = self.get_noblock()
if fut:
print("wait sync for future", mp.current_process().name)
item = fut.result(timeout) # block until
return item
def get_nowait(self):
return self._mp_queue.get_nowait()
def get(self, timeout=None):
if from_coroutine():
return self.get_async(timeout)
else:
return self.get_sync(timeout)
def put_noblock(self, item: T) -> Future | None:
fut = None
try:
if self.getters:
self.getters.pop().set_result(item)
else:
self._mp_queue.put(item, block=False)
except Full:
fut = Future()
self.putters.append(fut)
finally:
return fut
def put_sync(self, item:T, timeout=None):
while True:
fut = self.put_noblock(item)
if fut is None:
return
fut.result(timeout) # block until timeout
async def put_async(self, item:T, timeout=None):
while True:
fut = self.put_noblock(item)
if fut is None:
return
await asyncio.wait_for(asyncio.wrap_future(fut), timeout)
def put(self, item:T, timeout=None):
if from_coroutine():
return self.put_async(item, timeout)
else:
return self.put_sync(item, timeout)
def put_nowait(self, item):
return self._mp_queue.put_nowait(item)
def task_done(self):
return self._mp_queue.task_done()
async def join(self):
return await asyncio.to_thread(self._mp_queue.join)
def sync_join(self):
return self._mp_queue.join()
def flush(self):
self._mp_queue.put_nowait(None)
def empty(self):
return self._mp_queue.empty()
def qsize(self):
return self._mp_queue.qsize()
def full(self):
return self._mp_queue.full()
def close(self):
return self._mp_queue.close()
def __str__(self):
return self._mp_queue.__str__()
def __repr__(self):
return f"<Queue at {id(self)} maxsize={self.maxsize}>"Another idea would be separate communication channels for waiting getters and putters. Whether this works, and how to handle races, cancellation and shutdown, still needs implementation and testing. That's where my investigation stood.