diff --git a/docs/GatewayProcessDesign.md b/docs/GatewayProcessDesign.md index ae5b5ca..3437b38 100644 --- a/docs/GatewayProcessDesign.md +++ b/docs/GatewayProcessDesign.md @@ -481,7 +481,9 @@ Internally it owns: - pipe stream, - read loop, - write loop, +- event write loop, - outbound command/control channel serialized by the write loop, +- unbounded event staging channel drained by the event write loop, - bounded inbound event channel, - pending command dictionary keyed by correlation id, - heartbeat monitor, @@ -499,15 +501,29 @@ The read loop: 1. Reads one frame. 2. Parses `WorkerEnvelope`. 3. Validates protocol fields. -4. Dispatches by body type: +4. Dispatches by body type, **synchronously and without blocking**: - `WorkerCommandReply`: completes pending command. - - `WorkerEvent`: enqueues event. + - `WorkerEvent`: hands the event to the event write loop via a non-blocking staging write. - `WorkerHeartbeat`: updates heartbeat timestamp. - `WorkerFault`: faults session. 5. Stops when pipe closes or cancellation is requested. If the pipe closes while the session is not closing, fault the session. +The read loop never awaits event enqueue. Events are staged to the event write +loop with a non-blocking write, so a full inbound event channel (a slow or +absent `StreamEvents` consumer) cannot stall the read loop behind an event and +delay a command reply or heartbeat (GWC-04). The bounded-channel backpressure +window (`EventChannelFullModeTimeout`) and the sustained-overflow fault are +applied by the event write loop, not the read loop. + +### Event write loop + +The event write loop drains the staging channel and performs the timed write +into the bounded inbound event channel. When the inbound channel stays full past +`EventChannelFullModeTimeout` it faults the session (`ProtocolViolation`) — the +same overflow contract as before, moved off the read loop. + ### Write loop The write loop serializes all writes to the pipe. No other code should write to diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index b8e603a..342ae6d 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -24,6 +24,13 @@ public sealed class WorkerClient : IWorkerClient private readonly WorkerFrameWriter _writer; private readonly Channel _outboundEnvelopes; private readonly Channel _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 _eventStaging; private readonly ConcurrentDictionary _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( + 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; } - /// Routes received envelope to appropriate handler. + /// + /// 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). + /// /// The envelope to dispatch. - /// Cancellation token. - 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 } } + /// + /// 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 TryWrite 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 against the bounded consumer channel, + /// off the read loop (GWC-04). + /// + /// The event received from the worker. + private void StageWorkerEvent(WorkerEvent workerEvent) + { + if (workerEvent.Event is not null) + { + _metrics?.EventReceived(SessionId, workerEvent.Event.Family.ToString()); + } + + _eventStaging.Writer.TryWrite(workerEvent); + } + + /// + /// Drains staged worker events and applies the bounded-channel backpressure (and + /// sustained-overflow fault) on a dedicated task, so the timed write + /// never runs on the read loop (GWC-04). Mirrors for events. + /// + 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()) + { + } + } + /// /// Enqueues a worker event for client consumption. The channel is /// configured with @@ -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 /// Cancellation token. 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() .ToArray(); diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs index b3d38a5..8300759 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -218,6 +218,51 @@ public sealed class WorkerClientTests Assert.Equal(WorkerClientState.Faulted, client.State); } + /// + /// Verifies that a command reply arriving on the pipe after events is dispatched promptly even + /// when the event channel is full and has no consumer — event enqueue is decoupled from the read + /// loop, so a blocked event writer cannot delay a reply (GWC-04). The event full-mode timeout is + /// set far above the command timeout: without the decoupling the read loop would block behind the + /// full event channel and the in-flight InvokeAsync would hit CommandTimeout. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReadLoop_WhenEventChannelFull_DispatchesCommandReplyWithoutBlocking() + { + await using PipePair pipePair = await PipePair.CreateAsync(); + await using WorkerClient client = CreateClient( + pipePair, + new WorkerClientOptions + { + EventChannelCapacity = 1, + // Much larger than TestTimeout: the read loop must not block for this long behind the + // full event channel while a reply waits. + EventChannelFullModeTimeout = TimeSpan.FromSeconds(30), + HeartbeatGrace = TimeSpan.FromSeconds(60), + HeartbeatCheckInterval = TimeSpan.FromSeconds(60), + }); + await CompleteHandshakeAsync(client, pipePair); + + // No StreamEvents consumer is attached, so the capacity-1 event channel fills and the event + // writer blocks on the second event's timed WriteAsync. + await pipePair.WorkerWriter.WriteAsync( + CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange)); + await pipePair.WorkerWriter.WriteAsync( + CreateEventEnvelope(sequence: 12, MxEventFamily.OnDataChange)); + + Task invokeTask = client.InvokeAsync( + CreateCommand(MxCommandKind.Ping), + TestTimeout, + CancellationToken.None); + WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + await pipePair.WorkerWriter.WriteAsync( + CreateCommandReplyEnvelope(commandEnvelope.CorrelationId, MxCommandKind.Ping)); + + WorkerCommandReply reply = await invokeTask.WaitAsync(TestTimeout); + Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind); + Assert.Equal(WorkerClientState.Ready, client.State); + } + /// /// Verifies that when the client faults it kills the owned worker process. /// The assertion waits on , which