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.

IronMQ – Simple Warteschlangen in den Wolken

Wie können Sie Anwendungsteile kommunizieren lassen, wenn sie allemal über Netzwerke verteilt sind. Klar, irgendwie geht das mit WCF und Azure… Aber warum so kompliziert?

Stefan Lieser und ich verbringen gerade einige Tage in unserem jährlichen Clean Code Advisors Retreat, um das vergangene Jahr zu reflektieren und das nächste zu skizzieren. Neben viel Zeit zum Reden und Denken darf da aber natürlich auch die Programmierung nicht zu kurz kommen. Also haben wir uns eine Application Kata ausgedacht: AppZwitschern. Dabei geht es darum, Tweets mit einem Desktop (oder Mobile) Client zu erfassen und erst zu einem späteren Zeitpunkt zu versenden. Selbstverständlich muss der Client dafür nicht offen gehalten werden. Die terminierte Versendung übernimmt ein Server.

Das ist eine schöne Aufgabe, in der wir die neue Flow Runtime NPantaRhei einsetzen. Macht Spaß, funktioniert gut. Darüber hinaus kommen einige Technologien zum Einsatz, z.B. Twitter, bit.ly, NCron und auch eine Kommunikationstechnologie. Die Clients müssen ja dem Server die Tweets zur späteren Versendung schicken.

Zuerst haben wir dafür Amazon SQS benutzt. Das ist Cloud Queue-Service. Der funktioniert. Unsere verteilte App zwitschert damit verlässlich und skalierbar. Alle Clients schreiben in dieselbe SQS Queue. Und beliebig viele Server-Instanzen verarbeiten die Versandaufträge. Perfektes scale-out.

Aber Amazon SQS hat einen recht umständlichen API. Und interessanterweise sichert SQS nicht zu, die Nachrichten in einer Warteschlange in der Reihenfolge abzuliefern (dequeue), in welcher sie eingestellt wurden (enqueue). Das ist zwar für unser Szenario nicht so wichtig, dennoch irgendwie merkwürdig. Lohnt sich da wirklich eine nähere Beschäftigung mit diesem Service?

Auch wenn wir für AppZwitschern nicht mehr Leistung brauchen, habe ich dann aber mal geschaut, ob es Alternativen gibt. Und tatsächlich, die gibt es. Hängengeblieben bin ich ganz schnell bei IronMQ von iron.io.

Attraktiv war deren Anspruch an Einfachheit:

• Einfache Anmeldung ohne Kreditkartenangabe, da 250.000 Nachrichten pro Monat ohnehin frei sind. Sie wollen es Interessenten also leicht machen, IronMQ auszuprobieren.
• Einfacher API auf für .NET. Der ist als Quelle bei github zu finden: iron_mq_dotnet. Unter downloads findet sich allerdings auch eine binäre Version.

Und IronMQ bietet, was SQS nicht hat: echte FIFO-Semantik.

Also habe ich alternativ zu unseren SQS-Operationen für die AppZwitschern-Flows welche für IronMQ aufgesetzt. Das war in 20 Minuten gemacht. Hier unser IronMQ-Adapter, der simples Enqueue und Dequeue so anbietet, das es leicht bei der Flow Runtime zu registrieren ist:

public class IronMQOperations
{
    private readonly Client _client;
    private readonly Queue _queue;
 
    public IronMQOperations(string queueName, string projectId, string token)
    {
        _client = new Client(projectId, token);
        _queue = _client.queue(queueName);
    }
 
 
    public void Enqueue(string data)
    {
        _queue.push(data);
    }
 
 
    public void Dequeue(Action<string> onDataReceived)
    {
        do
        {
            var msg = _queue.get();
            if (msg == null) break;
 
            onDataReceived(msg.Body);
 
            _queue.deleteMessage(msg);
        } while (true);
 
        onDataReceived(null);
    }
}

Ich würde sagen, einfacher gehts kaum. Mit den paar Zeilen ist eine verlässliche Kommunikation via Cloud möglich. Die braucht ihre Zeit; Performance wie mit TCP im lokalen Netz ist da nicht zu erwarten. Aber das spielt für unser Szenario und viele andere keine Rolle. Es geht darum, serverseitig skalieren zu können.

Der SQS-Adapter hat knapp doppelt soviele LOC – und bietet eben nicht das, was man von einer Queue erwartet.

Nun bin ich gespannt auf die .NET Bindings für die anderen iron.io Dienste: den Key-Value-Store (Cache) und serverseitige Prozesse (Worker). Einstweilen habe ich in meinem Köcher einen no-brainer für die Verteilung von Arbeitspaketen.

PS: Der Support bei iron.io ist gut. Ich habe ein paar Fragen gehabt und den angebotenen Live Chat benutzt. Da war zu jeder Tageszeit jemand zu erreichen, der entweder Auskunft geben konnte oder dafür gesorgt hat, dass mir später geholfen wurde. I like that.

Asynchrone Kommunikation mit EBCs statt “Async-Pattern”

Freut mich, dass Thomas Christian im build broken Blog eine Lanze für die asynchrone Kommunikation mit Events gebrochen hat. Schade allerdings, dass der Code dafür so umständlich aussieht.

Kommunikation à la Async-Pattern

Ein kleines Formular fordert von einem Calculator die Quadratur einer Zahl an: m_calculator.CalcAsync(number). Was würden Sie dann im Calculator für Code erwarten? Irgendwo sollte number * number stehen. Klar. Und ansonsten? Schauen Sie einmal, wie der Calculator bei Thomas aussieht:

Puh… Da ist es gar nicht so leicht, das zu finden, worauf es eigentlich ankommt, finde ich. Sehen Sie die kalkulierende Businesslogik? Und was macht der Rest des Codes? Der ist dazu da, ein Pattern zu implementieren.

