Der Fortschritt in Datenflüssen

Datenflüsse sehen einfach aus, sind es auch in gewisser Weise – dennoch haben sie es in sich. Da habe ich neulich gemerkt, als ein Programm, das ich mit Flow-Design locker entworfen hatte, sich dann doch nicht so verhielt, wie erwartet. Es hat funktioniert, das war kein Problem. Aber sein Output war irgendwie, hm, strange.

Deshalb will ich hier einmal die Frage beleuchten, wie denn eigentlich die Verarbeitung in Datenflüssen fortschreitet. Oder besser: Welche unterschiedlichen Fortschrittsweisen kann es denn geben?

Ein Beispielszenario soll die unterschiedlichen Fortschrittsweisen vergleichbar machen. Es ist ganz einfach:

Nachrichten, die von einer Funktionseinheit verarbeitet werden, bezeichne ich mit deren Namen: a(x) kommt “von draußen” und wird von a() verarbeitet, b() erzeugt c(y), die von c() verarbeitet wird usw. Nachrichten tragen also das Ziel und nicht die Quelle im Namen. Für das Beispiel reicht das als Identifikation.

Ausgehen von einer Nachricht a() erzeugen die Funktionseinheiten nun diese Nachrichten:

• a(1)
• b(11)
• c(111)
• e(1111)
• e(1112)
• d(111)
• c(112)
• e(1121)
• b(12)
• d(121)
• d(122)
• c(121)
• b(13)
• c(131)
• e(1311)
• e(1312)
• d(131)
• c(132)
• e(1321)
• d(132)
• d(133)
• c(133)
• e(1331)
• e(1332)
• d(134)

Dieser Baum beschreibt Input-Output-Zusammenhänge, z.B. Input b(12) an b() führt zum Output d(121), d(122) und c(121). Das bedeutet auch, das c(121) nach d(122) erzeugt wird.

Allerdings steckt keine Aussage darüber in dem Baum, wann die Nachrichten verarbeitet werden. Verarbeitet c() die Nachricht c(132) während b() noch weiteren Output generiert oder erst nachdem d(134) ausgegeben wurde?

Darauf geben die Fortschrittsweisen Antwort.

Depth-first

Die naheliegende Vorstellung vom Fortschritt der Verarbeitungsweise des Flows ist wohl, dass jede Nachricht sofort verarbeitet wird. Auf einen Zeitstrahl aufgetragen sieht das so aus:

Hier spiegelt sich der Baum direkt wider. So würde die Verarbeitung auch verlaufen, wenn der Datenfluss als Call Stack interpretiert würde, also die Funktionseinheiten sich geschachtelt aufriefen.

Eine Schachtelung im Sinne einer Servicehierarchie soll ja aber gerade mit Flow-Design vermieden werden. Die Funktionseinheiten a()..e() sollen sich nicht kennen; c() soll nicht von e() abhängen, b() nicht von c() und d(), a() nicht von b().

Alternativ kann diese Fortschrittweise aber auch mit Event-Based Components (EBC) erzielt werden: Wenn a() mit b(11) einen Event feuert, dann arbeitet b() als Event-Handler den erst ab, bevor a() dazu kommt, b(12) auszugeben. Wenn b() währenddessen c(111) erzeugt, arbeitet c() die Nachricht erst ab, bevor b() d(111) erzeugt usw.

Alle Nachrichten werden also erstens sequenziell verarbeitet, d.h. nacheinander und auch noch streng in der Reihenfolge, in der sie erzeugt wurden. Und zweitens werden sie synchron verarbeitet, d.h. während der Verarbeitung wartet die erzeugende Funktionseinheit.

Die synchron-sequenzielle Verarbeitungsweise geht depth-first vor. Die Verarbeitung jeder Nachricht fließt sofort soweit nach rechts durch wie möglich.

Breadth-first

Ist der depth-first Fortschritt der richtige, der beste, der einzig wahre Fortschritt für Datenflüsse? Ich glaube nicht. Er mag der naheliegendste sein, doch das scheint mir für eine Bewertung zu wenig. Denn warum soll es richtiger sein als etwas anderes, Nachrichten sofort in der Tiefe zu verarbeiten und dabei Quellfunktionseinheiten darauf warten zu lassen?

Ich denke, zunächst einmal gleichberechtigt ist die breadth-first Verarbeitung von Nachrichten:

