Threading für Anfänger

[Sophus]

@BlackJack: Die Testdatei ist in Ordnung. Beim Entpacken tauchen keinerlei Fehler auf. Oder was genau soll mir da auffallen?

[Hyperion]

Hallo Hyperion und BlackJack, danke für die Hinweise, und für das beste Beispiel der Welt, Hyperion. Aber ich habe zu spät bemerkt, dass du mir ein Beispiel präsentierst, ...

Wieso "zu spät"? Das Beispiel rennt ja nicht weg

[Sirius3]

@Sophus: wenn Du Dir diese beiden Zeilen anschaust:

Code: Alles auswählen

               fd.write(chunk)
                downloaded_bytes += chunk_size

was ist da das naheliegendste, die Bytes, die gerade heruntergeladen wurden, zu ermitteln?

[Sophus]

@Sirius3: Meinst du etwa 'chunk'? Sprichst du hier das Problem an, was BlackJack meinte, von wegen nicht "Fehlerfrei"? Aber was spricht dagegen die chunk_size zu verwenden? Ich meine, die Häppchen werden ja solange heruntergeladen, bis die Datei komplett ist, also 100%. Und der Prozessbalken verhält sich auch "korrekt".

EDIT:

Ich habe mich mit dem 'chunk' auseinandergesetzt, udn wie zu vermuten war, hatte es auch nicht geklappt. Hier die beiden Versionen:

Version 1

Code: Alles auswählen

        with open(self.location, 'wb') as fd:
            for chunk in file.iter_content(chunk_size):
                fd.write(chunk)
                downloaded_bytes += chunk
                print (float(downloaded_bytes)/file_size*100)
                self.notify_progress.emit(float(downloaded_bytes)/file_size*100)

Hier dachte ich, könnte man die heruntergeladenen Bytes (chunk) zu den downloaded_bytes hinzuaddieren. Klappt aber nicht.

Version 2

Code: Alles auswählen

        with open(self.location, 'wb') as fd:
            for chunk in file.iter_content(chunk_size):
                fd.write(chunk)
                self.notify_progress_emit(chunk)

Hier wollte ich nicht addieren, sondern die 'chunks' direkt benutzen.

Beide Versionen funktionieren nicht.

[Sirius3]

@Sophus: Du hast dich nicht auseinandergesetzt, sondern wild herumgeraten.

Ja requests liefert immer exakt chunk_size Bytes (und es ist unglaublich schwierig, ihm das abzugewöhnen), bis auf den letzten Chunk.

[Sophus]

@Sirius3: Ich komme einfach nicht dahinter, was du mir da aufzeigen willst. Der letzte 'chunk_size'-Byte ist natürlich nicht gleich groß, sondern ein Rest vom Ganzen. Aber was genau möchtest du mir mit den zwei Zeilen aufzeigen? Vielleicht eine nette Erklärung? Habe gehört, dass man dadurch auch lernen kann, und den Lernenden nicht immer alleine lässt

[Sirius3]

Einen indirekten Wert zu nehmen, birgt immer das Risiko, dass irgendwo etwas nicht wie erwartet läuft und es deshalb zu Fehlern kommt. Wie zum Beispiel, dass der letzte Chunk kleiner und damit downloaded_bytes zu groß wird.
Wenn man also statt dessen

Code: Alles auswählen

downloaded_bytes += len(chunk)

schreibt, kann es gar nicht mehr zu Abweichungen kommen. Oder noch direkter

Code: Alles auswählen

downloaded_bytes = fd.tell()

entspricht exakt der Anzahl an Bytes die geschrieben wurden.

[Sophus]

@Sirius3: Danke für deine Erklärung. Und jetzt kapiere ich was du meinst. Jedoch muss ich zu meiner Schande gestehen, dass ich die Methode tell() gar nicht kannte. Und dass ich nicht selbst auf die Methode len() gekommen bin, ist mir schon etwas peinlich. Du sagtest, die tell()-Methode sei direkter. Inwiefern? Ich meine, die len()-Methode liefert genauso direkt die Größe oder?

[Sophus]

