Task ב-C#: Task.Run, WhenAll, WhenAny, ביטול ותהליכונים

Task היא אובייקט שמייצג עבודה שנמצאת בתהליך. היא יכולה לרוץ על תהליכון ממאגר התהליכונים, לחכות לטיימר או לחכות לתשובה מהרשת; בכל מקרה אפשר לשאול אם היא הסתיימה, לקבל את התוצאה שלה, לחכות לה או לחבר אליה עבודה נוספת. async ו-await הם התחביר של השפה מעל משימות. העמוד הזה מכסה את ה-API של Task עצמו, ואת התהליכונים.

Task.Run: עבודה על מאגר התהליכונים

Task.Run מקבלת ביטוי למדה ומכניסה אותו לתור של מאגר התהליכונים, קבוצת תהליכונים ש-.NET שומרת זמינים כדי לא ליצור תהליכון חדש לכל עבודה. היא מחזירה Task, או Task<T> אם הלמדה מחזירה ערך.

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

פלט:

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

Result חוסמת את התהליכון הקורא עד שהמשימה מסתיימת. זה מקובל ב-Main של תוכנית קונסולה, אבל בתוך קוד אסינכרוני הייתם כותבים int primes = await task;, שמחכה בלי לחסום. אותו דבר חל על task.Wait() עבור Task בלי תוצאה.

Task.Run מיועדת לעבודה שתלויה במעבד: חישובים שמעסיקים ליבה. ל-I/O, כמו קריאת קובץ או קריאה ל-API ברשת, השתמשו במתודה האסינכרונית שהספרייה כבר מספקת (File.ReadAllTextAsync, HttpClient.GetStringAsync) ובצעו עליה await ישירות. בזמן שה-I/O ממתין, קריאה אסינכרונית עם await לא מחזיקה שום תהליכון, כך שעטיפה שלה ב-Task.Run לא מרוויחה כלום ורק מוסיפה מעבר דרך מאגר התהליכונים. המקרה שבאמת תופס תהליכון מהמאגר הוא Task.Run סביב קריאה חוסמת סינכרונית כמו File.ReadAllText: התהליכון הזה יושב בטל עד שהדיסק או הרשת עונים.

כמה משימות בבת אחת: WhenAll

פצלו עבודה עצמאית למשימות, הפעילו את כולן, ושלבו את התוצאות עם Task.WhenAll. במחשב עם כמה ליבות החלקים רצים במקביל.

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

פלט:

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

כל חלק עובד על טווח משלו ומחזיר סכום משלו, כך שהמשימות לא חולקות כלום ולא צריכות נעילה. WhenAll מחזירה את התוצאות לפי סדר המשימות, לא משנה באיזה סדר התהליכונים סיימו. כשמשימות כן צריכות לעדכן מצב משותף, ראו lock.

WhenAny ו-timeouts

Task.WhenAny מסתיימת כשהראשונה מבין המשימות שלה מסתיימת, ומחזירה את המשימה הזו. תחרות בין העבודה האמיתית ל-Task.Delay נותנת 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();
    }
}

פלט:

fast mirror answered after 100 ms
Timed out after 300 ms

WhenAny מחזירה את המשימה המנצחת עצמה, לא את התוצאה שלה, אז מבצעים עליה await שוב כדי לקבל את הערך (או את החריגה, אם היא נכשלה). שימו לב שהמשימות המפסידות ממשיכות לרוץ; WhenAny לא עוצרת אותן. כדי באמת לעצור את העבודה האיטית, בטלו אותה עם token כמו שמוצג בהמשך. ב-.NET 6 ומעלה, await work.WaitAsync(TimeSpan.FromMilliseconds(300)) מבטא את אותו timeout בקריאה אחת.

המשכים עם ContinueWith

לפני await, הדרך להריץ קוד אחרי שמשימה הסתיימה הייתה ContinueWith. עדיין תראו אותה בקוד ישן:

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

פלט:

Report total: 42