Die hervorgehobenen Nachrichten zeigen den Unterschied gegenüber dem depth-first Fortschritt. a() erzeugt die Nachrichten b(11), b(12) und b(13) und die werden zuerst komplett abgearbeitet, bevor deren Output dran kommt. Und auch der wird zuerst abgearbeitet, bevor sein Output dran kommt usw. Die am weitesten rechts stehende Funktionseinheit e() erhält also hier als letzte Arbeit, weil sie am tiefsten im Baum liegt als der der Fluss angesehen werden kann, wenn man ihn um 90°dreht.

Die Verarbeitung eilt damit nicht mehr bei jeder Nachricht zum Ende der Verarbeitung, sondern schreitet sequenziell über alle “Flussarme” hinweg fort. Ich stelle mir das als eine breite Welle vor.

Vorteil von depth-first ist, dass nach Anstoß eines Flows schnell erste Ergebnisse am Ende heraustropfen. Das bedeutet aber nicht, dass die Verarbeitung von weit vorangeschritten ist. Bei breadth-first hingegen können Sie sicher sein, dass Arbeitsschritte abgeschlossen sind, wenn ihre Ergebnisse verarbeitet werden.

Das fühlt sich für mich mehr nach Datenfluss an: Ein Input kommt bei einer Funktionseinheit “auf dem Tisch”, wird verarbeitet, dabei wird Output erzeugt – und wenn das alles fertig ist, dann geht es bei der nächsten Funktionseinheit weiter.

Zumindest empfinde ich das als “fluss-mäßiger”, wenn ich die Verarbeitung als synchron denke. Weder findet Verarbeitung auf mehreren Threads innerhalb einer Funktionseinheit statt, noch arbeiten mehrere Funktionseinheiten parallel. Wenn Funktionseinheiten ihre Arbeit abschließen können, bevor ihr Output verarbeitet wird, dann sind sie auch hübsch unabhängig von einander.

Synchrone Verarbeitung ist für mich der default beim Flow-Design. Sowohl Flow-Design Implementierungen mit EBC wie mit der Flow Runtime folgen dem auch – auch wenn sie sich in der synchronen Verarbeitung unterscheiden, wie Sie hier sehen.

Dennoch war ich damit nicht zufrieden. Denn dieser breadth-first Fortschritt hat sich in der eingangs erwähnten Anwendung als etwas merkwürdig angefühlt. Warum?

Solange am Anfang eines Flusses nur eine Nachricht steht, macht breadth-first kein Problem. Dann läuft die große Welle langsam in die Tiefe.

Falls auf a(1) jedoch noch a(2), a(3) usw. folgen und a(i) nicht komplett in der Tiefe verarbeitet ist, bevor a(i+1) eintrifft, kann es zum Stau kommen [1]. Es geht dann zwar alles ganz gerecht zu im Sinne sequenzieller Verarbeitung. Doch solche Gerechtigkeit ist nicht in allen Fällen wünschenswert. Manchmal wäre es gut, wenn Nachrichten einander überholen könnten – zumindest wenn sie in unterschiedlichen Flussarmen fließen. Warum muss ein d(2…) auf ein e(1…) zwangsläufig warten?

Round-robin

Angesichts des merkwürdigen Verhaltens der Anwendung habe ich einen Mittelweg zwischen depth-first und breath-first Verarbeitung gesucht. Eingefallen ist mir eine round-robin Verarbeitung von Nachrichten. Um das zu verstehen, hier die grundsätzliche Arbeitsweise der Flow Runtime:

Nachrichten kommen von außen zur Flow Runtime, die sie asynchron verarbeitet. Nachrichten, die bei der Verarbeitung entstehen, fließen entweder hinaus, weil sie Endergebnisse darstellen – oder sie fließen zurück in die Runtime, um von folgenden Funktionseinheiten verarbeitet zu werden. a(i) ist eine Nachricht, die von außen zur Runtime kommt. b(i) usw. sind Nachrichten, die die Runtime quasi an sich selbst schickt.

Jede Nachricht, die bei der Runtime eintrifft, wird auf deren einzigem Thread abgearbeitet; das symbolisiert der Kreis in der Funktionseinheit. Sie ist insofern autonom gegenüber ihrer Umwelt.

Immer wenn eine Funktionseinheit eine Nachricht verarbeitet hat, schaut die Runtime nach, ob weitere Nachrichten zur Verarbeitung anliegen. Die stehen in einer Queue, über die die Runtime in einer Schleife läuft. Der Inhalt dieser Queue sieht für das Beispiel über die Zeit so aus (von links wird angehängt, von rechts entnommen):

