Messaging und Multithreading in Unit Tests
Derzeit arbeite ich an der Implementierung einer Webanwendung bei der wir das ASP.NET MVC Framework und Messaging mit MassTransit verwenden. Messaging eignet sich hervorragend um Services anzusprechen die eine lange Laufzeit haben. Dies ist immer dann der Fall, wenn Services über das Netz aufgerufen werden, sei es lokal im Intranet oder global im Internet. Damit der anfragende Thread nicht während der gesamten Wartezeit blockiert ist sollten solche Anfragen natürlich asynchron erfolgen.
Das Publish/Subscribe Modell liefert eine gute Abstraktion einer solchen asynchronen Anfrage. Der Client veröffentlicht (Publish) eine Message zu seiner Anfrage auf dem Messagebus. Der Service meldet sein Interesse an diesem Messagetyp am Bus an (Subscribe). Der Bus sorgt für den Transport der Messages.
Der Bus ist dafür verantwortlich Messages vom Publisher zu den Subscribern zu übertragen. Client und Service kennen einander nicht, die Schnittstelle zwischen beiden wird durch die Messagetypen und den Bus gebildet. Anfrage und Antwort werden jeweils als Messagetyp realisiert. Bei der Anfrage ist der Client der Publisher und der Service der Subscriber. Bei der Antwort kehrt sich dies um, der Service ist der Publisher, der Client der Subscriber.
Ein weiterer Vorteil von Messaging ist, dass die konkrete Verteilung der Services nur noch eine Frage der Konfiguration ist. Ob Client und Service im selben Prozess laufen, in unterschiedlichen Prozessen auf dem selben Rechner oder aber auf verschiedene Rechner verteilt werden spielt bei der Implementierung keine Rolle sondern muss nur konfiguriert werden. Dies ist ein enormer Vorteil bei der Skalierung von Anwendungen.
Laufen Client und Service im gleichen Prozess, müssen sie natürlich in eigenen Threads ausgeführt werden um asynchron zusammen zu arbeiten. Und damit kommen wir zur Frage wie man mit Multithreading in Unit Tests umgeht.
Die Implementierung eines asynchronen Client könnte etwa folgendermaßen aussehen:
public bool Login(string username, string passwordHash) {
responseEvent = new ManualResetEvent(false);
bus.Subscribe<LoginResponseMessage>(Consume);
bus.Publish(new LoginRequestMessage(username, passwordHash));
responseEvent.WaitOne(5.Seconds());
bus.Unsubscribe<LoginResponseMessage>(Consume);
if (response != null) {
return response.LoggedIn;
}
return false;
}
public void Consume(LoginResponseMessage responseMessage) {
response = responseMessage;
responseEvent.Set();
}
Nachdem der Client seine Callbackmethode Consume für Messages vom Typ LoginResponseMessage angemeldet hat, veröffentlicht er mit Publish seine Anfrage. Dann wartet er mit Hilfe eines ManualResetEvent maximal 5 Sekunden auf das Eintreffen einer Antwort.
ManualResetEvent hält den Thread solange an bis entweder der Event ausgelöst wird oder der Timeout erreicht ist. Beim Subscribe wird dem Bus die Callbackmethode Consume übergeben die aufzurufen ist wenn Messages vom entsprechenden Typ eintreffen. Die Consume Methode speichert die Nachricht und löst den Event aus. Dadurch wird der angehaltene Thread wieder frei gegeben. Zu beachten ist dabei, dass der Aufruf der Consume Methode durch das Messaging Framework auf einem anderen Thread erfolgt.
Um die Login Methode zu testen müssen wir sie aufrufen und auf einem anderen Thread die Consume Methode starten. Dazu habe ich eine Extension Method erstellt die eine Action nach einer einstellbaren Verzögerung auf einem neuen Thread startet:
public static class ActionExtensions
{
public static void DeferFor(this Action action, TimeSpan timeSpan) {
ThreadStart scheduledAction = () => {
Thread.Sleep(timeSpan);
action();
};
Thread thread = new Thread(scheduledAction);
thread.Start();
}
}
Mit dieser Extension Method sieht der Test wie folgt aus:
[Concern(typeof(AnmeldeGateway))]
public class Wenn_das_AnmeldeGateway_rechtzeitig_eine_Antwort_vom_Bus_erhaelt :
InstanceContextSpecification<AnmeldeGateway>
{
private LoginResponseMessage response;
private IServiceBus bus;
private bool result;
private const string passwordHash = "";
private const string username = "Lieser";
protected override void establish_context() {
response = new LoginResponseMessage(true);
bus = dependency<IServiceBus>();
}
protected override AnmeldeGateway create_sut() {
return new AnmeldeGateway(bus);
}
protected override void because() {
Action action = () => sut.Consume(response);
action.DeferFor(100.Milliseconds());
result = sut.Login(username, passwordHash);
}
[Observation]
public void wird_die_Antwort_als_Returnwert_geliefert() {
result.should_be_true();
}
}
Der Knackpunkt liegt in der because Methode. Dort wird für den Aufruf der Consume Methode mit action.DeferFor ein neuer Thread gestartet und für 100 Millisekunden verzögert. Parallel wird die Login Methode aufgerufen. Dadurch laufen im Test zwei Threads parallel, so wie es im Echtbetrieb der Komponente auch der Fall ist.
December 15th, 2008 at 19:46
Stefan,
schön zu sehen das ihr Fortschritte mit Messaging + MVC macht. Das gezeigte Beispiel für Asynchronität im Controller hat aber leider einen Haken: der gezeigte Controller ist nicht asynchron.
Durch WaitOne(..) wird der Thread blockiert und nicht wiederverwendet. Sobald ich dazu komme und ich fertig bin, poste ich einmal unsere Controller, die auf einem engl. Blogpost basieren. Diese Lösung mit einem echten asnychronen Controller sorgt dafür, das der Thread ganz wie bei der asynchronen Page-Verarbeitung in ASP.NET wiederverwendet wird, um andere Requests abzuarbeiten.
Grundsätzlich sollte man vielleicht auch sagen, dass man dieses Verhalten für besseres scale-up des Webservers braucht. Trotzdem kommt man um eine skalierbare (scale-out) Architektur wohl oft nicht herum, um sich für zukünftige Last abzusichern.
Grüße
Sebastian
December 15th, 2008 at 20:06
Hallo Sebastian,
dass der ASP.NET Thread des Controllers blockiert wird ist uns schon bewusst. Wir setzen da auf die Fortentwicklung des Framework. Im ersten Schritt ging es uns um eine Architektur die Asynchronität ermöglicht, und die haben wir mit der skizzierten Lösung gefunden. Der Rest ist… Implementierungsdetail
Grüße
Stefan
December 16th, 2008 at 10:24
Stefan,
Using deferred parallel calls in unit tests leads to unpredictable tests. You have to either increase the timeout (increasing the duration of tests) or deal with unpredictability (the test may fail if 100ms is not enough, this may easily happen when running the test on a busy build server). That was my experience.
Also Sebastian is right. You lost all the benefits of async messaging when built a synchronous facade around. Integrating async messaging with ASP.NET might be tricky though.
December 16th, 2008 at 10:43
Hi Stefan,
meiner Meinung nach hat Multithreading/Threadsynkronisation/Callbackmanagement nichts in einem UnitTest und noch weniger in einer BDD-Spezifikation zu suchen.
Für mein Empfinden läuft ein solcher Test zudem zu langsam um noch als UnitTest bezeichnet zu werden und enthält zu viel “Infrastukturelles” neben dem eigentlichen Verhalten unter Test. Die BDD-Spezifikation wird dadurch unklarer und enthält meiner Meinung nach mehr “Rauschen”
Wir haben bei uns ja was ähnliches mit Messaging gemacht. Multithreading, Synkronisation und Callback-Handling sind bei uns Aspekte, die vom Bus gemanged werden. Das diese Aspekte vom Bus korrekt gehandelt werden sichern wir in einem Integrationtest. Der Rest der Tests (hier das LoginGateway) wird bei uns ohne Multithreading gegen einen Mock des ServiceBus getestet, um sicher zu stellen, dass Konsumenten den Bus korrekt aufrufen. Dies macht die Tests besser lesbar und produziert deterministischere Ergebnisse (meiner Meinung nach).
Viele Grüße
Björn
December 16th, 2008 at 11:43
@Sergey You’re right, the timing on the build server might be a problem.
@Björn Der Bus ist in unserem Test ebenfalls gemockt. Solange sich das Gateway selber um das Timeout Management kümmert muss dieser Aspekt meiner Ansicht nach auch in einem Unit Test abgesichert werden. Eine Root Cause Analysis befördert allerdings zu Tage dass das Gateway sich nicht um Timeouts kümmern sollte sondern ausschließlich Messages vom Bus erwartet. Der Timeout müsste dann vom Bus per Message signalisiert werden.
Bleibt aber die Frage wie man das in ASP.NET MVC unterbringt. Der MVC Controller muss irgendwann ein ActionResult liefern. Solange er dies nicht vorliegen hat muss der Thread angehalten werden (sei es blockierend oder freigebend).
December 16th, 2008 at 13:11
>>Bleibt aber die Frage wie man das in ASP.NET MVC unterbringt. Der MVC >>Controller muss irgendwann ein ActionResult liefern. Solange er dies nicht >>vorliegen hat muss der Thread angehalten werden (sei es blockierend >>oder freigebend).
Wir hatten dafür zwei verschiedene Methoden im Bus. Send war asynchron und SendSynchronously macht das ganze quasi synchron und benutzt einen ähnlichen Mechanismus den Du beschrieben hast, um auf die Antwort zu einem Send zu warten.
Ich weiss nicht, ob das die beste Variante ist, um das ganze zu lösen, aber so konnten wir Stellen realisieren, wo wir blockierendes Verhalten brauchten.
December 16th, 2008 at 13:54
@Björn: SendSynchronously ist eine gute Idee, Danke!
December 17th, 2008 at 21:53
>>Bleibt aber die Frage wie man das in ASP.NET MVC unterbringt. Der MVC Controller muss irgendwann ein ActionResult liefern. Solange er dies nicht vorliegen hat muss der Thread angehalten werden (sei es blockierend oder freigebend).<<
Nein, muss er nicht!