CQRS - Commands parallel aber synchronisiert laufen lassen

'n Abend,

wir haben aktuell eine CQRS ähnliche Architektur.
Es gibt einen Server, der per WCF angesprochen wird.
Der hat X Commands, die alle irgendwas tun, mal mehr mal weniger komplex.
Jeder Command bekommt seine eigene Session.
Sind während der Ausführung Fehler aufgetreten, wird jede Änderung ausnahmslos verworfen und der komplette Command abgebrochen.
Wir haben an jeder Entity eine Version-Spalte. Wenn ein Command Daten abfragt, merkt sich NHibernate die Version und prüft die später beim Speichern gegen. Hat die Version sich geändert, wird nicht gespeichert.

Bisher waren diese Commands der Einfachheit wegen alle untereinander gelockt, es konnte also nur ein Command gleichzeitig ausgeführt werden.
Nun ist die Gesamtanwendung aber weit stärker gewachsen als erwartet, weshalb wir nach einem Weg suchen, die Commands möglichst parallel laufen zu lassen.

Bei Commands, die wenig tun, ist das kein Problem.
Wenn sie nur lesen oder nur einmal zu Beginn ein paar wenige Daten lesen und zum Schluss ein bisschen was schreiben, was wenige Abhängigkeiten hat, läuft das.
Das Problem sind Commands, die sehr komplexe Dinge tun, auf viele Daten zurück greifen und viel ändern. Solche Commands können, wenn sie parallel laufen, Probleme verursachen.

Ich suche nun nach einer Möglichkeit, wie man überschaubar aber flexibel managen kann, welche Commands wann parallel ausgeführt werden dürfen.

Eine Idee:
Jeder Command, der gestartet wurde, sagt dem CommandDispatcher, dass er läuft und umgekehrt, wann er fertig ist.
Wenn nun ein zweiter Command gestartet werden soll, dann kann der CommandDispatcher sehen, welcher Command gerade läuft.
Er kann den zweiten Command dann fragen, ob er sich mit denen "verträgt". Wenn ja, wird er parallel gestartet, andernfall landet er in einer Queue.
Standardmäßig sagt jeder Command "Ich mag niemanden", ich kann dann aber einzeln bestimmen, womit er sich nicht in die Quere kommen würde.
Eine Deadlock-Möglichkeit gibt es meines Erachtens nach nicht.

Eine weitere Idee:
Ein Command ruft irgendwelche Daten ab und sagt dann, dass er Entity X mit der ID Y braucht.
Jeder weitere Command, der genau das auch braucht, muss dann eben warten, bis die ID wieder freigegeben wurde.
Auf diese Weise kann ich theoretisch jeden Command parallel laufen lassen und die Commands warten nur aufeinander, wenn sie gleichzeitig den selben Datenbank-Eintrag brauchen.
Leider kann es auch sein, dass ich mir einen Deadlock baue.

Hat jemand eine Idee, wie man das gut lösen kann?
Oder Tipps zu den Ideen?

Mir ist dabei noch wichtig, dass die konkrete Umsetzung nicht von NHibernate abhängig ist.
Aktuell tendieren wir eher dazu, in Zukunft EntityFramework zu nutzen. Außerdem soll die Grundlage, die ich jetzt baue, später auch in anderen stark abweichenden Projekten genutzt werden können.

Beste Grüße und vielen Dank für die Hilfe

Google mal ein wenig nach Saga Persistance, CQRS Command Chaining oder Reliable Messaging.

Commands sollten sich bei CQRS auf >keinen Fall< untereinander locken oder ähnliches!
Besser ist es, auf Statusänderungen (durch Events) vom Commands oder Commandresultaten zu hören und dadurch einen neuen Command zu starten.

Besser ist es, auf Statusänderungen (durch Events) vom Commands oder Commandresultaten zu hören und dadurch einen neuen Command zu starten.

