Modul zur Standardisierung nebenläufiger Prozesse

[justme1968]

ich denke es sind drei punkte die zusammenspielen und deren lösung sich überlappen:
- blockieren verhindern
- aktives pollen verhindern
  - gestückelte nachrichten

select und und main loop kann die ersten beiden lösen und ermöglicht im child auch gleich noch InternalTimer und lesen von unterschiedlichen quellen sie z.b. anderen geräten und datenquellen

protokoll wie z.b. json würde den dritten punkt lösen.

nicht blockierende sockets würden nur bei 1. helfen. für 2. wäre trozdem select nötig und 3. geht ohne protokoll (egal wie minimal) sowieso nicht da es noch mehr ursachen gibt aus denen eine nachricht nicht vollständig zum lesen verfügbar ist oder mehr als einenachricht im buffer steht.

gruß
  andre

[Dr. Boris Neubert]

Exakt. In fronthem bin ich dazu den Weg gegangen das handle non-blocking aufzusetzen, beim write EWOULDBLOCK abzufragen und einen zusätzlichen Puffer zu pflegen.

Kannst Du mir bitte sagen, in welchen beiden Routinen im Fronthem diese Kommunikation stattfindet?

[Dr. Boris Neubert]

protokoll wie z.b. json würde den dritten punkt lösen.

Bevor ich google oder selber denke: gibt es für die Verarbeitung fragmentierter Nachrichten einen Königsweg? Was meine ich damit? Nehmen wir an, dass wir JSON sprechen. Dann kann es vorkommen, dass halbe, ganze, anderthalbe, ... Nachrichten übertragen werden:

Code Auswählen Erweitern


{ foo: "b
ar"
} { abcd: "hallo" } { foo: "

Simpel lässt sich das lösen, wenn man als Transprotokoll z.B. die Form [Nachrichtenlänge] [n Bytes Nachricht] wählt.

Will man nur JSON ohne Transportprotokoll nutzen, muss man schon parsen und sammeln. bis man ein vollständiges JSON-Objekt gefunden hat. Das wird dann aus dem Nachrichtenstream entfernt und an die höheren Gehirnregionen übergeben, während der Rest gepuffert wird. Das ist aufwendig selbst zu machen, aber vielleicht gibt es dafür ja ein Modul.

Im Moment favorisiere ich, in SubProcess das o.g. simple Protokoll i.V.m. mit dem Vorschlag von Jörg zu nichtblockierenden Handles.

Viele Grüße
Boris

[justme1968]

für json hatte ich bis lang nichts gefunden.

für xml gibt es eine lib die das könnte, die ist aber so schwergewichtig das ich im harmony modul auf ein paar regex für diesen zweck ausgewichen bin. da gibt es aber den vorteil das es keine beliebigen daten bzw. nachrichten sind.

sich das leben mit einem <länge><json> format einfacher zu machen ist der bessere ansatz.

ich würde dieses protokoll mit select und non blocking sockets vorschlagen. das select ist wegen InternalTimer und nicht-busy wait wichtig.

gruss
  andre

[herrmannj]

Kannst Du mir bitte sagen, in welchen beiden Routinen im Fronthem diese Kommunikation stattfindet?

Gern. (Da ist noch nicht alles top, das ist noch Prozess. Im Lehrbuch läßt sich das noch nicht abdrucken)

https://github.com/herrmannj/fronthem/blob/master/FHEM/01_fronthem.pm

Der parent schreibt via fronthem_ipcWrite(@) auf den fork
Der parent liest via fronthem_ipcRead($) vom fork.

Der (bisher eine existierende) fork liest via fronthem_wsIpcRead(@) vom parent
Die Kommunikation vom fork an den parent erfolgt derzeit noch (inkonsequent; eben gewachsen) in den einzelne Routinen, zB in fronthem_wsConnect(@) : my $size = $conn->server()->{ipc}->send($msg."\n",0); (todo:))

Um Dir das lesen zu erleichtern: weil ich sehr auf sicherheit achte speichere ich die PID beim forken und der client muss sich mit der PID authorisieren. So möchte ich verhindern das "böse" Prozesse die Schnittstelle mißbrauchen.

