From 13583322b53cfdabc385dc1493929f55508396c4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 17:10:54 -0400 Subject: [PATCH] perf(worker): launcher-configurable event queue capacity --- docs/GatewayConfiguration.md | 1 + docs/MxAccessWorkerInstanceDesign.md | 6 +- .../Configuration/GatewayOptionsValidator.cs | 12 ++++ .../Configuration/WorkerOptions.cs | 12 ++++ .../Workers/WorkerProcessLauncher.cs | 9 +++ .../GatewayOptionsValidatorTests.cs | 55 +++++++++++++++ .../Workers/WorkerProcessLauncherTests.cs | 37 +++++++++- .../MxAccess/MxAccessEventQueueTests.cs | 69 +++++++++++++++++++ .../MxAccess/MxAccessEventQueue.cs | 48 +++++++++++++ .../MxAccess/MxAccessStaSession.cs | 9 ++- 10 files changed, 253 insertions(+), 5 deletions(-) diff --git a/docs/GatewayConfiguration.md b/docs/GatewayConfiguration.md index 4231dfe..7d73c9e 100644 --- a/docs/GatewayConfiguration.md +++ b/docs/GatewayConfiguration.md @@ -115,6 +115,7 @@ launch CWD (SEC-01, SEC-33). | `MxGateway:Worker:StartupProbeRetryDelayMilliseconds` | `250` | Delay between transient startup probe retry attempts. | | `MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds` | `2000` | Per-attempt timeout used by the worker named-pipe connect retry path. The overall pipe connection still stays under the startup budget. | | `MxGateway:Worker:WriteCompletionWaitMilliseconds` | `1500` | Bounded wait the worker holds a unary write reply (`Write`/`Write2`/`WriteSecured`/`WriteSecured2`; bulk writes excluded) for the matching MXAccess `OnWriteComplete` callback, so the reply's `statuses` carry the real commit outcome. `0` disables the wait (pure fire-and-forget replies). Must be `>= 0`. The gateway conveys the value to the worker via the `MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS` environment variable. Consumers that time their own writes must budget above this wait: OtOpcUa's GalaxyDriver wraps gateway writes in a 2 s Tier A resilience timeout, so a deployment raising this option past ~2000 must raise that driver `ResilienceConfig` write timeout in step or slow-but-successful commits surface as consumer-side failures. | +| `MxGateway:Worker:EventQueueCapacity` | `10000` | Capacity, in events, of the worker's outbound MXAccess event queue. Must be between `1000` and `1000000`. This is burst headroom, not a throttle: the queue has no drop policy, so filling it records a `QueueOverflow` worker fault and faults the session. Raise it for sessions whose subscription set can outrun the drain loop (large advise sets, slow event consumers); the backing queue pre-allocates its slots, so the ceiling keeps a mistyped value from committing the 32-bit worker to an outsized allocation. The gateway conveys the value to the worker via the `MXGATEWAY_EVENT_QUEUE_CAPACITY` environment variable; a missing or unusable value leaves the worker on the 10000 default rather than failing the session. | | `MxGateway:Worker:ShutdownTimeoutSeconds` | `10` | Grace period for worker shutdown before the gateway treats shutdown as failed and may kill the worker process tree. | | `MxGateway:Worker:HeartbeatIntervalSeconds` | `5` | Worker heartbeat send interval and gateway heartbeat check cadence input. | | `MxGateway:Worker:HeartbeatGraceSeconds` | `15` | Maximum age of the last worker heartbeat before the gateway faults the worker. This must be greater than or equal to `HeartbeatIntervalSeconds`. | diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index 4372eba..c043110 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -416,7 +416,11 @@ type on buffered events. `OperationComplete` is only emitted from the native `MxAccessEventQueue` is the bounded outbound event queue for one worker session. It assigns the monotonic `WorkerSequence` and `WorkerTimestamp` when an event is accepted, preserving the order in which MXAccess handlers enqueue -events. The default capacity is `10000`. When the queue reaches capacity it +events. The capacity is `10000` by default and comes from +`MxGateway:Worker:EventQueueCapacity`, which the gateway stamps onto the worker +launch environment as `MXGATEWAY_EVENT_QUEUE_CAPACITY`; a missing, unparseable, +or out-of-range value (outside `1000`–`1000000`) leaves the worker on the +default rather than failing the session. When the queue reaches capacity it records a `WorkerFaultCategory.QueueOverflow` fault and rejects further events. The event handler catches conversion and enqueue failures, records the first fault on the queue, and returns to the STA message pump instead of writing to diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs index 29a3aa6..55e31c1 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs @@ -10,6 +10,12 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase MaximumWorkerEventQueueCapacity) + { + builder.Add( + $"MxGateway:Worker:EventQueueCapacity must be between {MinimumWorkerEventQueueCapacity} and {MaximumWorkerEventQueueCapacity}."); + } + if (options.MaxMessageBytes is < MinimumMaxMessageBytes or > MaximumMaxMessageBytes) { builder.Add( diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs index 78b1586..699a0e2 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs @@ -33,6 +33,18 @@ public sealed class WorkerOptions /// public int WriteCompletionWaitMilliseconds { get; init; } = 1500; + /// + /// Capacity of the worker's outbound MXAccess event queue, in events. + /// Default 10,000; must be between 1,000 and 1,000,000. This is + /// headroom, not a throttle: the queue has no drop policy, so a burst + /// that fills it faults the session with a QueueOverflow worker + /// fault. Raise it for sessions whose subscription set can outrun the + /// drain loop (large advise sets, slow event consumers). Conveyed to + /// the worker through the MXGATEWAY_EVENT_QUEUE_CAPACITY + /// environment variable. + /// + public int EventQueueCapacity { get; init; } = 10000; + /// The maximum time in seconds for graceful shutdown. public int ShutdownTimeoutSeconds { get; init; } = 10; diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs index e36fec8..96d07c6 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs @@ -44,6 +44,13 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher public const string WorkerMaxAlarmsPerFetchEnvironmentVariableName = "MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH"; + /// + /// Conveys MxGateway:Worker:EventQueueCapacity to the worker: the capacity + /// of the outbound MXAccess event queue, whose overflow faults the session. + /// + public const string WorkerEventQueueCapacityEnvironmentVariableName = + "MXGATEWAY_EVENT_QUEUE_CAPACITY"; + private readonly IWorkerProcessFactory _processFactory; private readonly IWorkerStartupProbe _startupProbe; private readonly GatewayMetrics _metrics; @@ -202,6 +209,8 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher _workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture); startInfo.Environment[WorkerWriteCompletionWaitEnvironmentVariableName] = _workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture); + startInfo.Environment[WorkerEventQueueCapacityEnvironmentVariableName] = + _workerOptions.EventQueueCapacity.ToString(System.Globalization.CultureInfo.InvariantCulture); startInfo.Environment[WorkerAlarmPollIntervalEnvironmentVariableName] = _alarmsOptions.PollIntervalMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture); startInfo.Environment[WorkerMaxAlarmsPerFetchEnvironmentVariableName] = diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs index 40a62b9..a5cb35a 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs @@ -1099,4 +1099,59 @@ public sealed class GatewayOptionsValidatorTests Assert.True(result.Failed); Assert.Contains(result.Failures!, f => f.Contains("MaxMessageBytes") && f.Contains("reserve")); } + + /// + /// Verifies the shipped worker event-queue capacity default passes validation and still matches + /// the worker-side MxAccessEventQueue.DefaultCapacity the environment variable falls back + /// to when the launcher value is unusable. + /// + [Fact] + public void Validate_Succeeds_WithDefaultEventQueueCapacity() + { + Assert.Equal(10000, new WorkerOptions().EventQueueCapacity); + + ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, ValidOptions()); + Assert.True(result.Succeeded); + } + + /// + /// Verifies a worker event-queue capacity outside the supported range fails validation. Too + /// small leaves no burst headroom (an overflow faults the whole session); too large commits the + /// 32-bit worker to an outsized pre-allocation. + /// + /// Capacity under test. + [Theory] + [InlineData(0)] + [InlineData(-1)] + [InlineData(999)] + [InlineData(1_000_001)] + public void Validate_Fails_WhenEventQueueCapacityOutOfRange(int eventQueueCapacity) + { + ValidateOptionsResult result = new GatewayOptionsValidator().Validate( + null, + WithWorkerAndProtocol( + new WorkerOptions { EventQueueCapacity = eventQueueCapacity }, + new ProtocolOptions())); + + Assert.True(result.Failed); + Assert.Contains( + result.Failures!, + f => f.Contains("MxGateway:Worker:EventQueueCapacity", StringComparison.Ordinal)); + } + + /// Verifies the worker event-queue capacity bounds themselves are accepted. + /// Capacity under test. + [Theory] + [InlineData(1000)] + [InlineData(1_000_000)] + public void Validate_Succeeds_AtEventQueueCapacityBounds(int eventQueueCapacity) + { + ValidateOptionsResult result = new GatewayOptionsValidator().Validate( + null, + WithWorkerAndProtocol( + new WorkerOptions { EventQueueCapacity = eventQueueCapacity }, + new ProtocolOptions())); + + Assert.True(result.Succeeded); + } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs index 6acef16..fb97c47 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs @@ -47,6 +47,13 @@ public sealed class WorkerProcessLauncherTests "1500", processFactory.LastStartInfo.Environment[ WorkerProcessLauncher.WorkerWriteCompletionWaitEnvironmentVariableName]); + // The worker sizes its outbound event queue from the launch environment; + // the queue has no drop policy, so this capacity is the session's burst + // headroom rather than a throttle. + Assert.Equal( + "10000", + processFactory.LastStartInfo.Environment[ + WorkerProcessLauncher.WorkerEventQueueCapacityEnvironmentVariableName]); // MxGateway:Alarms defaults reach the worker's alarm poll loop and its // GetXmlCurrentAlarms2 cap (which is also its truncation threshold) // through the launch environment, not the command line. @@ -64,6 +71,32 @@ public sealed class WorkerProcessLauncherTests Assert.Equal(0, metrics.GetSnapshot().WorkersRunning); } + /// + /// Verifies that a configured — not the shipped + /// default — is what reaches the worker, so the option is deployable without a worker rebuild. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task LaunchAsync_WithConfiguredEventQueueCapacity_ExportsItToTheWorkerEnvironment() + { + using TestDirectory directory = TestDirectory.Create(); + string executablePath = directory.CreateWorkerExecutable(machine: 0x014c); + FakeWorkerProcessFactory processFactory = new(new FakeWorkerProcess(processId: 1234)); + WorkerProcessLauncher launcher = CreateLauncher( + executablePath, + processFactory, + new SucceedingStartupProbe(), + eventQueueCapacity: 65536); + + using WorkerProcessHandle handle = await launcher.LaunchAsync(CreateRequest()); + + Assert.NotNull(processFactory.LastStartInfo); + Assert.Equal( + "65536", + processFactory.LastStartInfo.Environment[ + WorkerProcessLauncher.WorkerEventQueueCapacityEnvironmentVariableName]); + } + /// Verifies that a failed startup probe kills and disposes the worker process. /// A task that represents the asynchronous operation. [Fact] @@ -216,7 +249,8 @@ public sealed class WorkerProcessLauncherTests GatewayMetrics? metrics = null, int startupTimeoutSeconds = 30, int startupProbeRetryAttempts = 3, - int startupProbeRetryDelayMilliseconds = 250) + int startupProbeRetryDelayMilliseconds = 250, + int eventQueueCapacity = 10000) { GatewayOptions options = new() { @@ -227,6 +261,7 @@ public sealed class WorkerProcessLauncherTests StartupTimeoutSeconds = startupTimeoutSeconds, StartupProbeRetryAttempts = startupProbeRetryAttempts, StartupProbeRetryDelayMilliseconds = startupProbeRetryDelayMilliseconds, + EventQueueCapacity = eventQueueCapacity, }, }; diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs index e783fde..82eba17 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs @@ -419,6 +419,75 @@ public sealed class MxAccessEventQueueTests await Assert.ThrowsAnyAsync(async () => await wait); } + /// + /// The queue capacity comes from the launcher-set environment variable; + /// a missing, unparseable, below-floor, or above-ceiling value must fall + /// back to the 10,000 default rather than throw. A bad environment value + /// must never stop the worker's session from starting. + /// + /// Raw environment value under test. + /// Expected resolved capacity. + [Theory] + [InlineData(null, MxAccessEventQueue.DefaultCapacity)] + [InlineData("", MxAccessEventQueue.DefaultCapacity)] + [InlineData("not-a-number", MxAccessEventQueue.DefaultCapacity)] + [InlineData("0", MxAccessEventQueue.DefaultCapacity)] + [InlineData("-5", MxAccessEventQueue.DefaultCapacity)] + [InlineData("999", MxAccessEventQueue.DefaultCapacity)] + [InlineData("1000001", MxAccessEventQueue.DefaultCapacity)] + [InlineData("1000", 1000)] + [InlineData("1000000", 1000000)] + [InlineData("50000", 50000)] + public void ResolveCapacity_WithEnvironmentValue_FallsBackToDefaultWhenUnusable( + string? environmentValue, + int expected) + { + string? original = Environment.GetEnvironmentVariable( + MxAccessEventQueue.CapacityEnvironmentVariableName); + try + { + Environment.SetEnvironmentVariable( + MxAccessEventQueue.CapacityEnvironmentVariableName, + environmentValue); + + Assert.Equal(expected, MxAccessEventQueue.ResolveCapacity()); + } + finally + { + Environment.SetEnvironmentVariable( + MxAccessEventQueue.CapacityEnvironmentVariableName, + original); + } + } + + /// + /// A resolved capacity is what the queue is actually built with, so the + /// configured headroom reaches the overflow check rather than only the + /// resolver. + /// + [Fact] + public void Constructor_WithResolvedCapacity_UsesEnvironmentValue() + { + string? original = Environment.GetEnvironmentVariable( + MxAccessEventQueue.CapacityEnvironmentVariableName); + try + { + Environment.SetEnvironmentVariable( + MxAccessEventQueue.CapacityEnvironmentVariableName, + "2500"); + + MxAccessEventQueue queue = new(MxAccessEventQueue.ResolveCapacity()); + + Assert.Equal(2500, queue.Capacity); + } + finally + { + Environment.SetEnvironmentVariable( + MxAccessEventQueue.CapacityEnvironmentVariableName, + original); + } + } + // Mirrors MxAccessEventQueue's per-event repeated-field allowance. Kept local rather than made // public on the queue: the byte-budget tests state their budgets in units of that charge, so a // change to it should surface here as a failing bound instead of silently moving with the code. diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs index 17f1d67..c22288d 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Globalization; using System.Threading; using System.Threading.Tasks; using Google.Protobuf.WellKnownTypes; @@ -28,6 +29,31 @@ public sealed class MxAccessEventQueue /// public const int DefaultCapacity = 10000; + /// + /// Environment variable the gateway's WorkerProcessLauncher sets + /// from MxGateway:Worker:EventQueueCapacity. A missing, + /// unparseable, or out-of-range value falls back to + /// — a bad environment value must never + /// keep the session from starting. + /// + internal const string CapacityEnvironmentVariableName = "MXGATEWAY_EVENT_QUEUE_CAPACITY"; + + /// + /// Floor on the resolved capacity. Mirrors the gateway-side + /// MxGateway:Worker:EventQueueCapacity minimum so a value that + /// slipped past startup validation still leaves the headroom the queue + /// needs: it has no drop policy, so an overflow faults the session. + /// + internal const int MinimumCapacity = 1000; + + /// + /// Ceiling on the resolved capacity, mirroring the gateway-side maximum. + /// The backing queue pre-allocates its slots, so an absurd environment + /// value would otherwise cost the 32-bit worker that allocation at + /// startup. + /// + internal const int MaximumCapacity = 1_000_000; + // Extra per-event slack added to WorkerEvent.CalculateSize() when charging the byte budget in // Drain(maxEvents, maxTotalBytes). CalculateSize() already accounts for the event's own tag and // length-delimiter (the WorkerEvent wrapper serializes MxEvent as field 1, and the reply packs @@ -80,6 +106,28 @@ public sealed class MxAccessEventQueue events = new Queue(capacity); } + /// + /// Resolves the queue capacity from the launcher-provided environment + /// variable. A missing, unparseable, or out-of-range value falls back to + /// rather than throwing: running on the + /// default capacity is always safe, and a worker that refuses to start + /// over a mistyped environment variable is not. + /// + /// The capacity to construct the outbound event queue with. + internal static int ResolveCapacity() + { + string? value = Environment.GetEnvironmentVariable(CapacityEnvironmentVariableName); + return int.TryParse( + value, + NumberStyles.Integer, + CultureInfo.InvariantCulture, + out int capacity) + && capacity >= MinimumCapacity + && capacity <= MaximumCapacity + ? capacity + : DefaultCapacity; + } + /// /// The queue's maximum capacity. /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs index 20741b6..6c2a480 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs @@ -63,12 +63,15 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession /// /// Initializes a new instance of with default dependencies. + /// The outbound event queue is sized from the launcher-provided + /// MXGATEWAY_EVENT_QUEUE_CAPACITY (see ); + /// callers that pass their own queue keep full control of its capacity. /// public MxAccessStaSession() : this( new StaRuntime(), new MxAccessComObjectFactory(), - new MxAccessEventQueue()) + new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity())) { } @@ -84,7 +87,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession : this( new StaRuntime(), new MxAccessComObjectFactory(), - new MxAccessEventQueue(), + new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity()), alarmCommandHandlerFactory) { } @@ -99,7 +102,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession StaRuntime staRuntime, IMxAccessComObjectFactory factory, IMxAccessEventSink eventSink) - : this(staRuntime, factory, eventSink, new MxAccessEventQueue()) + : this(staRuntime, factory, eventSink, new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity())) { }