Modul SubThread für Threads in FHEM

[zap]

In der Vergangenheit habe ich mit dem Modul SubProcess.pm experimentiert. Leider gab es in Praxis Performance Probleme, wenn sehr viele Daten vom Subprozess an den Parent geschickt wurden. Daher habe ich mir überlegt, es mal mit Threads zu versuchen und für den Datenaustausch nicht wie SubProcess.pm Sockets zu verwenden sondern Shared Memory Queues, die das Modul Thread::Queue bereitstellt. Die Queues fangen hohes Datenaufkommen ab. Die Daten können verarbeitet werden, wenn "Zeit dazu ist".

Das Modul benötigt die Perl Module threads und Thread::Queue.

Ich würde das Modul gerne einchecken, wenn nichts dagegen spricht, um es auch in meinen Modulen HMCCUxxx verwenden zu können. Falls ein Einchecken nicht erwünscht ist (weil es ein Hilfsmodul ist), müsste ich die Funktionalität direkt in meinen Modulen einbauen. Ich denke jedoch, auch andere Entwickler könnten das Modul hilfreich finden.

Die Funktionsweise von SubThread.pm ist an die von SubProcess.pm angelehnt. Hier eine schnelle Übersicht der Funktionen:

new ({ onRun => \&OnRunFunction, onExit => \&OnExitFunction, termsig => "Signal", hash => $hash })

Legt ein neues SubThread Objekt an. Der Parameter onRun ist Pflicht, die anderen sind optional. Die Funktionen On... bekommen beim Aufruf als einzigen Parameter eine Referenz auf das SubThread-Objekt übergeben. Der Parameter termsig legt das Signal fest, das die Funktion terminate() an den Workerthread schickt, z.B. 'INT'. Wenn termsig nicht angegeben wird, hat terminate() keine Funktion. Mit dem Parameter hash kann der Hash eines FHEM Devices übergeben werden. Auf diesen kann in den onRun und onExit Funktionen per $worker->{hash} zugegriffen werden, wobei $worker der Funktionsparameter ist.

Das erstellte Objekt sollte im Master-Prozess (FHEM) z.B. im Hash des aufrufenden Device gespeichert werden, Beispiel:

Code Auswählen Erweitern


$hash->{helper}{thread} = SubThread->new ({ onRun => \&X_WorkerRun, onExit => \&X_WorkerExit, termsig => 'INT', hash => $hash });


workerThread ()

Gibt eine Referenz auf das eigentliche Thread-Objekt des Moduls threads zurück.

isRunning (), isDetached ()

Prüfen ob der Workerthread läuft oder ob er vom Master-Prozess (FHEM) abgekoppelt ist.

run ()

Workerthread starten. Es wird die mit onRun definierte Funktion aufgerufen und anschließend, sofern angegeben, die mit onExit definierte Funktion. Gibt 0 zurück, wenn der Workerthread nicht gestartet werden konnte. Sonst die ID des Workerthreads (Integer > 0).

terminate ()

Schickt das mit terminate definierte Signal an den Workerthread. Dieses Signal sollte in der onRun Funktion abgefangen und die onRun Funktion beendet werden. Beispiel:

Code Auswählen Erweitern


sub MyWorkerThread () {
  my $worker = shift;
  my $hash = $worker->{hash};
  my $run = 1;

  $SIG{'INT'} = sub { $run = 0; };

  while ($run) {
    # do things ...
    my $data = ...
    $worker->sendToMaster ($data);
  }
}


detach ()

Entkoppelt einen Workerthread vom Master-Prozess (FHEM). Nützlich, um Fehlermeldungen beim Beenden von FHEM zu vermeiden, wenn noch Workerthreads aktiv sind. Die Funktion terminate() entkoppelt den Workerthread automatisch.

signal ($sig, $detach)

Schickt das Signal sig an den Workerthread. Wenn detach = 1 ist, wird der Workerthread vom Master-Prozess entkoppelt.

pendingData ()

