| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using System.Threading.Channels; |
| | | 3 | | |
| | | 4 | | namespace 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<Func<Task>></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> |
| | | 17 | | internal 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> |
| | 3 | 30 | | private int PendingCount => Volatile.Read(ref _pending); |
| | | 31 | | |
| | | 32 | | /// <summary>Runs the ChannelSerialExecutor operation.</summary> |
| | 3 | 33 | | public ChannelSerialExecutor(ILogger logger, string channel, int capacity = DefaultCapacity) |
| | | 34 | | { |
| | 3 | 35 | | if (capacity <= 0) |
| | 3 | 36 | | throw new ArgumentOutOfRangeException(nameof(capacity)); |
| | | 37 | | |
| | 3 | 38 | | _logger = logger; |
| | 3 | 39 | | _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. |
| | 3 | 43 | | _queue = Channel.CreateBounded<Func<Task>>(new BoundedChannelOptions(capacity) |
| | 3 | 44 | | { |
| | 3 | 45 | | SingleReader = true, |
| | 3 | 46 | | SingleWriter = false, |
| | 3 | 47 | | AllowSynchronousContinuations = false, |
| | 3 | 48 | | FullMode = BoundedChannelFullMode.Wait |
| | 3 | 49 | | }); |
| | | 50 | | |
| | 3 | 51 | | _readerLoop = Task.Run(DrainAsync); |
| | 3 | 52 | | } |
| | | 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 | | { |
| | 3 | 61 | | var reader = _queue.Reader; |
| | 3 | 62 | | while (await reader.WaitToReadAsync().ConfigureAwait(false)) |
| | | 63 | | { |
| | 3 | 64 | | while (reader.TryRead(out var work)) |
| | | 65 | | { |
| | 3 | 66 | | Interlocked.Decrement(ref _pending); |
| | | 67 | | |
| | 3 | 68 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 69 | | _logger.LogDebug("Channel executor starting work for {Channel} (pending {PendingCount}).", _channel, |
| | | 70 | | |
| | | 71 | | try |
| | | 72 | | { |
| | 3 | 73 | | await work().ConfigureAwait(false); |
| | 3 | 74 | | } |
| | 3 | 75 | | catch (Exception ex) |
| | | 76 | | { |
| | 3 | 77 | | _logger.LogError(ex, "Channel executor error for {Channel} (pending {PendingCount}).", _channel, Pen |
| | | 78 | | // swallow, so the loop stays alive |
| | 2 | 79 | | } |
| | | 80 | | finally |
| | | 81 | | { |
| | 3 | 82 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 83 | | _logger.LogDebug("Channel executor completed work for {Channel} (pending {PendingCount}).", _cha |
| | | 84 | | } |
| | | 85 | | } |
| | | 86 | | } |
| | | 87 | | |
| | 3 | 88 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 89 | | _logger.LogDebug("Channel {Channel} executor completed (pending {PendingCount}).", _channel, PendingCount); |
| | 3 | 90 | | } |
| | | 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 | | { |
| | 3 | 99 | | ArgumentNullException.ThrowIfNull(work); |
| | 3 | 100 | | if (cancellationToken.IsCancellationRequested) |
| | 3 | 101 | | return Task.FromCanceled<bool>(cancellationToken); |
| | | 102 | | |
| | 3 | 103 | | return EnqueueCoreAsync(work, cancellationToken); |
| | | 104 | | } |
| | | 105 | | |
| | | 106 | | private async Task<bool> EnqueueCoreAsync(Func<Task> work, CancellationToken cancellationToken) |
| | | 107 | | { |
| | 3 | 108 | | Interlocked.Increment(ref _pending); |
| | | 109 | | try |
| | | 110 | | { |
| | 3 | 111 | | await _queue.Writer.WriteAsync(work, cancellationToken).ConfigureAwait(false); |
| | 3 | 112 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 113 | | _logger.LogDebug("Channel executor enqueued work for {Channel} (pending {PendingCount}).", _channel, Pen |
| | 3 | 114 | | return true; |
| | | 115 | | } |
| | 3 | 116 | | catch (ChannelClosedException) |
| | | 117 | | { |
| | 3 | 118 | | Interlocked.Decrement(ref _pending); |
| | 2 | 119 | | return false; |
| | | 120 | | } |
| | 2 | 121 | | catch |
| | | 122 | | { |
| | 2 | 123 | | Interlocked.Decrement(ref _pending); |
| | 2 | 124 | | throw; |
| | | 125 | | } |
| | 3 | 126 | | } |
| | | 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 | | /// </summary> |
| | | 132 | | public bool TryEnqueue(Func<Task> work) |
| | | 133 | | { |
| | 3 | 134 | | ArgumentNullException.ThrowIfNull(work); |
| | 2 | 135 | | Interlocked.Increment(ref _pending); |
| | 2 | 136 | | if (_queue.Writer.TryWrite(work)) |
| | | 137 | | { |
| | 2 | 138 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 2 | 139 | | _logger.LogDebug("Channel executor enqueued work for {Channel} (pending {PendingCount}).", _channel, Pen |
| | 2 | 140 | | return true; |
| | | 141 | | } |
| | | 142 | | |
| | 2 | 143 | | Interlocked.Decrement(ref _pending); |
| | 2 | 144 | | _logger.LogWarning("Channel executor could not enqueue work for {Channel}; queue is full or completed (pending { |
| | 2 | 145 | | return false; |
| | | 146 | | } |
| | | 147 | | |
| | | 148 | | /// <summary> |
| | | 149 | | /// Signals that no more work items will be posted and waits for queued work to complete. |
| | | 150 | | /// </summary> |
| | | 151 | | public async ValueTask DisposeAsync() |
| | | 152 | | { |
| | 3 | 153 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 154 | | _logger.LogDebug("Disposing channel executor for {Channel} (pending {PendingCount}).", _channel, PendingCoun |
| | | 155 | | |
| | | 156 | | // Complete the writer so the reader drains the remaining items and the loop exits; then wait |
| | | 157 | | // for the loop to finish so callers can rely on all queued work having run. |
| | 3 | 158 | | _queue.Writer.TryComplete(); |
| | 3 | 159 | | await _readerLoop.ConfigureAwait(false); |
| | | 160 | | |
| | 3 | 161 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 162 | | _logger.LogDebug("Disposed channel executor for {Channel} (pending {PendingCount}).", _channel, PendingCount |
| | 3 | 163 | | } |
| | | 164 | | } |