Mit der Klasse Task ist es möglich, weitere Tasks zu definieren, die ausgeführt werden, sobald der aktuelle Task beendet wurde. Dabei kann u.a. definiert werden, ob die Nachfolge-Aktion nur bei erfolgreicher Beendigung oder im Falle eines Fehlers ausgeführt werden soll. Hierdurch lassen sich mit der Task Parallel Library (TPL) sehr elegant umfangreiche Workflows definieren.

Die Methode ContinueWith()

Durch die Methode ContinueWith() der Klasse Task wird eine weitere Task angelegt. Die Methode ist mehrfach überladen. Hier die einfachste Variante:

public Task ContinueWith(Action<Task> continuationAction)

Bei dem folgenden Beispiel werden drei Objekte der Klasse Task erstellt. Das Objekt task1 wird über die Methode StartNew() erzeugt (Details hierzu im 2. Teil).

Die Methode ContinueWith() erwartet als Parameter ein Delegate Action<Task>, also einen Delegate der als Parameter ein Task-Objekt erwartet und keinen Rückgabewert besitzt.

Wird über das Objekt task1 die Methode ContinueWith() aufgerufen (Zeile 4), so wird nach Beendigung von task1 die Methode MethodeTask2 ausgeführt und als Parameter task1 übergeben (Zeile 13). Einem so erzeugten Task wird immer das Objekt des vorherigen Task übergeben.

public void Run()
{
    Task task1 = Task.Factory.StartNew(MethodeTask1);
    Task task2 = task1.ContinueWith(MethodeTask2);
    Task task3 = task2.ContinueWith(MethodeTask3);
    Console.ReadLine();
}
private void MethodeTask1()
{
    Console.WriteLine("MethodeTask1");
    Thread.Sleep(1000);
}
private void MethodeTask2(Task firstTask)
{
    Console.WriteLine("MethodeTask2");
    Thread.Sleep(1000);
}
private void MethodeTask3(Task secondTask)
{
    Console.WriteLine("MethodeTask3");
    Thread.Sleep(1000);
}

Die Schreibweise lässt sich mit Lambda-Ausdrücken noch weiter vereinfachen:

Task task1 = Task.Factory.StartNew(() =>
{
    // ...
});
Task task2 = task1.ContinueWith((firstTask) =>
{
    // ...
});
Task task3 = task2.ContinueWith((secondTask) =>
{
    // ...
});

Sehr häufig findet man auch die folgende Schreibweise:

Task task3 = Task.Factory.StartNew(() =>
{
    // ...
}).ContinueWith((firstTask) =>
{
    // ...
}).ContinueWith((secondTask) =>
{
    // ...
});

Rückgabeparameter

Da an den Task immer der vorherige Task übergeben wird, kann auf dessen Rückgabewert zugegriffen werden. Somit können die einzelnen Rückgabewerte durch die einzelnen Tasks weitergegeben und berücksichtigt werden.

Hierzu wird die folgende Variante von ContinueWith() genutzt:

public Task<TNewResult> ContinueWith<TNewResult>(Func<Task<TResult>, TNewResult> continuationFunction);

Bei diesem Beispiel besitzt der Rückgabewert in jedem Task einen anderen Datentyp. Die Methode StartNew<int>() legt zwar den ersten Task an, liefert aber ein Task<string> zurück; also den letzten Task im Workflow.

Task<string> task3 = Task.Factory.StartNew<int> (() =>
{
    int x = 1;
    Console.WriteLine("Task1: {0}", x);
    return x;   // int
}).ContinueWith<double>((firstTask) =>
{
    double x = firstTask.Result + 1.1;
    Console.WriteLine("Task2: {0}", x);
    return x;   // double
}).ContinueWith<string>((secondTask) =>
{
    string x = string.Format("{0}", secondTask.Result + 1);
    Console.WriteLine("Task3: {0}", x);
    return x;   // string
});
Console.WriteLine("end: {0}", task3.Result);
Console.ReadLine();

Beispiel 1 (Visual Studio 2012) auf GitHub

Jede Task greift über die Eigenschaft Result auf den Rückgabewert des vorherigen Task zu (Zeile 8 und Zeile 13), berechnet einen neuen Wert und gibt diesen zurück (Zeile 10 und Zeile 15).

Das endgültige Ergebnis liefert der letzte Task im Workflow, also task3. Dessen Ergebnis wird in Zeile 17 ausgegeben.

TaskContinuationOptions

