Threading, Multiprocessing richtig anwenden

[Nobuddy]

Hallo zusammen,

bei dem Thema Threading, Multiprocessing und Co, habe ich bisher vor mich her geschoben.
Nun möchte ich mich dem Thema widmen und Schritt für Schritt diese Sache angehen.

Ich habe für mich ein Programm geschrieben, das Kunden-, Lieferanten-, Bestell-, Lager-, Rechnug-, Buchhaltung-Wesen beinhaltet. Weiter aktualisiere ich auch Produktdaten verschiedener Lieferanten und Preise darüber. Als GUI verwende ich Tkinter.
Bei der Aktualisierung der Produktdaten (ca. 130.000 Datensätze), ist während des Prozesses kein weiterer Zugriff auf das Programm mehr möglich, was eigentlich nicht sein sollte.

Ich habe schon etliche Themen darüber gelesen, bin mir aber nicht darüber im Klaren, welches für meinen Bedarf das Richtige ist.
Auch die Vorgehensweise, wie und wo ich dies gezielt einsetze, ist ebenfalls unklar.
Starte ich mein Programm schon direkt in einem Thread/Multiprocess und weitere rechenintensive Vorgänge fließen dann dort ebenfalls ein ... ?

Vielleicht könnte ich da von Euch ein wenig Input erhalten, bevor ich dann mit dem Code loslege!

Grüße Nobuddy

[Sirius3]

@Nobuddy: bei so einer allgemeinen Beschreibung kann man nur eine allgemeine Antwort geben: normalerweise startet man pro Aufgabe einen Thread. Treten die Aufgaben regelmäßig ein, kann es sinnvoll sein, den Thread gleich am Anfang zu starten und in einer Queue einzelne Anfragen abzuarbeiten. Wichtig ist dabei, dass, wenn ein Prozesse auf eine Datenstruktur schreibend zugreift, nicht gleichzeitig ein anderer lesend oder schreiben zugreift. Das geht am einfachsten, wenn man generell auf geteilte Daten verzichtet und die ganze Kommunikation über kontrollierte Kanäle, wie z.B. Queues abläuft.

[Nobuddy]

Hallo Sirius3,

Deine Antwort ist momentan ausreichend.

normalerweise startet man pro Aufgabe einen Thread.

Eine Aufgabe ist dann z.B. das Starten meines Programmes (GUI Tkinter), eine weitere Aufgabe wäre ein weiteres Modul zu starten.
Ist das so richtig?

Treten die Aufgaben regelmäßig ein, kann es sinnvoll sein, den Thread gleich am Anfang zu starten und in einer Queue einzelne Anfragen abzuarbeiten..

Also zuerst mein Programm in einem Thread starten, weitere Aufgaben wie ein weiteres Modul, dann in einer Queue abarbeiten.
Ist das so richtig?

Wichtig ist dabei, dass, wenn ein Prozesse auf eine Datenstruktur schreibend zugreift, nicht gleichzeitig ein anderer lesend oder schreiben zugreift. Das geht am einfachsten, wenn man generell auf geteilte Daten verzichtet und die ganze Kommunikation über kontrollierte Kanäle, wie z.B. Queues abläuft.

Das ist ein wichtiger Aspekt, was nicht so einfach handhaben ist.
Hast Du zu geteilten Daten und kontrollierte Kanäle, eine kurze verständliche Erklärung?

Grüße Nobuddy

[Sirius3]

Eine GUI muß immer im Hauptthread gestartet werden. Was Du mit Modul meinst, versteh ich nicht.

[BlackJack]

@Nobuddy: „Module starten” ist eine ungewöhnliche Formulierung, ich denke da kann keiner etwas mit anfangen. Und was eine Aufgabe in Deinem Programm ausmacht kann Dir keiner sagen, das musst Du schon selbst wissen. „Aktualisierung der Produktdaten” scheint laut dem ersten Beitrag ja eine Aufgabe zu sein die parallel abgearbeitet werden soll.

Kurz:

Bei Threads: Geteilte Daten: Nicht machen! Kontrollierte Kanäle: `Queue`-Objekte.

Bei Multiprocessing: Geteilte Daten gibt's nicht. Kontrollierte Kanäle: Um die Interprocesskommunikation kümmert sich das `Pool`-Objekt schon.

Bei der Beispielaufgabe aus dem ersten Thread ist zu beachten das Du während die 130000 Datensätze aktualisiert werden, nicht auf die Datensätze zugegriffen werden darf wenn Du auf der sicheren Seite sein willst. Ansonsten musst Du Dir genau überlegen wie man den Zugriff regeln kann/muss, damit keine Inkonsistenzen entstehen wenn gleichzeitig zum Aktualisieren auf die Daten lesend oder gar schreibend zugegriffen wird. Auch hier kann man allgemein nicht mehr sagen. Ausser dass Nebenläufigkeit allgemein kein einfaches Thema ist.

[Nobuddy]

Eine GUI muß immer im Hauptthread gestartet werden.

Ok, das verstehe ich.

Was Du mit Modul meinst, versteh ich nicht.

Code: Alles auswählen

[Codebox=python file=product_update.py][/Codebox]
from product_update import update_start

Wenn ich z.B. 'update_start()' aus meiner GUI heraus starte.

[Nobuddy]

Hallo BlackJack ,

Das mit „Aktualisierung der Produktdaten” ist eine Aufgabe.
Um zu verstehen, wie ich das mit meinen Daten handle.
Meine Daten liegen als csv-Dateien vor. Diese werden beim Starten meiner Gui als Dictionarys geladen und bei Änderung werden die csv-Dateinen bei Beendigung der GUI aktualisiert.
Beim Aktualisierung der Produktdaten, werden zuerst die aktuellen Daten gesichert. Danach werden die csv-Dateien zur Aktualisierung verwendet.
Ist die Aktualisierung abgeschlossen, werden die neuen Daten in den Dictionarys der GUI aktualisiert. Danach kann ich mit den aktuellen Daten in der GUI weiterarbeiten. Sind das dann geteilte Daten?

Ein Problem, was schon angesprochen wurde, ist wie man den Zugriff regeln kann/muss, damit keine Inkonsistenzen entstehen wenn gleichzeitig zum Aktualisieren auf die Daten lesend oder gar schreibend zugegriffen wird.
Z.B. wenn ich während der Produktdaten-Aktualisierung, ein Produktinfo benötigen würde, also lesend darauf zugreifen würde, während in diesem Moment, die Daten in der GUI aktualisiert werden würden.

[snafu]

@Nobuddy: Das Importieren eines Moduls oder das Starten einer GUI gehört in den Haupt-Thread und nicht in irgendwelche zusätzlich gestarteten Threads. Denn welchen Sinn sollte es haben, die halbe Sekunde, die vielleicht beim Importieren eines Moduls vergeht, in einen neuen Thread auszulagern?

Im Eingangsbeitrag hattest du etwas von einem aufwändigen Update geschrieben. Exakt nur dies sollte einen eigenen Thread spendiert bekommen, da nur hier Threading Sinn ergibt. Der zusätzliche Thread startet also nicht beim Importieren des Moduls, welches die Aktualisierungen verwaltet, sondern er startet in dem Moment, wo eine Aktualisierung angestoßen wird.

[BlackJack]

@Nobuddy: Zumindest semantisch währen das geteilte Daten auch wenn beide Seiten (GUI und Aktualisierung) auf ”physisch” verschiedenen Daten arbeiten sollten beide Kopien konsistent bleiben. Wenn also während der Aktualisierung Daten mit der GUI verändert werden, dann wird diese Veränderung nicht berücksichtigt und am Ende der Aktualisierung beim laden der Daten einfach überschrieben.