Diesen Fall habe ich allerdings so gut wie nie, ein Command hört nicht auf Events anderer Commands und Statusänderungen gibts wenn dann auch nur am Ende, wenn gespeichert wird.
Wenn ich die drei Schlagworte von dir richtig verstanden habe, treffen die auch in diese Richtung und helfen mir leider nicht weiter.

Ein Command wird dann gestartet, wenn ein Benutzer z.B. auf einen Button klickt.
Jeder Command sucht zuerst nach Daten, von denen er abhängig ist und prüft, ob er überhaupt ausgeführt werden darf.
Er errechnet auf Basis dieser Daten dann ein Ergebnis, welches er natürlich auch speichern muss.
Stellt sich dabei aber heraus, dass die Daten, auf deren Basis er gearbeitet hat, zwischendurch geändert wurden (weil ein anderer Nutzer etwas getan hat), dann darf das Ergebnis nicht gespeichert werden.

Dieser Check passiert auf Basis einer Version, die bei jeder Änderung weiter gezählt wird.
Leider wird die auch bei Änderungen hoch gezählt, die für den aktuell laufenden Command gar nicht von Bedeutung sind.

Das bedeutet, dass manche Commands nicht parallel laufen können bzw. dürfen, da ein Command die Daten des anderen Commands ändern würde.
Ich muss diese Commands also locken, ob ich will oder nicht.
Leider sind das meist auch Commands, die sehr viel tun und entsprechend lange laufen. Bei denen würde es sich also lohnen, die parallel laufen zu lassen.

Und ich suche jetzt nach einer Möglichkeit, wie ich die Commands möglichst wenig locken muss, dass sie möglichst lange parallel arbeiten aber sich mit den Daten nicht in die Quere kommen.

Das würde ich erreichen, wenn ein Command explizit meldet, welchen Datensatz er braucht und dann solange wartet, bis niemand mehr damit arbeitet ist.
Leider gibt's da auch das unschöne Deadlock-Risiko und mir fällt nichts ein, wie ich das sinnvoll umgehen kann.

Oder ich sage aktiv, welche Commands nicht gleichzeitig laufen dürfen, was dann aber auch bedeutet, dass ein Command auch dann wartet, wenn er einen anderen Datensatz braucht und gar nicht warten müsste.

Commands sequentiell zu verarbeiten ist absolut normal und auch durchaus notwendig. Immerhin ist CQRS ja gerade ein Pattern für Collaboration.

Wichtig ist es v.a. Aggregate Roots gut zu wählen. Die wichtigste Bedingung ist Konsistenz: Choosing Aggregate Boundaries – Consistency. In einem Buchungssystem können z.B. Buchungen für ein Hotel nicht Parallel verarbeitet werden. Das heißt du machst aus einem Hotel ein Aggregate Root und verarbeitest die Commands für dieses Hotel sequentiell.

Ich würde bloß nicht locken sondern eher über Queues nachdenken und dann überlegen, wie du partitionierst. Am besten über AggregateId % NumPartitions oder pro Stadt in dem sich das Hotel befindet. Das funktioniert aber nur auf Aggregate Ebene, falls deine Operationen übergreifend sind würde ich das über einen ProcessManager abbilden (=Saga) und dann die EventHandler partitionieren.

Btw: Genau dafür gibt es ja auch den CommandBus, es ist einfach das Dispatching zu ändern.

CQRS besagt ein Command verändert genau ein Aggregate.

Also kann es nur einen Konflikt geben wenn ein anderer Command das gleiche Aggregate verändert hat. Diese auftretenden Concurrency-Conflicts kann man oft durch einfaches Wiederholen des Commands beheben.

Dafür legt man sich eine Strategie zurecht die im Zweifel eine ConcurrencyException wirft.

So eine Saga wird dann benötigt, wenn ein Command im Prinzip doch mehr als ein Aggregate verändern wollen würde wenn er dürfte.