Wenig Code für das Wesentliche, viel für Infrastruktur – das hört sich für mich nicht so gut an, auch wenn ich das Kommunikationsmodell voll unterstütze.

Ich schlage daher einen anderen Weg vor. Hier meine Version der Szenarios:

Kommunikation à la Event-Based Components

Zunächst die Businesslogik:

Hier finden Sie keine Rauschen durch Infrastrukturcode. Der Fokus ist ganz klar auf der Problemdomäne. Aber Achtung: Ich habe keine Funktion für Calc() geschrieben, sondern eine Prozedur, d.h. eine Methode ohne Rückgabewert. Das Resultat liefert die Methode dann über einen Event: DeliverCalcResult(). So ist das bei Event-Based Components.

Jetzt der Client-Code im Formular:

Auch hier kein Rauschen durch Infrastruktur. Nicht mal der Calculator wird als Komponenteninstanz injiziert. Das Formular kennt keinen Calculator. Es ist völlig unabhängig von einer solchen Funktionseinheit. Es äußert Berechnungswünsche lediglich über einen Event – CalcRequest – und empfängt Berechnungsresultate mit OnCalcResult().

Das ist im Prinzip sehr ähnlich wie bei Thomas. Im Detail ist es jedoch anders, da das Formular eben auf die Kenntnis einer Service-Funktionseinheit verzichtet. Die wird bei Thomas noch im Formular instanziert.

Und wie kommt es nun zu Berechnungen, wenn das Formular den Calculator nicht kennt? Beide Funktionseinheiten werden von außen verdrahtet. Damit sind Logik und Abhängigkeiten als Concerns sauber getrennt.

Voilà! Jetzt kann das Programm starten und berechnet Quadratzahlen. Zuerst werden die Funktionseinheiten instanziert. Dann werden die Eventhandler in die Events gesteckt. Mehr ist nicht nötig. Keine Funktionseinheit weiß von der anderen. Nur der Verdrahtungscode kennt sie.

Asynchrone Kommunikation

Die kleine Anwendung funktioniert nun zwar, aber synchron. Thomas Beispiel will ja aber gerade zeigen, wie einfach asynchrone Kommunikation mit Events ist. Was muss ich also tun, um meinen Nachbau ebenfalls asynchron zu machen?

Asynchronizität erfodert keine (!) Veränderung an der Geschäftslogik oder im Frontend. Das halte ich für den wesentlichen Vorteil meines Vorgehens. Ich muss lediglich die Verdrahtung anpassen:

Ich instanziere zwei kleine Standard-Funktionseinheiten, eine, die Events asynchron weiterleitet und eine, die Events zurück in den GUI-Thread bringt. Diese Funktionseinheiten schalte ich zwischen den Dialog und den Calculator. War die Verbindung z.B. bisher direkt

dlg.CalcRequest = calc.Calc;

so ist sie nun indirekt:

dlg.CalcRequest = asyncProcessAsynchronously;
async.OnAsynProcessing = calc.Calc;

Die Standard-Komponenten sind natürlich auch Event-Based Components, d.h. Input erhalten sie über Eventhandler und Output generieren sie mittels Events.

Das war´s. Mehr ist nicht zu tun. Die Geschäftslogik läuft jetzt im Hintergrund. Und die Ergebnisse kommen problemfrei im Vordergrund an. Wie das funktioniert? Egal :-) Das ist die Magie der Standard-Komponenten.

Fazit

Asynchrone Kommunikation mit Events ist eine gute Sache, da stimme ich Thomas zu. Aber man sollte diese Sache dann auch so einfach wie möglich halten. Das bedeutet für mich, dass ich in der Geschäftslogik (oder auch im Frontend) von technischen Klimmzügen, die die Asynchronizität herstellen, nichts sehen will. Geschäftslogik muss frei von diesem Concern sein. Der steckt am besten in wiederverwendbaren Standard-Komponenten. Event-Based Components bieten dafür eine gute Grundlage.

P.S.

Na, schön. Hier nun doch noch der Code für die Standard-Komponenten. Damit keiner sage, der sei bestimmt total kompliziert ;-)

Dieser Code ist wiederverwendbar. Er folgt nur dem Event-Based Component Muster für die Kommunikation. Deshalb kann er mit beliebigen Pins von anderen Komponenten zusammengesteckt werden, ohne dass die etwas davon merken. Das ist Composability.

Einfacher im Gleichschritt – Dienstaufrufe mit der CCR synchronisieren

In der dotnetpro 3/2010 gehen Tobias Richling et al. in einem Artikel der Frage nach, wie asynchrone WCF-Dienstaufrufe synchronisiert werden können. Sie überlegen also, wie ein Client, der Service S1 und Service S2 asynchron (!) aufruft, es so einrichten kann, dass er erst weiter macht, wenn beide (!) Services ein Resultat geliefert haben.

Bei synchronen Aufrufen stellt sich diese Frage nicht. Da geht es ohnehin erst weiter im Client, wenn beide Aufrufe zurückgekehrt sind.  Hier das Beispiel dazu aus dem Artikel:

Im Client geht es erst nach der letzten Zeile weiter, wenn sowohl GetArticleList() wie auch GetSpecialOfferArticle() synchron aufgerufen wurden.

Symptomkur

Was aber, wenn beide Aufrufe asynchron sind? Bei Silverlight ist das gar nicht anders möglich. Und bei anderen Client-Plattformen sollten Sie auch darüber nachdenken, denn asynchrone Aufrufe bringen Performancevorteile. Sie laufen ja parallel, egal ob der Client mit einem oder mehreren Prozessorkernen ausgestattet ist.

Im Artikel sieht ein asynchroner Aufruf mit WCF so aus:

Nicht ganz trivial, oder? Und davon gibt es zwei – die dann eben, wenn nacheinander ausgeführt, zum Problem führen: Der Client muss irgendwie dafür sorgen, dass er auf die Ergebnisse von proxy.GetArticleListAsync() und proxy.GetSpecialOfferArticleAsync() wartet, bevor er weiter macht.

Wie das gehen kann, dazu macht sich der Artikel Gedanken und stellt am Ende eine allgemeine Lösung vor. Mit der sieht das Warten dann so aus:

Man registriert einen Eventhandler (AllComplete), der gerufen wird, wenn schließlich beide Dienstaufrufe zurückgekehrt sind, registriert die Dienstaufrufe (AddServiceCall()), die getätigt werden sollen und startet sie schließlich (ExecuteAll()). Das alles macht möglich ein sog. ServiceCallSynchronizer.

Das sieht auf den ersten Blick doch ordentlich aus. Auf den zweiten schleicht sich bei mir aber zumindest kleines Unbehagen ein, weil die Dienstaufrufe nicht typsicher registriert werden. Die Methodennamen sind als Zeichenketten anzugeben. Ob es eine Methode gibt, stellt sich also erst zur Laufzeit heraus. Ebenso gibt es keine Argumentprüfung zur Compilezeit, weil ja AddServiceCall() nicht weißt, ob und welche Parameter eine Dienstmethode hat. Das ist beides nicht schön.

Wer mit WCF arbeiten muss, mag den Autoren aber dennoch danken. Sie lösen mit Liebe ein Problem, das ansonsten ein Projekt ins Stocken bringen kann. Immerhin umfasst der allgemeine Aufrufsynchronisierer einige Dutzend Zeilen, über die Sie sich nun nicht mehr den Kopf zerbrechen müssen:

Wer die Lösung im Detail sehen möchte, der lese den Artikel.

Das Übel an der Wurzel packen

Ich habe nichts gegen WCF – aber dieses Beispiel zeigt mir wieder, dass ich mit WCF nicht direkt arbeiten möchte in meinem Anwendungscode. Ich halte die im Artikel vorgestellte Lösung für eine Symptomkur. Das Symptom: WCF bietet keinen Synchronisationsmechanismus für asynchrone Aufrufe.

Und was ist das Grundproblem, wenn das nur ein Symptom ist? WCF ist das Grundproblem. Denn WCF ist ein Frameworkbolide, der immer noch in der synchronen Kommunikation verwurzelt ist. Asynchronizität geht zwar auch irgendwie, ist aber nicht der Fokus. Dabei ist Asynchronizität – zumindest aus meiner Sicht – eine unverbrüchliche Bedingung für die Kommunikation in verteilten Anwendungen.

Aber ich will hier nicht philosophisch werden ;-) Lieber zeige ich einfach, wie aus meiner Sicht das Übel an der Wurzel gepackt werden könnte. WCF will ich dabei gar nicht ersetzen. WCF soll nur für mich als Anwendungsprogrammierer unsichtbar werden.

Hier meine Lösung:

That´s it.

Nur eine Kleinigkeit fehlt darin, die ich aber nur ausgelagert habe, um das Abstraktionsniveau einheitlich zu halten. Doch das ist auch nur eine simple Erweiterungsmethode:

Sieht das übersichtlich aus? Ist das alles typsicher? War das “out of the box” möglich? Drei Mal Ja.

Übersichtlich, typsicher und “out of the box” ist das alles, weil ich eben nicht WCF benutze, sondern ein Kommunikationsframework, das konsequent auf Asynchronizität setzt. Das ist der Xcoordination Application Space (AppSpace), der als Open Source Komponente bei CodePlex liegt und die Verteilung von Anwendungen erlaubt, die mit Microsofts Concurrency Coordination Runtime (CCR) arbeiten.

Wie funktioniert nun mein Code? Dazu müssen Sie verstehen, wie Dienste im AppSpace aufgerufen werden. Das geschieht immer über sog. Ports. Das sind typisierte Nachrichtenkanäle, die die CCR zur Verfügung stellt.

Dienste halten Sie also nicht in Form von Methoden in der Hand, sondern als Ports. Sie können AppSpace-Dienste nur indirekt anstoßen vermittels von Nachrichten. AppSpace-Anwendungen haben also immer eine Event-Driven Architecture (EDA), denn Dienste registrieren Eventhandler auf ihren Ports, um an die Nachrichten ihrer Clients zu kommen.

Einen Dienst einzurichten, ist mit dem AppSpace sehr einfach. Hier die Dienstimplementation:

public class MyArticleService : PArticleService

{

    [XcoConcurrent]

    internal void ProcessGetArticleList(GetArticleList query)

    {

        …

    }

 

    [XcoConcurrent]

    internal void ProcessGetSpecialOfferArticle(GetSpecialOfferArticle query)

    {

        …

    }

}

Und hier der Code, mit dem der Dienst gehostet wird, so dass Clients ihn aufrufen können:

using(var server = new XcoAppSpace("wcf.port=12345"))