Prüft, ob Daten in der Queue vorhanden sind. Wenn diese Funktion im Master-Prozess aufgerufen wird, prüft sie, ob Daten vom Workerthread anstehen. Bei Aufruf im Workerthread wird geprüft, ob Daten vom Master-Prozess anstehen. Im Master-Prozess kann pendingData () z.B. in einer X_Ready() Funktion verwendet werden. Voraussetzung ist, dass der Hash des aufrufenden Moduls in den Hash %readyfnlist eingetragen wird. Beispiel:

Code Auswählen Erweitern


sub X_Ready () {
  return $hash->{helper}{thread}->pendingData ();
}

sub X_Read () {
  while (my $item = $hash->{helper}{thread}->receiveFromWorker ()) {
    # Process data
  }
}


sendToMaster ($data)

Wird vom Workerthread aufgerufen und schreibt data in die Queue des Master-Prozesses. Dieser kann die Daten mit receiveFromWorker () lesen.

receiveFromMaster ()

Wird vom Workerthread aufgerufen und liest Daten aus der Queue des Workerthreads, die der Master-Prozess mit sendToWorker () bereitgestellt hat.

sendToWorker ($data)

Wird vom Master-Prozess aufgerufen und schreibt data in die Queue des Workerthreads. Dieser kann die Daten mit receiveFromMaster () lesen.

receiveFromWorker ()

Wird vom Master-Prozess aufgerufen und liest Daten aus der Queue des Master-Prozesses, die der Workerthread mit sendToMaster () bereitgestellt hat.

Die Funktionen sendToWorker() und receiveFromMaster() können auch verwendet werden, um dem Workerthread zu signalisieren, dass er sich beenden soll. In diesem Fall könnte der Master-Prozess ein 'STOP' an den Workerprozess schicken, auf das dieser mit dem Verlassen der onRun Funktion reagiert.

[rudolfkoenig]

Ich habe gerade perlthrtut durchgelesen, und meine urspruenglichen Bedenken gegen Threads in FHEM sind jetzt  groesstenteils weg. Was mir unklar war, dass Variablen nur nach explizitem :shared Schluesselwort gemeinsam sind, alles andere wird kopiert.

Bei den Signalen bin ich noch nicht ganz sicher, es heisst, je nach Implementierung ist nicht festgelegt, welcher Thread die (echten, nicht "thread"-) Signale kriegt. Das wuerde bei den (vmtl. kaum genutzen) SIGHUP zu falschen Ergebnissen fuehren, wenn man sie nicht zum Hauptprozess weiterleitet. Ob es bei ALRM richtig funktioniert, weiss ich nicht, das ist vmtl. auch kaum relevant. Wenn doch, dann aber schwer zu debuggen.

Noch ein Punkt: manche Perl Installation unterstuetzen keine Threads.

Ich habe trotzdem nichts dagegen, wenn es eingecheckt wird, haette aber gerne die Meinung von Anderen gehoert.

[justme1968]

@zap: ganz unabhängig vom für und wieder zu threads und prozessen: auf den ersten blick verstehe ich nicht warum dieses modell so viel schneller sein soll als die lokalen sockets.

nach meinem verständniss:
- die socket version wird in die fhem event loop eingehängt, sobald daten da sind wird ReadFn automatisch aufgerufen. d.h. eingehende daten werden so schnell wie möglich verarbeitet. wenn der rechner zu langsam ist für die menge der daten macht er eventuell nicht mehr viel anderes

- in der thread version werden die daten nicht automatisch verarbeitet sondern man muss irgendwie selber regelmäßig pendingData() aufrufen und dann verarbeiten. daten werden nur im 'pooling' intervall verarbeitet (d.h. potentiell verzögert), es ist natürlich noch zeit für anderes aber wenn der rechner zu langsam ist füllt sich die queue.

stimmt das? oder verstehe ich etwas völlig falsch?

und die lokalen sockets sind doch auch nicht viel mehr als ein shared memory fifo pro richtung bei dem einer schreibt und einer liest.

