Task di C#: Task.Run, WhenAll, WhenAny, Cancellation dan Thread

Task adalah object yang mewakili pekerjaan yang sedang berlangsung. Pekerjaan itu mungkin sedang berjalan di thread dari thread pool, menunggu sebuah timer, atau menunggu balasan dari jaringan; apa pun kasusnya, kamu bisa menanyakan apakah ia sudah selesai, mengambil hasilnya, menunggunya, atau menempelkan pekerjaan lain padanya. async dan await adalah sintaks bahasa di atas task. Halaman ini membahas API Task itu sendiri, serta thread.

Task.Run: pekerjaan di thread pool

Task.Run menerima sebuah lambda dan memasukkannya ke antrean thread pool, kumpulan thread yang disimpan .NET agar tidak perlu membuat thread baru untuk setiap pekerjaan. Method ini mengembalikan Task, atau Task<T> jika lambda-nya mengembalikan nilai.

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);
    }
}

Output:

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

Result memblokir thread pemanggil sampai task selesai. Itu bisa diterima di Main sebuah program console, tetapi di dalam kode async kamu akan menulis int primes = await task;, yang menunggu tanpa memblokir. Hal yang sama berlaku untuk task.Wait() pada Task yang tidak punya hasil.

Task.Run dipakai untuk pekerjaan CPU-bound: perhitungan yang membuat sebuah core terus sibuk. Untuk I/O, seperti membaca file atau memanggil web API, pakai method async yang sudah disediakan library (File.ReadAllTextAsync, HttpClient.GetStringAsync) lalu await langsung. Selama I/O masih berjalan, pemanggilan async yang ditunggu dengan await sama sekali tidak memakai thread, jadi membungkusnya dengan Task.Run tidak memberi keuntungan apa pun dan hanya menambah satu lompatan lewat thread pool. Kasus yang benar-benar menahan sebuah thread pool adalah Task.Run di sekitar pemanggilan synchronous yang memblokir seperti File.ReadAllText: thread itu menganggur sampai disk atau jaringan menjawab.

Beberapa task sekaligus: WhenAll

Pecah pekerjaan yang saling independen menjadi beberapa task, mulai semuanya, lalu gabungkan hasilnya dengan Task.WhenAll. Di komputer dengan beberapa core, potongan-potongan itu berjalan secara paralel.

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());
    }
}

Output:

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

Setiap potongan mengerjakan rentangnya sendiri dan mengembalikan jumlahnya sendiri, jadi task-task itu tidak berbagi apa pun dan tidak butuh locking. WhenAll mengembalikan hasilnya sesuai urutan task-nya, bagaimanapun urutan selesainya thread. Jika task memang perlu mengubah state bersama, lihat lock.

WhenAny dan timeout

Task.WhenAny selesai saat task pertama di antara task-task itu selesai, lalu mengembalikan task tersebut. Mengadu pekerjaan yang sebenarnya dengan Task.Delay menghasilkan sebuah timeout:

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();
    }
}

Output:

fast mirror answered after 100 ms
Timed out after 300 ms

WhenAny mengembalikan task pemenangnya itu sendiri, bukan hasilnya, jadi kamu perlu melakukan await lagi untuk mendapatkan nilainya (atau exception-nya, jika gagal). Perhatikan bahwa task yang kalah tetap berjalan; WhenAny tidak menghentikannya. Untuk benar-benar menghentikan pekerjaan yang lambat, batalkan dengan token seperti di bawah. Di .NET 6 dan sesudahnya, await work.WaitAsync(TimeSpan.FromMilliseconds(300)) mengungkapkan timeout yang sama dalam satu pemanggilan.

Continuation dengan ContinueWith

Sebelum ada await, cara menjalankan kode setelah sebuah task selesai adalah ContinueWith. Kamu masih akan menemuinya di kode lama:

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);
    }
}

Output:

Report total: 42

