< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.ChannelSerialExecutor
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/ChannelSerialExecutor.cs
Line coverage
100%
Covered lines: 66
Uncovered lines: 0
Coverable lines: 66
Total lines: 170
Line coverage: 100%
Branch coverage
100%
Covered branches: 28
Total branches: 28
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_PendingCount()100%11100%
.ctor(...)100%22100%
DrainAsync()100%1010100%
Enqueue(...)100%22100%
EnqueueCoreAsync()100%22100%
TryEnqueue(...)100%88100%
DisposeAsync()100%44100%

File(s)

/_/src/AsyncResponse.Core/ChannelSerialExecutor.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using System.Threading.Channels;
 3
 4namespace AsyncResponse;
 5
 6/// <summary>
 7/// Executes asynchronous work items for a specific channel serially: a bounded
 8/// <see cref="Channel{T}"/> drained by a single reader loop guarantees per-channel ordering, so
 9/// progress messages for one correlation id are never processed concurrently or out of order.
 10/// <para>
 11/// This is a deliberately lean replacement for an earlier <c>ActionBlock&lt;Func&lt;Task&gt;&gt;</c>:
 12/// the rest of the hot path is hand-tuned to avoid allocations, and a single reader over a
 13/// <see cref="System.Threading.Channels"/> queue gives the same strict serial, in-order, one-at-a-time
 14/// semantics without TPL Dataflow's per-item task/scheduler overhead or the dependency it pulls in.
 15/// </para>
 16/// </summary>
 17internal sealed class ChannelSerialExecutor : IAsyncDisposable
 18{
 19    internal const int DefaultCapacity = 1024;
 20    private readonly Channel<Func<Task>> _queue;
 21    private readonly Task _readerLoop;
 22    private readonly ILogger _logger;
 23    private readonly string _channel;
 24
 25    // Items waiting for capacity or accepted into the queue but not yet pulled out for execution.
 26    // The item currently running is no longer "pending".
 27    private int _pending;
 28
 29    /// <summary>How many work items are currently waiting for capacity or waiting to run.</summary>
 1574930    private int PendingCount => Volatile.Read(ref _pending);
 31
 32    /// <summary>Runs the ChannelSerialExecutor operation.</summary>
 177933    public ChannelSerialExecutor(ILogger logger, string channel, int capacity = DefaultCapacity)
 34    {
 177935        if (capacity <= 0)
 236            throw new ArgumentOutOfRangeException(nameof(capacity));
 37
 177738        _logger = logger;
 177739        _channel = channel;
 40
 41        // Waiting writers apply backpressure instead of allowing an overloaded correlation id to
 42        // retain an unbounded delegate backlog. One reader preserves strict per-key ordering.
 177743        _queue = Channel.CreateBounded<Func<Task>>(new BoundedChannelOptions(capacity)
 177744        {
 177745            SingleReader = true,
 177746            SingleWriter = false,
 177747            AllowSynchronousContinuations = false,
 177748            FullMode = BoundedChannelFullMode.Wait
 177749        });
 50
 177751        _readerLoop = Task.Run(DrainAsync);
 177752    }
 53
 54    /// <summary>
 55    /// The single consumer: pulls work items in FIFO order and runs them one at a time. A work item
 56    /// that throws is logged and swallowed so the loop stays alive for the rest of the queue —
 57    /// exactly the resilience the old ActionBlock body provided.
 58    /// </summary>
 59    private async Task DrainAsync()
 60    {
 177761        var reader = _queue.Reader;
 427962        while (await reader.WaitToReadAsync().ConfigureAwait(false))
 63        {
 1805964            while (reader.TryRead(out var work))
 65            {
 1555766                Interlocked.Decrement(ref _pending);
 67
 1555768                if (_logger.IsEnabled(LogLevel.Debug))
 504169                    _logger.LogDebug("Channel executor starting work for {Channel} (pending {PendingCount}).", _channel,
 70
 71                try
 72                {
 1555773                    await work().ConfigureAwait(false);
 1554974                }
 275                catch (Exception ex)
 76                {
 277                    _logger.LogError(ex, "Channel executor error for {Channel} (pending {PendingCount}).", _channel, Pen
 78                    // swallow, so the loop stays alive
 279                }
 80                finally
 81                {
 1555182                    if (_logger.IsEnabled(LogLevel.Debug))
 504183                        _logger.LogDebug("Channel executor completed work for {Channel} (pending {PendingCount}).", _cha
 84                }
 85            }
 86        }
 87
 176988        if (_logger.IsEnabled(LogLevel.Debug))
 20689            _logger.LogDebug("Channel {Channel} executor completed (pending {PendingCount}).", _channel, PendingCount);
 176990    }
 91
 92    /// <summary>
 93    /// Queues a work delegate for execution. The returned task completes when the item has been
 94    /// accepted into the queue (not when the work is finished). Returns <c>false</c> when the
 95    /// executor is already shutting down (the queue was completed by <see cref="DisposeAsync"/>).
 96    /// </summary>
 97    public Task<bool> Enqueue(Func<Task> work, CancellationToken cancellationToken = default)
 98    {
 515899        ArgumentNullException.ThrowIfNull(work);
 5158100        if (cancellationToken.IsCancellationRequested)
 2101            return Task.FromCanceled<bool>(cancellationToken);
 102
 5156103        return EnqueueCoreAsync(work, cancellationToken);
 104    }
 105
 106    private async Task<bool> EnqueueCoreAsync(Func<Task> work, CancellationToken cancellationToken)
 107    {
 5156108        Interlocked.Increment(ref _pending);
 109        try
 110        {
 5156111            await _queue.Writer.WriteAsync(work, cancellationToken).ConfigureAwait(false);
 5152112            if (_logger.IsEnabled(LogLevel.Debug))
 301113                _logger.LogDebug("Channel executor enqueued work for {Channel} (pending {PendingCount}).", _channel, Pen
 5152114            return true;
 115        }
 2116        catch (ChannelClosedException)
 117        {
 2118            Interlocked.Decrement(ref _pending);
 2119            return false;
 120        }
 2121        catch
 122        {
 2123            Interlocked.Decrement(ref _pending);
 2124            throw;
 125        }
 5154126    }
 127
 128    /// <summary>
 129    /// Synchronously queues a work delegate, returning <c>false</c> when the executor is already
 130    /// shutting down or full. Use <see cref="Enqueue"/> when the producer can wait for capacity.
 131    /// <paramref name="logIfFull"/> is <c>false</c> for producers that treat a full queue as
 132    /// expected backpressure and come back later (the DB channels' dispatch sweep), so a busy
 133    /// correlation id does not log a warning per sweep tick.
 134    /// </summary>
 135    public bool TryEnqueue(Func<Task> work, bool logIfFull = true)
 136    {
 10415137        ArgumentNullException.ThrowIfNull(work);
 10415138        Interlocked.Increment(ref _pending);
 10415139        if (_queue.Writer.TryWrite(work))
 140        {
 10405141            if (_logger.IsEnabled(LogLevel.Debug))
 4740142                _logger.LogDebug("Channel executor enqueued work for {Channel} (pending {PendingCount}).", _channel, Pen
 10405143            return true;
 144        }
 145
 10146        Interlocked.Decrement(ref _pending);
 10147        if (logIfFull)
 4148            _logger.LogWarning("Channel executor could not enqueue work for {Channel}; queue is full or completed (pendi
 6149        else if (_logger.IsEnabled(LogLevel.Debug))
 2150            _logger.LogDebug("Channel executor for {Channel} is at capacity (pending {PendingCount}); the producer will 
 10151        return false;
 152    }
 153
 154    /// <summary>
 155    /// Signals that no more work items will be posted and waits for queued work to complete.
 156    /// </summary>
 157    public async ValueTask DisposeAsync()
 158    {
 1775159        if (_logger.IsEnabled(LogLevel.Debug))
 206160            _logger.LogDebug("Disposing channel executor for {Channel} (pending {PendingCount}).", _channel, PendingCoun
 161
 162        // Complete the writer so the reader drains the remaining items and the loop exits; then wait
 163        // for the loop to finish so callers can rely on all queued work having run.
 1775164        _queue.Writer.TryComplete();
 1775165        await _readerLoop.ConfigureAwait(false);
 166
 1769167        if (_logger.IsEnabled(LogLevel.Debug))
 206168            _logger.LogDebug("Disposed channel executor for {Channel} (pending {PendingCount}).", _channel, PendingCount
 1769169    }
 170}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

get_PendingCount()
.ctor(Microsoft.Extensions.Logging.ILogger,System.String,System.Int32)
DrainAsync()
Enqueue(System.Func`1<System.Threading.Tasks.Task>,System.Threading.CancellationToken)
EnqueueCoreAsync()
TryEnqueue(System.Func`1<System.Threading.Tasks.Task>,System.Boolean)
DisposeAsync()