Читать книгу: «Dojos für Entwickler 2», страница 4
Async/Sync
Aufgabe des Asynchronizer-Bausteins ist es, den eingehenden Datenfluss auf einen anderen Thread umzuleiten. Der Asynchronizer hat dazu eine Process-Methode als Eingang und ein Result-Event als Ausgang. Die Kernidee beim Asynchronizer ist: Beim Aufruf der Process-Methode wird der Result-Event ausgelöst, allerdings auf einem anderen Thread. Und natürlich wird der Parameter der Process-Methode an den Result-Event weitergereicht, sodass der Datenfluss durch den Asynchronizer hindurchfließt. Soll ein Baustein den Datenfluss einfach nur durchreichen, sieht die Implementation so wie in Listing 1 aus.
Listing 1
Den Datenfluss durchreichen.
public class Asynchronizer<T> { public void Process(T input) { Result(input); } public event Action<T> Result; }
Um den Event auf einem anderen Thread auszulösen, muss ein neuer Thread gestartet werden. Das kann dadurch erfolgen, dass Sie eine neue Instanz eines Threads anlegen oder einen freien Thread aus dem ThreadPool verwenden. Dabei kann eine Lambda-Expression übergeben werden, die auf dem neuen Thread ausgeführt wird. Wenn man einen neuen Thread anlegt, muss das Starten des Threads explizit durch Start erfolgen, siehe Listing 2.
Listing 2
Den neuen Thread explizit starten.
public class Asynchronizer<T> { public void Process(T input) { var thread = new Thread(() => Result(input)); thread.Start(); } public event Action<T> Result; }
Ganz einfach, oder? Die erste Frage lautet nun: Wie kann man den Asynchronizer automatisiert testen? Des Weiteren stellt sich die Frage, ob es eine gute Idee ist, jeweils einen neuen Thread zu erzeugen. Würde der Zugriff auf den ThreadPool Vorteile bringen? Doch zunächst zum Testen.
Asynchronen Code testen
Die Herausforderung beim automatisierten Testen von asynchronem Code liegt darin, dass der zu testende Code im Testablauf relativ schnell zum Aufrufer zurückkehrt. Denn schließlich wird die zu erledigende Aufgabe auf einen anderen Thread in den Hintergrund verschoben. Kehrt der zu testende Code jedoch in die Testmethode zurück, wird diese bis zu ihrem Ende ausgeführt und dann beendet. Damit ist dann auch der Test schon zu Ende, bevor es richtig losging. Die Herausforderung besteht also darin, in der Testmethode auf die Ausführung des Hintergrundthreads zu warten. Wie wäre es mit Thread.Sleep()? Listing 3 zeigt ein Beispiel.
Listing 3
Nicht optimal: Thread.Sleep.
[Test] public void Naive_approach_to_async_ tests() { var result = 0; sut.Result += x => result = x; sut.Process(42); Thread.Sleep(500); Assert.That(result, Is.EqualTo(42)); }
Aber wie lange soll auf den Hintergrundthread gewartet werden? Eine halbe Sekunde, sprich 500 ms wie oben gezeigt? Meist genügt das. Doch wenn der Test auf einem gut ausgelasteten Continuous Integration Server läuft, dann kann die Wartezeit zu gering sein. Also zur Sicherheit fünf Sekunden warten? Nein, natürlich ist das keine gute Idee. Der Test soll natürlich exakt so lange auf den Hintergrundthread warten, bis dieser fertig ist. Das erreichen Sie mit einem WaitHandle, Listing 4 zeigt den entsprechenden Code. Vergessen Sie Thread. Sleep für solche Fälle.
Listing 4
Besser : WaitHandle.
[TestFixture] public class AsynchronizerTests { private EventWaitHandle waitHandle; private Asynchronizer<int> sut; [SetUp] public void Setup() { sut = new Asynchronizer<int>(); waitHandle = new EventWaitHandle( false, EventResetMode.ManualReset); } [Test] public void Result_event_is _raised_with_parameter() { var result = 0; sut.Result += x => { result = x; waitHandle.Set(); }; sut.Process(42); Assert.That(waitHandle.WaitOne(500), Is.True); Assert.That(result, Is.EqualTo(42)); } }
Das WaitHandle wird verwendet, um nach Aufruf der Process-Methode darauf zu warten, dass der Result-Event auf dem neu gestarteten Thread seine Arbeit verrichtet hat. Das WaitHandle befindet sich anfangs im Zustand „nicht gesetzt“. Wird waitHandle.WaitOne aufgerufen, wird der aktuelle Thread so lange angehalten, bis das WaitHandle gesetzt ist. Daher muss das WaitHandle im Result-Event mit Set gesetzt werden. Auf diese Weise wacht der Thread, auf dem der Test läuft, dann wieder auf und kann sein Assert ausführen.
Damit der Test bei einem Fehler nicht unendlich lange wartet, habe ich bei WaitOne einen Timeout von 500 ms definiert. Dadurch wird das WaitHandle entweder durch Set innerhalb von 500 ms gesetzt oder durch einen Timeout. Der Timeout tritt automatisch ein, wenn das WaitHandle nicht nach der hinterlegten Zeit mit Set gesetzt wird, in diesem Fall nach 500 ms. Der Timeout wird dadurch signalisiert, dass WaitOne false zurückliefert.
Um den Timeout im Test zu erkennen, habe ich ein Assert ergänzt, welches prüft, ob WaitOne mit true beendet wurde. Wenn das der Fall ist, ist das WaitHandle ordnungsgemäß mit Set gesetzt worden, sprich, der Result-Event wurde ausgelöst. Liefert WaitOne jedoch false zurück, ist im Test etwas schiefgelaufen, weil das WaitHandle durch einen Timeout ausgelöst wurde.
Hier ist es nun tatsächlich angemessen, darüber nachzudenken, ob 500 ms als Timeout ausreichend sind. Auf einem Continuous Integration Server kann das schon mal knapp bemessen sein. Da die Timeout-Zeitspanne nur dann in Anspruch genommen wird, wenn etwas schiefläuft, können Sie den Timeout nun ohne Not auf mehrere Sekunden hochsetzen. Solange alles in Ordnung ist, und das sollte der Normalfall sein, läuft der Test dann durch, so schnell es eben geht. Nur für den hoffentlich seltenen Fall, dass jemand den Code „kaputt gemacht hat“, tritt die Timeout-Wartezeit in Kraft.
ThreadPool
Alternativ zum Erzeugen eines neuen Threads kann auch ein Thread aus dem ThreadPool verwendet werden. Dies hat Vorteile bei kurzlaufenden Operationen, da der ThreadPool weniger Overhead benötigt. Die Threads sind bereits erzeugt und warten nur auf einen neuen Auftrag.
Der wichtigste Unterschied liegt jedoch darin, dass Threads aus dem ThreadPool als Hintergrundthreads laufen, während neu erzeugte Threads standardmäßig Vordergrundthreads sind. Hintergrundthreads werden automatisch beendet, sobald alle Vordergrundthreads beendet sind. In der Regel hat ein Programm lediglich einen einzigen Vordergrundthread, der die UI übernimmt. Beendet man das Programm, werden automatisch alle Hintergrundthreads mitbeendet. Die selbst erzeugten Vordergrundthreads laufen jedoch weiter, sodass das Programm in diesem Fall nicht einfach dadurch beendet wird, dass das UI geschlossen wird.
Man kann natürlich auch selbst erzeugte Threads als Hintergrundthread starten, indem vor dem Start die Eigenschaft IsBackground auf true gesetzt wird. Was man im Einzelfall benötigt, hängt vom konkreten Anwendungsfall ab.
Und wieder synchron
Mithilfe des Asynchronizer-Bausteins kann die Ausführung des im Fluss nachfolgenden Bausteins auf einen anderen Thread verlagert werden. Doch was tun, wenn die Ergebnisse aus dem Hintergrundthread in das UI übernommen werden sollen? Dann ergibt sich bei Windows Forms, WPF und Silverlight das Problem, dass die Controls nur aus dem UI-Thread heraus angesprochen werden dürfen. Die Windows-Forms-Controls haben dafür einen Mechanismus: InvokeRequired und Invoke, siehe Listing 5.
Listing 5
InvokeRequired nutzen.
if (listBox1.InvokeRequired) { listBox1.Invoke(new MethodInvoker(() => listBox1.Items.Add(t))); } else { listBox1.Items.Add(t); }
Bevor in diesem Beispiel ein weiterer Eintrag in die ListBox eingefügt wird, wird durch Abfrage von InvokeRequired geprüft, ob der Aufruf mit Invoke erfolgen muss. Ist das der Fall, wird dem Invoke-Aufruf eine Lambda-Expression übergeben, welche den Eintrag in die ListBox schreibt. Invoke sorgt dafür, dass die Lambda-Expression auf dem UI-Thread ausgeführt wird.
Das kann man so machen, doch man sollte es nicht so machen. Denn auf diese Weise wird der UI-Code mit Aspekten der Parallelisierung kontaminiert. Außerdem müsste ein UI, das zunächst synchron und erst später asynchron verwendet wird, modifiziert werden. Das ist unschön. Zum einen verstößt diese Vorgehensweise gegen das Open Closed Principle (OCP), welches besagt, dass eine Klasse offen für Erweiterungen, aber geschlossen gegenüber Modifikationen sein sollte. Noch schwerer wiegt das Argument, dass auf diese Weise das Prinzip die Separation of Concerns (SoC) verletzt würde. Der Aspekt der Parallelisierung würde mit der eigentlichen Aufgabe des UIs vermischt.
Abhilfe schafft die Verwendung des SynchronizationContext bei Windows Forms bzw. des Dispatchers bei WPF und Silverlight. Beides sind Infrastrukturelemente, mit denen die Synchronisation auf einen Zielthread erfolgen kann. Nutzt man diese Infrastruktur, kann man auf einfache Weise einen Synchronizer-Baustein realisieren, der den Result-Event auf dem gewünschten Zielthread ausführt.
Der Synchronizer erhält dieselbe Schnittstelle wie der Asynchronizer: eine Process- Methode als Eingang sowie einen Result- Event als Ausgang. Auch hier sorgt der Baustein wieder dafür, dass der Parameter durchgereicht wird, der Datenfluss also einfach hindurchfließt. Damit der Baustein im Sinne eines Standardbausteins vielseitig einsetzbar ist, erhält er einen generischen Typparameter, der den Datentyp von Ein- und Ausgang bestimmt. Hier zahlt es sich aus, dass beim Flow-Design immer nur maximal ein Parameter verwendet wird. Auf diese Weise ist es nämlich leicht möglich, Standardbausteine zu erstellen und Flows neu zusammenzustöpseln. Zwar könnte man auch Varianten der Standardbausteine anlegen, die mehrere Parameter unterstützen, aber dadurch würde die ganze Sache doch etwas komplizierter.
Listing 6 zeigt die Implementation des Synchronizers für Windows Forms.
Listing 6
Synchronizer für Windows Forms.
public class Synchronizer<T> { private readonly SynchronizationContext synchronizationContext; public Synchronizer() { synchronizationContext = SynchronizationContext.Current ?? new SynchronizationContext(); } public void Process(T input) { synchronizationContext.Send(_ => Result(input), null); } public event Action<T>Result; }
Der SynchronizationContext wird im Konstruktor des Bausteins ermittelt und in einem Feld des Synchronizers abgelegt. Auf diese Weise steht er in der Process-Methode zur Verfügung, um mittels Send eine Lambda-Expression an den Zielthread zu übergeben. Die Lambda-Expression ruft wieder den Result-Event auf, wie das beim Asynchronizer schon der Fall war. Damit der Synchronizer funktioniert, muss sein Konstruktor auf dem Thread aufgerufen werden, auf den später synchronisiert werden soll. Normalerweise ist das der Hauptthread, auf dem das Programm initialisiert wird. Aber Obacht!
Der SynchronizationContext funktioniert nicht mit beliebigen Threads. Windows Forms treibt hinter den Kulissen etwas Magie. Es realisiert nämlich eine Message-Loop, die dafür sorgt, dass Aufrufe an den UI-Thread übergeben werden können. Zwischen beliebigen Threads zu synchronisieren ist auf diese Weise nicht möglich. Eine allgemeine Lösung dieses Problems würde allerdings den Rahmen der übung sprengen. Für die Praxis ist das kein Beinbruch, da die Synchronisierung in aller Regel im Zusammenhang mit dem UI auftritt. Genau in diesem Szenario funktionieren SynchronizationContext bzw. Dispatcher bei WPF und Silverlight wunderbar.
Doch wie testet man nun den Synchronizer? Ich sehe hier im Wesentlichen zwei Aspekte, die durch automatisierte Tests sichergestellt werden sollten :
Zum einen muss sichergestellt sein, dass der Eingangsparameter der Process-Methode an den Result-Event übergeben wird. Der Flow darf schließlich nicht unterbrochen werden.
Zum anderen sollte ein Test sicherstellen, dass der Eventhandler, der durch den Result-Event ausgelöst wird, auf dem korrekten Thread abläuft.
Für den zweiten Test ist zu berücksichtigen, dass dies nur im Rahmen einer Windows-Forms-Anwendung funktioniert. Der erste Test für den Synchronizer erfolgt analog zum Test des Asynchronizers: Die Process-Methode wird mit einem Parameter aufgerufen, um anschließend zu überprüfen, ob der Result-Event den Parameter wieder mitführt, siehe Listing 7.
Listing 7
Den Synchronizer testen.
[TestFixture] public class SynchronizerTests { private Synchronizer<int> sut; private int result; [SetUp] public void Setup() { sut = new Synchronizer<int>(); sut.Result += x => result = x; } [Test] public void Parameter_wird_durchgereicht() { sut.Process(42); Assert.That(result, Is.EqualTo(42)); } )
Da durch den Synchronizer kein weiterer Thread gestartet wird, der Synchronizer arbeitet synchron, muss im Test nicht auf die Ausführung eines anderen Threads mittels WaitHandle gewartet werden, wie das beim Asynchronizer der Fall war. Dieser Test war also ganz einfach.
Trickreicher Test
Lange beschäftigt hat mich dagegen die Frage, wie man automatisiert testet, ob der Synchronizer in einer Windows- Forms-Anwendung tatsächlich den Eventhandler auf dem UI-Thread ausführt. Dazu müssen zwei Voraussetzungen geschaffen werden:
Der Synchronizer darf erst instanziert werden, nachdem eine Windows- Forms-Form instanziert wurde. Andernfalls wird dem Vordergrundthread kein WindowsFormsSynchronization- Context hinzugefügt.
Die Windows-Forms-Message-Loop muss laufen. Dazu muss Application.Run aufgerufen werden.
Der Aufruf von Application.Run in Listing 8 erfolgt synchron, das heißt, er kehrt erst zum Aufrufer zurück, wenn die „Awendung“ beendet wird. Um die „Anwendung“ zu beenden, ist es erforderlich, die an Application.Run übergebene Form zu schließen. Ich rufe dazu in der Lambda- Expression des Result-Events, nach dem überprüfen der Thread-ID, die Close- Methode der Form auf. Dadurch wird Application.Run verlassen. Das ist ein etwas trickreicher Test, der einige Annahmen über das Verhalten von Windows Forms trifft. Aber genau dieses Szenario soll hier getestet werden, insofern bin ich mit dem Test zufrieden.
Listing 8
Test für Windows Forms.
[Test] public void Result_Event_wird_auf_Zielthread_ausgeführt_wenn_das_Ziel_WinForms_ist() { var myForm = new Form(); sut = new Synchronizer<int>(); var mainThreadId = Thread.CurrentThread. ManagedThreadId; sut.Result += _ => { Assert.That(Thread.CurrentThread. ManagedThreadId, Is.EqualTo(mainThreadId)); myForm.Close(); }; var thread = new Thread(() => sut.Process(1)); thread.Start(); Application.Run(myForm); }
Scatter/Gather
Nun zum zweiten Teil der übung. Häufig lassen sich Aufgaben schneller erledigen, wenn mehrere Elemente gleichzeitig bearbeitet werden. Zu einer echten Beschleunigung kommt es nur, wenn mehrere Kerne des Prozessors ausgenutzt werden können. Andernfalls kommt es sogar zu Verzögerungen, da das Wechseln von einem zum anderen Thread einen gewissen Overhead mit sich bringt.
Das Szenario für den Einsatz der Scatter- und Gather-Bausteine sieht so aus, dass in einer Liste mehrere Elemente zur Verfügung stehen, die von zwei parallel ausgeführten Bausteinen bearbeitet werden sollen. Scatter entnimmt also Elemente aus der Liste und reicht sie jeweils an den einen oder anderen Baustein weiter. Natürlich erfolgt die Weitergabe der Elemente auf einem anderen Thread, sodass die beiden Ausgänge des Scatter- Bausteins auf je einem eigenen Thread laufen.
Damit man diese Form der Parallelisierung zu einem späteren Zeitpunkt so einfach wie möglich in einen Flow integrieren kann, soll das API von Scatter und Gather ebenfalls auf Aufzählungen IEnumerable<T> arbeiten. So kann ein Baustein, der ursprünglich für die Bearbeitung aller Elemente zuständig war, unverändert mehrfach instanziert werden und mit je einem Ausgang des Scatter- Bausteins verbunden werden. Würde der Scatter-Baustein jeweils ein zu bearbeitendes Element am Ausgang zur Verfügung stellen statt eine Aufzählung, dann müsste der Bearbeitungsbaustein angepasst werden.
Die folgenden Codeausschnitte zeigen, wie die Scatter- und Gather-Bausteine verwendet werden. Listing 9 zeigt, wie die Logik für das Ermitteln von Stichwörtern aus JPEG-Dateien zusammengesteckt wird, ohne Parallelisierung oder Asynchronizität.
Listing 9
Stichwörter ermitteln, einfache Version.
public class Eindeutige_Stichwörter_ ermitteln { private readonly Action<Tuple<string, string>> process; public Eindeutige_Stichwörter_ermitteln() { var dateinamen_suchen = new Dateinamen_suchen(); var alle_Stichwörter_ermitteln = new Alle_Stichwörter_ermitteln(); var eindeutige_Stichwörter_filtern = new Eindeutige_Stichwörter_filtern(); dateinamen_suchen.Result += alle_Stichwörter_ermitteln.Process; alle_Stichwörter_ermitteln.Result += eindeutige_Stichwörter_ filtern.Process; eindeutige_Stichwörter_ filtern.Result += x => Result(x); process = path_und_SearchPattern => dateinamen_suchen.Process( path_und_SearchPattern); } public void Process(Tuple<string, string> path_und_SearchPattern) { process(path_und_SearchPattern); } public event Action<IEnumerable<string>> Result; }
In diese Platine werden nun Scatter- und Gather-Bausteine eingebracht, um das Auslesen der Stichwörter zu parallelisieren, siehe Listing 10. Dazu wird der Baustein Alle_Stichwörter_ermitteln zweimal instanziert. Der Baustein erwartet eine Liste von Dateinamen am Eingang und liefert dazu eine Liste aller gefundenen Stichwörter am Ausgang. Das Ermitteln der Dateinamen wird vom Baustein Dateinamen_suchen übernommen. Für die Parallelisierung der Aufgabe wird daher der Ausgang von Dateinamen_suchen mit dem Eingang eines Scatter-Bausteins verbunden. Dieser sorgt dafür, dass die Eingangsdaten auf die beiden Ausgänge verteilt werden. An die beiden Ausgänge des Scatter-Bausteins werden daher die beiden Alle_Stichwörter_ermitteln-Bausteine gehängt. Um die Ergebnisse wieder zusammenzuführen, wird jeweils der Ausgang von Alle_Stichwörter_ermitteln mit einem Eingang eines Gather-Bausteins verbunden. Dieser sammelt die Ergebnisse und stellt sie an seinem Ausgang zur Verfügung. Abbildung 1 zeigt den Flow vor, Abbildung 2 nach der Parallelisierung.
[Abb. 1]
Vor der Parallelisierung ...
[Abb. 2]
... und nach der Parallelisierung.
Listing 10
Stichwörter ermitteln, parallel.
public class Eindeutige_Stichwörter_ermitteln_parallel { private readonly Action<Tuple<string, string>> process; public Eindeutige_Stichwörter_ ermitteln_parallel() { var dateinamen_suchen = new Dateinamen_suchen(); var alle_Stichwörter_ermitteln1 = new Alle_Stichwörter_ermitteln(); var alle_Stichwörter_ermitteln2 = new Alle_Stichwörter_ermitteln(); var eindeutige_Stichwörter_filtern = new Eindeutige_Stichwörter_filtern(); var scatter = new Scatter<string>(); var gather = new Gather<string>(); dateinamen_suchen.Result += scatter.Process; scatter.Output1 += alle_Stichwörter_ ermitteln1.Process; scatter.Output2 += alle_Stichwörter_ ermitteln2.Process; alle_Stichwörter_ermitteln1.Result += gather.Input1; alle_Stichwörter_ermitteln2.Result += gather.Input2; gather.Result += eindeutige_Stichwörter_ filtern.Process; eindeutige_Stichwörter_filtern.Result += x => Result(x); process = path_und_SearchPattern => dateinamen_suchen.Process( path_und_SearchPattern); } public void Process(Tuple<string, string> path_und_SearchPattern) { process(path_und_SearchPattern); } public event Action<IEnumerable< string>> Result; }
Zu beachten ist, dass der Flow, der durch die Platine realisiert wird, durch die Verwendung von Scatter und Gather asynchron abläuft. Nach Aufruf der Process-Methode der Platine kehrt der Kontrollfluss zum Aufrufer zurück, während der Flow der Platine asynchron auf weiteren Threads noch läuft. Auch der Result- Event der Platine wird asynchron auf einem anderen Thread ausgelöst.
Die Parallelisierung der Stichwortsuche zeigt deutliche Geschwindigkeitsvorteile. Das ist natürlich nicht in jedem Fall so. Ein triviales Beispiel, bei dem einfach nur Zahlen multipliziert werden, wird durch die Parallelisierung nicht beschleunigt. Im Gegenteil! Durch den Overhead des Multithreadings wird der Durchsatz sogar verringert.