Setiap continuation menerima task sebelumnya dan berjalan saat task itu selesai. await melakukan hal yang sama dengan jauh lebih sedikit basa-basi, menangani exception secara alami, dan kembali ke context yang benar, jadi lebih baik pakai cara itu di kode baru. ContinueWith tetap berjalan meskipun task sebelumnya gagal atau dibatalkan, jadi kode lama harus memeriksa t.IsFaulted atau memberikan TaskContinuationOptions agar tidak memanggil .Result pada task yang gagal.

Pembatalan dengan CancellationToken

Task dibatalkan secara kooperatif. CancellationTokenSource mengeluarkan sebuah token; kode yang mengerjakan tugasnya memeriksa token itu dan menghentikan dirinya sendiri. Tidak ada yang dihentikan secara paksa.

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();
    }
}

Output:

Cancelled before finishing all 10 orders
Processed 3

Token itu diteruskan ke setiap lapisan, dan API async seperti Task.Delay, HttpClient, dan pemanggilan database juga menerimanya, jadi pembatalan bisa menghentikan mereka di tengah penantian. Tangkap OperationCanceledException; TaskCanceledException yang dilempar sebagian API diturunkan darinya. CancellationTokenSource adalah IDisposable, itulah sebabnya ada using.

Thread vs Task

Thread adalah thread sistem operasi yang kamu buat sendiri. Ia lebih berat: masing-masing punya stack sendiri (1 MB secara default di Windows), memulainya butuh waktu yang terasa, dan ia tidak memberimu hasil atau penanganan exception.

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);
    }
}

Output:

Thread started
Result: 42

Pakai Thread hanya jika kamu butuh sesuatu yang tidak diberikan pool: perulangan berumur panjang yang tidak boleh meminjam thread dari pool, prioritas tertentu, atau single-threaded apartment untuk COM. Untuk task yang berumur panjang, Task.Factory.StartNew(work, TaskCreationOptions.LongRunning) mendapat thread khusus sambil tetap memakai API task.

Parallel.For dan Parallel.ForEach

Untuk "lakukan ini pada setiap item, dengan memakai semua core", System.Threading.Tasks.Parallel lebih sederhana daripada membuat task secara manual. Class ini membagi rentangnya ke beberapa thread dan baru kembali setelah semua iterasi selesai:

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);
    }
}

Contoh output:

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

Baris untuk setiap gambar muncul dalam urutan apa pun sesuai urutan thread menjalankannya, dan ID thread-nya berbeda di setiap kali dijalankan; hanya totalnya yang tetap. Begitulah sifat perulangan paralel: iterasinya tidak boleh bergantung pada urutan satu sama lain, dan state bersama apa pun butuh sinkronisasi (di sini Interlocked.Add). Parallel.ForEach melakukan hal yang sama pada collection apa pun, dan .NET 6 menambahkan Parallel.ForEachAsync untuk body yang async.

Kesalahan umum

• .Result atau .Wait() di dalam kode async. Ini memblokir thread dan bisa menyebabkan deadlock di aplikasi UI; pakai await.
• Menangkap tipe exception asli di sekitar .Result. Kegagalan datang dalam bungkusan AggregateException; await membuka bungkusan itu.
• Fire and forget. Task.Run(...) tanpa await atau tanpa menyimpan task-nya akan menghilangkan exception-nya.
• Task.Run di sekitar I/O. Di sekitar method async, ini hanya menambah lompatan lewat thread pool; di sekitar method yang memblokir seperti File.ReadAllText, ini menahan sebuah thread dari pool selama seluruh waktu tunggu. Await langsung method I/O yang async.
• Mengabaikan token. Meneruskan CancellationToken tidak berpengaruh apa pun kecuali pekerjaannya memeriksa token itu.
• State bersama tanpa sinkronisasi di kode paralel. Pakai Interlocked, lock, atau beri setiap task datanya sendiri.