C#•12mo ago

✅ System.Threading.Channels - Consumer is blocking the main thread where the producer is

Hello,
The producer writes to the channel with

WriteAsync()

WriteAsync()

WriteAsync()

WriteAsync(), but the consumer blocks the termination of the producer.
The outcome should be a non-blocking read, so the producer would complete in the meantime.

The following is the consumer:

c#
while (await _reader.WaitToReadAsync())
{ 
    var metadata = await _reader.ReadAsync();
}

c#
while (await _reader.WaitToReadAsync())
{ 
    var metadata = await _reader.ReadAsync();
}

c#
while (await _reader.WaitToReadAsync())
{ 
    var metadata = await _reader.ReadAsync();
}

c#
while (await _reader.WaitToReadAsync())
{ 
    var metadata = await _reader.ReadAsync();
}

c#
public sealed class VisitMetadataChannel
{
    private readonly Channel<VisitMetadata> _channel = Channel.CreateUnbounded<VisitMetadata>(
        new UnboundedChannelOptions
        {
            SingleReader = true, 
            SingleWriter = false, 
            AllowSynchronousContinuations = false
        });
    
    public ChannelWriter<VisitMetadata> Writer => _channel.Writer;
    
    public ChannelReader<VisitMetadata> Reader => _channel.Reader;
}

c#
public sealed class VisitMetadataChannel
{
    private readonly Channel<VisitMetadata> _channel = Channel.CreateUnbounded<VisitMetadata>(
        new UnboundedChannelOptions
        {
            SingleReader = true, 
            SingleWriter = false, 
            AllowSynchronousContinuations = false
        });
    
    public ChannelWriter<VisitMetadata> Writer => _channel.Writer;
    
    public ChannelReader<VisitMetadata> Reader => _channel.Reader;
}

c#
public sealed class VisitMetadataChannel
{
    private readonly Channel<VisitMetadata> _channel = Channel.CreateUnbounded<VisitMetadata>(
        new UnboundedChannelOptions
        {
            SingleReader = true, 
            SingleWriter = false, 
            AllowSynchronousContinuations = false
        });
    
    public ChannelWriter<VisitMetadata> Writer => _channel.Writer;
    
    public ChannelReader<VisitMetadata> Reader => _channel.Reader;
}

c#
public sealed class VisitMetadataChannel
{
    private readonly Channel<VisitMetadata> _channel = Channel.CreateUnbounded<VisitMetadata>(
        new UnboundedChannelOptions
        {
            SingleReader = true, 
            SingleWriter = false, 
            AllowSynchronousContinuations = false
        });
    
    public ChannelWriter<VisitMetadata> Writer => _channel.Writer;
    
    public ChannelReader<VisitMetadata> Reader => _channel.Reader;
}

Sehra•1/23/25, 10:03 AM

how is the consumer started?

CoreOP•1/23/25, 10:10 AM

It's a quartz background job

CoreOP•1/23/25, 10:11 AM

The quartz scheduler is running separately from the .net app

Sehra•1/23/25, 10:13 AM

WaitToReadAsync will wait either for an item to be available (returns true) or that you do Writer.Complete(), when it returns false

Sehra•1/23/25, 10:14 AM

if you just want the consumer to process what is available and then exit, use while (Reader.TryRead(out var item)) {}while (Reader.TryRead(out var item)) {}

CoreOP•1/23/25, 10:16 AM

The consumer must run until the app stops. The producer should just write and ignore the rest

Sehra•1/23/25, 10:19 AM

so how would that block terminating the producer?

Sehra•1/23/25, 10:20 AM

normally you do Writer.Complete() to say no more data is coming, then await Reader.Completion to see when everything is processed

Sehra•1/23/25, 10:24 AM

var channel = Channel.CreateUnbounded<int>();
_ = Task.Run(async () =>
{
    while (await channel.Reader.WaitToReadAsync())
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
        Console.WriteLine(await channel.Reader.ReadAsync());
    }
});
await channel.Writer.WriteAsync(1);
await channel.Writer.WriteAsync(2);
await channel.Writer.WriteAsync(3);
channel.Writer.Complete();
await channel.Reader.Completion;
Console.WriteLine("done");

var channel = Channel.CreateUnbounded<int>();
_ = Task.Run(async () =>
{
    while (await channel.Reader.WaitToReadAsync())
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
        Console.WriteLine(await channel.Reader.ReadAsync());
    }
});
await channel.Writer.WriteAsync(1);
await channel.Writer.WriteAsync(2);
await channel.Writer.WriteAsync(3);
channel.Writer.Complete();
await channel.Reader.Completion;
Console.WriteLine("done");

CoreOP•1/23/25, 10:27 AM

I modified the code as follows:

c#
 while (_reader.TryRead(out var metadata))
{ 
    _logger.LogInformation("Hello");
}

c#
 while (_reader.TryRead(out var metadata))
{ 
    _logger.LogInformation("Hello");
}

c#
await _writer.WriteAsync(metadata, ct);
_writer.Complete();

c#
await _writer.WriteAsync(metadata, ct);
_writer.Complete();

The producer now termines. But the consumer never consumes the message

Sehra•1/23/25, 10:28 AM

in that case the consumer will exit at any point the channel is empty

Sehra•1/23/25, 10:30 AM

just keep in mind you can't write more to the channel after you complete it

CoreOP•1/23/25, 10:31 AM

