[API Proposal]: Add rendezvous channel without buffer

[Szer]

Background and motivation

Problem

let ch = Channel.CreateBounded 1
ch.Write "hi"
Console.WriteLine "consumer received"

The problem is - the last line is a lie, there is no guarantee neither that the consumer received anything, nor there is a consumer in the first place all because that channel has a BUFFER.

Reference

Channels are messaging primitives allowing processes/actors/threads/fiber/jobs/tasks to communicate and synchronize their work and data flow.

Synchronization is achieved by blocking both consumer and receiver until they are ready to communicate and pass data between each other. This is actually a default behaviour of channels in theory (pi-calculus) and practice (Go, Kotlin, F# Hopac, CML)

The problem above is solved in Go precisely as described because default Go channels don't have any buffer and they will block the producer until the consumer is ready and vice versa.

In Kotlin it is achieved by creating Channel with capacity = 0

In .NET you can't create BoundedChannel with capacity=0

• because capacity can't be lower than 1
• even if we could, it will be UB what will happen with capacity=0 and BoundedChannelFullMode other than Wait

So I propose to add
Channel.CreateRendezvous()
which will create a channel without any buffers to achieve messaging synchronization and probably some selective messaging (e.g. read multiple channels, read 1, leave all others intact)

API Proposal

namespace System.Threading.Channels;

public static class Channel
{
    public static Channel<T> CreateRendezvous<T>(RendezvousChannelOptions options)
}

API Usage

let fancyChannel = Channel.CreateRendezvous()
do! fancyChannel.WriteAsync 42 // blocked until consumer is ready
Console.WriteLine "consumer received our message!" // not a lie as we know it definitely received

Alternative Designs

Make BoundedChannel work with capacity=0

Risks

N/A

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

Problem

let ch = Channel.CreateBounded 1
ch.Write "hi"
Console.WriteLine "consumer received"

The problem is - the last line is a lie, there is no guarantee neither that the consumer received anything, nor there is a consumer in the first place all because that channel has a BUFFER.

Reference

Channels are messaging primitives allowing processes/actors/threads/fiber/jobs/tasks to communicate and synchronize their work and data flow.

Synchronization is achieved by blocking both consumer and receiver until they are ready to communicate and pass data between each other. This is actually a default behaviour of channels in theory (pi-calculus) and practice (Go, Kotlin, F# Hopac, CML)

The problem above is solved in Go precisely as described because default Go channels don't have any buffer and they will block the producer until the consumer is ready and vice versa.

In Kotlin it is achieved by creating Channel with capacity = 0

In .NET you can't create BoundedChannel with capacity=0

• because capacity can't be lower than 1
• even if we could, it will be UB what will happen with capacity=0 and BoundedChannelFullMode other than Wait

So I propose to add
Channel.CreateRendezvous()
which will create a channel without any buffers to achieve messaging synchronization and probably some selective messaging (e.g. read multiple channels, read 1, leave all others intact)

API Proposal

namespace System.Threading.Channels;

public static class Channel
{
    public static Channel<T> CreateRendezvous<T>(RendezvousChannelOptions options)
}

API Usage

let fancyChannel = Channel.CreateRendezvous()
do! c.WriteAsync 42 // blocked until consumer is ready
Console.WriteLine "consumer received our message!" // not a lie as we know it definitely received

Alternative Designs

Make BoundedChannel work with capacity=0

Risks

N/A

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

[Szer]

@eiriktsarpalis as a maintainer of Channels namespace, could you please take a look at this in your spare time?
Thank you!

[stephentoub]

We've considered such an implementation in the past, e.g. with CreateBounded and a capacity of 0 rather than a dedicated new name. The initial prototype of the library actually had it. But some of the APIs in the abstraction would not play well with it. For example, consider the default implementation of ReadAllAsync:

runtime/src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelReader.netcoreapp.cs

Lines 20 to 26 in 655b177

	while (await WaitToReadAsync(cancellationToken).ConfigureAwait(false))
	{
	while (TryRead(out T? item))
	{
	yield return item;
	}
	}

If a producer only ever uses TryWrite, this method will never be able to receive anything, as the only way it could is if the TryRead/TryWrite occurred at exactly the same time, and even then, most likely synchronization would prevent that from ever happening.

[Szer]

@stephentoub do you think it is a show stopper for making such a needed Channel?

Might there be a possibility to create a new abstract class ChannelSlim<T> without those methods?

[Lanayx]

while (await WaitToReadAsync(cancellationToken).ConfigureAwait(false))
{
while (TryRead(out T? item))
{
yield return item;
}
}

This should be possible to rewrite as

 while (await Read(cancellationToken, out T? item)) 
 { 
     yield return item; 
 }

People who want to squeeze all performance just won't be using that helper, but such change will allow for the suggestion to happen

[stephentoub]

do you think it is a show stopper for making such a needed Channel?

Showstopper, no, but the proposal needs to factor that in. Just a new implementation of the current abstraction could be a pit of failure.

Might there be a possibility to create a new abstract class ChannelSlim without those methods?

I'm not excited about implementing an entire new set of abstractions for this.

This should be possible to rewrite as

Such a Read method doesn't exist (and can't with such an out parameter while also being an async method). The implementation of such a method in terms of the existing APIs would need to either be in terms of WaitToReadAsync, as in the previous example, or ReadAsync, which will incur an exception when the channel hits eof, e.g.

public static async ValueTask<(bool, T)> ReadAsync<T>(this ChannelReader<T> reader, CancellationToken cancellationToken)
{
    try
    {
        return (true, await reader.ReadAsync(cancellationToken).ConfigureAwait(false));
    }
    catch (ChannelClosedException)
    {
         return (false, default(T));
    }
}

Beyond the akwardness of consuming that, it puts exceptions on the success path, which is something we try to avoid.

But even if that particular example gets rewritten, the problem isn't whether it can be rewritten, but that it needs to be: any code out there written against a ChannelReader<T> might interact very poorly with such a channel implementation, as if it uses WaitToReadAsync/TryRead (which is often the recommended pairing for both efficiency and if the channel isn't infinite) rather than ReadAsync, it would never work with someone using TryWrite on the other side.

