API proposal: Add IProducerConsumerCollection to System.Threading.Channels.CreateUnbounded(...)

[zcsizmadia]

Reason

Currently unbounded channels use a ConcurrentQueue to hold the queued items. This concurrent queue is created in the internal UnboundedChannel object and cannot be changed. Since the Channel reader and writer architecture is a producer/consumer pattern, it would be great if the object holding the queued items could be provided to the channel creation, using the IProducerConsumerCollection interface. If no such interface is specified the default ConcurrentQueue is created and used by the channel.

This addition would reflect the nature of the Channel architecture.

Use case

Use a priority queue instead of a FIFO.

API Proposal

namespace System.Threading.Channels
{
    public static class Channel
    {
        // existing
        public static Channel<T> CreateUnbounded<T>();
        public static Channel<T> CreateUnbounded<T>(UnboundedChannelOptions options);

        // new
        public static Channel<T> CreateUnbounded<T>(IProducerConsumerCollection<T> collection);
        public static Channel<T> CreateUnbounded<T>(IProducerConsumerCollection<T> collection, UnboundedChannelOptions options);
}

Implementation

The internal UnboundedChannel object is already using a ConcurrentQueue object to hold the queued items, and the used methods are only TryDequeue, Enqueue and Count.
The _items object must be changed to IProducerConsumerCollection and the Dequeue and Enqueue must be changed to TryTake and TryAdd.

Diff/PR

internal sealed class UnboundedChannel<T> : Channel<T>, IDebugEnumerable<T>
{
    private readonly IProducerConsumerCollection<T> _items;
        
    internal UnboundedChannel(bool runContinuationsAsynchronously)
            : this(runContinuationsAsynchronously, new ConcurrentQueue<T>())
    {
    }

    internal UnboundedChannel(bool runContinuationsAsynchronously, IProducerConsumerCollection<T> queue)
    {
        _runContinuationsAsynchronously = runContinuationsAsynchronously;
        _completion = new TaskCompletionSource<VoidResult>(runContinuationsAsynchronously ? TaskCreationOptions.RunContinuationsAsynchronously : TaskCreationOptions.None);
        _items = queue ?? throw new ArgumentNullException(nameof(queue));
        Reader = new UnboundedChannelReader(this);
        Writer = new UnboundedChannelWriter(this);
    }

    // Example changes:

    //if (parent._items.TryDequeue(out item)) ->
    //if (parent._items.TryTake(out item))

    //parent._items.Enqueue(item); ->
    //if (!parent._items.TryAdd(item))
    //{
    //    return false;
    //}
}

[stephentoub]

It's a reasonable proposal, though it would almost certainly need a different implementation rather than just changing the existing one, as I expect the overhead of going through the interface would be measurable on high-throughput scenarios. The implementation would also need to correctly handle a collection that came in non-empty, that could potentially change from other users while it was being used by the channel, etc. It would need to be fully prototyped to validate everything works correctly.

[zcsizmadia]

Makes sense. What is the performance implication of the following. When we add the TQueue to the UnboundedChannel. In this case the constructors do not change, we have the existing CreateUnbounded() and an additional new API of CreateUnbounded<T, TQueue>().

internal sealed class UnboundedChannel<T, TQueue> : Channel<T>, IDebugEnumerable<T> where TQueue : IProducerConsumerCollection<T>, new()
    {
        //...
        private readonly TQueue _items = new TQueue();
        //...
    }


    public static class Channel
    {
        // Existing
        public static Channel<T> CreateUnbounded<T>() =>
            new UnboundedChannel<T, ConcurrentQueue<T>>(...);

        // New
        public static Channel<T> CreateUnbounded<T, TQueue>() where TQueue : IProducerConsumerCollection<T>, new() =>
            new UnboundedChannel<T, TQueue>(...);
    }

[stephentoub]

What is the performance implication of the following

The runtime does not specialize generics for reference types, which TQueue is almost certainly going to be (as it is with ConcurrentQueue<T>). We'd pay the overheads associated with using generics in this fashion, we wouldn't be able to use ConcurrentQueue<T>.IsEmpty which is much faster than its IProducerConsumerCollection<T>.Count implementation, etc.

[JanEggers]

+1 stacklike behavior would be really great.

my usecase is timeouut handling. My workload has a timeout and once I hit it all queued operations will also timeout. If I could process the newest work first it is much more likely that im able to recover at some point.

[GF-Huang]

Sorry for the digression. Since I couldn't find any docs which about is it Channel is FIFO or not? This issue seems to be the closest to my question.

[stephentoub]

The channels created by both CreateUnbounded and CreateBounded are FIFO.

[GF-Huang]

The channels created by both CreateUnbounded and CreateBounded are FIFO.

By the way, is it possible to clear all items in Channel without ReadAllAsync?

[stephentoub]

There is no API for that today.