Ich denke hier fängt sich jetzt langsam an zu ”rächen” das Du selbst unbedingt ein DBMS mit CSV-Dateien nachbauen wolltest, denn nun musst Du Dich auch um so etwas wie Transaktionen mit eigenem Code kümmern, oder zumindest die gesamte Datenbasis effektiv gegen Veränderungen sperren solange von einem anderen Thread aus Veränderungen vorgenommen werden. Der Vorteil vom Threading wäre dann zumindest noch das Du die GUI selbst sperren kannst, zum Beispiel mit einem modalen Dialog der dem Benutzer sagt was gerade vor sich geht. Und eventuell auch den Fortschritt der Aktion anzeigt. Ist immerhin besser als eine GUI die eine ganze Zeit lang wirklich ”tot” ist so dass sich der Benutzer fragt ob das Programm abgestürzt oder hängen geblieben ist.

Das würde ich dann nicht auf verschiedene Repräsentationen der Daten verteilen sondern auch beim aktualisieren auf die Daten im Speicher zugreifen (sofern wirklich alles im Speicher gehalten wird).

Und was man noch bräuchte wäre ein Benachrichtigungsmechanismus das sich Daten geändert haben die gerade angezeigt werden. Oder man aktualisiert nach dem Aktualisieren der Daten grundsätzlich alle angezeigten Daten, damit der Benutzer nichts altes angezeigt bekommt.

[snafu]

Ein Problem, was schon angesprochen wurde, ist wie man den Zugriff regeln kann/muss, damit keine Inkonsistenzen entstehen wenn gleichzeitig zum Aktualisieren auf die Daten lesend oder gar schreibend zugegriffen wird.
Z.B. wenn ich während der Produktdaten-Aktualisierung, ein Produktinfo benötigen würde, also lesend darauf zugreifen würde, während in diesem Moment, die Daten in der GUI aktualisiert werden würden.

Ich würde in dem Thread für die Aktualisierung das Dictionary mit den Daten komplett neu aufbauen, d.h. unabhängig vom bisherigen Dictionary. Wenn dann zwischenzeitlich lesende Zugriffe vorkommen, dann arbeiten diese auf dem alten Datenbestand. Schreibzugriffe während einer Aktualisierung - wenn sie denn nicht ganz vermieden werden können - sollten als anstehende Aufgaben in eine Warteschlange gestellt werden.

Nachdem die Aktualisierung erfolgt ist, werden die Änderungen aus der Warteschlange in das Dictionary mit den aktualisierten Daten geschrieben. Falls gewünscht, kann dann dieser Datenbestand nochmal in die CSV-Datei geschrieben werden. Anschließend wird das neue Dictionary zur Verwendung freigegeben. Wie gesagt: So wäre grob meine Vorgehensweise bei einer solchen Aufgabe.

[Nobuddy]

Möchte mich mal bei Euch für Euren Input bedanken!

Dass sich nun manches rächt, was ich zuvor bei meiner Programmierung nicht beachtet bzw. nicht gewusst habe, muss ich hier eingestehen ...

Bei der Produkt-Aktualisierung in meinem Fall, sollte ein Lesen der Daten ohne Gefahr möglich sein. Ein Ändern der Daten muss allerdings verhindert werden, um keine Inkonsistenz zu erreichen.
Das mit dem Zugreifen der Produkt-Aktualisierung auf den Speicher, wo die gespeicherten Daten gehalten werden, wollte ich aus Sicht der Datensicherheit verhindern. Dies würde das ganze noch komplexer machen, als es ohnehin schon ist. Aber wahrscheinlich fehlt mir das Wissen zur entsprechenden Sichtweise.
Bei dem Zugreifen der Produkt-Aktualisierung auf den Speicher, wäre dann ein Benachrichtigungsmechanismus notwendig, das sich Daten geändert haben die gerade angezeigt werden.

Gibt es da evtl. schon einen Code, den man als Basis verwenden könnte?

[Nobuddy]

Habe hier im Forum einen Code gefunden, der evtl. als Anfang dienen könnte.

Code: Alles auswählen

[code]#!/usr/bin/env python
# -*- coding: utf-8 -*-
# For Python3.x

import queue
import threading
import time