Nun, ich möchte, dass der Anwender den Prozess des Herunterladens abbrechen kann - aus welchen Gründen auch immer. Dazu dachte ich sofort: "Nimm doch terminate()", und habe das so umgesetzt:

Code: Alles auswählen

class Download_Thread(QThread):
    finished_thread = pyqtSignal()
    notify_progress = pyqtSignal(int)
 
    def __init__(self, loc, link):
        QThread.__init__(self)
 
        self.url = link
        self.location = loc
 
    def run(self):
        print self.url
        print self.location
        file = requests.get(self.url, stream=True)
        file_size = int(requests.head(self.url).headers.get('content-length', [0]))
        print "%s Byte" %file_size
        result = file_size / (1024*5)
        print result
        chunk_size = int(result)
        downloaded_bytes = 0
            
        with open(self.location, 'wb') as fd:
            for chunk in file.iter_content(chunk_size):
                fd.write(chunk)
                downloaded_bytes = fd.tell() # sehr genau und direkt
                print (float(downloaded_bytes)/file_size*100)
                self.notify_progress.emit(float(downloaded_bytes)/file_size*100)

        print "Finish"
        self.finished_thread.emit()

    def stop(self):
        print "Cancel"
        self.finished_thread.emit()
        self.terminate()

Jedoch sagt die QT-Dokumentation folgendes:

Warning: This function is dangerous and its use is discouraged. The thread can be terminated at any point in its code path. Threads can be terminated while modifying data. There is no chance for the thread to clean up after itself, unlock any held mutexes, etc. In short, use this function only if absolutely necessary.

Welche Möglich gibt es, den Vorgang sauber abzubrechen? Mit quit()-Methode habe ich auch schon versucht - klappt leider nicht.

[Hyperion]

Welche Möglich gibt es, den Vorgang sauber abzubrechen? Mit quit()-Methode habe ich auch schon versucht - klappt leider nicht.

Ich verweise da noch mal auf das beste Beispiel der Welt zu diesem Thema

Nutze eben ein Flag-Attribut, welches in der ``run``-Methode periodisch abgefragt wird. Dies kann man z.B. über eine QSemaphore machen.

[Sophus]

@Hyperion: Besten dank. Durch deine Hilfestellung habe ich es wie folgt umgesetzt:

Code: Alles auswählen

class Download_Thread(QThread):
    finished_thread = pyqtSignal()
    notify_progress = pyqtSignal(int)
 
    def __init__(self, loc, link):
        QThread.__init__(self)
 
        self.url = link
        self.location = loc

        self._run_semaphore = QSemaphore(1)
 
    def run(self):
        print self.url
        print self.location
        file = requests.get(self.url, stream=True)
        file_size = int(requests.head(self.url).headers.get('content-length', [0]))
        print "%s Byte" %file_size
        result = file_size / (1024*5)
        print result
        chunk_size = int(result)
        downloaded_bytes = 0
            
        with open(self.location, 'wb') as fd:
            for chunk in file.iter_content(chunk_size):
                fd.write(chunk)
                downloaded_bytes = fd.tell() # sehr genau und direkt
                print (float(downloaded_bytes)/file_size*100)
                self.notify_progress.emit(float(downloaded_bytes)/file_size*100)
                if self._run_semaphore.available() == 0:
                    self._run_semaphore.release(1)
                    break
 
        print "Finish"
        self.finished_thread.emit()

    def stop(self):
        print "Cancel"
        self.finished_thread.emit()
        self._run_semaphore.acquire(1)
 
if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyCustomDialog()
    window.resize(600, 400)
    window.show()
    sys.exit(app.exec_())

Um zu überprüfen, ob ich es auch tatsächlich kapiert habe, und nicht alles nur stumpfsinnige kopiere wie ein Äffchen.

Zeile 11: Laut der QT-Dokumentation is die QSemaphore()-Klasse thread-sicher. Diese Klasse dient also zur allgemeinen Zählung. Dadurch werden bestimmte Anzahlen von Ressourcen zur Verfügung gestellt. Also wird QSemaphore an das Attribut self._run_semaphore gebunden.

