Task w C#: Task.Run, WhenAll, WhenAny, anulowanie i wątki

Task to obiekt, który reprezentuje trwającą pracę. Może działać na wątku z puli, czekać na timer albo na odpowiedź z sieci; w każdym przypadku możesz zapytać, czy się zakończył, pobrać jego wynik, poczekać na niego albo dołączyć do niego kolejną pracę. async i await to składnia języka nad taskami. Ta strona omawia samo API Task oraz wątki.

Task.Run: praca w puli wątków

Task.Run przyjmuje lambdę i kolejkuje ją w puli wątków, czyli zestawie wątków, które .NET trzyma w gotowości, żeby nie tworzyć nowego dla każdego zadania. Zwraca Task albo Task<T>, jeśli lambda zwraca wartość.

using System;
using System.Threading.Tasks;

class Program
{
    static int CountPrimes(int max)
    {
        int count = 0;
        for (int n = 2; n <= max; n++)
        {
            bool prime = true;
            for (int d = 2; d * d <= n; d++)
            {
                if (n % d == 0) { prime = false; break; }
            }
            if (prime) count++;
        }
        return count;
    }

    static void Main()
    {
        Task<int> task = Task.Run(() => CountPrimes(200000));
        Console.WriteLine("Started, completed yet: " + task.IsCompleted);

        int primes = task.Result;   // blocks until the task finishes
        Console.WriteLine("Primes up to 200,000: " + primes);
        Console.WriteLine("Status: " + task.Status);
    }
}

Wynik:

Started, completed yet: False
Primes up to 200,000: 17984
Status: RanToCompletion

Result blokuje wątek wywołujący, dopóki task się nie zakończy. W Main programu konsolowego to akceptowalne, ale w kodzie asynchronicznym pisze się int primes = await task;, co czeka bez blokowania. To samo dotyczy task.Wait() dla Task bez wyniku.

Task.Run służy do pracy obciążającej procesor: obliczeń, które zajmują rdzeń. Do wejścia-wyjścia, takiego jak czytanie pliku albo wywołanie web API, użyj metody asynchronicznej, którą biblioteka już udostępnia (File.ReadAllTextAsync, HttpClient.GetStringAsync), i czekaj na nią bezpośrednio przez await. Dopóki operacja wejścia-wyjścia trwa, asynchroniczne wywołanie z await nie zajmuje żadnego wątku, więc opakowanie go w Task.Run nic nie daje i tylko dodaje przeskok przez pulę wątków. Przypadek, który rzeczywiście blokuje wątek z puli, to Task.Run wokół synchronicznego blokującego wywołania, takiego jak File.ReadAllText: ten wątek bezczynnie czeka, aż odpowie dysk albo sieć.

Kilka tasków naraz: WhenAll

Podziel niezależną pracę na taski, uruchom je wszystkie i połącz wyniki przez Task.WhenAll. Na maszynie z kilkoma rdzeniami części działają równolegle.

using System;
using System.Linq;
using System.Threading.Tasks;

class Program
{
    static long SumRange(int from, int to)
    {
        long sum = 0;
        for (int i = from; i < to; i++) sum += i % 7;
        return sum;
    }

    static void Main()
    {
        int chunks = 4, size = 5000000;

        Task<long>[] parts = Enumerable.Range(0, chunks)
            .Select(c => Task.Run(() => SumRange(c * size, (c + 1) * size)))
            .ToArray();

        long[] results = Task.WhenAll(parts).Result;
        Console.WriteLine("Parts: " + string.Join(", ", results));
        Console.WriteLine("Total: " + results.Sum());
    }
}

Wynik:

Parts: 14999995, 14999999, 15000003, 15000000
Total: 59999997

Każdy fragment pracuje na własnym zakresie i zwraca własną sumę, więc taski niczego nie współdzielą i nie potrzebują blokad. WhenAll zwraca wyniki w kolejności tasków, niezależnie od tego, w jakiej kolejności skończyły wątki. Gdy taski muszą aktualizować współdzielony stan, zobacz lock.

WhenAny i limity czasu