Sehr vorteilhaft ist die Möglichkeit festzulegen, unter welcher Bedingung der Nachfolge-Task ausgeführt werden soll. So kann festgelegt werden, dass nur bei erfolgreicher Beendigung oder im Fall eines Fehlers die angegebene Aktion zur Ausführung kommt. Hierzu stehen spezielle Überladungen der Methode ContinueWith() bereit:

public Task ContinueWith(Action<Task<TResult>> continuationAction, TaskContinuationOptions continuationOptions);
public Task<TNewResult> ContinueWith<TNewResult>(Func<Task<TResult>, TNewResult> continuationFunction, TaskContinuationOptions continuationOptions);

Neu ist der Parameter vom Typ TaskContinuationOptions. Diese Aufzählung erlaubt die Festlegung verschiedener Bedingungen, mit der die Methode ContinueWith() den Task erzeugt. Hier ein Auszug:

Da die Aufzählung mit dem Attribut FlagAttribute dekoriert wurde, können mehrere Werte miteinander Oder-verknüpft werden. Aber nicht alle Werte lassen sich miteinander sinnvoll kombinieren. Wird eine fehlerhafte Kombination ausgewählt, so erzeugt die Methode die Ausnahme ArgumentOutOfRangException.

Der Einsatz dieser Option erlaubt interessante Möglichkeiten. Im folgenden Beispiel werden zwei Nachfolge-Tasks erzeugt (Zeile 10 und Zeile 15). Der eine soll nur dann erzeugt werden, wenn taskRoot ohne Fehler beendet wurde. Ein weiterer wird nur im Fehlerfall zur Ausführung gebracht.

int x = 0;
// int x = 1;

Task<int> taskRoot = new Task<int>((para) =>
{
    Console.WriteLine("taskRoot x: {0}", (int)para);
    int y = 1 / (int)para;
    return y;
}, x);

taskRoot.ContinueWith((taskSuccess) =>
{
    Console.WriteLine("taskSuccess");
}, TaskContinuationOptions.NotOnFaulted);

taskRoot.ContinueWith((taskError) =>
{
    Console.WriteLine("taskError");
}, TaskContinuationOptions.OnlyOnFaulted);

taskRoot.Start();

Console.ReadLine();

Der Fehler in taskRoot wird durch eine Division durch 0 erzeugt. Hierzu muss in Zeile 1 und Zeile 2 die Variable x auf 0 (Fehler) oder 1 (kein Fehler) gesetzt werden.

Beispiel 2 (Visual Studio 2012) auf GitHub

Zugriff auf das GUI

Auch beim Programmieren mit der TPL gilt die altbekannte Regel, dass nur der Thread auf Elemente einer Benutzeroberfläche zugreifen darf, der die Steuerelemente auch erzeugt hat. Hält man sich nicht an diese Regel, so kommt es im günstigsten Fall zur Ausnahme InvalidOperationException.

WindowsForms: Control.InvokeRequired und Control.Invoke()

Mit der Eigenschaft InvokeRequired eines Steuerelementes kann geprüft werden, ob dieses direkt angesprochen werden kann. Liefert die Eigenschaft false zurück, so sind keine besonderen Maßnahmen notwendig und es kann direkt auf das Steuerelement zugegriffen werden. Ist die Eigenschaft allerdings true, so ist ein Wechsel in dem Kontext des GUI-Thread erforderlich. Hierzu kann die Methode Invoke() genutzt werden. Als Parameter erwartet diese ein Delegate, dessen Methode im GUI-Thread ausgeführt wird.

Windows Presentation Foundation: Dispatcher.CheckAccess() und Dispatcher.Invoke()

WPF bietet mit der Klasse Dispatcher einen ähnlichen Mechanismus. Über die Methode CheckAccess() kann der Kontext geprüft werden. Mit Invoke() ist bei Bedarf ein Wechsel in den richtigen Kontext möglich.

TaskScheduler.FromCurrentSynchronizationContext()

Mit Hilfe der TPL ist ein Zugriff auf die GUI ebenfalls möglich. Hierzu wird der Methode Task.ContinueWith() ein sogenannter Synchronisationskontext übergeben. Erzeugt wird dieser mit der Klasse TaskScheduler und der statischen Methode FromCurrentSynchronizationContext(). Diese gibt ein Objekt zurück, das ebenfalls von TaskScheduler abgeleitet wurde. Aufgerufen werden muss die Methode natürlich im GUI-Thread, da wir ja den Kontext des GUI-Threads haben wollen.

Die TPL stellt hierzu die Methode ContinueWith() mit folgender Signatur bereit:

public Task ContinueWith(Action<Task> continuationAction, TaskScheduler scheduler);