Zeile 30 bis 32: Je jedem Schleifendurchlauf wird durch die If-Kaskade mittels der available()-Methode abgefragt, ob die Ressource noch zur Verfügung steht. Wenn nicht, dann wird durch die release()-Method die Ressource wieder zur Verfügung gestellt, und im Anschluss mit einem break der Verlauf abgebrochen. Die release()-Methode dient dazu, dass man zu einem späteren Zeitpunkt den Thread nochmal starten kann.

Zeile 40: In der stop()-Funktion wird die Ressource durch die acquire()-Methode die Ressource "erworben", und dadurch wie der Verlauf in der run()-Funktion unterbrochen.

Das wäre mein Verständnis.

[Hyperion]

Zeile 11: Laut der QT-Dokumentation is die QSemaphore()-Klasse thread-sicher. Diese Klasse dient also zur allgemeinen Zählung. Dadurch werden bestimmte Anzahlen von Ressourcen zur Verfügung gestellt. Also wird QSemaphore an das Attribut self._run_semaphore gebunden.

Na, ``QSemaphore`` ist einfach eine Semaphore :K

Zeile 30 bis 32: Je jedem Schleifendurchlauf wird durch die If-Kaskade mittels der available()-Methode abgefragt, ob die Ressource noch zur Verfügung steht. Wenn nicht, dann wird durch die release()-Method die Ressource wieder zur Verfügung gestellt, und im Anschluss mit einem break der Verlauf abgebrochen. Die release()-Methode dient dazu, dass man zu einem späteren Zeitpunkt den Thread nochmal starten kann.

Prinzipiell ja. Aber: Wieso kann man *dieses spezielle* Thread-Exemplar denn später erneut starten? Das musst Du imho genau verstehen... hier hast Du Dich an der expliziten Aussage vorbei "gemogelt", was mir suggeriert, dass Du es nicht wirklich genau verstanden hast...

Zeile 40: In der stop()-Funktion wird die Ressource durch die acquire()-Methode die Ressource "erworben", und dadurch wie der Verlauf in der run()-Funktion unterbrochen.

Japp.

Eine Anmerkung noch: Das Emittieren in Zeile 39 ist imho falsch! Denk da mal drüber nach

Je nach Anwendung könnte man aber auch drüber nachdenken, ein gesondertes Signal speziell für den Abbruch zu senden... dann müsste man natürlich Zeile 32 anders gestalten.

Das Senden einen solchen Signals aus der ``stop``-Methode halte ich aber in jedem Fall für falsch! (Einzig ein "Zwischensignal", welches dazu dient, andere Programmteile über den *eingeleiteten* Abbruch des Threads zu informieren, wäre an dieser Stelle richtig - aber das sollte wohl eher selten vorkommen...)

[Sophus]

Hallo Hyperion,

Prinzipiell ja. Aber: Wieso kann man *dieses spezielle* Thread-Exemplar denn später erneut starten? Das musst Du imho genau verstehen... hier hast Du Dich an der expliziten Aussage vorbei "gemogelt", was mir suggeriert, dass Du es nicht wirklich genau verstanden hast...

In Zeile 30-32 wird gesagt, wenn das Attribut _run_semaphore (mittels der available()-Methode) gleich Null ist, also keine n Ressource (n = 1) zur Verfügung steht, dann soll durch die release(1)-Methode dem Attribute _run_semaphore die n Ressource wieder zur Verfügung gestellt werden, und im Anschluss folgt der break-Befehl, also der Abbruch. Warum hier hinterher der Thread erneut gestartet werden kann liegt einzig und allein daran, weil die n Ressource vorhanden ist. In Zeile 11 sehen wir ja, wie dem Attribut _run_semaphore durch die QSemaphore()-Klasse eine n Ressource übergeben wird, also die Zahl 1.

Code: Alles auswählen

import os
import requests
import sys
 
from PyQt4.QtCore import QThread, pyqtSignal, Qt, QSemaphore
from PyQt4.QtGui import QVBoxLayout, QPushButton, QDialog, QProgressBar, QApplication

