Task trong C#: Task.Run, WhenAll, WhenAny, hủy tác vụ và Thread

Task là một đối tượng đại diện cho công việc đang diễn ra. Nó có thể đang chạy trên một luồng của thread pool, đang chờ một bộ hẹn giờ, hoặc đang chờ phản hồi qua mạng; dù thế nào bạn cũng có thể hỏi nó đã xong chưa, lấy kết quả, chờ nó, hoặc gắn thêm việc vào sau nó. async và await là cú pháp của ngôn ngữ xây trên task. Trang này trình bày chính API Task, cùng với luồng (thread).

Task.Run: chạy công việc trên thread pool

Task.Run nhận một lambda và đưa nó vào hàng đợi của thread pool, một tập các luồng mà .NET giữ sẵn để không phải tạo luồng mới cho mỗi việc. Nó trả về một Task, hoặc Task<T> nếu lambda trả về giá trị.

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

Đầu ra:

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

Result chặn luồng gọi cho đến khi task xong. Điều đó chấp nhận được trong Main của một chương trình console, nhưng trong code bất đồng bộ bạn sẽ viết int primes = await task;, cách này chờ mà không chặn luồng. Tương tự với task.Wait() cho một Task không có kết quả.

Task.Run dành cho công việc nặng về CPU: những phép tính giữ một nhân CPU luôn bận. Với I/O, như đọc tệp hay gọi web API, hãy dùng phương thức bất đồng bộ mà thư viện đã có sẵn (File.ReadAllTextAsync, HttpClient.GetStringAsync) và await trực tiếp. Trong lúc I/O đang chờ, một lời gọi bất đồng bộ được await không giữ luồng nào cả, nên bọc nó trong Task.Run chẳng được gì mà chỉ thêm một bước qua thread pool. Trường hợp thực sự chiếm một luồng của pool là Task.Run bọc quanh một lời gọi chặn đồng bộ như File.ReadAllText: luồng đó ngồi không cho đến khi ổ đĩa hoặc mạng trả lời.

Nhiều task cùng lúc: WhenAll

Chia công việc độc lập thành các task, khởi động tất cả, rồi gộp kết quả bằng Task.WhenAll. Trên máy có nhiều nhân, các phần chạy song song.

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

Đầu ra:

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

Mỗi phần làm việc trên khoảng riêng và trả về tổng riêng, nên các task không chia sẻ gì và không cần khóa. WhenAll trả về kết quả theo thứ tự của các task, bất kể các luồng xong theo thứ tự nào. Khi các task cần cập nhật trạng thái dùng chung, xem lock.

WhenAny và timeout

Task.WhenAny hoàn thành khi task đầu tiên trong số các task của nó hoàn thành, và trả về task đó. Cho công việc thật chạy đua với Task.Delay là có ngay một 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();
    }
}

Đầu ra:

fast mirror answered after 100 ms
Timed out after 300 ms

WhenAny trả về chính task thắng cuộc, không phải kết quả của nó, nên bạn await nó thêm lần nữa để lấy giá trị (hoặc ngoại lệ, nếu nó thất bại). Lưu ý rằng các task thua cuộc vẫn tiếp tục chạy; WhenAny không dừng chúng. Để thực sự dừng công việc chậm, hãy hủy nó bằng token như bên dưới. Từ .NET 6 trở đi, await work.WaitAsync(TimeSpan.FromMilliseconds(300)) diễn đạt cùng timeout đó trong một lời gọi.

Continuation với ContinueWith

Trước khi có await, cách để chạy code sau khi một task xong là ContinueWith. Bạn vẫn sẽ gặp nó trong code cũ:

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

Đầu ra:

Report total: 42

Mỗi continuation nhận task trước đó và chạy khi task đó hoàn thành. await làm điều tương tự với ít thủ tục hơn nhiều, xử lý ngoại lệ một cách tự nhiên, và quay về đúng ngữ cảnh, nên hãy ưu tiên nó trong code mới. ContinueWith vẫn chạy cả khi task trước thất bại hoặc bị hủy, nên code cũ phải kiểm tra t.IsFaulted hoặc truyền TaskContinuationOptions để tránh gọi .Result trên một task đã thất bại.

Hủy bằng CancellationToken

Task được hủy theo kiểu hợp tác. Một CancellationTokenSource phát ra token; code đang làm việc kiểm tra token và tự dừng. Không có gì bị cưỡng ép dừng.

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

Đầu ra:

Cancelled before finishing all 10 orders
Processed 3

Token được truyền xuống qua mọi tầng, và các API bất đồng bộ như Task.Delay, HttpClient hay các lời gọi cơ sở dữ liệu cũng nhận nó, nên việc hủy dừng chúng ngay giữa lúc chờ. Hãy bắt OperationCanceledException; TaskCanceledException mà một số API ném ra kế thừa từ nó. CancellationTokenSource cài đặt IDisposable, vì vậy có using.

Thread và Task

Thread là một luồng của hệ điều hành do bạn tự tạo. Nó nặng hơn: mỗi luồng có stack riêng (mặc định 1 MB trên Windows), khởi động một luồng tốn thời gian đáng kể, và nó không cho bạn kết quả hay cơ chế xử lý ngoại lệ.

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

Đầu ra:

Thread started
Result: 42

Chỉ dùng Thread khi bạn cần điều mà pool không cung cấp: một vòng lặp chạy rất lâu không nên mượn luồng của pool, một mức ưu tiên cụ thể, hoặc single-threaded apartment cho COM. Với các task chạy lâu, Task.Factory.StartNew(work, TaskCreationOptions.LongRunning) có được một luồng riêng mà vẫn giữ API của task.

Parallel.For và Parallel.ForEach

Với kiểu "làm việc này cho mọi phần tử, dùng hết các nhân", System.Threading.Tasks.Parallel đơn giản hơn việc tự tạo task. Nó chia khoảng lặp cho các luồng và trả về khi mọi lần lặp đã xong:

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

Đầu ra ví dụ:

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

Các dòng của từng ảnh xuất hiện theo bất kỳ thứ tự nào mà các luồng chạy, và ID luồng thay đổi giữa các lần chạy; chỉ có tổng là cố định. Đó là bản chất của vòng lặp song song: các lần lặp không được phụ thuộc vào thứ tự của nhau, và mọi trạng thái dùng chung đều cần đồng bộ hóa (ở đây là Interlocked.Add). Parallel.ForEach làm điều tương tự trên mọi collection, và .NET 6 bổ sung Parallel.ForEachAsync cho phần thân bất đồng bộ.

Lỗi thường gặp

• .Result hoặc .Wait() bên trong code bất đồng bộ. Chặn một luồng và có thể gây deadlock trong ứng dụng có giao diện; hãy dùng await.
• Bắt kiểu ngoại lệ gốc quanh .Result. Lỗi đến dưới dạng được bọc trong AggregateException; await mới mở lớp bọc.
• Chạy rồi bỏ mặc (fire and forget). Task.Run(...) mà không await hay lưu lại task sẽ làm mất các ngoại lệ của nó.
• Task.Run bọc quanh I/O. Bọc quanh một phương thức bất đồng bộ thì chỉ thêm một bước qua thread pool; bọc quanh một phương thức chặn như File.ReadAllText thì giữ một luồng của pool suốt thời gian chờ. Hãy await trực tiếp phương thức I/O bất đồng bộ.
• Bỏ qua token. Truyền CancellationToken vào chẳng có tác dụng gì nếu công việc không kiểm tra nó.
• Trạng thái dùng chung không được đồng bộ trong code song song. Dùng Interlocked, lock, hoặc cho mỗi task dữ liệu riêng.