Multi-Processing-Beispiel

Wir beginnen hier mit Code, der klar und einfach ist und von oben nach unten ausgeführt wird. Er ist einfach zu entwickeln und inkrementell zu testen:

[1]:
import requests
from multiprocessing.pool import ThreadPool as Pool

sites = [
    'https://github.com/veit/jupyter-tutorial/',
    'https://jupyter-tutorial.readthedocs.io/en/latest/',
    'https://github.com/veit/pyviz-tutorial/',
    'https://pyviz-tutorial.readthedocs.io/de/latest/',
    'https://cusy.io/en',
]

def sitesize(url):
    with requests.get(url) as u:
        return url, len(u.content)

pool = Pool(4)
for result in pool.imap_unordered(sitesize, sites):
    print(result)
('https://jupyter-tutorial.readthedocs.io/en/latest/', 6374)
('https://pyviz-tutorial.readthedocs.io/de/latest/', 6556)
('https://github.com/veit/pyviz-tutorial/', 164082)
('https://github.com/veit/jupyter-tutorial/', 183351)
('https://cusy.io/en', 26974)

Hinweis 1: Eine gute Entwicklungsstrategie ist die Verwendung von map, um den Code in einem einzelnen Prozess und einem einzelnen Thread zu testen, bevor zu Multi-Processing gewechselt wird.

Hinweis 2: Um besser einschätzen zu können, wann ThreadPool und wann Pool verwendet werden sollte, hier einige Faustregeln:

  • Für CPU-lastige Jobs sollte multiprocessing.pool.Pool verwendet werden. Üblicherweise beginnen wir hier mit der doppelten Anzahl von CPU-Kernen für die Pool-Größe, mindestens jedoch mit 4.

  • Für I/O-lastige Jobs sollte multiprocessing.pool.ThreadPool verwendet werden. Üblicherweise beginnen wir hier mit der fünffachen Anzahl von CPU-Kernen für die Pool-Größe.

  • Verwenden wir Python 3 und benötigen kein mit Pool identisches Interface, nutzen wir concurrent.future.Executor statt multiprocessing.pool.ThreadPool; er hat ein einfacheres Interface und wurde von Anfang an für Threads konzipiert. Da er Instanzen von concurrent.futures.Future zurückgibt, ist er kompatibel zu vielen anderen Bibliotheken, einschließlich asyncio.

  • Für CPU- und I/O-lastige Jobs bevorzugen wir multiprocessing.Pool, da hierdurch eine bessere Prozess-Isolierung erreicht wird.

[2]:
import requests
from multiprocessing.pool import ThreadPool as Pool

sites = [
    'https://github.com/veit/jupyter-tutorial/',
    'https://jupyter-tutorial.readthedocs.io/en/latest/',
    'https://github.com/veit/pyviz-tutorial/',
    'https://pyviz-tutorial.readthedocs.io/de/latest/',
    'https://cusy.io/en',
]

def sitesize(url):
    with requests.get(url) as u:
        return url, len(u.content)

for result in map(sitesize, sites):
    print(result)
('https://github.com/veit/jupyter-tutorial/', 183029)
('https://jupyter-tutorial.readthedocs.io/en/latest/', 6374)
('https://github.com/veit/jupyter-tutorial/', 183345)
('https://github.com/veit/pyviz-tutorial/', 164082)
('https://cusy.io/en', 26974)

Was ist parallelisierbar?

Amdahlsche Gesetz

Der Geschwindigkeitszuwachs vor allem durch den sequentiellen Anteil des Problems beschränkt, da sich dessen Ausführungszeit durch Parallelisierung nicht verringern lässt. Zudem entstehen durch Parallelisierung zusätzliche Kosten wie etwa für die Kommunikation und die Synchronisierung der Prozesse.

In unserem Beispiel können die folgenden Aufgaben nur seriell abgearbeitet werden:

  • UDP DNS request für die URL

  • UDP DNS response

  • Socket vom OS

  • TCP-Connection

  • Senden des HTTP Request für die Root-Ressource

  • Warten auf die TCP Response

  • Zählen der Zeichen auf der Website

[3]:
import requests
from multiprocessing.pool import ThreadPool as Pool

sites = [
    'https://github.com/veit/jupyter-tutorial/',
    'https://jupyter-tutorial.readthedocs.io/en/latest/',
    'https://github.com/veit/pyviz-tutorial/',
    'https://pyviz-tutorial.readthedocs.io/de/latest/',
    'https://cusy.io/en',
]

def sitesize(url):
    ''' Determine the size of a website '''
    with requests.get(url, stream=True) as u:
        return url, len(u.content)

pool = Pool(4)
for result in pool.imap_unordered(sitesize, sites):
    print(result)
('https://github.com/veit/pyviz-tutorial/', 164088)
('https://github.com/veit/jupyter-tutorial/', 183345)
('https://jupyter-tutorial.readthedocs.io/en/latest/', 6374)
('https://pyviz-tutorial.readthedocs.io/de/latest/', 6556)
('https://cusy.io/en', 26974)

Hinweis: imap_unordered wird verwendet, um die Reaktionsfähigkeit zu verbessern. Dies ist jedoch nur möglich, da die Funktion das Argument und das Ergebnis als Tuple zurückgibt.

Tipps

  • Macht nicht zu viele Trips hin und her

    Erhaltet ihr zu viele iterierbare Ergebnisse, ist dies ein guter Indikator für zu viele Trips, wie z.B. in

    >>>     def sitesize(url, start):
    ...         req = urllib.request.Request()
    ...         req.add_header('Range:%d-%d' % (start, start+1000))
    ...         u = urllib.request.urlopen(url, req)
    ...         block = u.read()
    ...         return url, len(block)
    
  • Macht auf jedem Trip relevante Fortschritte

    Sobald ihr den Prozess erhaltet, solltet ihr deutliche Fortschritte erzielen und euch nicht verzetteln. Das folgende Beispiel verdeutlicht zu kleine Zwischenschritte:

    >>> def sitesize(url, results):
    ...     with requests.get(url, stream=True) as u:
    ...         while True:
    ...             line = u.iter_lines()
    ...             results.put((url, len(line)))
    
  • Sendet und empfangt nicht zu viele Daten

    Das folgende Beispiel erhöht unnötig die Datenmenge:

    >>> def sitesize(url):
    ...     with requests.get(url) as u:
    ...         return url, u.content