Task.WhenAny kończy się, gdy zakończy się pierwszy z jego tasków, i zwraca ten task. Ściganie prawdziwej pracy z Task.Delay daje limit czasu:

using System;
using System.Threading.Tasks;

class Program
{
    static async Task<string> QueryServerAsync(string name, int ms)
    {
        await Task.Delay(ms);
        return name + " answered after " + ms + " ms";
    }

    static async Task RunAsync()
    {
        Task<string> fast = QueryServerAsync("fast mirror", 100);
        Task<string> slow = QueryServerAsync("main server", 800);
        Task<string> first = await Task.WhenAny(fast, slow);
        Console.WriteLine(await first);

        Task<string> work = QueryServerAsync("report service", 2000);
        Task timeout = Task.Delay(300);
        if (await Task.WhenAny(work, timeout) == timeout)
        {
            Console.WriteLine("Timed out after 300 ms");
        }
    }

    static void Main()
    {
        RunAsync().GetAwaiter().GetResult();
    }
}

Wynik:

fast mirror answered after 100 ms
Timed out after 300 ms

WhenAny zwraca sam zwycięski task, a nie jego wynik, więc czekasz na niego ponownie przez await, żeby dostać wartość (albo wyjątek, jeśli się nie powiódł). Pamiętaj, że przegrane taski działają dalej; WhenAny ich nie zatrzymuje. Żeby naprawdę zatrzymać wolną pracę, anuluj ją tokenem, jak pokazano niżej. W .NET 6 i nowszych await work.WaitAsync(TimeSpan.FromMilliseconds(300)) wyraża ten sam limit czasu w jednym wywołaniu.

Kontynuacje z ContinueWith

Przed await kod uruchamiany po zakończeniu taska zapisywało się przez ContinueWith. Nadal spotkasz to w starszym kodzie:

using System;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        Task<int> load = Task.Run(() => 21);

        Task<string> report = load
            .ContinueWith(t => t.Result * 2)
            .ContinueWith(t => "Report total: " + t.Result);

        Console.WriteLine(report.Result);
    }
}

Wynik:

Report total: 42

Każda kontynuacja otrzymuje poprzedni task i uruchamia się, gdy ten się zakończy. await robi to samo ze znacznie mniejszym narzutem, naturalnie obsługuje wyjątki i wraca do właściwego kontekstu, więc w nowym kodzie wybieraj właśnie jego. ContinueWith uruchamia się nawet wtedy, gdy poprzedni task się nie powiódł albo został anulowany, więc starszy kod musi sprawdzać t.IsFaulted albo przekazywać TaskContinuationOptions, żeby nie wywołać .Result na nieudanym tasku.

Anulowanie przez CancellationToken

Taski anuluje się kooperacyjnie. CancellationTokenSource wydaje token; kod wykonujący pracę sprawdza token i sam się zatrzymuje. Nic nie jest przerywane siłą.

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static async Task<int> ProcessOrdersAsync(int count, CancellationToken token)
    {
        int done = 0;
        for (int i = 0; i < count; i++)
        {
            token.ThrowIfCancellationRequested();   // stop between items
            await Task.Delay(100, token);           // pass the token on
            done++;
        }
        return done;
    }

    static async Task RunAsync()
    {
        using (var cts = new CancellationTokenSource())
        {
            cts.CancelAfter(350);                   // cancel in 350 ms
            try
            {
                int n = await ProcessOrdersAsync(10, cts.Token);
                Console.WriteLine("Processed " + n);
            }
            catch (OperationCanceledException)
            {
                Console.WriteLine("Cancelled before finishing all 10 orders");
            }
        }

        using (var cts = new CancellationTokenSource())
        {
            int n = await ProcessOrdersAsync(3, cts.Token); // never cancelled
            Console.WriteLine("Processed " + n);
        }
    }

    static void Main()
    {
        RunAsync().GetAwaiter().GetResult();
    }
}

Wynik:

Cancelled before finishing all 10 orders
Processed 3