Ich lese Deinen thread sehr aufmerksam mit weil ich gerade dabei bin fronthem für plots zu erweitern. Dazu forke ich weitere Prozesse (lesen aus dem filelog) die eben auch sehr die bidriektionale Kommunikation benötigen, ist im Prinzip exakt die gleiche Aufgabenstellung and der Du arbeitest.

Für das Protokoll verwende ich JSON. Das oben richtig beschriebene Problem mit den abgeschnittenen Nachrichten adressiere ich in dem ich jede vollständige JSON Nachricht mit crlf abschliesse und das in der msg maskiere.

Code Auswählen Erweitern

$rv = $serv->{'ipc'}->recv($msg, POSIX::BUFSIZ, 0);
unless (defined($rv) && length $msg) {
$serv -> shutdown();
return undef;
}
$serv->{buffer} .= $msg;
while (($serv->{buffer} =~ m/\n/) && (($msg, $serv->{buffer}) = split /\n/, $serv->{buffer}, 2))
{
eval {
$msg = decode_json($msg);
};
fronthem_forkLog3 ($serv->{ipc}, 1, "$serv->{id} ipc decoding error $@") if ($@);
fronthem_wsProcessInboundCmd($serv, $msg);
}

vg
jörg

[Dr. Boris Neubert]

Hallo,

ich habe nun die subprocess-Klasse um die folgenden Funktionen erweitert:

Code Auswählen Erweitern

writeToChild()
readFromChild()
writeToParent()
readFromParent()

Die bisherige Funktion read() wurde dabei zu readFromChild().

Ich benutze nicht-blockierende Sockets und select in der bidirektionalen symmetrischen Kommunikation zwischen Child und Parent. Getestet habe ich das über eine Set-Funktion im 98_SubProcessTester.pm, welches per

Code Auswählen Erweitern

set T send laberlaberlaber


einen Text zum Child sendet. Die Set-Funktion kehrt sofort wieder, auch wenn das Child in der onRun-Funktion noch in der fünfsekündigen Warteschleife (sleep 5) steht.

Zerissene Texte sollte es keine geben. Ich habe dazu ein simples Transportprotokoll implementiert, das jeder Übertragung vom Child zum Parent oder vom Parent zum Child die Länge des übermittelten Skalars als 32bit unsigned int (network byte order) übergibt. Auf diese Weise trenne ich das Transportprotokoll von einem etwaigen vom Anwender festzulegendem Kommunikationsprotokoll (z.B. JSON) in der Payload der Übertragung.

Jeder Aufruf von writeTo...() ist eine Übertragung. readFrom...() liefert entweder undef oder den mit writeTo...() gesendeten Skalar vollständig zurück.

Werde als nächstes versuchen, mit Hilfe von subprocess das VirtualBox-API in den für mich benötigten Teilen in FHEM zu integrieren.

Beste Grüße
Boris

[Markus Bloch]

Hallo Boris,

finde ich an sich eine tolle Sache. Das ermöglicht eine einfache non-blocking Kommunikation zu einem Child-Prozess (insbesondere bei dauerhaft laufenden Prozessen).

Jetzt stellt sich mir nur eine Frage: Funktioniert das auch unter Windows? So wie ich das sehe funktioniert das glaube ich nur unter Unix-basierten Systemen (Linux/Mac).

Hintergrund warum ich frage: Ich hatte in der Vergangenheit (besonders in der Anfangszeit von PRESENCE) immer wieder von Problemen bei Windows-Usern gehört. Damals ging es um das generelle Thema "forking unter Windows" was ja aktuell jeder Perl-Interpreter etwas anders implementiert. Der eine als Threads, der andere als neuer Hauptprozess. Ich weis nicht, ob solche Interpreter unter Windows auch UNIX-Sockets können.

Ich nehme mal an, SubProcess.pm ist ausschließlich für Unix-Systeme gedacht, gehe ich da richtig in der Annahme?

Viele Grüße

Markus

[Dr. Boris Neubert]

Hallo Markus,

ich habe beim Lesen der Doku zu fork und socketpair nichts spezielles gelesen, das systemspezifisch sei. Ich habe es aber auch nur unter Linux, genauer Ubuntu 15.04, getestet. Ich meine, dass am Anfang des Threads auch jemand unter OS X erfolgreich war. Windows müsste jemand testen. Freiwillige vor!

Viele Grüße
Boris