class Worker(threading.Thread):
    """
    Arbeitstier
    """
   
    def __init__(self, _queue, breaker):
        """
        Übernimmt die Queue und initialisiert den Thread
        """
       
        threading.Thread.__init__(self)
       
        self.queue = _queue
        assert(isinstance(self.queue, queue.Queue))
       
        self.breaker = breaker
   
   
    def run(self):
        """
        Wird bei "Worker.Start" aufgerufen.
        Wird gestoppt, wenn die Queue leer ist.
        """
       
        while True:
           
            if self.queue.empty():
                # Die Schleife und damit auch den Thread beenden
                break
           
            # Wert aus der Queue holen
            new_value = self.queue.get()
           
            # Arbeiten (natürlich nur simuliert)
            time.sleep(4)
            print(new_value,)
       
        # Ausschalter setzen
        self.breaker.set()


def main():
    """
    Hauptprozedur
    """
   
    # Queue erstellen
    _queue = queue.Queue()
   
    # Queue füllen
    t = ("a", "b", "c", "d", "e", "f", "g", "h", "i", "j")
    for item in t:
        _queue.put(item)

    # Ausschalter
    thread1_ausschalter = threading.Event()
    thread2_ausschalter = threading.Event()
    thread3_ausschalter = threading.Event()
   
    # Threads erstellen und starten
    thread1 = Worker(_queue, thread1_ausschalter)
    thread2 = Worker(_queue, thread2_ausschalter)
    thread3 = Worker(_queue, thread3_ausschalter)
   
    thread1.start()
    thread2.start()
    thread3.start()
   
    # Hauptprozedur so lange hier warten lassen, bis alle
    # Ausschalter gesetzt wurden. Es geht also erst weiter, wenn
    # alle Ausschalter von den Threads (symbolisch) umgelegt wurden.
    thread1_ausschalter.wait()
    thread2_ausschalter.wait()
    thread3_ausschalter.wait()
   
    # Fertig
    print("Fertig")
    return
   

if __name__ == "__main__":
    main()

[/code]

Oder lege ich mit diesem Code falsch?

[snafu]

@Nobuddy: Ich weiß nicht, ob der Code allzu viel für dich bringt. Mit mehreren Workern willst du ja anscheinend eher nicht arbeiten, sondern lediglich ein Thread soll die Aktualisierungen umsetzen, oder sehe ich das falsch? Mehrere Worker bedeuten auf jeden Fall mehr Komplexität, die da ja laut einem vorherigen Beitrag von dir lieber vermeiden möchtest (was ich auch verstehen kann).

Lies dich doch einfach mal in das Thema Threading ein, wenn du damit noch gar nicht vertraut bist. Mit den richtigen Suchbegriffen wirst du im Internet definitiv fündig.

[Nobuddy]

Weil Threading und Queue gefallen sind, dachte ich mit dem Code nicht verkehrt zu liegen.

Arbeite dies http://www.python-kurs.eu/threads.php mal durch.
Für weitere Empfehlungen, bin ich dankbar.

[Sirius3]

@Nobuddy: das verlinkte Tutorial ist, wie andere Tutorials auf dieser Seite, nicht empfehlenswert. Die Suche in diesem Forum wird Dir auch die Begründung liefern.

[snafu]

Im Kern musst du folgendes machen, um den Thread zu einer Ausgabe zu bewegen:

- Von `threading.Thread` ableiten und in `.run()` eine Schleife bauen, welche deine CSV-Daten ausliest.

- Innerhalb dieser Schleife die einzelnen Datensätze in die Queue ablegen.

- Benachrichtigen, wenn die Aktualisierung beendet wurde, damit man dann von außen auf die Daten zugreifen kann.

In Quelltext ausgedrückt, könnte das in etwa so aussehen (ungetestet):

Code: Alles auswählen