write happens during an enpoint call

CoreOP•1/23/25, 10:31 AM

I think it's clearer now, thank you

CoreOP•1/23/25, 10:31 AM

I will try to do the modifications

CoreOP•1/23/25, 10:37 AM

The Reader.WaitToReadAsync()Reader.WaitToReadAsync() causes the blocking. But I don't want that at all.
Is a while(true)while(true) block with delay in it a good solution?

Sehra•1/23/25, 10:49 AM

that's the whole idea, it will wait either for more data to be available, or no data to ever be available

Sehra•1/23/25, 10:50 AM

you have to explain how you want it to work

CoreOP•1/23/25, 10:59 AM

The producer writes to the channel, and it does not care what happens, it needs no response, so it must not be blocked (fire & forget)
The consumer reads from the channel and processes the data without blocking anyone. The consumer exists when the main application stops running

Sehra•1/23/25, 11:05 AM

if you put the consumer in a background service, you can use the stoppingToken to complete the writer and wait for reader completion

Sehra•1/23/25, 11:05 AM

or just hook the lifetime events yourself, but hosted service is quite easy

CoreOP•1/23/25, 11:06 AM

Do you refer to the built in backgroud service?

CoreOP•1/23/25, 11:07 AM

IHostedService?

Sehra•1/23/25, 11:07 AM

yeah, inherit from BackgroundService and it takes care of the lifetime events

CoreOP•1/23/25, 11:09 AM

Yes it might be more optimal for a job like this, since it's lifespan is bound to the app's lifetime

CoreOP•1/23/25, 11:09 AM

Many thanks, will try to implement it

CoreOP•1/23/25, 11:10 AM

Would this mean spawning separate backgound services?

Sehra•1/23/25, 11:11 AM

for what?

CoreOP•1/23/25, 11:12 AM

the stopping token, does it only signals the caller that the execution was canceled?

Sehra•1/23/25, 11:13 AM

in BackgroundService, it signals that the app is stopping

CoreOP•1/23/25, 11:52 AM

no matter what combination of write, read method I try, the producer is always blocked

c#
while (!context.CancellationToken.IsCancellationRequested)
{
    _reader.TryRead(out var metadata);

    if (metadata is not null) 
    {
      //handle
    }
}

c#
while (!context.CancellationToken.IsCancellationRequested)
{
    _reader.TryRead(out var metadata);

    if (metadata is not null) 
    {
      //handle
    }
}

c#
_writer.TryWrite(metadata)
_writer.Complete();

c#
_writer.TryWrite(metadata)
_writer.Complete();

CoreOP•1/23/25, 11:53 AM

after Complete()Complete() is called, the consumer instanly executes, blocking the producer

Sehra•1/23/25, 11:55 AM

that is bad code, it will spin as fast as it can

CoreOP•1/23/25, 11:56 AM

I know, but the delay will not solve the blocking

CoreOP•1/23/25, 11:58 AM

Why is the producer even blocked is beyond me, it should only write to the channel and call it a day

Sehra•1/23/25, 11:59 AM

hard to say when you only show two lines

CoreOP•1/23/25, 12:06 PM

Everything else is business logic, unrelated to the issue. There are 2 lines in the producer that are relevant. Once the TryWrite()TryWrite() is called, the TryRead()TryRead() executes, and the consumer is blocked

c#       
_writer.TryWrite(metadata);
_writer.Complete();

c#       
_writer.TryWrite(metadata);
_writer.Complete();

c#
public async Task Execute(IJobExecutionContext context)
{
    while (!context.CancellationToken.IsCancellationRequested)
    {
        _reader.TryRead(out var metadata);

        if (metadata is not null) 
        {
            Console.WriteLine("");
        }
    }
}

c#
public async Task Execute(IJobExecutionContext context)
{
    while (!context.CancellationToken.IsCancellationRequested)
    {
        _reader.TryRead(out var metadata);

        if (metadata is not null) 
        {
            Console.WriteLine("");
        }
    }
}

Sehra•1/23/25, 12:17 PM

but why are you running the consumer via Quartz?

Sehra•1/23/25, 12:19 PM

if you plan on running it periodically via Quartz, you can use while (_reader.TryRead(out var metadata))while (_reader.TryRead(out var metadata)), but i don't think Quartz is a good fit for something that should run for the duration of the application. if so use a hosted service

CoreOP•1/23/25, 12:21 PM

Sorry

CoreOP•1/23/25, 12:21 PM

I am just dumb. It was working all along

CoreOP•1/23/25, 12:21 PM

it's just in debug because there was no delay it automatically executed that part of the code....

CoreOP•1/23/25, 12:21 PM

:sadge:

CoreOP•1/23/25, 12:23 PM

Thank you for having the patience for me.... :))

Sehra•1/23/25, 12:24 PM

no worries, good that it's working

CoreOP•1/23/25, 12:24 PM

$close

CCore $close

MODiX•1/23/25, 12:24 PM

If you have no further questions, please use /close to mark the forum thread as answered

TeBeCo•1/23/25, 12:26 PM

if you need hight performance you'd need to use a double loop

TeBeCo•1/23/25, 12:26 PM

outer loop being async, inner loop using the sync one

TeBeCo•1/23/25, 12:26 PM

$channels

✅ System.Threading.Channels - Consumer is blocking the main thread where the producer is

Similar Threads

Similar Threads

Similar Threads