Parallelisierung / Schleife / Python

[ihPyP]

Einen schönen guten Abend,

ich habe mich mit dem Thema "Parallelisierung" (in Python) noch nicht beschäftigt und würde mir hier gerne ein paar gute Ratschläge und Anregungen geben lassen, mit denen ich dann weiter machen kann.
Im Grunde nutze ich Python, um ein anderes Programm im Batch-Modus aufzurufen und mein Programm/Skript ruft das externe Programm 4x sequentiell auf. Nun möchte ich dies gerne parallelisieren, denn jeder Prozess kann völlig autark arbeiten und am Ende soll mein Programm die Daten einsammeln und verarbeiten. Folglich möchte ich wissen:
a) Wie parallelisiert man einen entsprechenden Aufruf am einfachsten?
b) Wie bestimmt man dabei die Information, dass am Ende alle Prozesse beendet wurden, um dann mein Programm seriell weiterarbeiten zu lassen.

Für den Aufruf des externen Programmes benutze ich derzeit folgendes Kommando:

Code: Alles auswählen

Process = subprocess.Popen(cmdString, shell=DebugMode, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Output, errorCode = Process.communicate()

Ich bedanke mich schon einmal an dieser Stelle für die nützlichen Hinweise.

[BlackJack]

@ihPyP: Man könnte `concurrent.futures.ThreadPoolExecutor` verwenden. b) beantwortet sich mit der `map()`-Methode dann eigentlich von selbst.

`DebugMode` ist ein komischer Name für das `shell`-Argument und es müsste ja auch immer `True` sein wenn `cmdString` als Name nicht falsch sein soll.

[snafu]

a) Wie parallelisiert man einen entsprechenden Aufruf am einfachsten?

Am einfachsten mit einer Bibliothek, die das abstrahiert, damit man nicht direkt mit der Threading-API arbeiten muss. Nicht dass die schlecht wäre, aber sie ist für solche Aufgaben immer noch recht lowlevel. Das bereits vorgeschlagene map() aus dem `concurrent.futures`-Modul ist da ganz gut geeignet, zumal es in Pythons Standardbibliothek enthalten ist.

b) Wie bestimmt man dabei die Information, dass am Ende alle Prozesse beendet wurden, um dann mein Programm seriell weiterarbeiten zu lassen.

Wenn man direkt auf Threads arbeitet, dann würde man dafür die join()-Methode des jeweiligen Threads nutzen. Damit wartet der Programmfluss bis der neu erstellte Thread seine Arbeit beendet hat. Das übliche Schema wäre:

Code: Alles auswählen

for thread in deine_threads:
    thread.join()

Die Schleife läuft dann solange wie der längste Thread läuft (abgesehen von dem kleinen Overhead für die Codeausführung der Schleife). Falls das nicht sofort einleuchtet, dann ggf mal für 4 Threads, die nahezu gleichzeitig gestartet werden, so eine Art Zeitschema aufzeichnen. Dann sollte man schnell drauf kommen.

Noch eine wichtige Frage: Sind die zu parallelisierenden Abläufe eher I/O-lastig (d.h. Zugriffe auf Laufwerke oder Netzwerkverbindungen) oder sind es hauptsächlich aufwändige Berechnungen? Abhängig davon solltest du dich nämlich zwischen `ThreadPoolExecutor()` (für I/O) oder `ProcessPoolExecutor()` (für Berechnungen) entscheiden.

[snafu]

Wobei ich gerade beim nochmaligen Durchlesen sehe, dass es um den Aufruf eines Programms geht. Da brauchst du dann ja gar keine Python-seitige Parallelisierung, sondern du machst halt mehrere `Popen()`-Aufrufe und fragst in einer Schleife (ähnlich wie ich das zuvor gezeigt habe) die Ergebnisse ab. Oder habe ich etwas übersehen / falsch verstanden?

[BlackJack]

@snafu: Du hast übersehen das Standardausgabe und -fehlerausgabe von den externen Prozessen gepiped werden und `communicate()` blockiert. Also braucht man doch Parallelisierung im Python-Code um das gleichzeitig auszuführen.

[snafu]

Stimmt, und das Lesen von Prograummausgaben gehört natürlich zur Kategorie I/O. Demnächst lese ich aufmerksamer, bevor ich antworte...

[BlackJack]

Für Python 2 bekommt man `concurrent.futures` im Python Package Index und damit mit `pip` installiert; bei Python 3 ist es in der Standardbibliothek:

Code: Alles auswählen

#!/usr/bin/env python
# coding: utf8
from __future__ import absolute_import, division, print_function
from subprocess import PIPE, Popen
from concurrent.futures import ThreadPoolExecutor


def do_some_external_work(command):
    return Popen(command, stdout=PIPE, stderr=PIPE).communicate()


def main():
    commands = [['ls', './tmp'], ['du', '-sch', './tmp']]
    executor = ThreadPoolExecutor(len(commands))
    results = list(executor.map(do_some_external_work, commands))
    print(results)

 
if __name__ == '__main__':
    main()