[Szer]

I'm not excited about implementing an entire new set of abstractions for this.

not an entirely new set, but a more "core" Channel without all methods.

e.g.

// new core abstraction
abstract class ChannelCore<T> {
  // all core methods go here
}

// new fancy channel
class RendezvousChannel<T>: ChannelCore<T> {
  ...
}

// old abstraction, having all the same methods, where most of them come from ChannelCore
abstract class Channel<T>: ChannelCore<T> {
  // TryRead, TryWrite goes here
}

// old implementation, no changes here
class BoundedChannel<T>: Channel<T> {
  ...
}

I'm not sure whether this is a binary-breaking change, but that could make RendezvousChannel possible

[nazar554]

Is this similar to AsyncExchanger from dotNext?

[Szer]

Is this similar to AsyncExchanger from dotNext?

kind of, but the channel we are discussing is one-way

[xeromorph]

If a producer only ever uses TryWrite, this method will never be able to receive anything, as the only way it could is if the TryRead/TryWrite occurred at exactly the same time, and even then, most likely synchronization would prevent that from ever happening.

I don't think it's a valid point, cause FWIW there's already an abstraction leak present in case of UnboundedChannel, where you have implicit guarantee for TryWrite to always succeed. In general, I'd assume a fall back to async WriteAsync is pertinent in case of TryWrite returning false while channel isn't closed, and that's what unbuffered(rendezvous) channel would return unconditionally by design.

[stephentoub]

there's already an abstraction leak present in case of UnboundedChannel, where you have implicit guarantee for TryWrite to always succeed

That's not leaky. It's fine for TryWrite to always succeed, and it's fine for it to sometimes succeed. The problem is if it never succeeds, which is the concern here.

TryWrite / WaitToWriteAsync are the core of the abstraction; they're abstract, while WriteAsync is virtual (with a default implementation built on top of TryWrite and WaitToWriteAsync); same on the read side, with TryRead / WaitToReadAsync being abstract, and ReadAsync being a virtual with a default implementation layered on top. The expectation is that you can use WaitToWriteAsync, and when it completes with true, a TryWrite will typically succeed (whether it's yours or another competing write, or in the rare case where it's "consumed" by the operation being canceled). The difficulty with such a rendezvous channel is this may not be the case, depending on how the read side is implemented.

Imagine a method someone writes that will read one item from one of two channels, whichever has data first:

public static ValueTask<T> ReadAnyAsync<T>(ChannelReader<T> reader1, ChannelReader<T> reader2);

You can't just use ReadAsync on both reader1 and reader2, as doing so could end up consuming an item from both readers rather than just one of them, even if you cancel the second read the moment the first completes. So the typical implementation of such a method in terms of the abstraction would be to use WaitToReadAsync on both, effectively serving the same purpose as a "zero-byte read" common with networking and sockets, in order to be notified when data is available to read without actually consuming it. Only once you've been notified of data being available can you then consume it with TryRead. (And if you're dealing with concurrency and the possibility of someone else taking the item out from under you, you can loop.)

Same goes for the write side, e.g.

public static ValueTask WriteAnyAsync<T>(T item, ChannelWriter<T> writer1, ChannelWriter<T> writer2);

where you want to write the item to one and only one of the specified writers. If you were to WriteAsync on both, you could potentially end up writing the item to both, which would be erroneous. So you use WaitToWriteAsync to be notified of when space is available, and then TryWrite.

With a rendezvous channel, that ReadAnyAsync will work just fine if the channels are being written to with WriteAsync. And such a WriteAnyAsync will work just fine if the channels are being read from with ReadAsync. But if you were to use ReadAnyAsync with WriteAnyAsync, you'll deadlock, because none of the reads or writes will ever succeed.

My original prototype for the channels library had such a rendezvous channel, except it was exposed via a CreateUnbuffered method rather than a CreateRendezvous method, simply because I felt the terminology was less academic and more broadly understandable. But we removed it prior to shipping the initial release of System.Threading.Channels because of this issue. We can of course reconsider, but the core concern remains. See #24831 for reference.

[Szer]

Just for reference, Hopac library has both TryRead/TryWrite (as Ch.Try.give and Ch.Try.take) on Channel object (which can only be unbuffered) with such doc

module Ch =
  module Try =
    /// Creates a job that attempts to give a value to another job waiting on
    /// the given channel.  The result indicates whether a value was given or
    /// not.  Note that the other side of the communication must be blocked on
    /// the channel for communication to happen.
    val inline give: Ch<'x> -> 'x -> Job<bool>

    /// Creates a job that attempts to take a value from another job waiting on
    /// the given channel.  Note that the other side of the communication must
    /// be blocked on the channel for communication to happen.
    val inline take: Ch<'x> -> Job<option<'x>>