Das deutsche Python-Forum

Es gibt ein Simulationsprogramm, dessen Läufe von einem in Python geschriebenen Framework mittels subprocess gestartet wird. Anschließend wird die Ausgabe der Simulation über stdout geparst und das Ergebnis in eine Datenbank geschrieben. Das Framework ist mittels RPC-Aufrufen steuerbar. Es nimmt eine Befehlgruppe entgegen und führt dann die Befehle parallel aus.
Eine Webanwendung greift auf die Daten aus der Datenbank zu und stellt sie dem Benutzer zur Verfügung.

Zur Hierachie des RPC-Servers:
Der RPC-Server nimmt neue Jobs entgegen und packt sie in eine Queue. Ein (in seinem eigenen Thread laufender) Taskmanager entnimmt Aufträge aus der Queue und erstellt zu jeder Auftragsgruppe einen ResultHandler, und für jede Aufgabe einen neuen Task (der dann den Popen-Aufruf vornimmt). Sowohl ResultHandler als auch die Tasks laufen in ihrem eigenen Thread. Der ResultHandler verfügt über eine Queue, in die die Tasks nach beenden ihreres Jobs das Ergebnis packen. Der ResultHandler entnimmt die Ergebnisse, trägt diese in die Datenbank ein und verschickt am Ende eine Email.

Das Problem:
Es soll nun möglich sein, die Ausgabe des ``Popen``-Aufrufs live und asynchron mitverfolgen zu können. Das heißt, man kann sich dann in eine Ausführung nachträglich dazuschalten und die Ausgabe mitgeteilt bekommen. Das ganze möchte ich nun möglichst leichtgewichtig implementieren.

Meine Idee wäre jetzt einen neuen Handler zu erstellen, der mit einer neuen Queue die Zeilen der Ausgabe von den Tasks erhält und sich dann um das Broadcasting kümmert. Gibt es vielleicht einen eleganteren Weg? Ich weiß nicht, wie man am Besten mit hängenden Verbindungen umgehen soll. Zumal eine Zeile zwischen Millisekunden und mehreren Stunden generiert werden kann.

Weiss nicht, ob ich genau verstanden habe, was Du vorhast - die Sache klingt für mich nach Ausgabeumleitung auf mehrere Ziele ala Shellbefehl tee.

"Live"-Dazuschalten sollte recht einfach realisierbar sein, z.B. über eine Queue, die mittels Setzen eines Events beschrieben oder nicht beschrieben wird (innerhalb des Tasks, noch vorm Parsen). Das Event könntest Du einfach von Deinem Ausgabehandler aus triggern und auch die Queue dort auslesen. Dabei entstehen keine "hängenden" Verbindungen, solange Du sicherstellst, dass die Queue nicht vollaufen kann. Letzteres kannst Du verhindern, indem Du die Queue auf maxsize=0 setzt (und wartest bis der Speicher vollgelaufen ist ) oder schnell genug die Queue auslesen kannst, damit Dein Ausgabehandler nicht zum Flaschenhals wird (passiert z.B. auf der Linuxkonsole mit Programmen, die sehr viel Output generieren und die Konsole nicht hinterherkommt und so das Progamm ausgebremst wird).

Der asynchrone Fall ist schwieriger zu realisieren bzw. hängt von den Umständen ab. Problem ist, dass Du hier den Output irgendeiner Form zwischenspeichern musst in und zwar für alle laufende Tasks. Einfachste Variante mit vollständiger Historie (zurück bis Task-Start) wäre hier wohl das Rausschreiben des Ausgabestromes in getrennte Dateien, durch die Du später nach Belieben "seeken" kannst. Bist Du nicht an der kompletten Historie interessiert oder ist die mittlere Ausgabe * Tasks << RAM, kannst im Speicher arbeiten mit einem entsprechenden Puffer, z.B. Queue mit Größenbegrenzung (maxsize=0 wiederum auf eigene Gefahr ).

Ob das Ganze einen eigenen Ausgabehandler rechtfertigt, kann ich Dir nicht beantworten. Theoretisch könntest Du den Tasks einen HTTP-Server spendieren, der bei einkommender Verbindung die Ausgabe rausschickt.
Im Ernst, das hängt von den Gegebenheiten und Anforderungen ab. Deinen Ausführungen zufolge wäre der modulare Ansatz mit eigenem Handler wahrscheinlich der richtige Weg.

Edit:
Was ich vergessen habe - im asynchronen Fall mit beschränkter Historie ist es wahrscheinlich sinnvoll, einen Ringpuffer vor die Queue-Weitergabe zu hängen, z.B. mit deque, um das "Hängen" der Tasks zu vermeiden.

@jbs: Vielleicht wäre auch ein Messaging-System zumindest ein Teil einer Lösung!? Irgendwas zwischen ZeroMQ und RabbitMQ…

Danke für euere Antworten.

Ich habe jetzt eine neue Queue eingeführt, in die die Ausgabe geleitet und an anderer Stelle ausgewertet wird. Arbeitsspeicher ist zum Glück kein Problem, da ist (wirklich) genug vorhanden, und die Anzahl der Tasks ist beschränkt.