# Using Prioritized Channel

In .NET 9, a new `UnboundedPrioritized` channel type has been introduced to [System.Threading.Channels](https://www.nuget.org/packages/System.Threading.Channels). This feature has been available since version 8 of the Channel's NuGet package and is compatible with older .NET versions.

[Channels](https://learn.microsoft.com/en-us/dotnet/core/extensions/channels) provide thread-safe data structures for producer/consumer scenarios. In this pattern, one or more *producers* add items to a channel while one or more *consumers* read from it independently. 

Two types of channels exist:
- **Bounded channels**: Have a maximum size limit with customizable behavior when full
- **Unbounded channels**: Have no size limit (beyond system memory constraints)

## Prioritized Channels

Prioritized Channels add priority-based ordering to channel operations. Items carry priority information and are consumed in priority order. While single items are consumed immediately, when multiple items accumulate (when producers outpace consumers), they are read according to their priority.

The [UnboundedPriorityChannel](https://source.dot.net/#System.Threading.Channels/System/Threading/Channels/UnboundedPriorityChannel.cs,d7d606c88be10a9e) internally uses a `PriorityQueue` for backing storage. Each item is stored alongside a boolean flag, resulting in slight overhead in memory usage.

For example, if integers `1, 3, 2, 4, 2` are added to the channel (where the value serves as priority), consumers will read them in the order: `1, 2, 2, 3, 4`.

## Creating a Prioritized Unbounded Channel

To create a prioritized unbounded channel, use the static `CreateUnboundedPrioritized` method. This method requires:

1. A generic type parameter defining the type of objects that can be added and retrieved from the queue
2. An optional `UnboundedPrioritizedChannelOptions` type to customize the channel's behavior, including a `Comparer` property

```csharp
Channel.CreateUnboundedPrioritized<int>(new()
{
    SingleReader = true,
    SingleWriter = false,
    AllowSynchronousContinuations = true,
    Comparer = Comparer<int>.Default,
});
```

Items can be added to the channel with the regular `WriteAsync` or `TryWrite` methods:

```csharp
// Examples of writing to channel
await channel.Writer.WriteAsync(1);
await channel.Writer.WriteAsync(3);
await channel.Writer.WriteAsync(2);
await channel.Writer.WriteAsync(4);
await channel.Writer.WriteAsync(2);
```

Followed by a `channel.Writer.Complete();` when the writers complete writing items to channel. Reading this channel works with the same abstractions as the other channels, for example, all items can be read as `await foreach (var item in channel.Reader.ReadAllAsync())`. This foreach loop exits when the writers completed writing items and the readers consumed all items from the channel. 

## Custom Types and Comparers

While using integers as priorities is a straightforward example, more commonly, a custom type (`class` or `struct`) is used as the channel's item type. In these cases, the `Comparer` property becomes essential, requiring a developer to provide an `IComparer<T>` implementation with the channel's options, or a `IComparable<T>` type for the channel's generic type parameter. The corresponding comparer will be then used to order the items in the channel. I explored sorting with `IComparer<T>` or `IComparable<T>` [with reference types](https://blog.ladeak.net/posts/sorting-stable-unstable) and [with value types](https://blog.ladeak.net/posts/struct-sorting-stable-unstable) in my previous blog posts. Priority sorting in these channels is not stable, meaning items with the same priority might not retain their original order.

## Use Cases

The following scenarios, without being comprehensive, illustrate the usefulness of prioritized channels:

1. Protocol Implementation: When implementing HTTP/2 or HTTP/3 servers that use [Priority RFC 9218](https://www.rfc-editor.org/rfc/rfc9218), prioritized channels can order HTTP frames appropriately.
1. Network Communication: Applications often share network connections to databases, caches (e.g., Redis), or logging services. When multiple tasks use these shared resources, prioritized channels can help manage message ordering.
1. Real-time Processing: Scenarios like:
   - IO scheduling
   - Audio/video processing
   - Market data processing
   - Thread scheduling

## Performance Analysis

Using .NET 10 and [BenchmarkDotNet](https://benchmarkdotnet.org/), I compared the performance of priority channels versus regular channels. These tests focus on single reader with multiple writers scenarios.

The benchmarks use these helper methods:

```csharp
Task<int> Readers(Channel<int> channel)
{
    return Task.Run(async () =>
    {
        int i = 0;
        await foreach (var item in channel.Reader.ReadAllAsync())
        {
            i++;
        }
        if(i != 100_000)
            throw new Exception();
        return i;
    });
}

async Task Writers(Channel<int> channel)
{
    List<Task> tasks = new List<Task>();
    for (int i = 0; i < 10; i++)
    {
        tasks.Add(Task.Run(async () =>
        {
            for (int j = 10000; j > 0; j--)
            {
                await channel.Writer.WriteAsync(j);
            }
        }));
    }
    await Task.WhenAll(tasks);
    channel.Writer.Complete();
}
```

### Value Type Benchmark Results

*ReadPrio* test measures consuming 10000 elements from an unbounded priority channel of `int` with a single reader.
*ReadNonPrio* test measures consuming 10000 elements from an unbounded channel of `int` with a single reader.

```
| Method      | Mean     | Error     | StdDev    | Median   |
|------------ |---------:|----------:|----------:|---------:|
| ReadPrio    | 9.876 ms | 0.1968 ms | 0.4064 ms | 9.746 ms |
| ReadNonPrio | 1.648 ms | 0.0704 ms | 0.1950 ms | 1.552 ms |
```

*WritePrio* test measures adding 10000 elements to an unbounded priority channel of `int` with 10 writers.
*WriteNonPrio* test measures adding 10000 elements to an unbounded channel of `int` with 10 writers.

```
| Method          | Mean      | Error     | StdDev   | Median    |
|---------------- |----------:|----------:|---------:|----------:|
| WritePrio       |  9.296 ms | 0.4110 ms | 1.173 ms |  9.075 ms |
| WriteNonPrio    | 10.053 ms | 0.6427 ms | 1.895 ms |  9.783 ms |
```

*CompletePrio* test measures adding 10000 elements to an unbounded priority channel of `int` with 10 writers and consuming all with a single writer.
*CompleteNonPrio* test measures adding 10000 elements to an unbounded channel of `int` with 10 writers and consuming all with a single writer.

```
| Method          | Mean      | Error     | StdDev   | Median    |
|---------------- |----------:|----------:|---------:|----------:|
| CompletePrio    | 20.034 ms | 0.5927 ms | 1.738 ms | 19.752 ms |
| CompleteNonPrio | 11.664 ms | 0.5235 ms | 1.543 ms | 12.070 ms |
```

The key difference is between how items are consumed. An unbounded priority channel reads items a magnitude slower. The impact is significant enough that it is measurable on the overal *Complete* measurments too.

### Reference Type Benchmark Results

Let's repeat the same benchmarks with a reference type. In this case using an `IComparer<MyType>` gave a slight performance benefit:

```csharp
public class MyType
{
    public int Value { get; set; }
}

public class MyComparer : IComparer<MyType>
{
    public int Compare(MyType x, MyType y) => x.Value.CompareTo(y.Value);
}
```

*ReadPrio* test measures consuming 10000 elements from an unbounded priority channel of `MyType` with a single reader.
*ReadNonPrio* test measures consuming 10000 elements from an unbounded channel of `MyType` with a single reader.
*WritePrio* test measures adding 10000 elements to an unbounded priority channel of `MyType` with 10 writers.
*WritePrio* test measures adding 10000 elements to an unbounded channel of `MyType` with 10 writers.
*CompletePrio* test measures adding 10000 elements to an unbounded priority channel of `MyType` with 10 writers and consuming all with a single writer.
*CompleteNonPrio* test measures adding 10000 elements to an unbounded channel of `MyType` with 10 writers and consuming all with a single writer.

```
| Method          | Mean      | Error     | StdDev    |
|---------------- |----------:|----------:|----------:|
| ReadPrio        | 16.008 ms | 0.3146 ms | 0.4200 ms |
| ReadNonPrio     |  1.831 ms | 0.0932 ms | 0.2659 ms |
| WritePrio       | 10.437 ms | 0.3486 ms | 1.006 ms  |
| WriteNonPrio    |  8.782 ms | 0.4663 ms | 1.368 ms  |
| CompletePrio    | 25.626 ms | 0.5591 ms | 1.595 ms  |
| CompleteNonPrio | 12.699 ms | 0.3595 ms | 1.054 ms  |
```

Overall the measurements show that using a reference type might be slower, however please note, that this includes the instantiation of `MyType` objects.

## Conclusion

The new `UnboundedPrioritizedChannel` is a valuable addition to .NET's channel library, particularly for scenarios requiring message prioritization. Key takeaways:

- Writing performance is comparable to regular channels.
- Reading has significant overhead due to priority ordering.
- Use regular channels when priority ordering isn't required.
- Best suited for scenarios where message order matters more than raw throughput.