Beitrag vom 22. September 2022, für den Neubau übersetzt und fachlich ergänzt. Die Original-Codebeispiele sind historische Fragmente, teils absichtlich fehlerhaft oder unvollständig; nicht ungeprüft übernehmen. Der Beitrag dokumentiert eine Untersuchung, keine fertige produktionsreife Queue.
Dieser Beitrag beginnt mit ein paar asyncio-Mustern und Problemen asynchroner Programme. Danach geht es um multiprocessing, mehrere Event Loops und die Frage, wie Prozesse und Event Loops miteinander sprechen können. Am Ende steht eine Skizze mit offenen Problemen, keine fertige Lösung.
Als Einstieg passt auch mein Beitrag über horizontale und vertikale Parallelität. Grundkenntnisse zu Coroutinen, async und await setze ich voraus. Ich habe mit der Python-Dokumentation angefangen und später dieses Buch verwendet.
Der ursprüngliche Beitrag verwies auf ein Titelbild von r2hox / Wikimedia Commons, CC BY-SA 2.0; es wird hier nicht eingebunden.
Teil I: asyncio
Coroutinen sind leichter als eigene Threads oder Prozesse. Das ist ein Grund, mehrere I/O-Aufgaben nebenläufig zu starten. Ganz kostenlos sind sie aber nicht: Auch Tasks brauchen Speicher und können einen Server überlasten. Begrenzte Nebenläufigkeit und Backpressure sind oft sinnvoll.
Der Producer startet Tasks und hält Referenzen; anschließend warten wir auf abgeschlossene Aufgaben. Im Original fehlt 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)
Als Gegenbeispiel wollte ich zeigen, dass ein unmittelbares await auf jede einzelne Arbeit die Verarbeitung serialisieren kann. Fehler im Original: Der folgende zweite Block ist versehentlich eine Kopie des ersten und demonstriert dieses Gegenbeispiel nicht. Eine Queue ist übrigens nicht grundsätzlich ein Anti-Pattern: Sie kann mit mehreren Consumern und begrenzter Größe sehr nützlich sein.
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)Den Event Loop blockieren
CPU-lastiger Code oder eine blockierende Funktion verzögert alle anderen Tasks auf demselben Event Loop. Zum Ausprobieren kann man asyncio.sleep durch time.sleep ersetzen.
Blockierende I/O-Aufrufe lassen sich beispielsweise über loop.run_in_executor() oder asyncio.to_thread() in einen Thread auslagern. Korrektur: Das im Original genannte loop.to_task gibt es nicht; asyncio.to_thread ist die passende Hilfsfunktion.
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)Auch viele kleine Berechnungen können in Summe den Event Loop verzögern. Die frühere pauschale 30-%-CPU-Grenze war keine belastbare Regel; entscheidend sind Latenz, Arbeitslast und Messung. Thread-Auslagerung hilft vor allem bei blockierender I/O; für CPU-lastigen Python-Code kann ein Prozesspool geeigneter sein. Die asyncio-Dokumentation beschreibt diese Möglichkeiten.
Teil II: multiprocessing
Für größere CPU-Aufgaben kann man einen ProcessPoolExecutor nutzen. Mehrere Prozesse bringen Aufwand für Start, Kommunikation und Serialisierung mit. Nicht jedes Objekt lässt sich einfach übergeben. Je kleiner die Arbeitspakete, desto eher kann der Overhead die Vorteile auffressen – also messen statt raten.
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))Auch der Event Loop kann mit einem Prozesspool arbeiten. Das folgende Originalfragment setzt Imports und die Funktion aus dem vorherigen Beispiel voraus; ein realer Pool muss sauber beendet werden.
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)Eine andere Idee ist, mehrere Prozesse mit jeweils eigenem Event Loop laufen zu lassen. Das verteilt Arbeit nicht automatisch: Zuständigkeit, Kommunikation und Beenden müssen explizit geregelt werden.
Achtung: Das folgende historische Beispiel enthält eine fehlerhafte Warteschleife und kein korrektes join(). Es kann dauerhaft laufen und ist keine Vorlage zum Ausführen.
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()Teil III: The missing link
Für den Austausch zwischen Prozessen bietet Python unter anderem Pipe und multiprocessing.Queue. Queue-Objekte synchronisieren ihren Zugriff und serialisieren übergebene Objekte. Ein blockierendes get() direkt im Event Loop blockiert aber diesen Loop. Polling kann ebenfalls unnötig CPU-Zeit kosten.
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
Man kann get() in einem Thread ausführen lassen. Bei vielen wartenden Aufrufen kann das den Threadpool belegen. Das folgende Beispiel demonstriert ein Shutdown-Problem und darf nicht unkontrolliert ausgeführt werden:
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 pipeKorrektur der ursprünglichen Erklärung: Das Abbrechen des asyncio-Tasks beendet den bereits blockierenden Thread-Aufruf nicht zuverlässig. Der wartende Thread kann den Shutdown des Executors verhindern; je nach Umgang mit Prozessen und Queues können weitere Probleme entstehen. Ein automatisch abgebrochener Thread mit festgehaltener Queue-Sperre ist keine allgemeingültige Erklärung.
Als Nächstes habe ich die Queue in einen Adapter gepackt. Er macht das Interface bequemer, löst aber das Shutdown- und Threadpool-Problem nicht. Das Fragment braucht weitere Imports und Hilfsfunktionen und ist nicht produktionsreif:
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}>"
Ich habe daraufhin mit Events und Endsignalen experimentiert, um Start und Shutdown zu koordinieren. Bei vielen gleichzeitigen Aufrufen und gegenseitigen Abhängigkeiten kann ein begrenzter Threadpool trotzdem blockieren. Anzahl und Lebensdauer der Worker sowie eine explizite Shutdown-Strategie sind wichtig.
Eine Inspiration war ein Vortrag von David Beazley über asyncio und Threads. Bei Prozessen kommt hinzu, dass Objekte nicht einfach als gemeinsame Python-Referenzen behandelt werden können. Ein Future wird nicht dadurch fertig, dass man es mit asyncio.wrap_future() umhüllt: Irgendwer muss es sicher erfüllen, abbrechen und die Kommunikation zwischen Prozessen organisieren.
Für Experimente habe ich eine Shared-Memory-Queue begonnen. Im Stand des ursprünglichen Artikels fehlte ihr noch die nötige Synchronisierung.
Die folgende Skizze mit Futures und Deques ist ebenfalls unvollständig: Die Wartelisten sind zwischen Prozessen nicht gemeinsam verbunden, und sie enthält weitere Fehler. Nicht als funktionierende Queue verwenden.
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}>"Eine weitere Idee wären getrennte Kommunikationskanäle für wartende Getter und Putter. Ob das funktioniert und wie sich Race Conditions, Abbruch und Shutdown sauber lösen lassen, müsste erst implementiert und getestet werden. Genau hier war der Stand meiner Untersuchung.