API Proposal: Inform client code when a BoundedChannel drops an item

[odydoum]

Background and Motivation

The current Channels API clearly states in all drop modes that the message is "ignored and dropped". But consider that the message/item we try to send has some sort of cleanup code (maybe an IDisposable or a pooled object) that always need to be called eventually. Since the API forces the consumer to "ignore" the dropped messages, instances of those classes are currently unusable without external modifications.

Proposed API

Add a callback mechanism for dropped messages. As an example, since this only makes sense for Bounded channels, it could be provided as an optional argument during the construction of the channel. Something like the following:

namespace System.Threading.Channels
{
    public static class Channel
    {
        public static Channel<T> CreateBounded<T>(int capacity);
        public static Channel<T> CreateBounded<T>(BoundedChannelOptions options);
+       public static Channel<T> CreateBounded<T>(BoundedChannelOptions options, Action<T> itemDropped);

    }
}

[RamType0]

I really want this!

[omariom]

The beauty of the channels is the simplicity of the contract. It allows them to be very efficient.
I believe they should stay as dumb as possible and the "logic" should be implemented by the consumers of the library.

For more complex and rich semantics there is another library; Dataflow.

[RamType0]

The beauty of the channels is the simplicity of the contract. It allows them to be very efficient.
I believe they should stay as dumb as possible and the "logic" should be implemented by the consumers of the library.

Yeah,I also think that,but...
How can we implement the logic to disposing dropped item?

There's no such API for implement the logic.

Also,the overhead of this feature is really low.We can know whether on dropped action has been assigned by very simple null check,and if it is not assigned,it never causes any virtual call.

For more complex and rich semantics there is another library; Dataflow.

Dataflow has lower performance than Channels.

More specifically, Dataflow and Channels are made by the same author, and Channels seems to be Dataflow's upward compatibility, combining Dataflow's reflections and ValueTask as a replacement for Task.

[odydoum]

The beauty of the channels is the simplicity of the contract. It allows them to be very efficient.
I believe they should stay as dumb as possible and the "logic" should be implemented by the consumers of the library.

Although I generaly agree, I still believe this change is reasonable. Currently you are forced to ignore dropped items, so there is no consistent way to implement "cleanup logic" (for example, cleanup code is easy to implement for "dropWrite" but nearly impossible for "dropOldest").

What I am proposing basically changes the semantics from "drop and ignore" to "drop and optionally handle", while leaving the responsibility to the consumer of how exactly this is gonna be implemented. In other words the contract stays basically intact, it just enhances on the current semantics.

Regarding efficiency, the overhead seems minimal on first inspection, but it leaves room to the consumer of the API to abuse the cb mechanism with some expensive operation. I don't know if this is considered a big problem.

[ghost]

Tagging subscribers to this area: @tarekgh
Notify danmosemsft if you want to be subscribed.

[tarekgh]

CC @stephentoub

[omariom]

If you control creation of the channel you could make it non-dropping and then TryWrite on the sending side and dispose the object in the right context.

[RamType0]

If you control creation of the channel you could make it non-dropping and then TryWrite on the sending side and dispose the object in the right context.

It works only when DropWrite.

[Temppus]

I would love to use channels with DropOldest behavior, but currently I do not have way to cleanup/dispose object that was dropped.

I have got 2 alternatives which would be great to consider:

1. Channel itself could dispose object if it implements IDisposable. This would require runtime cast/check but this behavior could be opt in only (by boolean flag).
2. Create another factory method for channel with generic constraint to IDisposable. When channel was create with this method it would dispose dropped item without casting overhead and this would not require some "hacky" boolean flag.

Thank you for consideration.

[stephentoub]

Functionally, this is a reasonable request, though I think it should be exposed differently. Rather than it being a separate CreateBounded method, the delegate should be added to BoundedChannelOptions: that's the reason BoundedChannelOptions exists, to be able to add new options in the future without a proliferation of CreateBounded overloads. Further, the default mode for CreateBounded is to not drop anything, so the only way the delegate would ever be invoked is if you were supplying a BoundedChannelOptions anyway that specified a different blocking mode.

public sealed class BoundedChannelOptions : ChannelOptions
{
    public Action<T> ItemDropped { get; set; }
    ...
}

There's also a question of whether this should be an asynchronous callback, e.g. public Func<T, ValueTask> ItemDropped { get; set; }, but that would be problematic when the drop happens from TryWrite, as TryWrite would then need to implicitly block waiting for the callback to complete if it completes asynchronously. We could say the callback should then be queued, but such queueing could also just be done by the delegate itself. Or we could say that TryWrite should simply not block in such cases, but that implicitly leaves asynchronous work pending, which seems far less than ideal to do by default. And I expect the vast majority use case would be quick functionality anyway, such as disposing of an item, or queueing some data to be logged, or some such thing. @davidfowl, would ASP.NET use this callback anywhere, and if so, concerns about it being sync?

---

EDIT: Almost immediately after posting I realized my suggestion doesn't work, as BoundedChannelOptions is separate from the generic type parameter passed to CreateBounded 🤦 (need to drink coffee in the morning before replying to issues). That leaves two main options:

1. Unseal BoundedChannelOptions and add a new BoundedChannelOptions<T> that derives from it and adds the option: Channel.CreateBounded<int>(new BoundedChannelOptions<int> { ..., ItemDropped = ... }).
2. Stick with the original proposal that adds a CreateBounded overload.

[Temppus]

Hi Stephen, I am glad that you found time to look at it.
I will be glad to help and drop a PR when implementation design will be decided.

[terrajobst]

Video

• Looks good as proposed

namespace System.Threading.Channels
{
    public static class Channel
    {
        // Existing:
        // public static Channel<T> CreateBounded<T>(int capacity);
        // public static Channel<T> CreateBounded<T>(BoundedChannelOptions options);
        public static Channel<T> CreateBounded<T>(BoundedChannelOptions options, Action<T> itemDropped);
    }
}

[Temppus]

Great to see proposal being approved. I will start working on this.