Das nächste Beispiel führt die Methode Calc() über den Task taskCalc aus (Zeile 4). In Zeile 5 wird ein Nachfolge-Task mit der Methode ContinueWith() angelegt. Dieser Nachfolge-Task soll die Methode FeedbackGUI() im Synchronisationskontext des GUI-Threads ausführen, sobald taskCalc beendet wurde. Somit kann in der Methode FeedbackGUI() auf Steuerelemente zugegriffen werden.

Der Delegate des Nachfolge-Task besitzt einen Parameter vom Typ Task, welcher eine Referenz auf den Vorgänger-Task enthält. In diesem Beispiel eine Referenz auf taskCalc (Zeile 18). Da taskCalc das Rechenergebnis zurückliefert (Zeile 15), kann die Methode FeedbackGUI() auf dieses zugreifen (Zeile 20) und über ein Steuerelement zur Anzeige bringen.

private void Button_Click_1(object sender, RoutedEventArgs e)
{
    Label01.Content = "rechne...";
    var ui = TaskScheduler.FromCurrentSynchronizationContext();
    Task<double> taskCalc = new Task<double>(Calc);
    taskCalc.ContinueWith(FeedbackGUI, ui);
    taskCalc.Start();
}

private double Calc()
{
    double x = 0;
    for (int a = 1; a < 10000000; a++)
        x = Math.Log(a) / Math.Sqrt(a - Math.Sin(x));
    return x;
}

private void FeedbackGUI(Task<double> calc)
{
    Label01.Content = string.Format("fertig: {0}", calc.Result);
}

Dank Lambda-Ausdrücken lässt sich das Beispiel auch etwas kompakter formulieren:

private void Button_Click_1(object sender, RoutedEventArgs e)
{
    Label01.Content = "rechne...";
    var ui = TaskScheduler.FromCurrentSynchronizationContext();
    Task.Factory.StartNew<double> (() =>
    {
        double x = 0;
        for (int a = 1; a < 10000000; a++)
            x = Math.Log(a) / Math.Sqrt(a - Math.Sin(x));
        return x;
    }).ContinueWith((calc) =>
    {
        Label01.Content = string.Format("fertig: {0}", calc.Result);
    }, ui);
}

Beispiel 3 (Visual Studio 2012) auf GitHub

Das Angenehme hierbei ist die Unabhängigkeit zur Oberflächentechnologie. Komponenten die auf diese Weise den Zugriff auf die GUI bereitstellen, können unter WinForms genauso eingesetzt werden, wie auch unter WPF.

Kontrollierter Abbruch eines Workflows

Das Cancellation-Framework wurde im 2. Teil schon kurz vorgestellt. Bei Workflows kann dieses genutzt werden, um eine Prozesskette effektiv zu beenden. Hierbei gibt es allerdings einiges zu beachten.

Als Beispiel soll ein Workflow dienen, der aus zwei Tasks besteht. Der zweite Task wird durch die Methode ContinueWith() mit dem ersten Task verknüpft. Ist der erste Task beendet, wird der zweite automatisch ausgeführt. Beide Tasks führen für einige Sekunden eine Schleife aus und prüfen hierbei einen evtl. vorliegenden Abbruchwunsch. Tritt in einem der beiden Tasks ein Abbruch auf, so soll die gesamte Kette beendet werden.

public void Run()
{
    CancellationTokenSource cts = new CancellationTokenSource();
    CancellationToken ct = cts.Token;
    ct.Register(() => { Console.WriteLine("Token is canceled"); });

    Task task1 = Task.Factory.StartNew((ct1) =>
    {
        Console.Write("Task1 Start");
        CancellationToken ct1Local = (CancellationToken)ct1;
        int x = 0;
        try
        {
            while (x++ < 5)
            {
                Console.Write(".");
                Thread.Sleep(500);
                ct1Local.ThrowIfCancellationRequested();
            }
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Task1 OperationCanceledException");
            throw;
        }
        Console.WriteLine("Task1 End");
    }, ct, ct);

    Task task2 = task1.ContinueWith((firstTask, ct2) =>
    {
        Console.Write("Task2 Start");
        CancellationToken ct2Local = (CancellationToken)ct2;
        int x = 0;
        try
        {
            while (x++ < 5)
            {
                Console.Write(".");
                Thread.Sleep(500);
                ct2Local.ThrowIfCancellationRequested();
            }
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Task2 OperationCanceledException");
            throw;
        }
        Console.WriteLine("Task2 End");
    }, ct, ct);

    Console.ReadLine();
    cts.Cancel();
    Thread.Sleep(1000);
    Console.WriteLine("Status task1: {0}", task1.Status);
    Console.WriteLine("Status task2: {0}", task2.Status);
    Console.ReadLine();
}

