© sabbracadabra / 123RF.com

In der letzten Folge dieser Serie ging es um verschiedene Möglichkeiten, Nebenläufigkeit mit Python durch Multithreading zu realisieren. Diesmal dienen Multiprocessing und Asynchronous I/O dazu, dasselbe Ziel zu erreichen.

Multithreading bietet nicht die einzige Möglichkeit, die Performance durch Parallelarbeit zu steigern. Andere Varianten sind etwa das Multiprocessing, das mehrere Prozesse auf verschiedenen CPUs gleichzeitig startet, oder asynchrone Ein- und Ausgaben.

Multiprocessing

Beim Multiprocessing laufen die Prozesse vollkommen unabhängig voneinander ab. Das kann in bestimmten Situationen Vorteile bringen, birgt aber auch den Nachteil wachsender Komplexität, wenn die Prozesse sich untereinander austauschen müssen und deshalb Interprocess Communication (IPC) ansteht.

Das Beispiel aus Listing 1 zeigt eine typische Aufgabe, bei der Multiprocessing seine Stärken gut ausspielen kann: Das Skript fragt von verschiedenen Webseiten ihre Größe ab. Diese Sites lassen sich völlig unabhängig voneinander kontaktieren, die Prozesse brauchen also keine Informationen auszutauschen.

import urllib.request
from multiprocessing.pool import Pool
sites = [
  'https://github.com/veit/jupyter-tutorial-de/',
  'https://jupyter-tutorial.readthedocs.io/de/latest/',
  'https://github.com/veit/pyviz-tutorial/',
  'https://pyviz-tutorial.readthedocs.io/de/latest/',
  'https://cusy.io/de',
  ]
  def sitesize(url):
  with urllib.request.urlopen(url) as u:
    page = u.read()
    return url, len(page)
pool = Pool(4)
for result in pool.imap_unordered(sitesize, sites):
  print(result)

Der Code nutzt »imap_unordered« [1], um die Antwortzeiten zu verbessern. Dies klappt jedoch nur, weil die Funktion das Argument und das Ergebnis als Tuple zurückgibt.

Als Entwicklungsstrategie für Multiprocessing empfiehlt es sich, mit »map« zu beginnen, um den Code in einem einzelnen Prozess und einem einzelnen Thread zu testen, bevor man dann zu Multiprocessing wechselt. In unserem Beispiel sähe das dann so aus wie in Listing 2.

for result in map(sitesize, sites):
  print(result)

Vielleicht verwundert es, dass wir das Herunterladen der Seiten und das Berechnen der Größe in einem Prozess zusammengefasst haben. Tatsächlich sollte man beim Multiprocessing darauf achten, dass in »map« oder »imap_unordered« nicht zu viele Iterables stecken, da viele kurze Prozesse die Laufzeit insgesamt verlängern.

Zudem gilt es sicherzustellen, dass das Programm innerhalb eines Prozesses genügend Arbeit leistet. Beim Multiprocessing bestimmt letztlich der Betriebssystem-Kernel, wann zwischen Prozessen umgeschaltet wird – das lässt sich also nicht steuern. Leistet ein Prozess in einem Schritt zu wenig Arbeit, muss das Betriebssystem sehr häufig zwischen Prozessen hin und her schalten, was die Ausführungsgeschwindigkeit deutlich verlangsamt.

Um besser einschätzen zu können, wann man »ThreadPool« einsetzt und wann besser »Pool«, gibt es eine Reihe von Faustregeln. So sollte man für CPU-lastige Jobs »multiprocessing.pool.Pool« verwenden. Üblicherweise beginnt man hier mit der doppelten Anzahl von CPU-Kernen für die Pool-Größe, mindestens jedoch mit vier.

Für I/O-lastige Jobs eignet sich »multiprocessing.pool.ThreadPool«. Üblicherweise startet man mit der fünffachen Anzahl der CPU-Kerne als Größe. Kommt Python 3 zum Zug und kann man auf ein mit Pool identisches Interface verzichten, empfiehlt sich »concurrent.futures.ThreadPoolExecutor« statt »multiprocessing.pool.ThreadPool«. Ersteres hat ein einfacheres Interface und wurde von Anfang an für Threads konzipiert. Da es Instanzen von »concurrent.futures.Future« zurückgibt, bietet es Kompatibilität zu vielen anderen Bibliotheken, einschließlich Asyncio.

Für CPU- und I/O-lastige Jobs bevorzugen viele Python-Programmierer »multiprocessing.Pool«, da sich damit eine bessere Prozessisolierung erreichen lässt.

