[suggestion] add api to read/write multiple values from System.Threading.Channels

[JanEggers]

right now im using buffered channel and the overhead of locking is currently limiting throughput. i limit the buffer to 10000 items and most of the time the buffer is full so i i guess i could read more than one item at a time to improve.

[stephentoub]

In the past I've considered adding base virtuals like:

public virtual int TryRead(Span<T> destination)
{
    int i = 0;
    while (i < destination.Length)
    {
        if (!TryRead(out destination[i])) break;
        i++;
    }
    return i;
}

public virtual int TryWrite(ReadOnlySpan<T> source)
{
    int i = 0;
    while (i < source.Length)
    {
        if (!TryWrite(source[i])) break;
        i++;
    }
    return i;
}

and then overriding them on derived channel reader/writers where the implementation could do something more efficient.

Is that what you had in mind, or something else?

With regards to the locking itself, we should actually be able to get rid of some of that locking and use more lock-free approaches in various situations. I just haven't found the time to invest in / investigate doing so.

[JanEggers]

thx for the info.

if there will be a lockfree version/option later on ill just wait for that. i copied over the source and added a read method. it just gave 10% perf boost so i have to optimize the consumer of the channel after all to get better throughput...

here is what i did:

            public bool TryReadMany(out IEnumerable<T> items)
            {
                BoundedChannel<T> parent = _parent;
                lock (parent.SyncObj)
                {
                    parent.AssertInvariants();

                    // Get an item if there is one.
                    if (!parent._items.IsEmpty)
                    {
                        var list = new List<T>();

                        while (!parent._items.IsEmpty)
                        {
                            list.Add(DequeueItemAndPostProcess());
                        }
                        items = list;
                        return true;
                    }
                }

                items = Enumerable.Empty<T>();
                return false;
            }

that could be optimized if Dequeue had a method to get multiple items.

[AceHack]

lockfree and/or waitfree implementation would be great. Also bulk/multiple items operation would also be very great.

[kouvel]

CC @tarekgh

[dmitriyse]

Please also consider (zero-copy)able batch mode reader/writer interface.
For example:

    using System;
    using System.Threading.Channels;

    /// <summary>
    /// Provides a base class for reading from a channel in the batch mode possibly with the zero-copy.
    /// </summary>
    /// <typeparam name="T">Specifies the type of data that may be read from the channel.</typeparam>
    public abstract class ChannelBatchReader<T> : ChannelReader<T>, IDisposable
    {
        /// <summary>
        /// Finalizes an instance of the <see cref="ChannelBatchReader{T}"/> class.
        /// </summary>
        ~ChannelBatchReader()
        {
            Dispose(false);
        }

        /// <inheritdoc/>
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// Finalizes read operation.
        /// </summary>
        /// <param name="processedCount">The number of processed items.</param>
        /// <param name="slice">The buffer slice.</param>
        public abstract void CompleteRead(int processedCount, ref ChannelReaderBufferSlice<T> slice);

        /// <summary>
        /// Releases a part of previously allocated buffer slice.
        /// </summary>
        /// <param name="newCount">The number of items to remain allocated.</param>
        /// <param name="slice">The buffer slice.</param>
        public abstract void PartialFree(int newCount, ref ChannelReaderBufferSlice<T> slice);

        /// <summary>
        /// Allocates a buffer slice with data.
        /// </summary>
        /// <remarks>
        /// Resulting slice can has less length than requested.
        /// </remarks>
        /// <param name="count">The preferred amount of items to read.</param>
        /// <param name="slice">The buffer slice with the data.</param>
        /// <returns>
        /// <see langword="true"/> if the buffer slice was allocated; otherwise, <see langword="false"/> if it wasn't
        /// allocated.
        /// </returns>
        public abstract bool TryStartRead(int count, out ChannelReaderBufferSlice<T> slice);

        /// <summary>
        /// Releases the unmanaged resources used by the Component and optionally releases the managed resources.
        /// </summary>
        /// <param name="disposing">
        /// <see langword="true"/> to release both managed and unmanaged resources; <see langword="false"/>
        /// to release only unmanaged resources.
        /// </param>
        protected virtual void Dispose(bool disposing)
        {
        }
    }

    public struct ChannelReaderBufferSlice<T>
    {
        private T[] _array;

        private int _start;

        private int _length;

        private int _id;

        public ChannelReaderBufferSlice(T[] array, int start, int length, int id)
        {
#if DEBUG

// TODO: checks
#endif
            _array = array;
            _start = start;
            _length = length;
            _id = id;
        }

        public void DecreaseLength(int newSize)
        {
            _length = newSize;
        }

        public ReadOnlySpan<T> Span => new ReadOnlySpan<T>(_array, _start, _length);

        public int Id => _id;
    }

This zero-copy can give significant speedup in up to half of cases.
Otherwise batch mode will infect code with double-copy (on reader and on writer side).

Implementation of zero-copy batch mode channel is very complicated, but possible (I have some implementation in alpha stage).

[dmitriyse]

Batch mode, Queues (Channels), DataFlow-like framework, zero-copy is the "must have" tricks to get real-world 100k+ items/sec performance from a single Server Node.

[dmitriyse]

Also TPL is too heavy to work with single items, even with all kind of tricks expected in 2.1.
TPL have limit ~1M items/sec (per one CPU-core).
C# Async/await state machines have ~60M items/sec max through put on a single CPU-core.
Simple loop over batch of items have 1000M+ items/sec on a single CPU-core (at that speeds every IL instruction play role).
Batch-mode is a simplest way to workaround TPL limits.

(To get maximal theoretical performance, TPL should be avioded, all processing should be performed cooperatively (one event loops per CPU-Core), item processing workflows should be bound to event-loops to avoid as much synchronizations as possible, and only after that C#/IL will be the bottleneck and C++ can give some benefits)

[dmitriyse]

See more general proposal about batch processing: dotnet/corefxlab#2154