Für die Methode Task.StartNew() wird die folgende Variante genutzt (Zeile 7):

public Task StartNew(Action<object> action, object state, CancellationToken cancellationToken);

Der erste Parameter der Methode ist ein Action-Delegate mit einem Parameter vom Typ Object. In diesem Beispiel wird der Delegate als Lambda-Ausdruck angegeben.

Der zweite Parameter, ist der Wert, der beim Aufruf des Delegate an die Methode übergeben wird. Hier übergebe ich das Cancellation-Token an die Task-Methode. Es wäre auch möglich, das Token global anzusprechen, also aus der Task-Methode heraus direkt auf die Variable ct (Zeile 4) zuzugreifen. Solche globalen Objekte fördern nicht unbedingt die Übersicht, deshalb übergebe ich lieber das Cancellation-Token an die Task-Methode. In Zeile 18 wird die Methode ThrowCancellationRequested() der Klasse CancellationToken genutzt, um zu prüfen ob ein Abbruch vorliegt. Intern fragt die Methode die Eigenschaft IsCancellationRequested ab und löst die Ausnahme OperationCanceledException aus, sobald diese auf true ist.

Über den dritten Parameter wird das Cancellation-Token an das Task-Objekt übergeben. Dieses ist wichtig, damit das Task-Objekt seinen Status richtig setzen kann. Nur so kann der Task mit dem Status Canceled beendet werden.

Ähnlich sind Parameter der Methode Task.ContinueWith() (Zeile 29):

public Task ContinueWith(Action<Task, object> continuationAction, object state, CancellationToken cancellationToken);

Der Action-Delegate erhält als ersten Parameter immer eine Referenz auf den Vorgänger-Task. Aus diesem Grund hat das Action-Delegate zwei Parameter.

Über die Methode Register() kann der Klasse CancellationTokenSource ein Delegate übergeben werden. Dieser wird aufgerufen, sobald das Cancellation-Token gesetzt wurde. In Zeile 5 wird diese Methode genutzt, um eine Meldung auszugeben.

Am Ende des Programms wird das Cancellation-Token durch die Methode Cancel() gesetzt und die Zustände der beiden Task-Objekte werden ausgegeben.

Beispiel: Der Abbruch erfolgt erst dann, wenn alle Task-Objekte beendet wurden Die beiden Task-Objekte besitzen den Zustand RanToCompletion.
Beispiel: Der Abbruch erfolgt im ersten Task OperationCancelledException wird im ersten Task-Objekt ausgelöst.Das zweite Task-Objekt kommt nicht mehr zur Ausführung.Beide Tasks besitzen den Zustand Canceled.
Beispiel: Der Abbruch erfolgt im zweiten Task Die Ausnahme OperationCancelledException wird im zweiten Task ausgelöst.Da das erste Task-Objekt ohne Unterbrechung beendet wurde, besitzt dieses den Zustand RanToCompletion. Der zweite Task dagegen den Status Canceled.

Beispiel 4 (Visual Studio 2012) auf GitHub

Es ist wichtig zu wissen, dass es sich um einen kooperativen Abbruchmechanismus handelt. Die Task-Methode entscheidet selber, wann genau dieser beendet wird. Somit können alle notwendigen Aufräumarbeiten (z.B. Locks freigeben) durchgeführt werden.

Ich muss zugeben, dass ich einige Versuche benötigte, bis das Beispiel das machte, was es sollte. Zwei Punkte wurden mir zum Verhängnis:

1) Das Cancellation-Token muss an das Task-Objekt übergeben werden

Wird dieses nicht beachtet, so wird der Nachfolge-Task gestartet, selbst wenn der erste Task abgebrochen wurde. Da das Cancellation-Token weiterhin aktiv ist, wird der Nachfolge-Task unmittelbar nach dem Starten ebenfalls abgebrochen. Beide Tasks besitzen am Ende den Status Faulted.

2) Die Ausnahme OperationCanceledException muss weitergeleitet werden

Wird um die Methode ThrowIfCancellationRequested() ein try-catch gesetzt, so muss die Ausnahme OperationCanceledException im catch-Bereich erneut ausgelöst werden (Zeile 24 und Zeile 46). Ist dieses nicht der Fall, so wird zwar der Nachfolge-Task beendet, der Status wird aber nicht richtig gesetzt. Statt Canceled ist der Status des ersten Task RanToCompletion. Der Status ist dann von Bedeutung, wenn bei der Methode ContinueWith() mit der Aufzählung TaskContinuationOptions weitere Bedingungen für den Nachfolge-Task definiert werden.