{

    server.RunWorker<PArticleService, MyArticleService>("TheArticleService");

    …

Das ist nicht sehr aufwändig, oder? Und ich muss gar nicht auf die Segnungen von WCF verzichten, wie der Konfigurationsstring des AppSpace zeigt. Nur sind mir die ganzen lästigen WCF-Details einerlei. Ich kann mich auf das Wesentliche konzentrieren: Geschäftslogik.

Und wo sind die ominösen CCR Ports? Die stecken im Kontrakt des Dienstes, denn natürlich hat der AppSpace auch sein ABC: Address, Binding, Contract. Die Adresse des Dienstes ist localhost:12345/TheArticleService. Das Binding ist Net.Tcp von WCF. Und der Contract sieht so aus:

[Serializable]

public class Article

{

    public string description;

}

 

public class PArticleService : PortSet<PArticleService.GetArticleList,

                                       PArticleService.GetSpecialOfferArticle>

{

    [Serializable]

    public class GetArticleList

    {

        public Port<Article[]> response;       

    }

 

    [Serializable]

    public class GetSpecialOfferArticle

    {

        public int forMonth;

        public int forYear;

 

        public Port<Article> response;

    }

}

Das sieht für Sie sicher ungewöhnlich aus. Aber im Grunde ist es nicht viel anders als bei WCF. Ein AppSpace-Dienst hat auch eine Schnittstelle. Die besteht aber nicht aus einem interface sondern aus einer Liste von CCR Ports (PArticleService). Auf jedem Port “lauscht” dann ein Eventhandler (s. Process…() Methoden in der Dienstimplementation).

Ports haben keine Signatur, sondern transportieren nur Nachrichten eines Typs. Deshalb müssen die üblichen Dienstmethodensignaturen in Nachrichtentypen übersetzt werden. Aus

Article GetSpecialOfferArticle(int forMonth, int forYear);

wird dann ein Port für GetSpecialOfferArticle-Nachrichten. Und jede Nachricht enthält die ursprünglichen Methodenparameter als Felder. Ein Resultat liefert die Dienstmethode schließlich via einem Antwort-Port (response).

Das ist anders als bei WCF und Webservices. Aber ich würde sagen, es ist nicht schwer zu verstehen. Vor allem ist es aber 1. symmetrisch und 2. weniger trugschlussbehaftet.

Die Symmetrie zwischen Client und Service ergibt sich daraus, dass die Kommunikation in beide Richtungen via Ports läuft. Dadurch werden z.B. Notifkationen ein Kinderspiel. Und Sie unterliegen weniger Trugschlüssen bzgl. der Kommunikation, weil Sie sofort sehen, dass sie verteilt/asynchron ist, dass sie damit mal länger dauern kann, dass der Dienst auch mal nicht antworten kann usw.

Der Client holt sich dann einen “Proxy” für einen entfernten AppSpace-Dienst mit ConnectWorker<T>(). Der bietet Ports entsprechend dem Kontrakt, an die der Client seine Nachrichten schickt.

Und jetzt der Trick mit der Synchronisation: Dafür muss ich keinen Aufwand treiben, weil die CCR einen Synchronisationsoperator bietet, ein Join. Mit ihm kann der Client ganz einfach darauf waren, dass Nachrichten in mehreren Ports angekommen sind. Im Beispiel sind das die Antwort-Ports, die der Client in die Nachrichten an den Dienst gesteckt hat.

Eine Anweisung (client.Join()) genügt daher, um den Client-Code, der mit den Ergebnissen beider Dienstaufrufe weiter arbeiten sollen, erst dann auszuführen, wenn auch beide Dienste geliefert haben. Egal, in welcher Reihenfolge sie fertig geworden sind.

Fazit

Ich kann mir nicht helfen, aber ich finde die AppSpace/CCR-Lösung verständlicher, kürzer und “ehrlicher”. Das liegt für mich daran, dass AppSpace/CCR für Asynchronizität gemacht sind. Wo WCF sich strecken muss, um seinem synchronen Erbe zu entwachsen, da sind AppSpace/CCR schon lange angekommen.

Asynchronen Code zu schreiben, ist ungewohnt. Aber mit ein wenig Übung werden Sie feststellen, dass schon in relativ einfachen verteilten Szenarien dadurch viele Vorteile entstehen. Gerade für den “kleinen Verteilungshunger zwischendurch”, d.h. dort, wo Sie bisher nicht an Verteilung zu denken gewagt haben, da bieten AppSpace/CCR Ihnen eine Plattform, die es Ihnen einfach macht, den Einstieg zu finden. Ob die Kommunikation dann “auf dem Draht” mit WCF (TCP, Named Pipes) oder MSMQ oder Jabber oder Named Pipes läuft, das ist Ihnen egal. Ihr Programmiermodell ist immer gleich. Probieren Sie es mal aus.

CCR Flows - Asynchrone Prozesse mit der CCR verdrahten

Neulich habe ich eine Lanze dafür gebrochen, zwei Probleme der Softwareentwicklung auf einen Streich zu lösen. Das würde Software zukunftsfähiger machen. Denn Abhängigkeiten und Synchronizität sind Behinderungen auf dem Weg in eine glückliche Projektzukunft.

Das Mittel für diese “Wundertat”? Asynchrone Flows, d.h. Funktionseinheiten nicht mehr statisch voneinander abhängig machen und auch nicht mehr synchron miteinander kommunizieren lassen. Stattdessen Verarbeitungsschritte in einem expliziten, getrennten “Bereich” (Separation of Concerns) lose mit eigenständigen Verbindungsgliedern “zusammenstöpseln”.

Dazu hatte ich dann ein wenig über einen API spekuliert, der das möglich machen könnte. Damit lag ich – wie sich nun herausgestellt hat – wohl nicht ganz daneben. Denn nach einigen Versuchen habe ich nun so einen Flow API implementiert. Ich nenne ihn CCR Flows, weil er intern auf CCR (Microsoft Concurrency Coordination Runtime) Ports als “Verbindungsglieder” zwischen Prozessschritten setzt.

Die CCR Flows sind jetzt Open Source (sogar inkl. Dokumentation sowie Unit Tests) und liegen bei CodePlex:

http://ccrflows.codeplex.com

Über Feedback und Diskussion dort im Forum würde ich mich freuen. Es gibt natürlich noch etwas daran zu tun. Aber als Einstieg in ein anderes Programmiermodell finde ich den API nicht ganz schlecht. Ein Beispielprogramm in den Sourcen realisiert auch den Beispielprozess meines vorherigen Blogartikels. An dieser Stelle zum Schnuppern aber nur ein kleiner Prozess, der die Worte eines Textes extrahier und dann in zwei Schritten transformiert:

Flow<string>.Do<string>(SplitTextIntoWords).Do<string>(w=>w.ToUpper()).Do<string>(Reverse)

Meine Vermutung, solcher Code lässt sich besser weiterentwickeln, weil schon bei jeder Prozessstufe (stage) viel entkoppelter gedacht wird. Denn diese Stufen kennen ihren Vorgänger und Nachfolger nicht! Sie haben keine Abhängigkeiten.

Wie das genau geht, erklärt die Doku bei CodePlex. Ansonsten fragt mich einfach.

Viel Spaß damit!

Wider die Geißeln zukunftsfähiger Software: Abhängigkeiten und Synchronizität

Was macht Software so schwer zu evolvieren? Abhängigkeiten. Was beschränkt den Nutzen morgiger Prozessorgenerationen für heutige Software: Synchronizität.

Funktionseinheiten, die von anderen abhängig sind, die insofern einen bestimmten Kontext voraussetzen, lassen sich mühsamer weiterentwickeln als solche, die frei und unabhängig sind. Möglichkeiten zur Abhängigkeit gibt es natürlich viele und nicht alle lassen sich immer kappen. Aber das Streben nach immer geringerer Kopplung lohnt sich - wenn Evolvierbarkeit gefragt ist. Wo hingegen zweifelsfrei Effizienz nötig ist, da müssen womöglich Kopplungen eng bleiben oder gar enger werden. Im Zweifelsfall bin ich jedoch der Meinung, dass unsere Software heute eher unter zu engen, als zu losen Kopplungen leidet.

Funktionseinheiten, die synchron arbeiten, arbeiten notwendig auch sequenziell. Von einer steigenden Zahl an Prozessorkernen können sie nicht profitieren. Die nützen ja nur, wenn es auch etwas parallel auszuführen gibt.

Abhängigkeiten und Synchronizität stehen uns also im Weg bei unserer Reise in eine glücklichere Softwarezukunft. Was tun? Ich beschreibe mal ein paar Gedanken anhand eines Beispiels. Hier etwas synchroner und abhängiger Code:

class MyBusinessLogik : IBusinessLogik
{
    IDatenquelle dq;
    IDatensenke ds;
    IValidator v;

    public BusinessLogik(IDatenquelle dq, IDatensenke ds, IValidator v)
    {
        this.dq = dq;
        this.ds = ds;
        this.v = v;
    }

    public int Aktualisiere(Parameter p)
    {
        v.Validiere(p);
        Datencontainer dc = this.datenquelle.LadeDaten(p.Query);
        int n = AktualisiereDaten(dc, p.Request);
        this.datensenke.SpeichereDaten(dc);
        return n;
    }

   private int AktualisiereDaten(Datencontainer dc, Aktualisierungsanfrage req) { ... }
}

Statische und dynamische Abhängigkeiten

Der Code ist statisch und dynamisch abhängig von anderen Funktionseinheiten:

Diese statischen Abhängigkeiten der Businesslogik sind statisch in der Businesslogik-Implementation verdrahtet. Das ist ein Punkt, den es deutlich herauszustellen gilt. Ausdruck der statischen Abhängigkeiten sind die Felder der Klasse MyBusinessLogik und die Aufrufe von Methoden auf deren Instanzen.

Die dynamischen Abhängigkeiten hingegen, sind nicht in der Businesslogik-Implementation zu finden. Sie kennt nur abstrakte Dienstleister wie IDatenSenke oder IValidator. Die zur Laufzeit relevanten Implementationen, werden ihre hingegen über den Ctor injiziert.

In puncto Abhängigkeiten hat also schon eine gewisse Separation of Concerns (SoC) stattgefunden: die laufzeitrelevante Bindung übernimmt eine andere Codeeinheit.

Vollständig ist die SoC allerdings nicht. Denn - wie gesagt - die BusinessLogik hat neben ihrer funktionalen Aufgabe auch noch eine nicht-funktionale: die statische Bindung. Ihr dienen die Felder und die Aktualisiere()-Methode. Ihre funktionale Aufgabe erfüllt AktualisiereDaten().

Während der Code also durch Injektion konkreter Abhängigkeiten gegenüber der früheren Praktik evolvierbarer geworden ist, weil er nicht mehr an Implementationen gebunden ist. So ist die Kopplung noch nicht wirklich lose. Die statischen Abhängigkeiten halten ihn starr. Das schränkt seine Evolvierbarkeit ein.

Synchronizität und Sequenzialität

Dass die BusinessLogik synchron und sequenziell ist, liegt auf der Hand. Ohne Hilfsmittel kann sie in C# nicht definiert werden. Ein Aufrufer von Aktualisiere() muss also auf das Ergebnis warten. Und während Aktualisiere() läuft, kann ein Prozessor nichts anderes tun; andere Aufgaben, andere Programme müssen auch warten. (Preemptives Multitasking lasse ich hier außen vor. Damit kann zwar doch quasi-parallel weiteres geschehen, aber jede zusätzliche Aufgabe verlangsamt alle schon laufenden.)

Auf innerhalb der BusinessLogik gibt es keine Parallelität. Mehrere Prozessorkerne bringen hier überhaupt keinen Nutzen. Es könnte ja sein, dass Laden, Verarbeiten und Speichern der Daten im Rahmen der Aktualisierung zumindest überlappen dürfen. Während noch Daten geladen werden, kann die Verarbeitung schon beginnen. Und während die Verarbeitung noch läuft, können erste Ergebnisse schon gespeichert werden.

Die synchrone Notation einer Sprache wie C# lässt das jedoch ohne Hilfsmittel nicht zu. Und weil wir letztlich "in C# denken", kommen wir auch nicht so recht auf die Idee, dass es auch anders sein könnte. Die beschränkten Ausdrucksmttel von C# stehen für uns zu sehr im Vordergrund. Das objektorientierte, synchrone Programmierparadigma ist wie ein Korsett um unsere Vorstellungen.

Das schränkt den Nutzen ein, den die Software, zu der die BusinessLogik gehört, aus der zukünftig wachsenden Zahl an Prozessorkernen ziehen kann.

Ausweg Asynchronizität

Ich glaube nun, dass uns ein anderes Paradigma einen Weg aus dieser beschränkten Zukunftsfähigkeit weisen kann. Durch asynchrone Programmstrukturen können wir beide Fliegen mit einer Klappe schlagen. Und das geht so:

1. Abhängigkeiten beseitigen

Im ersten Schritt schütteln wir auch noch die statischen Abhängigkeiten ab. Oder genauer: Wir separieren den Aufbau von Abhängigkeiten komplett von der problemdomänenorientierten Funktionalität. Dazu führe ich mal den Begriff "Prozessdefinition" ein. Ich trenne die Abfolge von Arbeitsschritten ganz klar von den Arbeitsschritten selbst. Alle Arbeitsschritte werden damit frei von Abhängigkeiten:

Die BusinessLogik v2 enthält jetzt nur noch die Funktionalität von AktualisiereDaten()! Aufgabe der Prozessdefinition ist es nun, die Arbeitsschritte ohne Abhängigkeiten in eine nützliche Reihenfolge zu bringen. Dass sie selbst viele Abhängigkeiten enthält, ist nicht schlimm. Sie ist im Vergleich zu den Arbeitsschritten trivial.

Dynamische und statische Abhängigkeiten stehen damit auf derselben Stufe: sie sind separierte Concerns.

Die Konsequenz solcher Beseitigung von Abhängigkeiten ist, dass Funktionalität nicht mehr geschachtelt ist. Funktionalität der Problemdomäne wie die BusinessLogik oder auch Infrastrukturfunktionalität wie eine Datenquelle sind Blätter im Abhängigkeitsbaum. Ergebnisse reichen sie also nicht tiefer hinunter in einem Aufrufbaum. Unter ihnen gibt es ja keine Ebene mehr. Stattdessen sind Ergebnisse immer Rückgabewerte in irgendeiner Form. Nur wer die Funktionalität aufruf bzw. sie zusammengesteckt hat, weiß ja, was weiter mit Ergebnisse geschehen soll. Die Wiederverwendbarkeit ist damit gestiegen.

2. Vom Callstack zum Fluss

Wenn nun alle Funktionalitäten ohne Abhängigkeiten sind, dann sind wir plötzlich sehr frei, was ihre Verschaltung angeht. Müssen wir sie denn wirklich noch synchron "ineinanderstecken"?

Hier aber zunächst ein erster Schritt. Ich nehme die refaktorisierten Funktionalitäten 1:1 und füge sie zu einem Prozess zusammen. Eine Funktion scheint mir da der passende Ausdruck für einen Prozess. In den geht etwas hinein und am Ende kommt etwas heraus.

class Prozessdefinition
{
    public Func<Parameter, int> Prozess { get; private set; }

    public Prozessdefinition(IDatenquelle dq, IDatensenke ds, IValidator v, IBusinessLogikV2 blv2)
    {
        this.Prozess = new Func<Parameter, int>(p =>
           {
               v.Validiere(p);
               Datencontainer dc = dc.LadeDaten(p.Query);
               int n = blv2.AktualisiereDaten(dc, p.Request);
               ds.SpeichereDaten(dc);
               return n;
           });
    }
}

Soweit das synchrone Programmierparadigma. Um weiter zu kommen, ist nun eine Richtungsänderung nötig. Asynchrones Denken und codieren ist nötig. Dafür zunächst eine etwas andere Darstellung der Funktionseinheiten:

Die Funktionseinheiten haben immer noch keine Abhängigkeiten untereinander. Aber es ist ihnen nun deutlich anzusehen, was reingeht und was rauskommt. Eine implementationsunabhängige Darstellung im Sinne von EVA (Eingabe-Verarbeitung-Ausgabe).

Das bisherige Abhängigkeitsdiagramm war nicht so detailliert. Darin war nur zu sehen, ob eine Funktionalität von einer anderen abhängig ist. Indem nun die Funktionalitäten aber darstellen, wie man von ihnen abhängig sein kann, ist Klarheit gewonnen. (Dass ich Ausgaben bei Validator und Datensenke eingeführt habe, ist hier vernachlässigbar. Sie machen den späteren asynchronen Prozess etwas einfacher.)

Die Prozessdefinition ist nun nicht mehr in einer Black Box eingeschlossen, sondern kann im Grunde das trivial gewordene Abhängigkeitsdiagramm ersetzen:

Diese Grafik zeigt nicht nur die grundsätzlichen Zusammenhänge der Funktionsbausteine, sondern auch den Zweck, zu dem sie zusammenhängen: die Abarbeitungen eines Prozesses. Und bitte im Hinterkopf behalten: Die Funktionsbausteine haben keine statischen Abhängigkeiten mehr. Ihre dynamische Zusammenarbeit im Prozess sieht man ihnen selbst nicht an. Die ist Sache des Prozesses.

Jetzt der Trick: Wenn der Prozess erstmal so dargestellt ist und eben nicht sofort als Code wie in der obigen ersten Prozessdefinition, dann... ja, dann ist es leicht, von einer synchronen Kopplung abzusehen. Warum sollte ich dies:

übersetzen in jenes:

if (v.Validiere(p))
{
    Datencontainer dc = dq.LadeDaten(p.Query);
    ...

Das Prozessdiagramm ist aber nicht zu verwechseln mit einem simplen Flowchart, das auch ein Kind des synchronen Programmierparadigmas ist. Es ist allgemeiner als Fluss zu verstehen: Die Funktionseinheiten sind verbunden zu einem Fluss, auf dem Daten zwischen ihnen als Verarbeitungsstationen fließen.

3. Asynchrone Flüsse

Die Darstellung eines Prozesses bietet nun die Chance, das Paradigma zu wechseln. Die Frage ist nur, wie kann solch bisher synchroner, sequenzieller Code in asynchronen, sequenziellen oder parallelen Code gewandelt werden?

Der Schlüssel liegt in expliziten Verbindungsstücken!

Bei der üblichen synchronen Programmierung sind die Funktionsbausteine quasi aneinandergeschweißt. Die ursprüngliche BusinessLogik war untrennbar verbunden mit einem Validator usw. Sie bildeten eine Einheit - auch wenn die konkreten Abhängigkeiten erst zur Laufzeit dynamisch eingespritzt wurden.

Das ist zwar effizient - aber eben auch inflexibel. Und es ist synchron. Damit steht solche Verschweißung der Evolvierbarkeit im Wege.

Ganz anders das Bild, wenn die Funktionsbausteine nicht verschweißt, sondern verschraubt sind. Mit expliziten Verbindungsstücken - wieder eine Separation of Concerns - können Beziehungen viel flexibler aufgebaut werden.

Ein Mittel dafür sind die Ports der Microsoft Concurrency Coordination Runtime (CCR). Mit ihnen ließe sich ein Prozess ganz anders beschreiben. Aus den bisherigen synchronen Methoden

interface IValidator
{
    bool Validiere(Parameter p);
}

interface IDatenquelle
{
    Datencontainer LadeDaten(string query);
}

könnten solche werden:

void Validiere(Parameter p, Port<bool> output) {...}

void LadeDaten(string query, Port<Datencontainer> output) {...}

Ich habe sie aus zwei Gründen nicht in eine Klasse oder ein Interface eingetragen: Zum einen möchte ich an dieser Stelle keine bestimmte Lösung jenseits explizit verbundener Funktionsbausteine vorschlagen. Auch die CCR Ports sind nur eine mögliche Variante für explizite "Schraubverbindungen". Zum anderen sollen die beiden freigestellten Routinen unterstreichen, dass Funktionale Programmierung - also die Konzentration auf Funktionen statt Klassen - einen Beitrag leisten kann, um Software zukunftsfähig zu machen.

Die Übersetzung einer bisher synchronen Methode könnte mit Ports also geradlinig sein: Eingabeparameter bleiben, Rückgabewerte werden ersetzt durch einen Output-Port. Der ist mit der nächsten Verarbeitungsstation verbunden. Zu jeder Methode gehört also ein Port für die Eingabe. Für die obige Methode Validiere() könnte das in vereinfachter und roher Form so aussehen:

Port<bool> pOutput = ...;

var pValidiere = new Port<Parameter>();
pValidiere.ReceiveSequentially(p => Validiere(p, pOutput));

Der Port für die Ausgabe, der gehört dann zum nächsten Arbeitsschritt im Prozessfluss.

Die Methode ReceiveSequentially() ist eine Extension Method für Ports, die ich mir ausgedacht habe. Sie bindet einen Eventhandler so an einen Port, dass immer nur ein Element zur Zeit verarbeitet wird. Das sichert eine sequenzielle, allerdings asynchrone Verarbeitung durch die Stationen in einem Prozessfluss zu. Bei Bedarf können Stationen aber natürlich auch parallel verarbeiten. Ihre vielen Ergebnisse müssen dann nur auch wieder eingesammelt werden. Map-Reduce ist dafür ein berühmtes Beispiel.

Viel wichtiger als Parallelität ist jedoch die Asynchronizität. Dadurch, dass Arbeitsschritte jetzt explizit über puffernde Ports gekoppelt sind, geben sie immer wieder ihre Prozessorressource (Thread auf einem Kern) frei. Das skaliert wunderbar. Es ist kooperatives Multitasking, das alle Kerne oder potenziell auch viele Maschinen überspannt.

Damit ist das eingangs beschriebene Ziel im Grunde erreicht: Die Abhängigkeiten sind minimiert, der Umgang mit ihnen ist herausfaktorisiert. Und die Asynchronizität ist eingeführt. Damit ist der Code zukunftsfähiger geworden:

• Viele Flüsse können in dieser Weise gut skalierbar abgehandelt werden und nutzen dabei alle Prozessorkerne. (Und falls es mal nicht viele Flüsse durch die Last auf einem Rechner zu bearbeiten geben sollte, dann zeige ich in einem späteren Blog-Posting, wie so ein unterforderter Rechner anderen seine Leistung anbieten kann.)
• Abhängigkeitsfreie Funktionseinheiten lassen sich viel einfacher weiterentwickeln und neu kombinieren.

Technische Umsetzung mit Notationsschwierigkeiten

Jetzt aber noch kurz konkret zur Umsetzung des obigen Prozesses:

• Es kommen Parameter an, die in mehreren sequenziellen Schritten durch einen Prozess laufen sollen.
• Am Anfang werden die Parameter validiert. Nur wenn die Validation erfolgreich ist, werden sie zum Laden von Daten weitergeschickt. Ein Fork-Station spaltet die Daten dafür auf: sie werden gleichzeitig zur Validation und zu zwei Joins weitergeleitet.
• Der erste Join führt das Ergebnis der Validation und die Query in den Parametern zusammen und leitet die Query weiter an die Datenbeschaffung - aber natürlich nur, wenn die Validation erfolgreich war. Validation und Datenbeschaffung laufen also nicht parallel, sondern immer noch sequenziell.
• Nach der Datenbeschaffung führt ein weiterer Join deren Ergebnis (Datencontainer) und die Parameter zusammen und leitet beide weiter an die eigentliche Geschäftslogik.
• Die Geschäftslogik verändert die Daten und reicht sie weiter zum Speichern. Gleichzeitig erzeugt sie ein Ergebnis (z.B. Anzahl veränderter Datensätze), das jedoch nur am Ende aus dem Prozess herauskommt, wenn auch die Datenspeicherung erfolgreich war.

Dieses Szenarion mit Ports zu bauen, ist nicht schwierig. Es ist nur etwas umständlich. Eine wirklich gute textuelle Notation oder ein Fluent Interface, dass gerade diese Fork-Join-Kombinationen plastisch macht, ist mir noch nicht eingefallen.

Ohne Fork-Join oder auch Scatter-Gather oder Select/Choice und andere Muster, bei denen mehrere Ports beteiligt sind, wäre es einfach. Mit Pipes könnte ein vereinfachter Fluss ohne Fork-Join so aussehen:

Validator | Datenquelle | BusinessLogikv2 | Datensenke

Jede Station würde ihre Ergebnisse einfach nur weiterschieben an die nächste. Allerdings müsste z.B. der Validator alle Parameter weitergeben und nicht nur sein boolean-Resultat, weil ja nachfolgende Stationen davon mehr oder weniger für ihre Arbeit brauchen.

Etwas realistischer ließe sich so ein Fluss auch mit einem Fluent Interface beschreiben, z.B.

new Stage<Parameter>(Validiere)
    .Stage<Parameter, Tupel<Datencontainer, Parameter>>(LadeDaten)
    .Stage<Tupel<Datencontainer, Parameter>, Tupel<Datencontainer, int>>(Aktualisieren)
    .Stage<Tupel<Datencontainer, int>, int>(SpeichereDaten);

Dabei stehen die generischen Typparameter von Stage<TIn, TOut> für den Typ des Input- und des Output-Ports. Als Verarbeitungsschritt wird dann Action<TIn, TOut> erwartet.

Die Schwierigkeit der Notation - nicht der Technologie! - beginnt jedoch, wenn Flüsse sich teilen und wieder zusammenfließen.

Die Fork am Anfang des obigen Prozesses ließe sich so formulieren:

var fork = new Fork<Parameter, Parameter, string, Parameter>(
    (i, o0, o1, o2) => {oo.Post(i); o1.Post(i.Query); o2.Post(i);});

Der erste Typparameter definiert den Input-Typ, die weiteren die Typen für die Output-Ports, d.h. die Input-Ports der nachfolgenden Schritte. Die Aufgabe des Lambda-Funktion ist also die Aufspaltung oder Verteilung des Input auf die Outputs.

Die einzelnen Verarbeitungsschritte sind ebenfalls für sich genommen leicht zu formulieren:

var val = new Stage<Parameter, bool>(Validiere);
var laden = new Stage<string, Datencontainer>(LadeDaten);
var akt = new Stage<Tupel<Datencontainer, Parameter>,
                                          Datencontainer, int>(Aktualisieren);
var speichern = new Stage<Tupel<Datencontainer, int>, int>(SpeichereDaten);

Und wie die Joins formulieren, die Zusammenführungen von mehreren Flussarmen?

var joinFürLaden = new Join<bool, string, Datencontainer>((i0, i1, o0) => oo.Post(i1));

var joinFürLogik = new Join<Datencontainer, Parameter, Tupel<Datencontainer, Parameter>>(
    (i0, i1, oo) => oo.Post(new Tupel<Datencontainer, Parameter>(i0, i1)));

var joinFürEnde = new Join<bool, int, int>((i0, i1, oo) => oo.Post(i1));

Zum Schluss noch die Prozessschritte "zusammenstöpseln":

fork.Output[0] = val;
fork.Output[1] = joinFürLaden.Input[1];
fork.Output[2] = joinFürLogik.Input[1];

val.Output = joinFürLaden.Input[0];

joinFürLaden.Output = laden;

laden.Output = joinFürLogik.Input[0];

joinFürLogik.Output = akt;

akt.Output[0] = speichern;
akt.Output[1] = joinFürEnde.Input[1];

speichern.Output = joinFürEnde.Input[0];

Der Input geht dann in den Prozess bei fork.Input hinein und das Ergebnis kommt bei joinFürEnde.Output heraus. Das ist technologisch nicht kompliziert - aber die vorstehende Formulierung ist sicher nicht so gut zu lesen wie die synchrone Variante ganz am Anfang.

Was tun? Hier ist wahrscheinlich eine DSL angezeigt. Eine textuelle könnte schon ein wenig Erleichterung bringen; am Ende geht es aber wohl nicht ohne Diagramme. Ja, das wäre doch mal was: Eine grafische DSL mit einem hübschen Designer, die aus solchen Prozessdiagrammen eine Assembly produziert, die wir nur noch mit Prozessschritten parametrisieren müssen, wenn das nicht schon der Designer getan hat, weil wir ihm Referenzen auf Prozessschritte übergeben haben.

Ist das dann nicht aber die Microsoft Workflow Foundation neu erfunden? Nein. Die Prozesse, von denen ich hier geschrieben habe, sind leichtgewichtiger. Sind sollen nicht lange laufen. WF scheint mir da Overkill.

Und vielleicht... findet sich ja doch auch noch eine textuelle Beschreibung z.B. in Form eines Fluent Interface? Das würde mir sehr gefallen.

Aber all dessen ungeachtet ist mir an dieser Stelle wichtig gezeigt zu haben (oder zumindest laut gedacht zu haben), dass asynchrone explizite Kopplung mit soetwas wie CCR Ports hilft, Software zukunftsfähig in zweierlei Hinsicht zu machen. Wer asynchrone Prozesse denkt, der geht anders mit Abhängigkeiten um und macht Software fit für Mehrkernprozessoren. Da müssen wir gar nicht erst auf Axum & Co warten. Das geht hier und heute.