gruss
  andre

[zap]

@zap: ganz unabhängig vom für und wieder zu threads und prozessen: auf den ersten blick verstehe ich nicht warum dieses modell so viel schneller sein soll als die lokalen sockets.

Nicht schneller. Durch die Verwendung von Shared Memory Queues für den Datenaustausch können aber mehr Daten gepuffert werden, sodass große Datenmengen nicht zu Datenverlust führen. Bei den Sockets habe ich sogar mit diversen TCP Parametern auf Linux Ebene experimentiert. Dadurch konnte ich das Problem etwas entschärfen, jedoch nicht komplett eliminieren. Immer wenn der Child-Prozess eine großen Menge Daten in das Socket-Handle geschrieben hat, kam es zu I/O Fehlern und Daten gingen verloren. Ich hätte alles mühsam zwischenspeichern und nach und nach nochmal an den Parent-Prozess schicken müssen.

- in der thread version werden die daten nicht automatisch verarbeitet sondern man muss irgendwie selber regelmäßig pendingData() aufrufen und dann verarbeiten. daten werden nur im 'pooling' intervall verarbeitet (d.h. potentiell verzögert), es ist natürlich noch zeit für anderes aber wenn der rechner zu langsam ist füllt sich die queue.

Nein, die Daten werden nicht verzögert verarbeitet. In der Thread Version nutzt man den X_Ready()/X_Read() Mechanismus. D.h. das Thread nutzende Modul/Device trägt sich in %readfnlist ein. FHEM ruft in seinem Loop dann die X_Ready() auf. In X_Ready() wird per pendindData() geprüft, ob Daten anliegen. Wenn die Rückgabe wahr ist, ruft FHEM X_Read() auf. Beispiel s.o. bei der Beschreibung von pendingData().

und die lokalen sockets sind doch auch nicht viel mehr als ein shared memory fifo pro richtung bei dem einer schreibt und einer liest.

Im Prinzip ja, jedoch nur mit eingeschränktem Einfluss auf die Größe des I/O Puffers.

[justme1968]

hmmm... ich glaube ich verstehe es immer noch nicht ganz.

zum  einen verstehe ich nicht wie daten verloren gehen können. wenn man auf einer seite etwas in das socket steckt ist garantiert das sie auf der anderen seite ankommen. auf der schreibenden seite muss man bei einem blockierenden socket garnichts weiter tun, bei einem nicht blockierenden muss man schauen ob das socket 'voll' ist und gegebenenfalls warten. bei beidem muss man nicht zwischenspeichern.

und zum anderen hängt die readyFn nicht direkt im select sondern wird je nach platform (windows alle 0.1 sekunden, unix alle 5 sekunden) aufgerufen. unabhängig davon ob daten da sind. d.h. unter linux würden bei einem fhem das keine anderen sockets offen hat die per select überwacht werden würde die readyFn nur alle 5 sekunden aufgerufen. d.h. nicht sobald daten da sind sondern verzögert. das ist im prinzip pollen.

ein socket das direkt in die %selectlist eingehängt wird triggert dagegen die readFn automatisch sofort sobald daten da sind.

wie viele daten sind denn viel?

[zap]

zum  einen verstehe ich nicht wie daten verloren gehen können. wenn man auf einer seite etwas in das socket steckt ist garantiert das sie auf der anderen seite ankommen. auf der schreibenden seite muss man bei einem blockierenden socket garnichts weiter tun, bei einem nicht blockierenden muss man schauen ob das socket 'voll' ist und gegebenenfalls warten. bei beidem muss man nicht zwischenspeichern.

Die Situation ist so: Ich starte einen RPC-Server in einem Subprozess, der Daten von einem externen Gerät (Homematic CCU) empfängt. Sobald der RPC-Server Daten bekommt, muss er sie in den Socket schreiben. Alternative wäre zu puffern. Problem: Direkt nach dem Start des RPC-Servers schickt die CCU Daten für alle ihre Geräte. Wenn man viele Geräte hat, kommen da schnell 7-10 Kb zusammen. Beim Wegschreiben an den Parent per Socket kommt es dann zu Write Errors. Im "Normalbetrieb" bei gelegentlichen Updates von der CCU funktioniert alles.