Das erklärt die breadth-first Verarbeitung: Jede Funktionseinheit wird für eine Nachricht abgearbeitet und stellt ihren Output ans Ende der Queue. Der kommt dann erst dran, wenn der Output vorheriger Funktionseinheiten verarbeitet wurde.

Dieses System habe ich nun aufgebrochen, indem nun jede Funktionseinheit eine eigene Queue besitzt:

Über diese vielen Queues läuft nun die Flow Runtime im round-robin Verfahren. Das bedeutet, für jede Nachricht geht sie eine Queue weiter. Es entsteht folgendes Muster:

Sie sehen, die Verarbeitung wird Nachricht für Nachricht gleichmäßig über die Funktionseinheiten verteilt. Die Verarbeitungsreihenfolge hat im Grunde nichts mehr mit der Tiefe einer Funktionseinheit im Fluss zu tun. Wo Output erzeugt wird, da wird er auch abgearbeitet.

Käme nun ein a(2) zwischendurch an, so würde es alsbald zur Verarbeitung gebracht, wenn seine Queue an der Reihe ist. Es müsste nicht warten, bis alles, was vorher schon aufgelaufen war, abgearbeitet ist.

Dieses Verfahren scheint mir noch gerechter als breadth-first. Es hat allerdings eine Besonderheit, derer man sich bewusst sein muss: Aufs Ganze betrachtet, erfolgt die Abarbeitung der Nachrichten nicht mehr notwendig streng in Erzeugungsreihenfolge. Nachrichten können einander überholen: e(1112) wird zum Beispiel vor b(13) verarbeitet.

Bei depth-first Fortschritt ist b(13) noch nicht erzeugt, wenn e(1112) abgearbeitet wird. Bei breadth-first wurde b(13) erzeugt und schon abgearbeitet lange vor e(1112). Bei round-robin jedoch steht b(13) noch unverarbeitet in der b()-Queue, während e(1112) schon in Arbeit ist.

Asynchron im Kreis

Auch bei round-robin findet innerhalb der Flow Runtime noch keine Parallelverarbeitung statt. Trotzdem geht es überall voran, sobald die Runtime Gelegenheit hat, eine Nachricht zu verarbeiten. Das ist kein pre-emptive Multitasking, weil ja jede Funktionseinheit so lange an einer Nachricht herumlaborieren darf, wie sie mag. Insgesamt auf den ganzen Fluss gesehen, fühlt es sich dennoch so an, als würde quasi parallel gearbeitet.

Richtig ernst wird das, wenn einzelne Funktionseinheiten asynchron arbeiten. Dann kann der Output während ihrer Laufzeit von der Runtime schon weiterverarbeitet werden.

Beispielhaft setze ich mal b() auf asynchrone Verarbeitung, damit Sie sehen, wie sich das Muster dann verändern könnte. Die Länge der Nachrichtenkästen soll nun die Verarbeitungsdauer andeuten.

 

Jetzt kommt es natürlich auch darauf an, wann b() Output erzeugt. c(111) wird parallel abgearbeitet und erzeugt e(1111). Währenddessen arbeitet b() weiter! Wann fließt dort aber d(111) heraus? Während c() am Werk ist und e(1111) generiert oder erst später? Denn danach richtet sich, ob auf c() unmittelbar e() folgt wie im Bild oder zuerst d().

Fazit

Die Verarbeitung in Datenflüssen ist anders als die in Servicehierarchien. Anders, doch deshalb nicht schlechter. Sie müssen sich umgewöhnen. Das mag schwer fallen, weil die “Stack-Denke” so tief in uns allen drin steckt. Doch ich meine immer noch, dass sich das lohnt.

Denn anders bedeutet hier chancenreich. So strange das Verhalten der eingangs erwähnten Anwendung war, es hat mich wieder beeindruckt, wie leicht die Flow-Operationen zu testen waren, weil sie unabhängig von einander sind. Und es war ganz leicht, individuell für jede zu entscheiden, ob sie synchron oder asynchron laufen soll.

Und letztlich finde ich es auch gut, überhaupt die Wahl zu haben zwischen Verarbeitungsweisen. Die könnte eine Runtime womöglich sogar zur Auswahl anbieten. Sogar den depth-first Fortschritt hatte ich schon einmal implementiert.

Fußnoten

[1] Dass weitere Nachrichten vor Abarbeitung beim Fluss eintreffen, setzt natürlich bei aller Synchronizität seiner Funktionseinheiten voraus, dass der Fluss als Ganzes gegenüber seiner Umwelt asynchron arbeitet. Das ist bei der Flow Runtime der Fall.