class MyCustomDialog(QDialog):
 
    def __init__(self, parent=None):
        super(MyCustomDialog, self).__init__(parent)
        layout = QVBoxLayout(self)
 
        # Create a progress bar and a button and add them to the main layout
        self.progressBar = QProgressBar(self)
        self.progressBar.setAlignment(Qt.AlignCenter)
        #self.progressBar.setValue(0)
        #self.progressBar.setRange(0, 1)
        layout.addWidget(self.progressBar)
 
        button = QPushButton("Start", self)
        layout.addWidget(button)
        buttonCnacel = QPushButton("Cancel", self)
        layout.addWidget(buttonCnacel)
 
        button.clicked.connect(self.check_folder_exists)
 
        # Set data for download and saving in path
        self.location = os.path.abspath(os.path.join('temp', 'example-app-0.3.win32.zip'))
        self.url = 'http://sophus.bplaced.net/download/example-app-0.3.win32.zip'
 
        self.download_task = Download_Thread(self.location, self.url)
        self.download_task.notify_progress.connect(self.on_progress)
        self.download_task.finished_thread.connect(self.on_finished)

        buttonCnacel.clicked.connect(self.download_task.stop)
 
    def on_progress(self, i):            
        self.progressBar.setValue(i)
 
    def on_start(self):
        self.download_task.start()
 
    def check_folder_exists(self):
        location = os.path.abspath(os.path.join('temp'))
        if not os.path.exists(location):
            os.makedirs(location)
            print "Folder was created"
            self.on_start()
        else:
            print "Folder already exists"
            self.on_start()
 
    def on_finished(self):
        print "stop"
        self.progressBar.setValue(0)

    def closeEvent(self, evnt):
        self.connect(self.download_task.stop)
 
 
class Download_Thread(QThread):
    finished_thread = pyqtSignal()
    notify_progress = pyqtSignal(int)
 
    def __init__(self, loc, link):
        QThread.__init__(self)
 
        self.url = link
        self.location = loc

        self._run_semaphore = QSemaphore(1)
 
    def run(self):
        file = requests.get(self.url, stream=True)
        file_size = int(requests.head(self.url).headers.get('content-length', [0]))
        print "%s Byte" %file_size
        result = file_size / (1024*5)
        print result
        chunk_size = int(result)
        downloaded_bytes = 0

        with open(self.location, 'wb') as fd:
            for chunk in file.iter_content(chunk_size):
                fd.write(chunk)
                downloaded_bytes = fd.tell() 
                print (float(downloaded_bytes)/file_size*100)
                self.notify_progress.emit(float(downloaded_bytes)/file_size*100)
                if self._run_semaphore.available() == 0:
                    self._run_semaphore.release(1)
                    break
 
        print "Finish"
        self.finished_thread.emit()

    def stop(self):
        print "Cancel"
        self.finished_thread.emit()
        self._run_semaphore.acquire(1)
 
if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyCustomDialog()
    window.resize(600, 400)
    window.show()
    sys.exit(app.exec_())

Zeile 98 halte ich deshalb für richtig, weil hier ein Signal gesendet wird, der in der GUI-Klasse etwas bewirken soll. Wir sehen in Zeile 34, dass das Signal mit der on_finished()-Funktion verbunden ist. Und in dieser Funktion wird der Prozessbalken einfach wieder auf Null gesetzt, und eine Print-Anweisung wird ausgegeben.

Einzig, was ich nicht verstanden habe, ist, dass man den break-Befehl auch anders gestalten könnte, und gesonderte Abbruch-Signale senden kann. Kannst du mir da bitte auf die Sprünge helfen?

[Hyperion]

Schau Dir doch noch mal genau an, was die Zeilen 91, 94 und 98 bewirken... und zwar bezogen auf den Aufruf der ``stop``-Methode.

Noch etwas: So langen Code lagere doch bitte in ein Paste-bin aus! Man kann den Thread dann nicht mehr so gut lesen und insebsondere beim Antworten ist es doof, weil ich den Code nicht in einem separaten Tab sehen kann

Deine ``check_folder_exists`` ist imho immer noch sinnfrei! Wird die überhaupt aufgerufen?

Der Klassenname ``MyCustomDialog`` ist auch wenig aussagekräftig

[Sophus]

