[API Proposal]: Support a lower bound on BoundedChannelOptions

[davidfowl]

Background and motivation

Today bounded channels let the channel creator set a maximum capacity for the queue inside of the channel. This critical for making sure that backpressure can be applied to the consumer once the bounds have been hit (it can affects the reliability and performance of system using channels). When the consumer hits the capacity, the channel can be configured to behave in different ways by configuring the BoundedChannelFullMode. When this mode is set to wait, if the writer is using WaitToWriteAsync or WriteAsync, the method call will return an incomplete task until the reader consumes the one element that keeps it under the capacity. This normally isn't an issue but when building systems that want to react to this back pressure with a more heavy weight process, flipping from no wait to wait start on every single write operation could be costly.

The idea is to add a ResumeCapacity that applies when BoundedChannelFullMode.Wait is set that wouldn't resume the writer until that capacity is reached. By default it would be set to Capacity - 1, but somebody could set it to any value >= 0 and < Capacity. This would make it so that the writer could only start writing again after the queue size reached ResumeCapacity after it reach the original Capacity.

Today this is somewhat possible by polling the Count on the channel reader but it has a bunch of problems:

• The writer might not have access to the reader
• The writer doesn't want to think about bounds in wait mode, just what to do with the item when TryWrite returns false (or just have WriteAsync do the right thing).

API Proposal

namespace System.Threading.Channels;

public class BoundedChannelOptions
{
    public int? DrainCapacity { get; set; }
}

API Usage

var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(10) {
   DrainCapacity = 5 
});

async Task WriteNumbersAsync()
{
    for (var i = 0; ; i++)
    {
        // This will write 10 numbers, then wait until 5 of them has been consumed, then will write 5 more
        await channel.Writer.WriteAsync(i);
    }
}

_ = WriteNumbersAsync();

await foreach (var value in channel.Reader.ReadAllAsync())
{
     Console.WriteLine(value);
     // The reader is a little slower than the writer
     await Task.Delay(1000);
}

Alternative Designs

Instead of a setting, it could be done at the callsite with an overload of WaitToWriteAsync.

namespace System.Threading.Channels;

public abstract class ChannelWriter
{
    public virtual ValueTask<bool> WaitToWriteAsync(int numberOfItems);
}

var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(10));

async Task WriteNumbersAsync()
{
    for (var i = 0; ; i++)
    {
        // This will write 10 numbers, then wait until 5 of them has been consumed, then will write 5 more
        await WriteNumberAsync(i);
    }

    async Task WriteNumberAsync(int n)
    {
        if (!channel.TryWrite(i))
        {
           // We hit back pressure, now poll until the reader consumes enough...
           await channel.Writer.WaitToWriteAsync(5);
           channel.TryWrite(i); // Now we should be able to write
        }
    }
}

_ = WriteNumbersAsync();

await foreach (var value in channel.Reader.ReadAllAsync())
{
     Console.WriteLine(value);
     // The reader is a little slower than the writer
     await Task.Delay(1000);
}

Risks

No response

Tagging subscribers to this area: @dotnet/area-system-threading-channels
See info in area-owners.md if you want to be subscribed.

Issue Details

---

Background and motivation

Today bounded channels let the channel creator set a maximum capacity for the queue inside of the channel. This critical for making sure that backpressure can be applied to the consumer once the bounds have been hit (it can affects the reliability and performance of system using channels). When the consumer hits the capacity, it can the channel can be configured to behave in different ways by configuring the BoundedChannelFullMode. When this mode is set to wait, if the writer is using WaitToWriteAsync or WriteAsync, the method call will return an incomplete task until the reader consumes the one element that keeps it under the capacity. This normally isn't an issue but when building systems that want to react to this back pressure with a more heavy weight process, flipping from no wait to wait start on every single write operation could be costly.

The idea is to add a ResumeCapacity that applies when BoundedChannelFullMode.Wait is set that wouldn't resume the writer until that capacity is reached. By default it would be set to Capacity - 1, but somebody could set it to any value >= 0 and < Capacity. This would make it so that the writer could only start writing again after the queue size reached ResumeCapacity after it reach the original Capacity.

Today this is somewhat possible by polling the Count on the channel reader but it has a bunch of problems:

• The writer might not have access to the reader
• The writer doesn't want to think about bounds in wait mode, just what to do with the item when TryWrite returns false (or just have WriteAsync do the right thing).

API Proposal

namespace System.Threading.Channels;

public class BoundedChannelOptions
{
    public int? ResumeCapacity { get; set; }
}

API Usage

var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(10) {
   ResumeCapacity = 5 
});

async Task WriteNumbersAsync()
{
    for (var i = 0; ; i++)
    {
        // This will write 10 numbers, then wait until 5 of them has been consumed, then will write 5 more
        await channel.Writer.WriteAsync(i);
    }
}

_ = WriteNumbersAsync();

await foreach (var value in channel.Reader.ReadAllAsync())
{
     Console.WriteLine(value);
     // The reader is a little slower than the writer
     await Task.Delay(1000);
}

Alternative Designs

No response

Risks

No response

Author:	davidfowl
Assignees:	-
Labels:	api-suggestion, area-System.Threading.Channels, untriaged
Milestone:	-

[theodorzoulias]

I can surely imagine myself having the need for this API, and using it if it was available. I can also imagine myself reviewing my code a month later, and struggling to remember what exactly this esoteric ResumeCapacity configuration means!

[buyaa-n]

For me ResumeCapacity or ResumeWritersAtCapacity might sound meaningful, anyway looks like we can mark the proposal ready for review and discuss the naming there

Moving to Future milestone for now, @davidfowl if you need it in 7.0 feel free to move

[fernandozago]

I think we have something like this in System.IO.Pipelines don't we?

[theodorzoulias]

async Task WriteNumbersAsync()
{
    for (var i = 0; ; i++)
    {
        // This will write 10 numbers, then wait until 5 of them has been consumed,
        // then will write 5 more
        await channel.Writer.WriteAsync(i);
    }
}

@davidfowl supporting this scenario implies that the ChannelWriter.WriteAsync method will have a dual personality. It will complete immediately when the channel has space available, otherwise it will complete when the channel contains less than DrainCapacity items. This behavior might have unexpected consequences, in case the channel has more than one producers:

• The producer 1 calls WriteAsync("A") at a time that the channel is full.
• Some time later the producer 2 calls WriteAsync("B"), at a time that the channel has a few free slots.
• The value "B" is added in the channel before the value "A".
• The consumer gets the value "B" before the value "A".

My suggestion is to avoid this scenario, and limit the effect of the DrainCapacity only on the WaitToWriteAsync method (the existing, not the proposed overload). The WriteAsync should ignore this setting. The original example modified:

async Task WriteNumbersAsync()
{
    for (var i = 0; ; i++)
    {
        // This will write 10 numbers, then wait until 5 of them has been consumed,
        // then will write 5 more
        if (channel.Writer.TryWrite(i) continue;
        await channel.Writer.WriteAsync(i);      // DrainCapacity-oblivious
        await channel.Writer.WaitToWriteAsync(); // DrainCapacity-sensitive
    }
}

---

Edit: Actually the scenario that I mentioned above, with the wrong order of "A" and "B", can be avoided if the ChannelWriter.WriteAsync has the behavior of writing the item in the channel as soon as possible, and then idling until the DrainCapacity has been reached. It would be a weird behavior though, because the completion of the channel or cancellation of the token during the idling period could not be propagated by faulting the resulting ValueTask. After writing successfully the item in the channel, the ValueTask could only complete with success (to communicate the successful write), even though the completion of the ValueTask might happen a long time after the writing of the item.