Token jest przekazywany w dół przez każdą warstwę, a asynchroniczne API, takie jak Task.Delay, HttpClient i wywołania bazy danych, też go przyjmują, więc anulowanie zatrzymuje je w trakcie czekania. Łap OperationCanceledException; TaskCanceledException, który rzucają niektóre API, dziedziczy po nim. CancellationTokenSource implementuje IDisposable, stąd using.

Thread a Task

Thread to wątek systemu operacyjnego, który tworzysz sam. Jest cięższy: każdy ma własny stos (domyślnie 1 MB w Windows), jego uruchomienie zajmuje zauważalny czas i nie daje ci wyniku ani obsługi wyjątków.

using System;
using System.Threading;

class Program
{
    static void Main()
    {
        int result = 0;
        var worker = new Thread(() =>
        {
            Thread.Sleep(100);
            result = 6 * 7;
        });
        worker.Start();
        Console.WriteLine("Thread started");
        worker.Join();          // wait for it to finish
        Console.WriteLine("Result: " + result);
    }
}

Wynik:

Thread started
Result: 42

Po Thread sięgaj tylko wtedy, gdy potrzebujesz czegoś, czego pula nie daje: długo działającej pętli, która nigdy nie powinna pożyczać wątku z puli, konkretnego priorytetu albo jednowątkowego apartamentu (STA) dla COM. Dla długo działających tasków Task.Factory.StartNew(work, TaskCreationOptions.LongRunning) dostaje dedykowany wątek, zachowując API tasków.

Parallel.For i Parallel.ForEach

Do zadania "zrób to z każdym elementem, używając wszystkich rdzeni" System.Threading.Tasks.Parallel jest prostszy niż ręczne tworzenie tasków. Dzieli zakres między wątki i kończy się, gdy wszystkie iteracje są gotowe:

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        int[] sizes = { 120, 340, 90, 410, 275, 60 };
        long totalPixels = 0;

        Parallel.For(0, sizes.Length, i =>
        {
            long pixels = (long)sizes[i] * sizes[i];      // "process" image i
            Interlocked.Add(ref totalPixels, pixels);      // thread-safe add
            Console.WriteLine("image " + i + " on thread " + Thread.CurrentThread.ManagedThreadId);
        });

        Console.WriteLine("Total pixels: " + totalPixels);
    }
}

Przykładowy wynik:

image 4 on thread 7
image 3 on thread 6
image 5 on thread 8
image 0 on thread 1
image 1 on thread 4
image 2 on thread 5
Total pixels: 385425

Linie dla poszczególnych obrazów pojawiają się w takiej kolejności, w jakiej wykonały je wątki, a identyfikatory wątków zmieniają się między uruchomieniami; stała jest tylko suma. Taka jest natura pętli równoległych: iteracje nie mogą zależeć od wzajemnej kolejności, a każdy współdzielony stan wymaga synchronizacji (tutaj Interlocked.Add). Parallel.ForEach robi to samo na dowolnej kolekcji, a .NET 6 dodaje Parallel.ForEachAsync dla asynchronicznych ciał pętli.

Częste błędy

• .Result albo .Wait() w kodzie asynchronicznym. Blokuje wątek i w aplikacjach z interfejsem może spowodować zakleszczenie; zamiast tego użyj await.
• Łapanie oryginalnego typu wyjątku wokół .Result. Błędy przychodzą opakowane w AggregateException; await je rozpakowuje.
• Uruchom i zapomnij. Task.Run(...) bez await i bez zapisania taska gubi jego wyjątki.
• Task.Run wokół wejścia-wyjścia. Wokół metody asynchronicznej dodaje tylko przeskok przez pulę wątków; wokół metody blokującej, takiej jak File.ReadAllText, blokuje wątek z puli na cały czas oczekiwania. Czekaj bezpośrednio przez await na asynchroniczną metodę wejścia-wyjścia.
• Ignorowanie tokenu. Przekazanie CancellationToken nic nie daje, jeśli praca go nie sprawdza.
• Niesynchronizowany współdzielony stan w kodzie równoległym. Użyj Interlocked, lock albo daj każdemu taskowi własne dane.