In dem Fall wird der Command auf ein Aggregate angewandt und die daraus erzeugten Events gehen an die Saga, die wiedrum neue Commands auslöst, die dann wieder jeweils ein Aggregate ändern.

Vielleicht habe ich Saga Persistance auch falsch verstanden
Oder CQRS ist hier einfach der flasche Begriff für unseren Fall

Die Idee mit den Aggregate Roots gefällt mir aber ganz gut.
Ich würde dann von außen einige Aggregates definieren, zu denen ich Commands z.B. per Attribut zuordne.
Pro Aggregate gibt es dann eine Queue, aus der die Commands nacheinander abgearbeitet werden.

Wenn ich diese Zuordnung auch zur Laufzeit vor der Ausführung erlaube, dann könnte sich ein Command auch anhand einer übergebenen ID zuordnen.
Ob das so praktikabel ist - mal schauen - es hätte auf jeden Fall das Ergebnis, dass ich ziemlich detailliert definieren kann, welche Commands parallel ausgeführt werden dürfen, ohne dass ich die anderen Commands dabei kennen muss.

Das klingt auf jeden Fall danach, dass sich das auf unsere Software anwenden lässt ohne zu viel ändern zu müssen.
Ich danke 😃

Falls ihr noch keine Aggregate Roots habt, ist es wirklich das wichtigste diese Konsistenz Grenzen zu definieren und dann hat man auch fast schon die Aggregate Roots.

Du brauchst aber keine Queue pro Aggregate Root, das wäre eine krasse Partitionierung. Die Queue verringert in erster Linie die Anforderungen an transaktionale Änderungen der Datenbank (und natürlich ein bisschen Last). Falls du das ganze Aggregate Root transaktional speichern kanns, kannst du es auch einfach mit einem Retry versuchen.

Das hängt auch damit zusammen was ihr noch innerhalb eines Commands macht. Falls z.B. noch mit Externen Systemen kommuniziert wird geht es fast nicht ohne Queues (auch in CQRS würde man das oft sequentiell verarbeiten, aber eher dann in den Event Handlern oder Process Managern).

Ein Retry fällt vermutlich raus
Es kann sein, dass das an zu vielen wartenden Commands lag, aber teilweise dauert ein Command einige Sekunden bis eine halbe Minute.
Daher auch der Versuch, das parallel laufen zu lassen, bei ersten Versuchen lief der Command dann "nur" noch zwei Sekunden.
Der Datenbankserver hat auch ohne Ende Leistung, die er aber einfach nicht ausnutzen kann, wenn pauschal alles gelockt wird.

Du brauchst aber keine Queue pro Aggregate Root

Wäre das denn ein Problem?

Ich würde dann einfach pro Aggregate ein Dictionary mit Queues führen.
Ohne jetzt einen kompletten Überblick zu haben, klingt das eher einfacher mit kaum Nachteilen.

Falls z.B. noch mit Externen Systemen kommuniziert wird

Ja, das fängt bei Email oder Drucken an und endet bei externen Kunden-Schnittstellen und internen Produktionsanlagen.
Jetzt wo Du es sagst, könnte das vermutlich auch Probleme machen, wenn manche Commands parallel laufen würden 😄

Emails, Druckausgabe, etc. haben aber nichts mehr mit der Command-Verarbeitung zu tun. Die werden eigentlich durch die Events ausgelöst und darum kümmert sich dann ein nachgelagerter EventHandler

Mag schon sein, aber das auch noch anzupassen, lohnt einfach nicht, besonders da wir keinen expliziten Nachteil dadurch haben.

Das nehmen wir dann irgendwann später in Angriff, bis dahin ist ein vernünftiges asynchrones Abarbeiten der Commands wichtiger.

Naja der Nachteil is ganz einfach, dass Du Dir es teilweise jetzt schon verbaust und im Zweifel nen Deadlock hast.
Warum also nicht 10% mehr investieren und es direkt richtig machen? 😉

