ChannelReader Completion Task is never completed after OperationCanceledException

Question

If I call Stop(), OperationCanceledException is happened and _writer.TryComplete(exp) is true. But _reader.Completion Task is still not completed.

Is it desired behavior of Channels? If yes can someone tell me how to stop a Channel without waiting till it's empty and have its Completion Task in Completed state?

public interface IItem
{
    Uri SourceUri { get; }

    string TargetPath { get; }
}

public class Item : IItem
{
    public Item(Uri sourceUri, string targetPath)
    {
        SourceUri = sourceUri;
        TargetPath = targetPath;
    }

    public Uri SourceUri { get; }

    public string TargetPath { get; }
}

public class TestService
{
    private readonly ChannelWriter<IItem> _writer;
    private readonly ChannelReader<IItem> _reader;

    private readonly CancellationTokenSource _cts;

    public TestService()
    {
        _cts = new CancellationTokenSource();
        Channel<IItem> channel = Channel.CreateUnbounded<IItem>();
        _reader = channel.Reader;
        _writer = channel.Writer;
    }

    public async Task QueueDownload(IItem information)
    {
        await _writer.WriteAsync(information);
    }

    public void StartDownload()
    {
        Task.Factory.StartNew(async () =>
        {
            await ProcessDownloadAsync();
        }, TaskCreationOptions.LongRunning);
    }

    public void Stop()
    {
        _cts.Cancel();
        //_writer.Complete();
        //_writer = null;
        Console.WriteLine("Stop");
    }

    public async Task Wait()
    {
        await _reader.Completion;
    }

    private async Task ProcessDownloadAsync()
    {
        try
        {
            while (await _reader.WaitToReadAsync(_cts.Token))
            {
                IItem information = await _reader.ReadAsync(_cts.Token);
                using (WebClient webClient = new WebClient())
                {
                    Console.WriteLine(information.TargetPath);
                    await webClient.DownloadFileTaskAsync(information.SourceUri,
                        information.TargetPath);
                }
            }
        }
        catch (OperationCanceledException exp)
        {
            bool res = _writer.TryComplete(exp);
        }

    }
}

static class Program
{
    static async Task Main(string[] args)
    {
        TestService tSvc = new TestService();
        await tSvc.QueueDownload(new Item(new Uri(@"https://images.pexels.com/" +
            @"photos/753626/pexels-photo-753626.jpeg"), @"D:\\Temp\1.png"));
        await tSvc.QueueDownload(new Item(new Uri(@"https://images.pexels.com/" +
            @"photos/753626/pexels-photo-753626.jpeg"), @"D:\\Temp\1.png"));
        await tSvc.QueueDownload(new Item(new Uri(@"https://images.pexels.com/" +
            @"photos/753626/pexels-photo-753626.jpeg"), @"D:\\Temp\1.png"));
        await tSvc.QueueDownload(new Item(new Uri(@"https://images.pexels.com/" +
            @"photos/753626/pexels-photo-753626.jpeg"), @"D:\\Temp\1.png"));

        tSvc.StartDownload();
        Task t = tSvc.Wait();
        tSvc.Stop();

        await t;

        Console.WriteLine("Finished");
    }
}

Answer 1

The ChannelWriter.Complete method behaves a bit differently than one would expect. It is not invalidating instantly the contents of the channel. Instead, it just prevents adding more items in the channel. The existing items are still valid for consumption, and the ChannelReader.Completion property will not complete before all stored items are consumed.

The example below demonstrates this behavior:

var channel = Channel.CreateUnbounded<int>();
channel.Writer.TryWrite(1);
channel.Writer.Complete(new FileNotFoundException());
//channel.Reader.TryRead(out var data);
var completed = channel.Reader.Completion.Wait(500);
Console.WriteLine($"Completion: {(completed ? "OK" : "Timed-out")}");

Output:

Completion: Timed-out

You can uncomment the channel.Reader.TryRead line, to see the FileNotFoundException to emerge.