ביטול ויסות וסיום

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

ה־Channel מפריד בין צד כתיבה לצד קריאה. במצב bounded עם FullMode.Wait המפיק ממתין למקום במקום לצבור עוד נתונים בתור. זאת אינה מגבלה על כל זיכרון המערכת: המפיק עדיין יכול להחזיק פריט שהוא בונה, והצרכן יכול להחזיק פריט בעיבוד. כאשר כל פריט מכיל buffer, גם גודלו וגם מספר הפריטים הפעילים צריכים להיכנס לתקציב.

using System.Threading.Channels;

using var stop = new CancellationTokenSource(TimeSpan.FromSeconds(5));
var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(4)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleReader = true,
    SingleWriter = true
});
Task producer = Produce(channel.Writer, stop);
Task consumer = Consume(channel.Reader, stop);
try { await Task.WhenAll(producer, consumer); }
catch (Exception error) { Console.WriteLine(error.GetType().Name); }

static async Task Produce(ChannelWriter<int> writer, CancellationTokenSource stop)
{
    Exception? failure = null;
    try
    {
        for (int i = 0; i < 20; i++) await writer.WriteAsync(i, stop.Token);
    }
    catch (Exception error)
    {
        failure = error;
        stop.Cancel();
        throw;
    }
    finally { writer.TryComplete(failure); }
}
static async Task Consume(ChannelReader<int> reader, CancellationTokenSource stop)
{
    try
    {
        await foreach (int value in reader.ReadAllAsync(stop.Token))
        {
            await Task.Delay(5, stop.Token);
            Console.WriteLine(value);
        }
    }
    catch { stop.Cancel(); throw; }
}

השלמת ה־writer מודיעה שלא יתווספו פריטים. במסלול הרגיל הצרכן מרוקן את מה שכבר נכתב ומסיים. בכשל של הצרכן הוא מבטל את token המשותף, כך שהמפיק לא יישאר תקוע לנצח בהמתנה לתור שאין מי שיקרא ממנו. ה־WhenAll ממתין לשתי הפעולות; הוא אינו מבטל אותן בעצמו. אם נרצה לשמר את כל החריגות, נבחן גם את Exception של משימת WhenAll אחרי שהסתיימה, ולא נניח שהחריגה היחידה ש־await הציף היא כל הסיפור.

ה־CancellationToken הוא בקשה לשיתוף פעולה. הוא אינו הורג thread ואינו מוכיח שפעולת I/O כבר הפסיקה לגעת ב־buffer. פעולה יכולה להשלים בהצלחה במקביל לבקשת הביטול. לכן נתכנן מצב terminal אחד שמתקבל מן הפעולה, ולא נחליט בעצמנו שכל בקשת cancel הופכת תוצאה קיימת לכישלון.

ה־WaitAsync עם timeout מגביל את משך ההמתנה למשימה, ולא בהכרח את חיי הפעולה המקורית. אם יצאנו מן ההמתנה והחזרנו buffer ל־pool, הפעולה המקורית עשויה עדיין לכתוב אליו. יש להעביר token לפעולה שתומכת בו, לבקש ביטול ואז להמתין לאישור ההשלמה שלה לפני שחרור המשאב. כשאין API לביטול, ה־owner חייב להישאר חי עד ההשלמה גם אם הצרכן כבר איבד עניין בתוצאה.

ה־ValueTask מאפשר לייצג גם השלמה זולה שכבר קיימת, ולעיתים פעולה שמגובה במקור reusable. לכן החוזה הכללי הוא צריכה אחת, למשל await יחיד; אין לשמור אותו ולהמתין שוב ושוב בלי שה־API מבטיח שזה מותר. אם צריך לשתף תוצאה בין צרכנים ניצור Task פעם אחת באמצעות AsTask ונשתמש בו. ההמרה יכולה להקצות, ולכן זאת החלטת חוזה לפני שהיא החלטת ביצועים.

במערכת עם buffers נצטרך מדיניות מפורשת לפריטים שנותרו בתור אחרי ביטול. אפשר לרוקן ולשחרר אותם לאחר שכל היצרנים חדלו לפעול, או להמשיך לעבד אותם לפי חוזה drain. אסור לצפות ש־Channel יבצע Dispose לפריטים שלנו. התיעוד על תורים אסינכרוניים ועל המתנה עם timeout עוזר להפריד בין סיום התור, סיום המשימות וסיום הבעלות על כל פריט.