C# の Channel を学ぶ (2)

前回 C# の Channel を学ぶ (1) ではチャネルの基礎を学んだ。今回はキャンセルされた場合や、タイムアウトの実装を見てみる。CancellationToken をAPIがサポートしているので簡単だ。

今回も

• C# Channels - Timeout and Cancellation

のブログの内容を実施している。

Timeout

タイムアウトのケースでは、ReadAsync() が CancellationToken を受け付けるAPIがあるので、それを使うと、キャンセルが発生したときに OperationCanceledException が発生するようになっている。

public async Task ExecuteAsync()
{
    var joe = CreateMessenger("Joe", 10);
    var cts = new CancellationTokenSource();
    cts.CancelAfter(TimeSpan.FromSeconds(5));

    try
    {
        await foreach (var item in joe.ReadAllAsync(cts.Token))
            Console.WriteLine(item);
        Console.WriteLine("Joe sent all of his messages.");
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Timeout happens. 5 sec.");
    }
}

Quit Channel

上記のように、Exception を発生させるのではなく、チャネルが途中で中断されたという風にコードを書いてみよう。実際にキャンセルが発生した場合に、token.IsCancellationRequested を使って、チャネルを Complete するようにすると、うまくそのチャネルだけが中断されて、終了するという感じでコードを書くことが出来る。これは、Producer 側で処理を入れておくことになる。

static ChannelReader<string> CreateMessenger(string msg, int count, CancellationToken token = default)
{
    var ch = Channel.CreateUnbounded<string>();
    var rnd = new Random();

    Task.Run(async () =>
    {
        for (int i = 0; i < count; i++)
        {
            if (token.IsCancellationRequested)
            {
                await ch.Writer.WriteAsync($"{msg} says good bye!");
                break;
            }
            await ch.Writer.WriteAsync($"[{DateTime.Now}] {msg} {i}");
            await Task.Delay(TimeSpan.FromSeconds(rnd.Next(3)));
        }
        ch.Writer.Complete();

    });

    return ch.Reader;
}

public async Task ExecuteAsync()
{
    var cts = new CancellationTokenSource();
    var joe = CreateMessenger("Joe", 10, cts.Token);
    cts.CancelAfter(TimeSpan.FromSeconds(5));

    await foreach (var item in joe.ReadAllAsync())
        Console.WriteLine(item);
}

Web Search

最後に総まとめで、Webサイトにアクセスする場合を想定したキャンセレーショントークンの使い方を見てみよう。Webへのリクエストをシュミレートするため、ランダムなスピードで処理が実行される Search メソッドを作成する。検索の結果は、チャネルに書き込まれるというのを想定している。

private async Task Search(string source, string term, Channel<string> ch,  CancellationToken token)
{
    await Task.Delay(TimeSpan.FromSeconds(new Random().Next(5)), token);
    await ch.Writer.WriteAsync($"Result from {source} for {term}", token);
}

３つのサイトにアクセスしているが、トータルタイムで、タイムアウトが設定されている。タイムアウトになると例外が発生する。

public async Task ExecuteAsync()
{
    var ch = Channel.CreateUnbounded<string>();
    var term = "teststrone";
    var token = new CancellationTokenSource(TimeSpan.FromSeconds(3)).Token;
    
    var search1 = Search("Google", term, ch, token);
    var search2 = Search("Quora", term, ch, token);
    var search3 = Search("Wikipedia", term, ch, token);

    try
    {
        for (int i = 0; i < 3; i++)
            Console.WriteLine(await ch.Reader.ReadAsync(token));

        Console.WriteLine("All searchs have completed.");
    } 
    catch (OperationCanceledException)
    {
        Console.WriteLine("Timeout!");
    }

}

予想通りタイミングによって、タイムアウトになったり、ならなかったりする。

まとめ

C#の CancellationToken が強力なので、実装がシンプルでとても良い感じ。次回はチャネルの学習の最終回。