כל המשך מקבל את המשימה הקודמת ורץ כשהיא מסתיימת. await עושה את אותו הדבר עם הרבה פחות טקסים, מטפלת בחריגות באופן טבעי וחוזרת להקשר הנכון, אז העדיפו אותה בקוד חדש. ContinueWith רצה גם כשהמשימה הקודמת נכשלה או בוטלה, כך שקוד ישן צריך לבדוק את t.IsFaulted או להעביר TaskContinuationOptions כדי לא לקרוא ל-.Result על משימה שנכשלה.

ביטול עם CancellationToken

משימות מבוטלות בשיתוף פעולה. CancellationTokenSource מנפיק token; הקוד שעושה את העבודה בודק את ה-token ועוצר את עצמו. שום דבר לא נקטע בכוח.

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

פלט:

Cancelled before finishing all 10 orders
Processed 3

ה-token מועבר למטה דרך כל שכבה, ו-APIs אסינכרוניים כמו Task.Delay, HttpClient וקריאות למסד נתונים מקבלים אותו גם כן, כך שביטול עוצר אותם באמצע ההמתנה. תפסו OperationCanceledException; ה-TaskCanceledException שחלק מה-APIs זורקים יורשת ממנה. CancellationTokenSource הוא IDisposable, ומכאן ה-using.

Thread מול Task

Thread הוא תהליכון של מערכת ההפעלה שאתם יוצרים בעצמכם. הוא כבד יותר: לכל אחד יש מחסנית משלו (1 MB כברירת מחדל ב-Windows), ההפעלה שלו לוקחת זמן מדיד, והוא לא נותן תוצאה או טיפול בחריגות.

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

פלט:

Thread started
Result: 42

פנו ל-Thread רק כשצריך משהו שהמאגר לא נותן: לולאה ארוכת טווח שלעולם לא צריכה לשאול תהליכון מהמאגר, עדיפות מסוימת, או single-threaded apartment עבור COM. למשימות ארוכות טווח, Task.Factory.StartNew(work, TaskCreationOptions.LongRunning) מקבלת תהליכון ייעודי ושומרת על ה-API של משימות.

Parallel.For ו-Parallel.ForEach

עבור "עשו את זה לכל פריט, עם כל הליבות", System.Threading.Tasks.Parallel פשוט יותר מיצירת משימות ידנית. הוא מחלק את הטווח בין תהליכונים וחוזר כשכל האיטרציות הסתיימו:

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

פלט לדוגמה:

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

השורות של כל תמונה יוצאות בסדר שבו התהליכונים הריצו אותן, ומזהי התהליכונים משתנים מהרצה להרצה; רק הסכום קבוע. זה הטבע של לולאות מקבילות: אסור שאיטרציות יהיו תלויות בסדר של אחרות, וכל מצב משותף צריך סנכרון (Interlocked.Add כאן). Parallel.ForEach עושה את אותו הדבר על כל אוסף, ו-.NET 6 מוסיפה את Parallel.ForEachAsync לגופים אסינכרוניים.

טעויות נפוצות

• .Result או .Wait() בתוך קוד אסינכרוני. חוסם תהליכון ועלול לגרום ל-deadlock באפליקציות עם ממשק משתמש; השתמשו ב-await במקום.
• תפיסת טיפוס החריגה המקורי סביב .Result. כישלונות מגיעים עטופים ב-AggregateException; await פורקת אותם.
• Fire and forget. Task.Run(...) בלי await ובלי לשמור את המשימה מאבד את החריגות שלה.
• Task.Run סביב I/O. סביב מתודה אסינכרונית זה רק מוסיף מעבר דרך המאגר; סביב מתודה חוסמת כמו File.ReadAllText זה תופס תהליכון מהמאגר לכל זמן ההמתנה. בצעו await ישירות על מתודת ה-I/O האסינכרונית.
• התעלמות מה-token. העברת CancellationToken לא עושה כלום אם העבודה לא בודקת אותו.
• מצב משותף לא מסונכרן בקוד מקבילי. השתמשו ב-Interlocked, ב-lock, או תנו לכל משימה נתונים משלה.