From 5e2e40a9276e944b2c7b4d7fc1430fbf85171228 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 15:12:43 -0400 Subject: [PATCH] perf(gateway): trim event/command hot-path allocations (GWC-06/07/15, IPC-05) Behavior-preserving allocation cuts on the per-event/per-command path: - GWC-06: StreamEvents send-timing uses Stopwatch.GetTimestamp() + GetElapsedTime() instead of a per-event Stopwatch allocation (same measured span). - GWC-07/IPC-05 (event): MapEvent transfers ownership of the inner MxEvent instead of .Clone()-ing it. Safe: WorkerEvent is parsed fresh per pipe frame with the distributor pump as its single consumer (GWC-01), MapEvent runs once before fan-out, and every downstream consumer (subscribers, replay ring) only READS the event (WorkerSequence is stamped upstream; verified no post-mapping mutation). Comment documents the invariant + restore-clone caveat if a second consumer is added. - IPC-05 (command): CreateCommandEnvelope no longer re-clones; MapCommand already isolated the graph from the caller-owned gRPC message. - GWC-15: grpc_stream_queue.depth converts from a per-event push counter to an ObservableGauge summing registered channel sources at scrape time only (name/semantics unchanged); removes all per-event .Count/lock work. Kept every load-bearing isolation clone (MapCommand, Invoke, bulk filters, MapCommandReply). Server build clean (0 warnings); EventStream/Metrics/ Distributor/Mapper tests 62/62 (incl. formerly-flaky queue-depth tests, now green under the lazy gauge, + 2 new MapEvent ownership tests). Docs: Metrics.md, Grpc.md. Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW --- docs/Grpc.md | 4 ++ docs/Metrics.md | 21 +++--- .../Grpc/EventStreamService.cs | 41 +++++------ .../Grpc/MxAccessGatewayService.cs | 10 ++- .../Grpc/MxAccessGrpcMapper.cs | 11 ++- .../Metrics/GatewayMetrics.cs | 72 ++++++++++++++++--- .../Workers/WorkerClient.cs | 7 +- .../Gateway/Grpc/MxAccessGrpcMapperTests.cs | 31 ++++++++ .../Metrics/GatewayMetricsTests.cs | 4 +- 9 files changed, 155 insertions(+), 46 deletions(-) diff --git a/docs/Grpc.md b/docs/Grpc.md index 9c71aee..fc3b851 100644 --- a/docs/Grpc.md +++ b/docs/Grpc.md @@ -215,6 +215,10 @@ public WorkerCommand MapCommand(MxCommandRequest request) } ``` +The command clone above is the only isolating copy on the command path. `WorkerClient.CreateCommandEnvelope` no longer re-clones the `WorkerCommand` into the outbound envelope: the mapper's clone already produced a fresh graph owned by the invoke pipeline, and the envelope is built and owned entirely inside `WorkerClient`, so a second copy would be redundant. + +`MapEvent` takes the opposite stance from `MapCommand`: it transfers ownership of the inner `MxEvent` rather than cloning it. The enclosing `WorkerEvent` is parsed fresh from a single pipe frame and the `SessionEventDistributor` pump is its single consumer, so nothing else aliases or mutates it. The pump then shares that one `MxEvent` across every subscriber and the replay ring, but that fan-out is read-only, so a single instance is safe. If a second consumer of the `WorkerEvent` is ever introduced, restore the clone in `MapEvent`. + When the worker reply or event payload is missing, the mapper returns a synthetic public message with `ProtocolStatusCode.ProtocolViolation` (for replies) or a sentinel `MxEvent` with `MxEventFamily.Unspecified` (for events). The gateway never relays a partial frame to clients — anything missing is reported as a protocol violation against the worker, not a transport error against the client. The mapper also exposes static factory methods for every `ProtocolStatusCode` (`Ok`, `InvalidRequest`, `SessionNotFound`, `SessionNotReady`, `WorkerUnavailable`, `Timeout`, `Canceled`, `ProtocolViolation`) so that handlers and tests can produce status payloads without duplicating the enum-to-string mapping. diff --git a/docs/Metrics.md b/docs/Metrics.md index 6b40611..1ffea39 100644 --- a/docs/Metrics.md +++ b/docs/Metrics.md @@ -73,7 +73,7 @@ Observable gauges are pull-based; the `Meter` invokes the supplied callback when | `mxgateway.sessions.open` | `_openSessions` | Currently open sessions tracked by `SessionManager`. | | `mxgateway.workers.running` | `_workersRunning` | Worker clients in a running state. | | `mxgateway.events.worker_queue.depth` | `_workerEventQueueDepth` | Last reported depth of the worker-side event queue. | -| `mxgateway.events.grpc_stream_queue.depth` | `_grpcEventStreamQueueDepth` | Backlog held by `EventStreamService` for the active gRPC stream consumer. | +| `mxgateway.events.grpc_stream_queue.depth` | `_eventStreamBacklogSources` (summed on demand) | Live backlog buffered across every active `EventStreamService` subscriber, summed from the subscribers' channel `Count` at collection time. | ## Snapshot Shape @@ -173,14 +173,13 @@ This is the only fault path where the factory itself owns the kill decision; onc ### gRPC event stream service -`Grpc/EventStreamService.cs` records the dashboard/client event-stream backlog and disconnect counters: +`Grpc/EventStreamService.cs` reports the event-stream backlog through a live backlog source and records the disconnect/overflow counters: ```csharp -metrics.AdjustGrpcEventStreamQueueDepth(1); +IDisposable backlogRegistration = metrics.RegisterEventStreamBacklogSource( + () => subscriber.Reader.CanCount ? subscriber.Reader.Count : 0); ... -metrics.AdjustGrpcEventStreamQueueDepth(-1); -... -metrics.AdjustGrpcEventStreamQueueDepth(-remainingDepth); +backlogRegistration.Dispose(); metrics.StreamDisconnected("Detached"); ... metrics.QueueOverflow("grpc-event-stream"); @@ -189,14 +188,16 @@ metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString()); metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString()); ``` -The service tracks per-message enqueues and dequeues, so `AdjustGrpcEventStreamQueueDepth` updates the aggregate stream backlog. The `Math.Max(0, ...)` clamp inside the adjuster prevents a negative depth if the bookkeeping ever drifts. +Rather than pushing a running total on every streamed event (which took the subscriber channel's `Count` lock and the metric lock per event), each subscriber registers its channel as a backlog source for the duration of the stream. The gauge callback and `GetSnapshot` sum those sources' `Count` on demand — only when the metric is scraped or projected — so the streaming hot path does no per-event gauge bookkeeping (GWC-15). Disposing the registration in the `finally` removes the subscriber's contribution, so the gauge reflects only the remaining subscribers and returns to zero once the last stream detaches. Negative source returns are clamped to zero when summed. -`Grpc/MxAccessGatewayService.cs` records gRPC event send latency around each response-stream write: +`Grpc/MxAccessGatewayService.cs` records gRPC event send latency around each response-stream write using the allocation-free timestamp API (no per-event `Stopwatch` object): ```csharp -Stopwatch stopwatch = Stopwatch.StartNew(); +long sendStartTimestamp = Stopwatch.GetTimestamp(); await responseStream.WriteAsync(publicEvent).ConfigureAwait(false); -metrics.RecordEventStreamSend(publicEvent.Family.ToString(), stopwatch.Elapsed); +metrics.RecordEventStreamSend( + publicEvent.Family.ToString(), + Stopwatch.GetElapsedTime(sendStartTimestamp)); ``` ## Dashboard Consumption diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs index 5afce48..ae0c3fe 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs @@ -116,11 +116,23 @@ public sealed class EventStreamService( options.Value.Sessions.MaxEventSubscribersPerSession); } - int streamQueueDepth = 0; IAsyncEnumerator reader = subscriber.Reader .ReadAllAsync(cancellationToken) .GetAsyncEnumerator(cancellationToken); + // GWC-15: register this subscriber's channel as a live backlog source instead of + // reconciling the queue-depth gauge on every event. The gauge previously read the + // bounded channel's Count (which takes the channel's internal lock) and adjusted the + // metric under its own lock on every streamed event. Now the metric reads Count only + // when it is scraped (ObservableGauge callback) or projected (GetSnapshot), summing the + // live backlog across every registered subscriber — the same "buffered, not yet + // delivered" aggregate the per-event push reported, but with no per-event lock traffic. + // Disposing the registration in the finally removes this subscriber's contribution, so + // the gauge returns to the other subscribers' backlog (zero when none remain) on + // disconnect. CanCount guards a channel that ever cannot report Count (contributes 0). + IDisposable backlogRegistration = metrics.RegisterEventStreamBacklogSource( + () => subscriber.Reader.CanCount ? subscriber.Reader.Count : 0); + try { // Emit order for a resume: the ReplayGap sentinel FIRST (only when events were @@ -179,32 +191,21 @@ public sealed class EventStreamService( continue; } - // Queue-depth gauge tracks events the pump has fanned into this subscriber's - // channel but the client has not yet consumed — the same "buffered, not yet - // delivered" quantity the original per-RPC channel reported. The bounded - // subscriber channel supports counting, so reconcile the gauge to the current - // backlog; falling back to a no-op delta if a channel ever cannot count. - int backlog = subscriber.Reader.CanCount ? subscriber.Reader.Count : streamQueueDepth; - int delta = backlog - streamQueueDepth; - if (delta != 0) - { - streamQueueDepth = backlog; - metrics.AdjustGrpcEventStreamQueueDepth(delta); - } - + // The queue-depth gauge is maintained lazily via the backlog registration above + // (GWC-15): the metric reads this subscriber's channel Count only when scraped, + // so there is no per-event gauge bookkeeping on this hot path. yield return mxEvent; } } finally { await reader.DisposeAsync().ConfigureAwait(false); - subscriber.Dispose(); - if (streamQueueDepth != 0) - { - metrics.AdjustGrpcEventStreamQueueDepth(-streamQueueDepth); - streamQueueDepth = 0; - } + // Remove this subscriber's live backlog contribution before disposing the lease so + // the gauge stops counting a channel that is about to be completed; after this the + // gauge reflects only the remaining subscribers (zero when none remain). + backlogRegistration.Dispose(); + subscriber.Dispose(); metrics.StreamDisconnected("Detached"); } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs index 22675f1..7e756fa 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs @@ -152,9 +152,15 @@ public sealed class MxAccessGatewayService( .WithCancellation(context.CancellationToken) .ConfigureAwait(false)) { - Stopwatch stopwatch = Stopwatch.StartNew(); + // GWC-06: measure send latency with the allocation-free timestamp API rather + // than allocating a Stopwatch object per event per subscriber on the highest- + // volume gateway path. Stopwatch.GetTimestamp/GetElapsedTime measure the exact + // same wall-clock span the former Stopwatch.StartNew()/.Elapsed did. + long sendStartTimestamp = Stopwatch.GetTimestamp(); await responseStream.WriteAsync(publicEvent).ConfigureAwait(false); - metrics.RecordEventStreamSend(publicEvent.Family.ToString(), stopwatch.Elapsed); + metrics.RecordEventStreamSend( + publicEvent.Family.ToString(), + Stopwatch.GetElapsedTime(sendStartTimestamp)); } } catch (Exception exception) when (exception is not RpcException) diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcMapper.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcMapper.cs index 4333b8c..ea6d0f7 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcMapper.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcMapper.cs @@ -65,7 +65,16 @@ public sealed class MxAccessGrpcMapper { ArgumentNullException.ThrowIfNull(workerEvent); - return workerEvent.Event?.Clone() ?? new MxEvent + // GWC-07 / IPC-05: ownership transfer, not a deep clone. The enclosing WorkerEvent is + // parsed fresh from a single pipe frame in WorkerClient's read loop and is discarded + // immediately after this mapping — the SessionEventDistributor pump is its single + // consumer (GWC-01 claims the worker event channel as single-reader), so nothing else + // aliases or mutates workerEvent.Event. We therefore move the inner MxEvent into the + // outbound graph instead of cloning it. Downstream the pump fans this one MxEvent to + // every subscriber and retains it in the replay ring, but that sharing is READ-ONLY + // (subscribers only yield/filter it), so a single shared instance is safe. If a second + // consumer of WorkerEvent is ever added, restore a .Clone() here to re-isolate. + return workerEvent.Event ?? new MxEvent { Family = MxEventFamily.Unspecified, RawStatus = "Worker event did not contain a public event payload.", diff --git a/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs b/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs index c42468f..859b73d 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs @@ -32,10 +32,17 @@ public sealed class GatewayMetrics : IDisposable private readonly ConcurrentDictionary _eventsBySession = new(StringComparer.Ordinal); private readonly Dictionary _retryAttemptsByArea = new(StringComparer.OrdinalIgnoreCase); + // GWC-15: live backlog sources for the gRPC event-stream queue-depth gauge. Each active + // StreamEvents subscriber registers a delegate returning its channel's current backlog; the + // gauge sums them ONLY when scraped or projected (GetGrpcEventStreamQueueDepth), so nothing is + // read on the per-event streaming hot path. Concurrent because subscribers register/unregister + // from arbitrary gRPC threads while the gauge callback / GetSnapshot enumerate. + private readonly ConcurrentDictionary> _eventStreamBacklogSources = new(); + private long _nextEventStreamBacklogSourceId; + private int _openSessions; private int _workersRunning; private int _workerEventQueueDepth; - private int _grpcEventStreamQueueDepth; private int _alarmProviderMode; private long _sessionsOpened; private long _sessionsClosed; @@ -292,15 +299,24 @@ public sealed class GatewayMetrics : IDisposable } /// - /// Adjusts the gRPC event stream queue depth by the given delta. + /// Registers a live backlog source for the gRPC event-stream queue-depth gauge and + /// returns a handle that removes it when disposed. The gauge sums every registered + /// source's current value on demand (when scraped or via ), + /// rather than tracking a pushed running total, so the streaming hot path does no + /// per-event gauge bookkeeping (GWC-15). /// - /// Amount to adjust the queue depth by. - public void AdjustGrpcEventStreamQueueDepth(int delta) + /// + /// Returns this subscriber's current channel backlog. Invoked only at collection time; + /// must be cheap and non-blocking (a bounded channel's Count). Negative returns + /// are clamped to zero when summed. + /// + /// A handle whose disposal unregisters the source. Safe to dispose more than once. + public IDisposable RegisterEventStreamBacklogSource(Func backlog) { - lock (_syncRoot) - { - _grpcEventStreamQueueDepth = Math.Max(0, _grpcEventStreamQueueDepth + delta); - } + ArgumentNullException.ThrowIfNull(backlog); + long id = Interlocked.Increment(ref _nextEventStreamBacklogSourceId); + _eventStreamBacklogSources[id] = backlog; + return new EventStreamBacklogRegistration(this, id); } /// @@ -429,13 +445,16 @@ public sealed class GatewayMetrics : IDisposable /// The current metrics snapshot. public GatewayMetricsSnapshot GetSnapshot() { + // Compute the live gRPC stream backlog outside _syncRoot: the sources are the subscriber + // channels' Count (their own locks) and must not run under this lock. GWC-15. + int grpcEventStreamQueueDepth = GetGrpcEventStreamQueueDepth(); lock (_syncRoot) { return new GatewayMetricsSnapshot( OpenSessions: _openSessions, WorkersRunning: _workersRunning, WorkerEventQueueDepth: _workerEventQueueDepth, - GrpcEventStreamQueueDepth: _grpcEventStreamQueueDepth, + GrpcEventStreamQueueDepth: grpcEventStreamQueueDepth, SessionsOpened: _sessionsOpened, SessionsClosed: _sessionsClosed, CommandsStarted: _commandsStarted, @@ -495,12 +514,28 @@ public sealed class GatewayMetrics : IDisposable } } + // Sums the live backlog across every registered event-stream subscriber. Runs at collection + // time (ObservableGauge scrape) or when GetSnapshot projects the value — never on the + // per-event path. Enumerating ConcurrentDictionary.Values never throws on concurrent + // register/unregister; a source removed mid-enumeration simply drops from this sample. private int GetGrpcEventStreamQueueDepth() { - lock (_syncRoot) + int total = 0; + foreach (Func source in _eventStreamBacklogSources.Values) { - return _grpcEventStreamQueueDepth; + int value = source(); + if (value > 0) + { + total += value; + } } + + return total; + } + + private void UnregisterEventStreamBacklogSource(long id) + { + _eventStreamBacklogSources.TryRemove(id, out _); } private int GetAlarmProviderMode() @@ -521,4 +556,19 @@ public sealed class GatewayMetrics : IDisposable { values.AddOrUpdate(key, 1, static (_, currentValue) => currentValue + 1); } + + // Handle returned by RegisterEventStreamBacklogSource. Disposal (once) removes the source + // from the gauge's live sum. Idempotent so a double dispose from a stream teardown is safe. + private sealed class EventStreamBacklogRegistration(GatewayMetrics metrics, long id) : IDisposable + { + private int _disposed; + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) == 0) + { + metrics.UnregisterEventStreamBacklogSource(id); + } + } + } } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index 342ae6d..0a61536 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -997,9 +997,14 @@ public sealed class WorkerClient : IWorkerClient string correlationId, WorkerCommand command) { + // IPC-05: no second clone. MxAccessGrpcMapper.MapCommand already deep-cloned the command + // out of the caller-owned gRPC MxCommandRequest, so this WorkerCommand is a fresh graph + // owned by the invoke pipeline and referenced by no other consumer. The envelope is built + // and owned entirely inside WorkerClient and the command is not touched again after this + // point, so transferring it into the envelope introduces no aliasing hazard. return CreateEnvelope( correlationId, - envelope => envelope.WorkerCommand = command.Clone()); + envelope => envelope.WorkerCommand = command); } /// Creates shutdown envelope. diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGrpcMapperTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGrpcMapperTests.cs index 92ae3e9..b7249ba 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGrpcMapperTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGrpcMapperTests.cs @@ -76,4 +76,35 @@ public sealed class MxAccessGrpcMapperTests Assert.Equal(ProtocolStatusCode.ProtocolViolation, publicReply.ProtocolStatus.Code); } + + /// + /// Verifies MapEvent transfers ownership of the inner MxEvent (GWC-07 / IPC-05): the + /// returned reference is the same instance carried by the WorkerEvent, not a clone. The + /// WorkerEvent is discarded after mapping and the distributor pump is its single consumer, + /// so moving the inner event out is safe and avoids a per-event deep copy. + /// + [Fact] + public void MapEvent_TransfersOwnershipOfInnerEventWithoutCloning() + { + MxEvent innerEvent = new() + { + Family = MxEventFamily.OnDataChange, + RawStatus = "OK", + }; + WorkerEvent workerEvent = new() { Event = innerEvent }; + + MxEvent mapped = new MxAccessGrpcMapper().MapEvent(workerEvent); + + Assert.Same(innerEvent, mapped); + } + + /// Verifies that a worker event with no public payload maps to the unspecified-family sentinel. + [Fact] + public void MapEvent_WhenEventMissing_ReturnsUnspecifiedFamilySentinel() + { + MxEvent mapped = new MxAccessGrpcMapper().MapEvent(new WorkerEvent()); + + Assert.Equal(MxEventFamily.Unspecified, mapped.Family); + Assert.Equal("Worker event did not contain a public event payload.", mapped.RawStatus); + } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs index 2dec8bd..2985e4f 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs @@ -20,7 +20,9 @@ public sealed class GatewayMetricsTests metrics.EventReceived("session-1", "OnDataChange"); metrics.EventReceived("session-1", "OnDataChange"); metrics.SetWorkerEventQueueDepth(7); - metrics.AdjustGrpcEventStreamQueueDepth(3); + // GWC-15: the gRPC stream queue-depth gauge sums live backlog sources at collection time + // rather than tracking a pushed running total. Register a source reporting 3. + using IDisposable backlogSource = metrics.RegisterEventStreamBacklogSource(static () => 3); metrics.QueueOverflow("session-events"); metrics.Fault("CommandTimeout"); metrics.WorkerKilled("CommandTimeout");