und zum anderen hängt die readyFn nicht direkt im select sondern wird je nach platform (windows alle 0.1 sekunden, unix alle 5 sekunden) aufgerufen. unabhängig davon ob daten da sind. d.h. unter linux würden bei einem fhem das keine anderen sockets offen hat die per select überwacht werden würde die readyFn nur alle 5 sekunden aufgerufen. d.h. nicht sobald daten da sind sondern verzögert. das ist im prinzip pollen.

ein socket das direkt in die %selectlist eingehängt wird triggert dagegen die readFn automatisch sofort sobald daten da sind.

wie viele daten sind denn viel?

Mist. Das mit den 5 Sekunden wußte ich nicht. In dem Fall müsste ich doch pollen. Oder ich missbrauche %selectlist als Trigger. Hat jetzt aber alles nicht direkt etwas mit dem generellen Ansatz zu tun, Threads in FHEM zu unterstützen.

[justme1968]

nein. hat alles nicht prinzipiell etwas mit threads zu tun.

aber so lange es um einen einzigen parallelen code pfad geht sollte alles genau so einfach auch ohne threads gehen.

der default socket buffer sollte 4k groß sein und sich bis auf 65k vergrößern lassen. das sollte also eigentlich gehen. ist es wirklich so das die daten von der ccu schneller übers netz kommen als fhem sie verarbeiten kann? übers netz kommen die daten doch auch in mehreren häppchen oder?

sind das beim schreiben wirklich fehler? oder nur der hinweis das das socket 'voll' ist und man warten soll? das selber buffern sind zwei zeilen perl.

was passiert wenn du das socket vom child zum parent blockierend machst? dann wird automatisch gewartet.

[zap]

Ich hatte den Puffer vergrößert und auch diverse Timeouts hochgestuft. Richtig zuverlässig hat das alle nicht funktioniert. Die CCU schickt leider direkt nach der Registrierung des RPC Servers alle ihre Devices mit allen Kanälen in einem grossen RPC Request an den Server. Dazu kommt, dass der Server selbst nicht multithreaded ist (RPC::XML::Server Funktion server_loop()). Während der Verarbeitung der Daten nimmt er keine neuen entgegen. Wenn das zu lange dauert wg der Pufferung gibt es Fehler in der CCU. Im Extremfall hört sie sogar auf, weiter Daten zu schicken.

Also alles sehr zeitkritisch. Daher gehe ich derzeit den Umweg über Filequeues. übrigens wird jeRPC Schnittstelle der CCu ein Prozess/Thread gestartet. Das können bis zu 4 sein. Die Umstellung auf Threads hätte 2 Vorteile:

1. Threads benötigen weniger Speicher, da nicht jedes Mal der komplette fhem Prozess geklont wird
2. Ich komme von den Filequeues Weg (ok, ggf kann man Thread::Queue auch ohne Threads einsetzen, habe ich nicht getestet)

Eventuell kann ich auch das Polling durch den FHEM Standard IO ersetzen. Das ist durch die X_Ready Thematik mit den 5 Sekunden allerdings fraglich.

anyway: das eine oder andere Modul nutzt ja bereits Threads. Aber wenn das Hilfsmodul nicht gewünscht ist,  baue ich das eben direkt in mein Modul ein (klingt jetzt vielleicht etwas frustriert, ist aber nicht so gemeint. Ich sehe das wirklich leidenschaftslos)

P.S. das mit den 5 Sekunden macht aber FHEM unter Windows weitgehend unbrauchbar. Da wird ja X_Ready zwingend verwendet, wenn ich das richtig verstanden habe

[Wuppi68]

ich persönlich sehe mehr Vorteile als Nachteile mit der Unterstützung von Threats

mögliche Szenarien sehe ich wie folgt:

Umstellung der IO Devices auf Threats --> weniger Timingprobleme
HTTP Calls --> entspannteres Timing bei langsamen/offline Seiten (Blocking Probleme)
Sonos --> Entfall vom Neustart des Forks, wenn Änderungen am Device vorgenommen wurden

Workaround für Windows evtl. notwendig; Fritzbox wird offiziell nicht mehr unterstützt
Wie Rudi auch schon schrieb, Debugging kann wirklich Freude bereiten

[rudolfkoenig]

der default socket buffer sollte 4k groß sein und sich bis auf 65k vergrößern lassen.

Auf einem Ubuntu VM schauen die Zahlen anders aus (min, default, max):

Code Auswählen Erweitern

$ cat /proc/sys/net/ipv4/tcp_wmem
4096    16384    4194304

Unter OSX scheint es auch groesser zu sein:

Code Auswählen Erweitern

# sysctl net.inet.tcp.recvspace
net.inet.tcp.recvspace: 131072


1. Threads benötigen weniger Speicher, da nicht jedes Mal der komplette fhem Prozess geklont wird

Tippe auf das Gegenteil. Laut perlthrtut werden fuer perl-threads alle Perl-Variablen kopiert (was mich beruhigt ). Beim fork/clone wird COW verwendet, ich vermute, dass bei perl-threads diese Technik (wg. zu teuer, da keine MMU-Unterstutzung) nicht gemacht wird. Zahlen wurden mich wirklich interessieren, wie man sowas "richtig" misst, ist mir aber nicht ganz klar, vlt. reicht ein free.

Aber wenn das Hilfsmodul nicht gewünscht ist

Ich habe das nicht gesagt, und auch (soweit ich es verstehe) andre nicht. Es waere nur nett zu verstehen, warum das Problem mit den herkoemmlichen Mtteln wie SubProcess oder BlockingCall nicht realisiert werden kann, evtl. koennen wir das dann fixen.

P.S. das mit den 5 Sekunden macht aber FHEM unter Windows weitgehend unbrauchbar. Da wird ja X_Ready zwingend verwendet, wenn ich das richtig verstanden habe

Unter Windows wird ReadyFn zum Pruefen der seriellen Schnittstellen verwendet, da diese nicht mit select geprueft werden koennen. Hier ist der Timeout maximal 0.1s. Auf anderen Plattformen ist er maximal 5s, da ReadyFn hier  (etwas vereinfacht) nur zum wiederauffinden ausgesteckter USB-Geraete verwendet wird. Maximal heisst: wenn auf den per select ueberwachbaren Verbindungen nichts passiert.

[justme1968]

ich bin nicht generell gegen threads und ich habe auch nicht prinzipiell etwas gegen ein generelles hilfsmodul.

aber ich sehe threads eher fhem intern für die diversen io devices die timing kritisch sind und um blockieren generell zu vermeiden.

in der reinen kommunikation zwischen instanzen sehe ich nicht warum threads besser sind als zwei prozesse. bzw. wenn sie es tatsächlich ob es auf fhem seite ein problem gibt das wir beheben sollten.

was den speicherverbrauch oder die geschwindigkeit beim forken angeht: ich glaube nicht das dies für aktuelle betriebssysteme auf aktuellen rechner für diesen anwendungsfall (ein permanenter nebenläufiger codepfad) relevant ist.

ohne das es negativ klingen oder zu frustration führen soll: threads haben (manchmal) ihre daseinsberechtigung. aber ich habe schon mehr als einmal gesehen das threads als das allheilmittel mittel angesehen wurden die sie dann nicht waren weil das problem konzeptionell ganz woanders lag. manchmal haben threads das eigentliche problem auch nur zugekleistert und noch schwieriger zu finden gemacht.  sehr vieles ist mit 'altmodischen' und einfacher zu debuggenden methoden genau so möglich. oder den effekt den man durch die threads sieht ist ein zufälliger nebeneffekt der konventionell genau so umsetzbar wäre.