Queues solltem zudem nicht unbedingt In-Process sein, da Du Dir damit die Skalierung verbaust.
Queue lieber mit einem entsprechenden Queue-System wie RabbitMQ oder sowas.

Ein Deadlock durch synchron versendete Emails? O.o

Queues solltem zudem nicht unbedingt In-Process sein, da Du Dir damit die Skalierung verbaust.

Da bin ich allerdings anderer Meinung.
Mag schon sein, dass ich mir damit die Skalierbarkeit einschränke, aber es ist sehr unwahrscheinlich, dass wir jemals irgendeinen Vorteil dadurch hätten, eine Command-Queue in einem anderen Prozess laufen zu lassen.
Das bringt nur Mehraufwand und Nachteile, aber keine direkten Vorteile.

In unserem Fall reicht ein Thread völlig aus
Dazu kommt noch, dass ein kleineres und einfacher gestaltetes System deutlich einfacher bzw. schneller angepasst werden kann. Wenn dann noch ein großes Framework wie RabbitMQ oben drauf kommt, dann schränke ich mir in erster Linie meinen eigenen Handlungsfreiraum ein.

Das reine In-Process-Queuen von Commands ist ein Risiko, wenn mal der Prozess abraucht (oder gewollt runter fahrt) und Du zig Commands in der Queue hast.
Dadurch kannst Du ungültige, nicht nachvollziehbare Stati bilden, die sich im Zweifel nur durch manuellen Eingriff lösen lassen.

Mit so einem Risiko kann man meist leben, solange es nicht eintritt.

Das umgehen wir so, dass nichts gespeichert wird, bis der Command am Ende ist.
Gibt es irgendeinen Fehler, dann wird eben nichts gespeichert und es entsteht auch kein kaputter Zustand.

Mehrere Prozesse würden wahrscheinlich Sinn machen, wenn wir Commands eventgesteuert starten würden.
Dann würde ein Command mitteilen dass Command 2 ausgeführt werden soll, bricht dabei aber der Prozess ab, dann fehlt die Arbeit von Command 2.

Bei uns gibt es diesen Fall aber nicht, der eine Command tut alles, was getan werden muss, am Stück.

Das mit dem Schreiben ist ja seit es Transaktionen auf dem SQL-Server gibt das geringste Problem.

Was machst du aber wenn die Command-Queue noch voller Commands steckt und dann raucht dein System ab und damit auch die enthaltenen Commands? Dann schaut man auf die abgerauchte Queue und in die Röhre.

Darum sollten Commands auch schon in einer transaktionssicheren Queue persistiert werden bis diese endlich verarbeitet werden (wenn das Servernetzteil ausgetauscht wurde).

Wenn ich das so weiter denke, macht Abts Vorschlag wohl doch Sinn

Ich hab es erst so verstanden, dass für eine Queue ein Prozess gestartet wird und jedere weitere Command dort rüber geschoben werden muss.

Es wäre aber genauso gut möglich, wenn ein Command bzw. alle nötigen Informationen für einen Command erst einmal persistent (z.B. als Datei) serialisiert werden.
An anderer Stelle schaue ich dann dort nach und führe alle Commands aus, die da liegen. An der Stelle ist es dann auch egal, ob das ein eigener Thread oder Prozess ist.

Stürzt irgendeiner der Prozesse ab, geht kein Command verloren.

Etwas umständlicher wird dann nur die Frage, wie ich an eventuelle Rückgabewerte komme, aber auch da findet sich eine Lösung.

Du musst dir aber dann überlegen, ob du Commands asynchron verarbeiten kannst - auch im Client - und wie dann (hast du ja erwähnt) Antworten verarbeitet werden. Was ist z.B. wenn Commands abgelehnt werden, weil die Validierung fehlschlägt? z.B. Uniqueness Constraints?

Das ist dann aber vom Client her motiviert und nicht vom Server.