@Hyperion: Du meinst, hier wird zweimal das finished_thread.emit()-Signal gesendet, was unnötig ist. EInmal wird es nach dem break-Befehl und einmal in der stop()-Funktion aufgerufen. Das heißt also, das finished_thread.emit()-Signal hinter dem break-Befehl kann gelöscht werden, und dies in der stop()-Funktion belassen. Aber was meintest du mit dem break? Du meintest, ja nach Anwendung könnte man es anders gestalten, bzw. gesonderte Signale senden. Ich bin neugierig geworden

[Hyperion]

@Sophus: genau falsch herum gedacht! Wenn Du den Thread *nicht* abbrichst, dann soll das Signal doch auch gesendet werden, gel?

Das Problem ist, dass es in der ``stop``-Methode gesendet wird! Das Signal heißt doch ``finished_thread``. Zum Zeitpunkt des Aufrufs von ``stop`` *läuft* der Thread aber ja noch... nun reagieren evtl. angeschlosene Methode auf den Aufruf und erledigen Dinge, die sie noch nicht erledigen sollten. Im schlimmsten Fall kommen sie sich dabei mit dem laufenden Thread in die Quere, der ja noch aktiv ist.

Daher meine ich ja, dass man ein Signal, welches "ich bin zu ende" aussendet nur dann senden darf, wenn das auch der Fall ist, also im Code nur in Teilen der ``run``_Methode, die wirklich inhaltlich deren Ende markieren und nach denen keine externe Interaktion durch den Thread mehr stattfindet. In Zeile 94 ist das ja definitiv der Fall.

Und bezüglich des Einwands mit ``break`` meine ich, dass man stattdesen ja auch ein ``return`` nutzen könnte. Zuvor kann man dann ein *anderes* Signal ``thread_aborted`` z.B. aussenden, welches angeschlosenen Komponenten ermöglicht, auf den *Abbruch* anders zu reagieren als auf den *Erfolg*. Dies ist aber eine Frage des Anwendungsfalls. Wenn es egal ist, ob der Thread erfolgreich bearbeitet worden ist oder nicht, dann reicht ja auch ein Signal.

[Sophus]

@Hyperion: Aaaaaaawwwwwwwww danke. Jetzt wo du es sagst, klingt es logisch. Ich sehe sehr oft den Wald vor lauter Bäumen nicht. Übrigens, da du zu meiner Erklärung:

In Zeile 30-32 wird gesagt, wenn das Attribut _run_semaphore (mittels der available()-Methode) gleich Null ist, also keine n Ressource (n = 1) zur Verfügung steht, dann soll durch die release(1)-Methode dem Attribute _run_semaphore die n Ressource wieder zur Verfügung gestellt werden, und im Anschluss folgt der break-Befehl, also der Abbruch. Warum hier hinterher der Thread erneut gestartet werden kann liegt einzig und allein daran, weil die n Ressource vorhanden ist. In Zeile 11 sehen wir ja, wie dem Attribut _run_semaphore durch die QSemaphore()-Klasse eine n Ressource übergeben wird, also die Zahl 1.

nichts gesagt hast, gehe ich davon aus, dass ich es diesmal richtig verstanden habe?

Aber mir fiel da noch eine weitere Frage ein. Für den Fall, dass das Internet abbricht oder sonstige abrupt eintretende äußere Ereignisse zustande kommen, wollte ich im Thread mit der Try-Except-Ausnahme arbeiten. Aber irgendwie scheint es nicht zu funktionieren. Also, wenn ich spaßeshalber meine W-LAN-Verbindung mitten im Prozess des Herunterladens trenne, sollte schon eine Meldung kommen. Trenne ich vor dem Prozess die Internetverbindung, dann greift die Try-Except-Ausnahme, aber nicht während des Prozesses. Hier! (Pastebin )

[Hyperion]

Übrigens, da du zu meiner Erklärung ... nichts gesagt hast, gehe ich davon aus, dass ich es diesmal richtig verstanden habe?

Ja

Aber mir fiel da noch eine weitere Frage ein. Für den Fall, dass das Internet abbricht oder sonstige abrupt eintretende äußere Ereignisse zustande kommen, wollte ich im Thread mit der Try-Except-Ausnahme arbeiten. Aber irgendwie scheint es nicht zu funktionieren.