einem wirklich vorteil von threads sehe ich in der kommunikation der haupt instanz mit möglichen io instanzen über shared memory. aber nur wenn das global für alle ios und ohne polling umgesetzt wird. und das ist leider auch genau das szenario das mit potentiellen deadlocks am schwierigsten zu debuggen ist.

[herrmannj]

ich verstehe das buffer problem nicht.

Alles was ich senden möchte landet zu erst in eine perl var.
Von dort aus schreibe ich (so lange möglich) in das handle und lösche entsprechend aus der perl var.
Wenn der dazugehörige IO buffer voll ist bekomme ich ein E_WOULDBLOCK. Dann schreibe ich nichts mehr in das handle sondern warte bis dessen select wieder Bereitschaft meldet.
Die perl var nimmt weiterhin Daten entgegen.

Mit diesem Vorgehen bin ich bisher immer sehr gut gefahren und das gilt doch als etablierte Vorgehensweise. Wo treten denn in dieser Kette die locks auf ?

vg
joerg

[rudolfkoenig]

ich verstehe das buffer problem nicht.

Beispiel: eine Bibliotheksroutine, die mir Daten erst zurueckliefert, wenn sie komplett da sind, und das kann Minuten oder Stunden dauern. Und sie wird sehr ungemuetlich, wenn ich die gelieferten Daten nicht sofort abnehme, (z.Bsp. weil ich noch die alten Daten an FHEM weitergebe, und das dauert, da FHEM gerade Plots rechnet), dann bricht die andere Seite der Bibliotheksroutine die Verbindung ab.

Da faellt mir nur ein einen weiteren Thread zu starten, um die Daten langsam an dem Haupt-FHEM weiterzugeben, damit der andere Thread die Bibliotheksroutine bedienen kann. Man diesen Extra-Schreib-Thread sparen, wenn die Daten im TCP-Puffer passen (je nach OS 100kb-4mb)

[justme1968]

wobei das mit subProcess oder Blocking genau so geht. forken, (stunden) warten bis die daten da sind und dann per socket langsam an fhem übergeben. threads nicht nötig.

[zap]

ich verstehe das buffer problem nicht.

Alles was ich senden möchte landet zu erst in eine perl var.
Von dort aus schreibe ich (so lange möglich) in das handle und lösche entsprechend aus der perl var.
Wenn der dazugehörige IO buffer voll ist bekomme ich ein E_WOULDBLOCK. Dann schreibe ich nichts mehr in das handle sondern warte bis dessen select wieder Bereitschaft meldet.
Die perl var nimmt weiterhin Daten entgegen.

Mit diesem Vorgehen bin ich bisher immer sehr gut gefahren und das gilt doch als etablierte Vorgehensweise. Wo treten denn in dieser Kette die locks auf ?

vg
joerg

Das verhält sich im vorliegenden Fall so:

- RPc Server Subprozess akzeptiert eingehende Verbindung und nimmt Datenpaket entgegen
- Nun versucht er, die Daten an den FHEM Hauptpozess weiter zu geben (write auf socket)
- wichtig: während dieser Datenweitergabe ist der RPc Server blockiert und nimmt keine weiteren Daten entgegen
- jede verzögerte Weiterleitung der Daten erhöht die Gefahr, dass der Sender aufgibt oder Daten verwirft.

Wie Rudi geschrieben hat: einzige Möglichkeit ist in dem Fall, die Weiterleitung in einen weiteren Prozess auszulagern. Oder eben meine Löshng, die Daten nicht in die IO Schleife einhukippen sondern erst mal in eine Queue zu schreiben.

Ich habe vor einigen Monaten wirklich lange mit SubProcess experimentiert, auch mit Unterstützung von Boris. Mit Änderung von Timeouts und Puffergrössen. War leider nicht erfolgreich. Natürlich würde das Problem mit Threads auch auftreten, daher die Verwendung von den Shared Memory Queues quasi als Ersatz für die IO Puffer des Betriebssystems