Aus der Perspektive des Servers geht es bei den Queues v.a. um die sequentielle Verarbeitung. Wenn dein Client auf das die Bearbeitung des Commands wartet hat er ja auch eine Erwartung an die Verarbeitungsgeschwindigkeit, dann kannst du alte Commands auch einfach wegwerfen, egal ob dein Server crasht oder nicht.

Kurz: Ich würde mir genau überlegen ob du synchrone oder asynchrone Command-Verarbeitung willst.

Die Frage, ob synchron oder asynchron ist so ziemlich entschieden.
Bei der Menge an Nutzern, die wir haben, ist das auch durchaus sinnvoll.
Außerdem kann unser ziemlich starker Server dann endlich mal seine Leistung ausspielen 😄

Wir machen es jetzt wie folgt:

Der CommandDispatcher fragt bei jedem Command nach, ob er parallel zu den Commands A, B und C ausgeführt werden darf.
Das ist etwas anders als die ursprüngliche Aggregates-Idee, aber bedeutend einfacher und vor allem sicherer auf unseren aktuellen Stand umsetzbar, da alle anderen Commands dann per Default synchron laufen.

Jeder Command, der in der Queue landet, wird nach einem bestimmten Timeout persistent gespeichert. Dafür reichen ja der Command-Typ und die Parameter aus.

Der CommandDispatcher kann mir dann den Status über die Ausführung liefern.
Exceptions oder Rückgabewerte speicher ich nicht.

Euer ziemlich starker Server muss nicht unbedingt ein Vorteil sein, denn jetzt stehst Du vor dem Problem der horizontalen Skalierung und vor Dingen wie Ausfallsicherheit und Co.
Ich hab die Erfahrung gemacht, dass mehrere kleine Instanzen im Umgang mit der gesamten CQRS Umgebung einfacher zu handhaben ist, als ein großer mit Recovery-Szenarien.

Normalerweise legt beim bei der persistenten Speicherung eines Commands auch Ausführkriterien ab, um zB. Race Conditions zu vermeiden.

Exceptions solltest Du über einen anderen Kanal ableiten (zB, eine Exception Handler Queue), aber nicht fallen lassen.
Nur so als Tipp 😉

Wie sieht das bei mehreren kleinen Prozessen denn aus, wenn wir den Server wegen eines Updates beenden müssen? Werden die kleinen Prozesse dann auch beendet?

Denn wenn ich alles Datei-basiert oder über die Datenbank als Queue vorhalte und das Ergebnis/die Exception daneben liegt, dann ist es völlig egal, ob das nun ein eigener Thread, Prozess oder ein ganz anderer Server ist.

Bei einem eigenen Prozess kann ich den Aufrufer nur nicht so leicht über ein Ergebnis informieren, da ich sowas wie Events nicht mehr habe. Ich muss als Aufrufer dann regelmäßig fragen, ob sich etwas geändert hat.
Oder gibt es da eine einfachere Lösung?

Normalerweise legt beim bei der persistenten Speicherung eines Commands auch Ausführkriterien ab, um zB. Race Conditions zu vermeiden.

Was meinst Du damit?
Ich würde da den Command bzw. den Namen und die Parameter serialisieren.
Da neben dann noch Status, Result und Exception. Für Analyse-Zwecke könnte dann auch noch eine ExecutionTime dazu

Exceptions solltest Du über einen anderen Kanal ableiten (zB, eine Exception Handler Queue), aber nicht fallen lassen.
Nur so als Tipp 😉

Die werde ich ganz sicher nicht fallen lassen 😄
Ich stelle mir das so vor, dass ich neben dem Command einen Status liegen habe.
Außerdem gibt's da dann die Felder Result und Exception, die je nach Status ihren entsprechenden Inhalt haben.
Und wenn der Status nicht "Waitin" oder "Running" ist, dann gilt der Command als beendet und es kann das Ergebnis oder die Exception irgendwann an anderer Stelle wieder abgefragt und behandelt werden.