Asyncio

Asyncio schaltet zwischen den Tasks kooperativ um. Darum muss man explizit »yield« oder »await« hinzufügen, um einen Task-Switch herbeizuführen. Auf diese Weise lässt sich kontrollieren, wann diese Task-Wechsel und gegebenenfalls Locks und Synchronisationen stattfinden sollen. Der Aufwand für Task-Switches lässt sich so sehr niedrig halten.

Zudem ist die erneute Anfrage eines Generators [2] oder Awaitable [3] günstiger als der erneute Aufruf einer Python-Funktion, da das die CPU-Auslastung deutlich verbessert. Bei komplexen Systemen ist Asyncio auch übersichtlicher als Threads mit Locks, benötigt jedoch eine große Menge an Werkzeugen – Futures [4], Event Loops [5] und nicht blockierende Versionen von fast allem.

Das Code-Beispiel aus Listing 3 zeigt, wie Asyncio zwischen zwei Aufgaben hin und her schaltet. Dabei ist die zweite Aufgabe (»consume«) auf das Ergebnis der ersten (»produce«) angewiesen. Zur Entdeckung möglicher Race Conditions dienen zufällige Wartezeiten, die I/O-Operationen simulieren. Via »queue« landen die Ergebnisse von »produce« in einer Warteschlange, wo sie der »consume«-Prozess aufnimmt und weiterverarbeitet. Dabei simuliert »sleep« aufwendigere I/O-Operationen, und »None« signalisiert, dass die Warteschlange in »produce« vollständig abgearbeitet wurde. In »consume« wird mit »await« auf eingehende Nachrichten gewartet, bei »None« bricht der Prozess ab.

Asyncio bringt einen Debug-Modus [6] mit. Er enthält einen Profiler, der alle langsamen asynchronen Aufrufe protokolliert. Für den Produktivbetrieb eignet sich jedoch Aiodebug [7] besser. Als weiteres Werkzeug sollte man das nicht blockierende Logging Aiologger [8] verwenden.

import asyncio
import random
async def produce(queue, n):
  for x in range(1, n + 1):
    # produce an item
    print('producing {}/{}'.format(x, n))
    # simulate i/o operation using sleep
    await asyncio.sleep(random.random())
    item = str(x)
    # put the item in the queue
    await queue.put(item)
  # indicate the producer is done
  await queue.put(None)
async def consume(queue):
  while True:
    # wait for an item from the producer
    item = await queue.get()
    if item is None:
      # the producer emits None to indicate that it is done
      break
    # process the item
    print('consuming {}'.format(item))
    # simulate i/o operation using sleep
    await asyncio.sleep(random.random())
loop = asyncio.get_event_loop()
queue = asyncio.Queue(loop=loop)
asyncio.ensure_future(produce(queue, 10), loop=loop)
loop.run_until_complete(consume(queue))

Vor- und Nachteile

Die Tabelle “Vor- und Nachteile” vergleicht die drei besprochenen Lösungen für Nebenläufigkeit (Multithreading, Multiprocessing und Asynchronous I/O) anhand der folgenden vier Kriterien:

• Trennung: Wird ein gemeinsamer Status geteilt, und kann das zu Race Conditions führen? Erfolgen Berechnungen unabhängig voneinander? Entsteht zusätzlicher Kommunikationsaufwand, wenn die Berechnungsergebnisse untereinander ausgetauscht werden müssen?
• Wechsel: Wie wechseln die Aufgaben – präemptiv oder kooperativ?
• Tooling: Wie viele Werkzeuge benötigt man, um nebenläufige Prozesse auszuführen? Wie verständlich bleibt der Code auch bei komplexeren Berechnungen?
• Performance: Welche Art der Nebenläufigkeit eignet sich bei I/O-lastigen, welche bei CPU-lastigen Anwendungen? Wo liegen die Leistungsgrenzen?

Abschließend lässt sich festhalten, dass es kein Patentrezept für Nebenläufigkeit unter Python gibt. Jeder Ansatz hat seine spezifischen Vor- und Nachteile. Vor der Entscheidung, wie man die Nebenläufigkeit in Python organisieren will, sollte man die Problemstellung gut analysieren. Selbst für den Fall, dass sich eine bestimmte Methode besser für die Problemstellung eignet als andere, muss man alle kennen, um Projekte erfolgreich umzusetzen: In realen Szenarien ist es nicht ungewöhnlich, dass man in einer Anwendung das gesamte Arsenal an Werkzeugen und Stilen für Gleichzeitigkeit und Nebenläufigkeit benötigt. (jcb)