dohertj2/mxaccessgw
1
0
Fork 0
Code Issues Pull Requests 2 Actions 29 Packages Projects Releases Wiki Activity

feat(gateway): decouple worker event enqueue from the read loop (GWC-04)

Browse Source 
The read loop awaited EnqueueWorkerEventAsync inline, which blocks in the bounded
event channel's timed WriteAsync (up to EventChannelFullModeTimeout, default 5s)
when the channel is full with no/slow consumer. A WorkerCommandReply or heartbeat
queued behind an event frame was then stalled, so an in-flight InvokeAsync could
hit CommandTimeout even though the worker replied in time.

Mirror the existing outbound WriteLoopAsync: add an unbounded event staging
channel and a dedicated EventWriteLoopAsync. DispatchEnvelope is now fully
synchronous — the WorkerEvent branch hands the event off with a non-blocking
TryWrite and the read loop never awaits. The event write loop owns the timed
WriteAsync into the bounded channel and the sustained-overflow ProtocolViolation
fault (unchanged contract). Registered in WaitForBackgroundTasks + completed on
close/fault/dispose.

Test: reply arriving after events with a full, consumer-less event channel is
dispatched promptly (no CommandTimeout) — pipe-harness, verified on windev.
Docs: GatewayProcessDesign read/write/event-loop section.
This commit is contained in:
Joseph Doherty
2026-07-09 09:28:13 -04:00
parent bfbfcdd112
commit 32d6b48910
3 changed files with 132 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs
+69 -13
View File
@@ -24,6 +24,13 @@ public sealed class WorkerClient : IWorkerClient
private readonly WorkerFrameWriter _writer;
private readonly Channel<WorkerEnvelope> _outboundEnvelopes;
private readonly Channel<WorkerEvent> _events;
// Staging hand-off between the read loop and the dedicated event writer. The read loop writes
// here with a non-blocking TryWrite so a full consumer channel (_events) can never stall the read
// loop behind an event — replies and heartbeats keep flowing (GWC-04). Unbounded, but only fills
// during the bounded EventChannelFullModeTimeout window before EventWriteLoopAsync faults on a
// sustained backlog, after which the read loop stops.
private readonly Channel<WorkerEvent> _eventStaging;
private readonly ConcurrentDictionary<string, PendingCommand> _pendingCommands = new(StringComparer.Ordinal);
private readonly SemaphoreSlim _pendingCommandSlots;
private readonly CancellationTokenSource _stopCts = new();
@@ -35,6 +42,7 @@ public sealed class WorkerClient : IWorkerClient
private int _eventsReaderClaimed;
private Task? _readLoopTask;
private Task? _writeLoopTask;
private Task? _eventWriteLoopTask;
private Task? _heartbeatLoopTask;
private bool _workerStartRecorded;
private bool _disposed;
@@ -82,6 +90,14 @@ public sealed class WorkerClient : IWorkerClient
FullMode = BoundedChannelFullMode.Wait,
AllowSynchronousContinuations = false,
});
_eventStaging = Channel.CreateUnbounded<WorkerEvent>(
new UnboundedChannelOptions
{
// The read loop is the only writer; EventWriteLoopAsync is the only reader.
SingleReader = true,
SingleWriter = true,
AllowSynchronousContinuations = false,
});
_lastHeartbeatAt = _timeProvider.GetUtcNow();
}
@@ -144,6 +160,7 @@ public sealed class WorkerClient : IWorkerClient
MarkReady(readyEnvelope.WorkerReady);
_readLoopTask = Task.Run(ReadLoopAsync);
_eventWriteLoopTask = Task.Run(EventWriteLoopAsync);
_heartbeatLoopTask = Task.Run(HeartbeatLoopAsync);
}
@@ -344,6 +361,7 @@ public sealed class WorkerClient : IWorkerClient
KillOwnedProcess("Dispose");
_stopCts.Cancel();
_outboundEnvelopes.Writer.TryComplete();
_eventStaging.Writer.TryComplete();
_events.Writer.TryComplete();
CompletePendingCommands(
new WorkerClientException(
@@ -398,7 +416,7 @@ public sealed class WorkerClient : IWorkerClient
while (!_stopCts.IsCancellationRequested)
{
WorkerEnvelope envelope = await _reader.ReadAsync(_stopCts.Token).ConfigureAwait(false);
await DispatchEnvelopeAsync(envelope, _stopCts.Token).ConfigureAwait(false);
DispatchEnvelope(envelope);
}
}
catch (OperationCanceledException) when (_stopCts.IsCancellationRequested || IsTerminalState())
@@ -497,12 +515,14 @@ public sealed class WorkerClient : IWorkerClient
return true;
}
/// <summary>Routes received envelope to appropriate handler.</summary>
/// <summary>
/// Routes a received envelope to its handler. Every branch dispatches synchronously and
/// immediately — the event branch only stages the event for the dedicated writer — so a full
/// event channel can never delay a command reply, heartbeat, fault, or shutdown ack behind an
/// event backlog (GWC-04).
/// </summary>
/// <param name="envelope">The envelope to dispatch.</param>
/// <param name="cancellationToken">Cancellation token.</param>
private async Task DispatchEnvelopeAsync(
WorkerEnvelope envelope,
CancellationToken cancellationToken)
private void DispatchEnvelope(WorkerEnvelope envelope)
{
switch (envelope.BodyCase)
{
@@ -510,7 +530,7 @@ public sealed class WorkerClient : IWorkerClient
CompleteCommand(envelope);
break;
case WorkerEnvelope.BodyOneofCase.WorkerEvent:
await EnqueueWorkerEventAsync(envelope.WorkerEvent, cancellationToken).ConfigureAwait(false);
StageWorkerEvent(envelope.WorkerEvent);
break;
case WorkerEnvelope.BodyOneofCase.WorkerHeartbeat:
MarkHeartbeat(envelope.WorkerHeartbeat);
@@ -533,6 +553,45 @@ public sealed class WorkerClient : IWorkerClient
}
}
/// <summary>
/// Hands a received worker event to the dedicated event writer without blocking the read loop.
/// The staging channel is unbounded and this is the only writer, so <c>TryWrite</c> always
/// succeeds unless the channel has been completed during shutdown — in which case the event is
/// safely dropped because the client is closing. Backpressure and the sustained-overflow fault
/// are applied by <see cref="EventWriteLoopAsync"/> against the bounded consumer channel,
/// off the read loop (GWC-04).
/// </summary>
/// <param name="workerEvent">The event received from the worker.</param>
private void StageWorkerEvent(WorkerEvent workerEvent)
{
if (workerEvent.Event is not null)
{
_metrics?.EventReceived(SessionId, workerEvent.Event.Family.ToString());
}
_eventStaging.Writer.TryWrite(workerEvent);
}
/// <summary>
/// Drains staged worker events and applies the bounded-channel backpressure (and
/// sustained-overflow fault) on a dedicated task, so the timed <see cref="Channel"/> write
/// never runs on the read loop (GWC-04). Mirrors <see cref="WriteLoopAsync"/> for events.
/// </summary>
private async Task EventWriteLoopAsync()
{
try
{
await foreach (WorkerEvent workerEvent in
_eventStaging.Reader.ReadAllAsync(_stopCts.Token).ConfigureAwait(false))
{
await EnqueueWorkerEventAsync(workerEvent, _stopCts.Token).ConfigureAwait(false);
}
}
catch (OperationCanceledException) when (_stopCts.IsCancellationRequested || IsTerminalState())
{
}
}
/// <summary>
/// Enqueues a worker event for client consumption. The channel is
/// configured with <see cref="BoundedChannelFullMode.Wait"/>
@@ -552,11 +611,6 @@ public sealed class WorkerClient : IWorkerClient
WorkerEvent workerEvent,
CancellationToken cancellationToken)
{
if (workerEvent.Event is not null)
{
_metrics?.EventReceived(SessionId, workerEvent.Event.Family.ToString());
}
if (_events.Writer.TryWrite(workerEvent))
{
int queueDepth = Interlocked.Increment(ref _eventQueueDepth);
@@ -747,6 +801,7 @@ public sealed class WorkerClient : IWorkerClient
_stopCts.Cancel();
_outboundEnvelopes.Writer.TryComplete();
_eventStaging.Writer.TryComplete();
_events.Writer.TryComplete();
CompletePendingCommands(
new WorkerClientException(
@@ -780,6 +835,7 @@ public sealed class WorkerClient : IWorkerClient
_stopCts.Cancel();
_outboundEnvelopes.Writer.TryComplete(fault);
_eventStaging.Writer.TryComplete(fault);
_events.Writer.TryComplete(fault);
KillOwnedProcess(errorCode.ToString());
CompletePendingCommands(fault);
@@ -1005,7 +1061,7 @@ public sealed class WorkerClient : IWorkerClient
/// <param name="cancellationToken">Cancellation token.</param>
private async Task WaitForBackgroundTasksAsync(CancellationToken cancellationToken)
{
Task[] tasks = new[] { _readLoopTask, _writeLoopTask, _heartbeatLoopTask }
Task[] tasks = new[] { _readLoopTask, _writeLoopTask, _eventWriteLoopTask, _heartbeatLoopTask }
.Where(task => task is not null)
.Cast<Task>()
.ToArray();