ContinueWhenAll() und ContinueWhenAny()

Neben der Methode ContinueWith() bietet die Klasse TaskFactory noch weitere hilfreiche Methoden für das Erstellen von Workflows an. Zwei davon sind ContinueWhenAll() und ContinueWhenAny().

public Task ContinueWhenAll(Task[] tasks, Action<Task[]> continuationAction);
public Task ContinueWhenAny(Task[] tasks, Action<Task> continuationAction);

Die Methode ContinueWhenAll() startet einen Task, nach dem alle übergebenen Tasks abgeschlossen wurden. Dagegen startet ContinueWhenAny() den Task, sobald einer der angegebenen Tasks beendet wurde. Jede Methode besitzt zahlreiche Überladungen.

An die Task-Methode werden die Task-Objekte weitergeben, die für das Ausführen des neuen Task-Objektes geführt haben. Somit lassen sich sehr gut Ergebnisse von den Vorgänger-Tasks über die Eigenschaft Result an den Nachfolge-Task weitergeben.

Beispiel

Die folgende Graphik zeigt den zeitlichen Verlauf des Beispiels. Die Tasks M01 und M02 werden parallel gestartet. Erst wenn M02 beendet wurde, sollen M03 und M04 zur Ausführung kommen. Da M05 die Ergebnisse von M01, M03 und M04 benötigt, darf dieser erst nach Beendigung dieser Tasks gestartet werden.

Jede Task-Methode erhält als Parameter einen Integer-Wert, der verändert und als Rückgabewert dem Nachfolge-Task zur Verfügung gestellt wird.

Die Methode ContinueWhennAll() sorgt für die Synchronisierung der Tasks. In der Grafik wurde dieser Sync Point entsprechend markiert.

Man sollte sich bei diesem Beispiel vor Augen halten, welcher Aufwand für die gleiche Aufgabenstellung ohne den Einsatz der TPL nötigt wäre. So sind in dem Beispiel keinerlei Wait-Handles, Semaphoren, Lock-Objekte oder If-Abfragen enthalten.

Ich denke, dass das Beispiel keine weiteren Erklärungen benötigt.

public void Run()
{
    Task<int> taskM01 = Task.Factory.StartNew<int>(M01, 1);
    Task<int> taskM02 = Task.Factory.StartNew<int>(M02, 2);
    Task<int> taskM03 = taskM02.ContinueWith<int>(M03);
    Task<int> taskM04 = taskM02.ContinueWith<int>(M04);
    Task[] tasks = new Task[] { taskM01, taskM03, taskM04 };
    Task<int> taskM05 = Task.Factory.ContinueWhenAll<int>(tasks, M05);

    taskM05.Wait();
    Console.WriteLine("taskM05: {0}", taskM05.Result);
    Console.ReadLine();
}
private int M01(object x)
{
    Thread.Sleep(800); // Rechenzeit simulieren
    return (int)x + 1;
}
private int M02(object y)
{
    Thread.Sleep(600); // Rechenzeit simulieren
    return (int)y + 2;
}
private int M03(Task<int> task)
{
    Thread.Sleep(500); // Rechenzeit simulieren
    return task.Result + 3;
}
private int M04(Task<int> task)
{
    Thread.Sleep(300); // Rechenzeit simulieren
    return task.Result + 4;
}
private int M05(Task[] tasks)
{
    int result = 0;
    foreach (Task<int> task in tasks)
        result += task.Result;
    return result;
}

Beispiel 5 (Visual Studio 2012) auf GitHub

Auf eine Besonderheit möchte ich aber noch hinweisen. Die Methode M05 enthält als Parameter ein Objekt vom Typ Task[]. Es gibt keine Überladung von ContinueWhennAll() die ein Parameter vom Typ Task<T>[] ermöglicht. Da Task<T> von Task abgeleitet wurde, ist dieses auch nicht unbedingt notwendig.

Man sollte in Betracht ziehen, den genauen Typ in der Task-Methode abzufragen. Erst wenn sichergestellt wurde, dass es sich wirklich um ein Objekt vom Typ Task<T> handelt, kann ohne Bedenken auf die Eigenschaft Result zugegriffen werden.

private int M05(Task[] tasks)
{
    int result = 0;
    foreach (Task task in tasks)
        if (task is Task<int>)
            result += (task as Task<int>).Result;
    return result;
}

Author: Stefan Henneken

I’m Stefan Henneken, a software developer based in Germany. This blog is just a collection of various articles I want to share, mostly related to Software Development. View all posts by Stefan Henneken