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
This commit is contained in:
@@ -32,10 +32,17 @@ public sealed class GatewayMetrics : IDisposable
|
||||
private readonly ConcurrentDictionary<string, long> _eventsBySession = new(StringComparer.Ordinal);
|
||||
private readonly Dictionary<string, long> _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<long, Func<int>> _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
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <see cref="GetSnapshot"/>),
|
||||
/// rather than tracking a pushed running total, so the streaming hot path does no
|
||||
/// per-event gauge bookkeeping (GWC-15).
|
||||
/// </summary>
|
||||
/// <param name="delta">Amount to adjust the queue depth by.</param>
|
||||
public void AdjustGrpcEventStreamQueueDepth(int delta)
|
||||
/// <param name="backlog">
|
||||
/// Returns this subscriber's current channel backlog. Invoked only at collection time;
|
||||
/// must be cheap and non-blocking (a bounded channel's <c>Count</c>). Negative returns
|
||||
/// are clamped to zero when summed.
|
||||
/// </param>
|
||||
/// <returns>A handle whose disposal unregisters the source. Safe to dispose more than once.</returns>
|
||||
public IDisposable RegisterEventStreamBacklogSource(Func<int> 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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -429,13 +445,16 @@ public sealed class GatewayMetrics : IDisposable
|
||||
/// <returns>The current metrics snapshot.</returns>
|
||||
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<int> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user