Channel causing memory hikes

Question

I am learning about C# Channel library and trying to create a Producer/Consumer pattern. I wrote a simple example and find that there is huge memory hike when running it. Please help me find out why this is happening. I am using .NET 8, Visual Studio 2022.

public class Program
{
    static async Task Main(string[] args)
    {
        ProducerConsumerManager<int> manager = new ProducerConsumerManager<int>();
        DataGenerator generator = new DataGenerator();
        int data = 0;
        _ = Task.Run(async () =>
        {
            for (int i = 0; ; i++)
            {
                data = generator.GenerateData();
                await manager.ProduceAsync(data);
            }
        });

        while (true)
        {
            var message = await manager.ConsumeAsync();
            Console.WriteLine(message);
        }
    }
}
public class DataGenerator
{
    private Random rand = new Random();

    public int GenerateData()
    {
        return rand.Next(0, 200);
    }
}

public class ProducerConsumerManager<T>
{
    private static Channel<T> _channel = Channel.CreateUnbounded<T>();
    private ChannelWriter<T> _writer = _channel.Writer;
    private ChannelReader<T> _reader = _channel.Reader;

    public async Task ProduceAsync(T data)
    {
        await _writer.WriteAsync(data);
    }
    public async Task<T> ConsumeAsync()
    {
        return await _reader.ReadAsync();
    }
}

I tried added await Task.Delay(500) at the end of for loop but the memory is still going up.

Answer 1

As others have commented, the "producer" produces millions of items per second while the consumer reads at a much slower pace. There's no leak. To avoid this situation either use a bounded channel or add a delay to the producer loop.

The code is far more complicated than it needs to be. Channels have separate writers and readers for a reason. A producer only needs a ChannelWriter. A consumer only needs a ChannelReader. In fact, the producer could "own" the channel and only expose a reader to callers.

The code could be as simple as the following code, which creates the channel internally and only returns a reader:

ChannelReader<int> Producer(CancellationToken ct)
{
    var channel=Channel.CreateBounded<int>(10);
    var writer=channel.Writer;

    _ = Task.Run(()=>{
        var generator=new DataGenerator();
        while(!ct.CancellationRequested)
        { 
            var message=generator.GenerateData();
            await writer.WriteAsync(message);
        }
    }).ContinueWith(t=>channel.Writer.TryComplete(t.Exception));

    return channel.Reader;
}

All that's needed in the consumer code is to iterate over the IAsyncEnumerable<T> returned by ChannelReader<T>.ReadAllAsync :

// Stop automatically after 5 minutes
var cts=new CancellationTokenSource(TimeSpan.FromMinutes(5));

var reader=Producer(cts.Token);

await foreach(var message in reader.ReadAllAsync())
{
    Console.WriteLine(message);
}

You could return the IAsyncEnumerable<T> from the producer too:

IAsyncEnumerable<int> Producer(CancellationToken ct)
{
    var channel=Channel.CreateBounded<int>(10);
    var writer=channel.Writer;

    _ = Task.Run(()=>{
    ...
    }).ContinueWith(t=>channel.Writer.TryComplete(t.Exception));

    return channel.Reader.ReadAllAsync();
}

Pipelines

This pattern ensures we can control the lifetime of the channel and complete even in case of exceptions. It also makes it easier to create pipelines of steps that each consume one reader and produce another, eg :

public static ChannelReader<int> Square(this ChannelReader<int> reader,CancellationToken ct)
{
    var channel=Channel.CreateBounded<int>(10);
    var writer=channel.Writer;

    _ = Task.Run(()=>{
        await foreach (var msg in reader.ReadAllAsync(ct))
        { 
            var msg2=Math.Square(msg);
            await writer.WriteAsync(msg2);
        }
    }).ContinueWith(t=>channel.Writer.TryComplete(t.Exception));

    return channel.Reader;
}
...
var results=Producer(ct).Square(ct).....;
await foreach(var msg in results.ReadAllAsync())
{
...
}

You can use Async LINQ methods with the IAsyncEnumerable returned by ChannelReader. In fact, you could return an `IAsyncEnumerable