class Updater(threading.Thread):
    def __init__(self, queue, csv_filename):
        self.queue = queue
        self.csv_filename = csv_filename
        self._finished = threading.Event()

    def run(self):
        with open(csv_filename) as csv_file:
            for row in csv.reader(csv_file):
                # Nur als Beispiel gedacht
                self.queue.put({
                    'name': row[0], 'menge': row[1], 'preis': row[2]
                })
        self._finished.set()

    def has_finished(self):
        return self._finished.is_set()


class MyTkWindow(object):
    # ...

    def update_data(self):
        if not self.updater:
            # Erster Durchlauf -> Thread muss gestartet werden.
            self.updater = Updater(self.queue, self.csv_filename)
            self.updater.start()
        if not self.updater.has_finished():
            # Aktualisierung läuft noch -> versuche es in 500ms erneut.
            self.root.after(500, self.update_data)
        else:
            while not self.queue.empty():
                # Elemente der Queue nacheinander an GUI übermitteln.
                self.set_gui_data(self.queue.get())
            # Thread-Objekt "wegschmeißen", da es nicht wiederverwendet werden kann.
            # Dient auch als Hinweis, dass ein neuer Thread gestartet muss, wenn die
            # Aktualisierung mittels `.update_data()` später erneut angestoßen wird.
            self.updater = None

Wichtig ist halt, zu verstehen, dass der aufrufende Thread weiterhin Zugriff auf die Queue benötigt. Ich gehe hier davon aus, dass die Queue vorab erstellt und an`self.queue` gebunden wurde, z.B. in `__init__()`.

So wie ich dich verstanden habe, willst du die Überführung der aktualisierten Daten in die GUI in einem Rutsch erledigen. Daher wird die `update_data()`-Methode in meinem Beispiel immer wieder aufgerufen bis der Thread mitgeteilt hat, dass nun alle Daten aus der CSV-Datei in den Speicher gelesen wurden. In dem 500ms dazwischen, kann die GUI etwas anderes tun. Eine `while`-Schleife würde die GUI an dieser Stelle einfrieren und den Effekt, den man durch das Threading erhalten möchte, letztlich nutzlos machen.

Und am Ende wird dann wird der Zwischenspeicher ausgelesen und die einzelnen Änderungen an die GUI zum Einpflegen übergeben.

[DasIch]

Muss es wirklich sein dass bei Threads jedes mal das threading Modul und Queues ausgekramt werden auf dessen Basis dann in jedem Forenthread zum Thema das Rad neuerfunden wird? Es gibt schon seit *Jahren* concurrent.futures in der Standard Library mit dem wesentlich einfacher zu benutzenden ThreadPoolExecutor. Es gibt auch einen backport für Python 2, wenn man Python 3 nicht benutzen will oder kann.

[Nobuddy]

Hallo snafu, Danke für Deinen Code!
Bin gerade noch auf einer anderen Baustelle ..., daher konnte ich mich im Moment noch nicht weiter dem Thema widmen.
Dies werde ich bald nachholen, denn für das Umsetzen Deines Inputś, brauche ich Zeit damit nichts schief läuft.

Hallo DasIch, habe bei meiner Recherche auch von concurrent.futures gelesen, soll aber nicht ganz so leistungsfähig sein.
Vielleicht war ich auch nur auf der falschen Seite ..., konnte aber nicht so viel über concurrent.futures erfahren.

PS: Ich verwende Python3!

[BlackJack]

@Nobuddy: Was ist denn an der Stelle mit Leistungsfähigkeit gemeint? Und selbst wenn: Das etwas fehlerfrei funktioniert ist doch auch etwas Wert. Das ist getestet und wird von vielen benutzt, im Gegensatz zu dem x-ten selbst gebastelten Pool-Manager.

[Nobuddy]

Hallo BlackJack,

das mit concurrent.futures, werde ich mir nochmals genauer anschauen.
Leider konnte ich dazu nicht all zu viele Beiträge dazu finden, daher habe ich Threading, Multiprocessing favorisiert, das was laut Beiträgen mehr angewendet wird.

Wo und welche Anleitungen, Beiträge sind empfehlenswert?

Grüße Nobuddy