C# Task 排程與異常狀況練習

Program.cs

using System.Collections.Concurrent;

namespace ConsoleSandbox
{
    internal class Program
    {
        const int STOP_AFTER = 3000;
        const int SLEEP_BEFORE_WAIT = 200;
        const int ENQUEUE_DELAY = 1000;
        const int ENQUEUE_INTERVAL = 200;

        // 是否模擬殘留執行緒問題
        static bool SIM_WONT_RUN_TASK = false;

        // 使用 ConcurrentQueue 確保多緒安全
        static ConcurrentQueue<NamedTask<int>> TaskQueue = new();

        // 排程時, 所有工作都要用 CTS 管理, 當 Worker 結束時標註取消
        // 如果沒使用 CTS, 而且 Worker 結束, 未完成的工作會卡在 Wait() 導致 Thread 停滯無法結束
        static CancellationTokenSource workerCts = new CancellationTokenSource();

        static void Main()
        {
            int[] delay = new int[5];
            for (int i = 0; i < 5; i++)
            {
                delay[i] = ENQUEUE_DELAY + ENQUEUE_INTERVAL * i;
            }

            // 模擬具名函數執行中丟出例外
            new Thread(() => {
                var taskName = "task #1";
                var task = new NamedTask<int>(taskName, SyncFuncWithEx, workerCts.Token);
                Thread.Sleep(delay[0]);
                Console.WriteLine($"thread #{Tid()} / {taskName} / enqueue");
                TaskQueue.Enqueue(task);
                Thread.Sleep(SLEEP_BEFORE_WAIT);
                try
                {
                    // 如果非同步工作丟出例外, 在 Wait() 時會連帶觸發例外
                    task.Wait();
                    Console.WriteLine($"thread #{Tid()} / {taskName} / return {task.Result}");
                }
                catch (AggregateException agex)
                {
                    Console.WriteLine($"thread #{Tid()} / {taskName} / throws exception {agex.InnerException?.GetType().Name}: {agex.InnerException?.Message}");
                }
            }).Start();

            // 模擬具名函數執行中有妥善處理例外
            new Thread(() => {
                var taskName = "task #2";
                var task = new NamedTask<int>(taskName, SyncFuncWithTry, workerCts.Token);
                Thread.Sleep(delay[1]);
                Console.WriteLine($"thread #{Tid()} / {taskName} / enqueue");
                TaskQueue.Enqueue(task);
                Thread.Sleep(SLEEP_BEFORE_WAIT);
                task.Wait();
                Console.WriteLine($"thread #{Tid()} / {taskName} / return {task.Result}");
            }).Start();

            // 模擬具名函數正常狀況
            new Thread(() => {
                var taskName = "task #3";
                var task = new NamedTask<int>(taskName, SyncFunc, workerCts.Token);
                Thread.Sleep(delay[2]);
                Console.WriteLine($"thread #{Tid()} / {taskName} / enqueue");
                TaskQueue.Enqueue(task);
                Thread.Sleep(SLEEP_BEFORE_WAIT);
                task.Wait();
                Console.WriteLine($"thread #{Tid()} / {taskName} / return {task.Result}");
            }).Start();

            // 模擬匿名函數正常狀況
            new Thread(() => {
                var taskName = "task #4";
                var task = new NamedTask<int>(taskName, () =>
                {
                    Console.WriteLine($"thread #{Tid()} / {taskName} / task with anonymous func");
                    return 0;
                }, workerCts.Token);
                Thread.Sleep(delay[3]);
                Console.WriteLine($"thread #{Tid()} / {taskName} / enqueue");
                TaskQueue.Enqueue(task);
                Thread.Sleep(SLEEP_BEFORE_WAIT);
                task.Wait();
                Console.WriteLine($"thread #{Tid()} / {taskName} / return {task.Result}");
            }).Start();

            // 模擬被自定義 CTS 取消
            new Thread(() =>
            {
                var taskName = "task #5";
                CancellationTokenSource myCts = new CancellationTokenSource();
                myCts.Cancel();
                var task = new NamedTask<int>(taskName, () =>
                {
                    Console.WriteLine($"thread #{Tid()} / {taskName} / cancelled task");
                    return 0;
                }, myCts.Token);
                Thread.Sleep(delay[4]);
                Console.WriteLine($"thread #{Tid()} / {taskName} / enqueue");
                TaskQueue.Enqueue(task);
                Thread.Sleep(SLEEP_BEFORE_WAIT);
                try
                {
                    task.Wait();
                    Console.WriteLine($"thread #{Tid()} / {taskName} / return {task.Result}");
                }
                catch (AggregateException agex)
                {
                    Console.WriteLine($"thread #{Tid()} / {taskName} / cancellation detected {agex.InnerException?.GetType().Name}: {agex.InnerException?.Message}");
                }
            }).Start();

            // 模擬 Worker 已結束後被排程
            new Thread(() =>
            {
                var taskName = "task #6";
                NamedTask<int> task;

                if (SIM_WONT_RUN_TASK)
                {
                    // 不提供 CancellationToken, 並且 Worker 結束後才排程, 會在 Wait() 階段卡住造成執行緒殘留
                    task = new NamedTask<int>(taskName, () =>
                    {
                        Console.WriteLine($"thread #{Tid()} / {taskName} / cancelled task");
                        return 0;
                    });
                }
                else
                {
                    task = new NamedTask<int>(taskName, () =>
                    {
                        Console.WriteLine($"thread #{Tid()} / {taskName} / cancelled task");
                        return 0;
                    }, workerCts.Token);
                }
                Thread.Sleep(STOP_AFTER + 500);
                Console.WriteLine($"thread #{Tid()} / {taskName} / enqueue");
                TaskQueue.Enqueue(task);
                Thread.Sleep(SLEEP_BEFORE_WAIT);
                try
                {
                    task.Wait();
                    Console.WriteLine($"thread #{Tid()} / {taskName} / return {task.Result}");
                }
                catch (AggregateException agex)
                {
                    Console.WriteLine($"thread #{Tid()} / {taskName} / cancellation detected {agex.InnerException?.GetType().Name}: {agex.InnerException?.Message}");
                }
            }).Start();

            var workerThread = new Thread(() => {
                var begin = DateTime.Now;
                bool done = false;

                Console.WriteLine($"thread #{Tid()} / Worker started.");

                while (!done)
                {
                    NamedTask<int>? task = null;
                    try
                    {
                        TaskQueue.TryDequeue(out task);
                        if (task != null)
                        {
                            if (!task.IsCompleted)
                            {
                                task.RunSynchronously();
                                Console.WriteLine($"thread #{Tid()} / {task.Name} / run task gracefully.");
                            }
                            else
                            {
                                // 已完成或是已取消的會跑到這裡
                                Console.WriteLine($"thread #{Tid()} / {task.Name} / cannot run task.");
                            }
                        }
                        else
                        {
                            // 佇列中沒工作
                            Thread.Sleep(500);
                        }
                    }
                    catch (Exception ex)
                    {
                        if (task != null)
                        {
                            // 原則上不應該跑到這裡, 就算 task 本身有丟出例外, 也只會觸發在
                            // 1. task function 內部
                            // 2. task.Wait()
                            Console.WriteLine($"thread #{Tid()} / {task.Name} / Done with exception.");
                        }
                        Console.WriteLine($"{ex.GetType().Name}: {ex.Message}");
                    }
                    double elapsed = (DateTime.Now - begin).TotalMilliseconds;
                    done = elapsed > STOP_AFTER;
                }

                Console.WriteLine($"thread #{Tid()} / Worker stopped.");
                workerCts.Cancel();
            });
            workerThread.Start();

            Console.WriteLine($"thread #{Tid():00} / Main() stopped.");
        }

        static int SyncFuncWithEx()
        {
            Console.WriteLine($"thread #{Tid()} / task #1 / task with named func & exception.");
            int a = 1;
            int b = 0;
            return a / b;
        }

        static int SyncFuncWithTry()
        {
            Console.WriteLine($"thread #{Tid()} / task #2 / task with named func & try.");
            try
            {
                int a = 1;
                int b = 0;
                return a / b;
            }
            catch
            {
                return -1;
            }
        }

        static int SyncFunc()
        {
            Console.WriteLine($"thread #{Tid()} / task #3 / task with named func.");
            return 0;
        }

        static int Tid()
        {
            return Thread.CurrentThread.ManagedThreadId;
        }
    }

    internal class NamedTask<T>: Task<T>
    {
        public readonly string Name;

        public NamedTask(string name, Func<T> action) : base(action)
        {
            Name = name;
        }

        public NamedTask(string name, Func<T> action, CancellationToken token): base(action, token)
        {
            Name = name;
        }
    }
}