Also zum ersten sollte man *nie* ein nacktes ``except`` im Code stehen haben, sondern wirklich nur die Fehler abfangen, die auch wirklich auftreten können bzw. die man an der Stelle wirklich *behandeln* will! Desweiteren was bedeutet denn das "scheint nicht zu funktionieren"? Wie äußerst sich das denn?

Man kann so etwas übrigens simulieren, indem man mal stumpf eine solche erwartete Ausnahme wirft, also etwa so:

Code: Alles auswählen

try:
    # some piece of Code
    # now we simulate an error:
    throw SomeSpecialError("something went wrong")
    # more Code
except SomeSpecialError:
    # react on that special error somehow...

Das kann man mal temporär einbauen, um die Behandlung zu testen. Eleganter sind dann Dinge wie Mock-Objekte, die *gezielt* solche Fehler werfen. Das würde man bei einem Unit Test machen.

Dein Code *im* ``try``-Block ist übrigens *viel zu lang*. Zum einen kann man dann schwer überblicken, was da ausgeschlossen werden soll, zum anderen erhöht so viel Code die Chance, dass man viel mehr potenzielle Quellen drin hat, die Exceptions werfen können, so dass man u.U. dann an der falschen Stelle nach dem Fehler guckt... und ganz allgemein sollte man logische Code-Blöcke (wie eben Methoden und Funktionen) *kurz* halten. (Ich erwähnte "Clean Code" ja wohl schon öfter? Das gibt 's sogar auf deutsch, sollte das Englische Dir Angst machen)

Du musst Dir also nun überlegen, welcher Aufruf nun einen "Internetverbindungsfehler" auslösen könnte. Umgib dann möglichst nur diese Stelle mit einem ``try... except``-Block und überlege Dir dann auch, was Du da *sinnvoll* zur Behandlung tun möchtest. Einfach nur etwas printen ist keine Lösung!

[Sophus]

@Hyperion: Die nackte except-Behandlung habe ich auf die Schnelle hingesetzt, weil ich keine Behandlung für eine fehlerhafte Internetverbindung wusste. Mir ist also klar, dass man nicht nur das nackte except benutzen soll, weil er sonst unüberlegt alle Fehler "verschluckt". Also muss ich mich auf die Suche machen, welche Fehler-Behandlung es bei Internet-Fehler gibt. Hier siehst du den Quelltext.

Im Quelltext habe ich mich dazu entschieden Zeile 85 und 86 in die Try-Except-Behandlung aufzunehmen, denn dieser Teil ist für die Verbindung ins Internet zuständig. Ist man offline und versucht die Datei herunterzuladen, so bekommt man eine Meldung über die QMessagebox. Wie man in Zeile 87 sieht, dachte ich, ich könnte mit dem HTTPError-Fehler arbeiten. Leider greift der Block nicht , so dass ich vorerst wieder auf den nackten except übergegangen bin. Dann ging ich weiter, und setzte einen zweiten Try-Block zwischen Zeile 97 und 111 ein, denn dieser Bereich ist dafür zuständig, dass die einzeln heruntergeladenen chunks zum speichern geschrieben werden. Sobald also die Verbindung abbricht, sollte nach meiner Erwartung eine Messagebox ausgegeben werden. In beiden Try-Blocks werden bei einem Fehler die error_http.emit()-Signale gesendet, die in der GUI-Klasse dazu veranlasst, dass in der on_HTTPError()-Funktion (Zeile 40-48) die Messagebox ausgegeben wird. Jedoch ist hier der Erfolg auch sehr begrenzt. Denn wenn ich mitten im Prozess die Verbindung trenne bleibt der Prozessbalken stehen, keine Fehlermeldung, keine Reaktion, nichts.

[Hyperion]

Python hat doch einen hervorragenden Stacktrace! Entferne einfach die ``except``s und ziehe dann das Kabel... dann solltest Du ja sehen, *welcher* Fehlertyp da geworfen wird und dann kannst Du diesen auch gezielt behandeln.

Alternativ hilft auch durchaus das Lesen der Dokus... da wird man sicherlich etwas dazu finden, welche Exceptions da geworfen werden - evtl. werden ja auch keine geworfen?