From 8b1cc6b7e3921e05e263ccfd2549480234eda314 Mon Sep 17 00:00:00 2001 From: romanett Date: Wed, 28 Oct 2026 12:00:01 +0100 Subject: [PATCH] Server: run SharedOnly requests on concurrency permits instead of queue workers (#4919) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Q7 of the request-queue comparison with ASP.NET Core/Kestrel. Stacked on #4917 (Q4 + Q5): review only the last two commits. ## Q7: requests run on concurrency permits instead of queue workers Without an isolation provider (the SharedOnly profile), the request queue used to grow workers one at a time under `m_workerGate`. Workers were never retired before stop. Requests went through a bounded `Channel`, and the "thread" settings were really limits on concurrent requests. Now: - `ScheduleIncomingRequest` takes one of `MaxRequestThreadCount` permits (a counter under the existing gate) and queues the handler on the ThreadPool directly (`IThreadPoolWorkItem` on .NET, `WaitCallback` on net48). A `Queue` holds only the requests that are waiting for a permit. Waiting requests go first, so ordering stays FIFO. - A request that finishes, or parks with `DecoupleHeldPublishRequests`, hands its permit to the oldest waiting request and runs it on the same thread, as a worker loop did. Otherwise it returns the permit. - External behaviour is unchanged: same configuration, `BadServerTooBusy` once `MaxQueuedRequestCount` requests wait, parking releases the permit. `StopCoreAsync` rejects waiting requests with `BadServerHalted` and waits for executing and parked handlers. It also waits for permits whose request has not started yet, so `m_cts` is disposed only after every permit is back. - `FairRequestQueue` and its worker loop are unchanged; `WorkerLoopAsync` now serves only the fair queue. - `InitializeRequestQueue` no longer raises `MaxRequestThreadCount` values below 100 to 100. The value is still raised to `MinRequestThreadCount`, which is at least 1. - The doc comments of `MinRequestThreadCount` and `MaxRequestThreadCount` and `docs/ServerScalability.md` now describe the settings as concurrency limits. I considered `System.Threading.RateLimiting.ConcurrencyLimiter` and did not use it: every `AttemptAcquire` and queued `AcquireAsync` allocates a lease. A plain counter under the gate the queue already takes per request costs nothing extra. ### Release note (behaviour change) > `MaxRequestThreadCount` values below 100 are no longer raised to 100: the configured value is the limit on requests executing at once. With the SharedOnly isolation profile, requests no longer run on retained queue workers; `MinRequestThreadCount` only starts workers for the other isolation profiles. ## Measurements Paired A/B with `perf/ab.sh`: baseline = #4917 head, candidate = this branch. 20 sessions, SharedOnly. The second commit adds `--isolation sharedonly` to the load harness, because `StandardServer`'s default Balanced profile uses the fair queue and never reaches this path. Quiet run (7 paired rounds, medians, server side): | Scenario | req/s | p99 | server B/req | server CPU µs/req | lock contention | |---|---|---|---|---|---| | read 1 value, 10 in flight | 358k → 407k (+14 %, 7/7 rounds) | 6.2 → 2.8 ms | 2253 → 2241 | 23.2 → 24.1 | −94 % | | read 100 values | 47.0k → 47.4k (+1 %, 7/7 rounds) | 1.98 → 1.96 ms | 52550 → 52480 | 229 → 228 | −99 % | | sub 5×100, pub 100 ms | flat | 9.4 → 9.2 ms | 1821 → 1818 | 24.4 → 25.3 | +12 % | A first run (5 rounds, 6 scenarios) overlapped other sessions' builds, so its throughput varied 4x within each side. Its deterministic numbers agree. Server bytes per request: read 1 value 2336 → 2272, write 100 values 71.7k → 71.8k, browse 32.9k → 32.8k, sub 1825 → 1827. Lock contention was −96 % on every read scenario. ## Tests - New `RequestQueueTests`: - the shared queue starts no workers - permits limit concurrency, and waiters are served in FIFO order (with the executing/queued gauges) - a configured limit below 100 is honoured - stop rejects waiting requests and drains executing and parked handlers - a concurrent stop either halts or runs every request - `Opc.Ua.Core.Tests` `Stack.Server`: 1951 passed (net10.0). net48 build compiles. - `Opc.Ua.Stress.Tests` `Category=Contract|Integration` (not in PR CI): 45/45 passed. - To run the permit path under real servers, I temporarily forced the default isolation mode to SharedOnly (local patch, not committed): - Stress Contract/Integration: 45/45 passed. - `Opc.Ua.Sessions.Tests`: 865 passed, 0 failed, 379 skipped. This includes the load tests. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- docs/ResourceIsolation.md | 4 +- docs/ServerScalability.md | 8 +- perf/README.md | 1 + .../NativeWorkerBenchmark.cs | 4 +- perf/ServerLoadHarness/Program.cs | 6 +- .../Schema/ApplicationConfiguration.cs | 22 +- .../Stack/Server/ServerBase.RequestQueue.cs | 309 ++++++++++++++---- src/Opc.Ua.Core/Stack/Server/ServerBase.cs | 5 - .../ServerResourceIsolationOptions.cs | 11 +- .../Stack/Server/RequestQueueTests.cs | 189 +++++++++++ .../IsolationConfigurationValidationTests.cs | 36 +- 11 files changed, 506 insertions(+), 89 deletions(-) diff --git a/docs/ResourceIsolation.md b/docs/ResourceIsolation.md index 03b5538302..7339b87046 100644 --- a/docs/ResourceIsolation.md +++ b/docs/ResourceIsolation.md @@ -593,7 +593,9 @@ bootstrap, reconnect, control and trusted floors, so the shared capacity equals the configured value. For example, with `MaxQueuedRequestCount = 10000`, ten clients can each keep 1,000 requests outstanding. Balanced then sizes the RequestQueue stage at 10,003 entries. RequestExecution keeps its floors inside -the worker limit, so an idle worker remains available for protected requests. +the worker limit, so an idle worker remains available for protected requests: +its total is `MaxRequestThreadCount`, and startup fails when that leaves no +worker for ordinary requests (Balanced needs at least 4). An explicit `Stages[].Capacity` is still an absolute total that includes the floors, and it cannot exceed the default total. diff --git a/docs/ServerScalability.md b/docs/ServerScalability.md index 7c04c3c7a5..686895b3df 100644 --- a/docs/ServerScalability.md +++ b/docs/ServerScalability.md @@ -219,13 +219,13 @@ The following settings size a single node for its hardware (see also the sizing ### Request workers and the request queue -A decoded request waits in the server's request queue until a request worker takes it. Workers are asynchronous loops on the .NET thread pool, not dedicated threads; a worker counts as busy while it executes a request and, with `DecoupleHeldPublishRequests`, is released when a `Publish` parks. With the FairShare, Balanced (default) or TrustedReservations [isolation profile](ResourceIsolation.md) the queue serves the callers in weighted turns; with SharedOnly it is one FIFO. +The "thread" settings are concurrency limits: requests run on the .NET thread pool, not on dedicated threads. With SharedOnly, a decoded request takes one of `MaxRequestThreadCount` permits and runs at once; when every permit is in use it waits in a FIFO queue, and a finished request hands its permit to the oldest waiting one. With the FairShare, Balanced (default) or TrustedReservations [isolation profile](ResourceIsolation.md) the request waits until a request worker (an asynchronous loop on the thread pool) takes it, and the queue serves the callers in weighted turns. In both modes, with `DecoupleHeldPublishRequests`, a `Publish` that parks releases its permit or worker. | Setting | Configuration default | Current behaviour | |---|---|---| -| `MinRequestThreadCount` | 10 | Workers started with the server (at least 1). The thread pool's limits are not changed; up to 2.0.1 the server raised the thread pool's minimum worker and I/O threads to this value. | -| `MaxRequestThreadCount` | 100 | Upper bound of workers (values below 100, or below `MinRequestThreadCount`, are raised). A worker is added only when a request is queued and every worker is busy; workers are not retired. | -| `MaxQueuedRequestCount` | 200 | Requests of established sessions that can wait for a worker (values below 100 are raised to 100). A request that finds the queue full is answered with `BadServerTooBusy`. | +| `MinRequestThreadCount` | 10 | At least 1. With an isolation profile, the workers started with the server; with SharedOnly no workers are started. `MaxRequestThreadCount` is raised to it. The thread pool's limits are not changed; up to 2.0.1 the server raised the thread pool's minimum worker and I/O threads to this value. | +| `MaxRequestThreadCount` | 100 | Requests executing at once. With an isolation profile, the upper bound of workers; a worker is added only when a request is queued and every worker is busy, and workers are not retired. Up to 2.0.1 values below 100 were raised to 100. | +| `MaxQueuedRequestCount` | 200 | Requests of established sessions that can wait to execute (values below 100 are raised to 100). A request that finds the queue full is answered with `BadServerTooBusy`. | Guidance: diff --git a/perf/README.md b/perf/README.md index 91ba69b5ab..9e6449cbe8 100644 --- a/perf/README.md +++ b/perf/README.md @@ -69,6 +69,7 @@ perf/ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe run --scenario | `--duration`, `--warmup` | 20, 10 | Measurement window and warm-up in seconds. | | `--diag`, `--audit` | 0 | Server diagnostics and auditing. | | `--threads`, `--queue` | 200, 2000 | `MaxRequestThreadCount` and `MaxQueuedRequestCount` of the server; `MinRequestThreadCount` is the smaller of 50 and `--threads`. | +| `--isolation` | `default` | `sharedonly`: run the server with the SharedOnly [resource isolation](../docs/ResourceIsolation.md) mode (no isolation provider, one FIFO request queue) instead of the default profile. | | `--trace ` | | Record the measurement window with `dotnet-trace`. | | `--trace-profile` | `cpu` | `cpu` (sampled thread time), `alloc` (allocation ticks), `contention` or `gc`. | | `--trace-side` | `server` | The process to record: `server` (the child) or `client` (the harness itself). | diff --git a/perf/ServerLoadHarness/NativeWorkerBenchmark.cs b/perf/ServerLoadHarness/NativeWorkerBenchmark.cs index 644d68a695..ed408ba557 100644 --- a/perf/ServerLoadHarness/NativeWorkerBenchmark.cs +++ b/perf/ServerLoadHarness/NativeWorkerBenchmark.cs @@ -413,7 +413,9 @@ namespace Opc.Ua.Perf.ServerLoadHarness FormattableString.Invariant( $"--counters {(counters ? Program.Arg("nodes", "1") : "0")} --sampling {Program.Arg("sampling", "10")} ") + FormattableString.Invariant( - $"--counter-lock {Program.Arg("counter-lock", "1")} --async-node-manager {Program.Arg("async-node-manager", "1")}"), + $"--counter-lock {Program.Arg("counter-lock", "1")} --async-node-manager {Program.Arg("async-node-manager", "1")} ") + + FormattableString.Invariant( + $"--isolation {Program.Arg("isolation", "default")}"), // the benchmark's .NET workers run with Server GC; SERVER_DOTNET_gcServer=0 overrides it new Dictionary(StringComparer.Ordinal) { ["DOTNET_gcServer"] = "1" }); string ready = handle.m_lines.Take(); diff --git a/perf/ServerLoadHarness/Program.cs b/perf/ServerLoadHarness/Program.cs index 4107c468d7..ab6c5f65b6 100644 --- a/perf/ServerLoadHarness/Program.cs +++ b/perf/ServerLoadHarness/Program.cs @@ -318,6 +318,10 @@ namespace Opc.Ua.Perf.ServerLoadHarness : contract ? new BenchmarkContractServer(telemetry, ArgInt("async-node-manager", 1) != 0) : new ReferenceServer(telemetry); + if (Arg("isolation", "default") == "sharedonly") + { + server.ResourceIsolationOptions.Mode = Opc.Ua.Server.ServerResourceIsolationMode.SharedOnly; + } server.TransportBindings = TestTransportBindings.WithAllSchemes(); await application.StartAsync(server).ConfigureAwait(false); Console.WriteLine(FormattableString.Invariant($"READY {endpointUrl} {Environment.ProcessId}")); @@ -391,7 +395,7 @@ namespace Opc.Ua.Perf.ServerLoadHarness }; (Process startedServer, BlockingCollection startedLines) = StartServerChild( FormattableString.Invariant( - $"server --port {port} --diag {Arg("diag", "0")} --audit {Arg("audit", "0")} --threads {Arg("threads", "200")} --counters {(scenario == "counters" ? Arg("nodes", "1") : "0")} --sampling {Arg("sampling", "10")} --queue {Arg("queue", "2000")}"), + $"server --port {port} --diag {Arg("diag", "0")} --audit {Arg("audit", "0")} --threads {Arg("threads", "200")} --counters {(scenario == "counters" ? Arg("nodes", "1") : "0")} --sampling {Arg("sampling", "10")} --queue {Arg("queue", "2000")} --isolation {Arg("isolation", "default")}"), serverEnvironment); using Process server = startedServer; using BlockingCollection serverLines = startedLines; diff --git a/src/Opc.Ua.Core/Schema/ApplicationConfiguration.cs b/src/Opc.Ua.Core/Schema/ApplicationConfiguration.cs index 3cf1a5fc64..36d170669d 100644 --- a/src/Opc.Ua.Core/Schema/ApplicationConfiguration.cs +++ b/src/Opc.Ua.Core/Schema/ApplicationConfiguration.cs @@ -1017,21 +1017,33 @@ namespace Opc.Ua } /// - /// The minimum number of threads assigned to processing requests. + /// The lower bound of the request concurrency limit; values below 1 are raised to 1. /// - /// The minimum request thread count. + /// + /// Requests do not run on dedicated threads. With a resource isolation profile + /// other than SharedOnly, this many request workers are started with the server. + /// is raised to this value. + /// + /// The minimum request concurrency. [DataTypeField(Order = 3)] public int MinRequestThreadCount { get; set; } = 10; /// - /// The maximum number of threads assigned to processing requests. + /// The maximum number of requests that execute at once (the request concurrency limit). /// - /// The maximum request thread count. + /// + /// Requests run on the .NET thread pool, not on dedicated threads. A request that + /// finds the limit reached waits in the request queue (see + /// ). With + /// a held request, such as a Publish waiting for notifications, does not + /// count against the limit while it waits. + /// + /// The maximum request concurrency. [DataTypeField(Order = 4)] public int MaxRequestThreadCount { get; set; } = 100; /// - /// The maximum number of requests that will be queued waiting for a thread. + /// The maximum number of requests that will be queued waiting to execute. /// /// The maximum queued request count. [DataTypeField(Order = 5)] diff --git a/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs index 0f08f9288a..e75407211f 100644 --- a/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs +++ b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs @@ -40,22 +40,34 @@ namespace Opc.Ua public partial class ServerBase { /// - /// Asynchronously manages a queue of requests. + /// Limits how many requests execute at once and queues the requests that wait. /// + /// + /// Without an isolation provider a request takes one of + /// maxThreadCount permits and runs on the thread pool; the queue holds only + /// the requests waiting for a permit, and a finished request hands its permit to + /// the oldest of them. With an isolation provider, workers dequeue from the + /// . + /// protected class RequestQueue : IDisposable { /// /// Initializes a new instance of the class. /// /// The server. - /// The minimum number of threads in the pool. - /// The maximum number of threads in the pool. + /// + /// The number of fair-queue workers started up front. Without an isolation + /// provider no workers are started, and it only has to be positive. + /// + /// + /// The maximum number of requests executing at once. + /// /// The maximum number of requests that will placed in the queue. /// /// When true (the default) a request that parks (for example a held - /// Publish waiting for notifications) releases its worker at the park - /// point instead of occupying it for the whole wait, so the worker/thread - /// budget need not scale with the number of concurrently-held requests. + /// Publish waiting for notifications) releases its permit or worker at the + /// park point instead of occupying it for the whole wait, so the concurrency + /// limit need not scale with the number of concurrently-held requests. /// /// /// The server-scoped isolation provider, or null for the SharedOnly compatibility path. @@ -81,15 +93,8 @@ namespace Opc.Ua m_minThreadCount = minThreadCount; m_maxThreadCount = maxThreadCount; m_decoupleHeldPublishRequests = decoupleHeldPublishRequests; - - var options = new System.Threading.Channels.BoundedChannelOptions(maxRequestCount) - { - SingleWriter = false, - SingleReader = false, - FullMode = System.Threading.Channels.BoundedChannelFullMode.Wait - }; - - m_queue = System.Threading.Channels.Channel.CreateBounded(options); + m_maxRequestCount = maxRequestCount; + m_waiting = new Queue(); if (resourceIsolationProvider != null) { m_fairQueue = new FairRequestQueue( @@ -111,25 +116,30 @@ namespace Opc.Ua description: "Time a request waited in the request queue before it was dequeued."); m_meter.CreateObservableGauge( "opcua.server.request_queue.queued", - () => m_fairQueue?.Count ?? m_queue.Reader.Count, + () => m_fairQueue?.Count ?? Volatile.Read(ref m_waitingCount), description: "Requests waiting in the request queue."); m_meter.CreateObservableGauge( "opcua.server.request_queue.executing", () => Volatile.Read(ref m_activeThreadCount), - description: "Requests executing and holding a worker."); + description: "Requests executing and holding a permit or worker."); m_meter.CreateObservableGauge( "opcua.server.request_queue.parked", - () => Math.Max(0, Volatile.Read(ref m_processingCount) - Volatile.Read(ref m_activeThreadCount)), - description: "Requests parked (for example held Publish requests) after releasing their worker."); + () => Volatile.Read(ref m_parkedCount), + description: "Requests parked (for example held Publish requests) after releasing their permit or worker."); m_workers = []; m_cts = new CancellationTokenSource(); + m_stopToken = m_cts.Token; m_activeThreadCount = 0; m_totalThreadCount = 0; m_stopped = false; // The process-wide ThreadPool limits are left to the host: workers are // asynchronous and do not occupy a pool thread while they wait. + if (m_fairQueue == null) + { + return; + } // Start worker tasks. Increment m_totalThreadCount before Task.Run so that // ScheduleIncomingRequest sees the correct worker count immediately and does @@ -186,22 +196,35 @@ namespace Opc.Ua } else { + var queued = new QueuedRequest(request, GetEnqueueTimestamp(), cancellationToken); + bool dispatch = false; lock (m_workerGate) { serverStopped = m_stopped; if (!serverStopped) { - queueFull = !m_queue.Writer.TryWrite( - new QueuedRequest(request, GetEnqueueTimestamp(), cancellationToken)); - if (!queueFull && - m_totalThreadCount < m_maxThreadCount && - m_activeThreadCount >= m_totalThreadCount) + // A free permit goes to a waiting request first, so a request + // runs at once only when nothing is waiting. + if (m_waitingCount == 0 && m_activeThreadCount < m_maxThreadCount) { - Interlocked.Increment(ref m_totalThreadCount); - m_workers.Add(Task.Run(() => WorkerLoopAsync(m_cts.Token), CancellationToken.None)); + m_activeThreadCount++; + dispatch = true; + } + else if (m_waitingCount < m_maxRequestCount) + { + m_waiting.Enqueue(queued); + m_waitingCount++; + } + else + { + queueFull = true; } } } + if (dispatch) + { + Dispatch(queued); + } } if (serverStopped) @@ -258,6 +281,94 @@ namespace Opc.Ua } } + /// + /// Runs a request that holds a permit on the thread pool, so that the caller, + /// usually a transport receive loop, does not run the handler. + /// + private void Dispatch(QueuedRequest queued) + { + var item = new DispatchItem(this, queued); +#if NETCOREAPP3_0_OR_GREATER + ThreadPool.UnsafeQueueUserWorkItem(item, preferLocal: false); +#else + ThreadPool.UnsafeQueueUserWorkItem(static state => ((DispatchItem)state!).Execute(), item); +#endif + } + + /// + /// Runs requests while this permit is handed from one request to the next + /// waiting one, then returns the permit. + /// + /// + /// Never throws: completes every request. + /// + private async Task RunWithPermitAsync(QueuedRequest queued) + { + while (true) + { + RecordTimeInQueue(queued.EnqueueTimestamp); + CancellationToken ct = m_stopToken; + if (ct.IsCancellationRequested || Volatile.Read(ref m_stopped) || + queued.CancellationToken.IsCancellationRequested) + { + RejectRequest( + queued.Request, + ct.IsCancellationRequested || Volatile.Read(ref m_stopped) ? + StatusCodes.BadServerHalted : StatusCodes.BadRequestCancelledByClient); + } + else if (RequestQueueBudget.HasExpired(queued.Request)) + { + // the client gave up while the request was queued + CompleteRequest(queued.Request, StatusCodes.BadTimeout); + } + else if (m_decoupleHeldPublishRequests && + queued.Request is IParkableIncomingRequest { ParkSink: RequestParkSink parkSink }) + { + // A request that parks keeps running detached; the permit is + // returned at the park point. + Task processing = ProcessRequestSafeAsync(queued, null, ct); + if (!processing.IsCompleted) + { + await Task.WhenAny(processing, parkSink.ParkedTask).ConfigureAwait(false); + CountWhileParked(processing); + } + } + else + { + await ProcessRequestSafeAsync(queued, null, ct).ConfigureAwait(false); + } + + if (!TryHandOffPermit(out queued)) + { + return; + } + } + } + + /// + /// Hands the permit of a finished or parked request to the oldest waiting + /// request, or returns it when none is waiting or the queue has stopped. + /// + private bool TryHandOffPermit(out QueuedRequest next) + { + lock (m_workerGate) + { + if (!m_stopped && m_waitingCount > 0) + { + next = m_waiting.Dequeue(); + m_waitingCount--; + return true; + } + m_activeThreadCount--; + if (m_stopped && m_activeThreadCount == 0 && Volatile.Read(ref m_processingCount) == 0) + { + m_processingDrained?.TrySetResult(true); + } + } + next = default; + return false; + } + /// /// Starts shared asynchronous disposal without blocking the caller. /// @@ -309,9 +420,17 @@ namespace Opc.Ua private async Task StopCoreAsync(Task[] workers) { Task cancellation = m_cts.CancelAsync(); - m_queue.Writer.TryComplete(); m_fairQueue?.Dispose(); - while (m_queue.Reader.TryRead(out QueuedRequest request)) + QueuedRequest[] waiting; + lock (m_workerGate) + { + // Admission and hand-off check m_stopped under the gate, so nothing + // enters the waiting queue after this. + waiting = [.. m_waiting]; + m_waiting.Clear(); + m_waitingCount = 0; + } + foreach (QueuedRequest request in waiting) { RejectRequest(request.Request, StatusCodes.BadServerHalted); } @@ -327,11 +446,12 @@ namespace Opc.Ua Task processing; lock (m_workerGate) { - // Parked handlers may still finish after the workers; the last one to + // Parked handlers may still finish after the workers, and requests that + // hold a permit may not have started yet; the last handler or permit to // finish completes the drain, unless none is left already. m_processingDrained = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - if (Volatile.Read(ref m_processingCount) == 0) + if (Volatile.Read(ref m_processingCount) == 0 && m_activeThreadCount == 0) { m_processingDrained.TrySetResult(true); } @@ -343,57 +463,43 @@ namespace Opc.Ua } /// - /// Ran by the worker threads to process requests. + /// Ran by the fair-queue workers to process requests. /// /// private async Task WorkerLoopAsync(CancellationToken ct) { // Stopping disposes the fair queue, which ends every parked wait, so a parked // worker does not register on the shared stop token every time it parks. - WorkerWakeSignal.Waiter? waiter = m_fairQueue?.CreateWaiter(); + FairRequestQueue fairQueue = m_fairQueue!; + WorkerWakeSignal.Waiter waiter = fairQueue.CreateWaiter(); try { while (!ct.IsCancellationRequested && !Volatile.Read(ref m_stopped)) { - FairRequestQueue.Entry? entry = null; - QueuedRequest queued; - if (m_fairQueue != null) + FairRequestQueue.Entry entry; + try { - try - { - entry = await m_fairQueue.DequeueAsync(waiter!, CancellationToken.None) - .ConfigureAwait(false); - } - catch (Exception ex) when (ex is not OperationCanceledException) - { - m_server.m_logger.RequestQueueProcessingFailed(ex); - continue; - } - queued = new QueuedRequest( - entry.Request, entry.EnqueueTimestamp, entry.CancellationToken); + entry = await fairQueue.DequeueAsync(waiter, CancellationToken.None) + .ConfigureAwait(false); } - else + catch (Exception ex) when (ex is not OperationCanceledException) { - queued = await m_queue.Reader.ReadAsync(ct).ConfigureAwait(false); + m_server.m_logger.RequestQueueProcessingFailed(ex); + continue; } + var queued = new QueuedRequest( + entry.Request, entry.EnqueueTimestamp, entry.CancellationToken); RecordTimeInQueue(queued.EnqueueTimestamp); if (ct.IsCancellationRequested || Volatile.Read(ref m_stopped) || queued.CancellationToken.IsCancellationRequested) { - entry?.Dispose(); + entry.Dispose(); RejectRequest( queued.Request, ct.IsCancellationRequested || Volatile.Read(ref m_stopped) ? StatusCodes.BadServerHalted : StatusCodes.BadRequestCancelledByClient); continue; } - // the fair queue drops expired requests when it selects them - if (entry == null && RequestQueueBudget.HasExpired(queued.Request)) - { - // the client gave up while the request was queued - CompleteRequest(queued.Request, StatusCodes.BadTimeout); - continue; - } if (m_decoupleHeldPublishRequests && queued.Request is IParkableIncomingRequest { ParkSink: RequestParkSink parkSink }) { @@ -418,10 +524,6 @@ namespace Opc.Ua { // Graceful shutdown } - catch (System.Threading.Channels.ChannelClosedException) - { - // Graceful shutdown - } finally { Interlocked.Decrement(ref m_totalThreadCount); @@ -452,6 +554,7 @@ namespace Opc.Ua if (!processing.IsCompleted) { await Task.WhenAny(processing, parkSink.ParkedTask).ConfigureAwait(false); + CountWhileParked(processing); } } finally @@ -465,6 +568,26 @@ namespace Opc.Ua // required here because ProcessRequestSafeAsync observes all faults. } + /// + /// Counts a request that released its permit or worker at the park point as + /// parked until its handler finishes. + /// + private void CountWhileParked(Task processing) + { + if (processing.IsCompleted) + { + return; + } + Interlocked.Increment(ref m_parkedCount); + // ProcessRequestSafeAsync never throws, so the continuation only counts. + _ = processing.ContinueWith( + static (_, state) => Interlocked.Decrement(ref ((RequestQueue)state!).m_parkedCount), + this, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + /// /// Invokes the server request handler and traps any unexpected error the same /// way the legacy inline path did, guaranteeing the request is always faulted @@ -513,12 +636,16 @@ namespace Opc.Ua { entry?.Dispose(); // Only shutdown waits for the drain, and it starts once every worker - // has exited, so the gate is needed only after the stop. + // has exited, so the gate is needed only after the stop. A permit that + // is still held completes the drain when it is returned. if (Interlocked.Decrement(ref m_processingCount) == 0 && Volatile.Read(ref m_stopped)) { lock (m_workerGate) { - m_processingDrained?.TrySetResult(true); + if (m_activeThreadCount == 0) + { + m_processingDrained?.TrySetResult(true); + } } } } @@ -573,30 +700,66 @@ namespace Opc.Ua long EnqueueTimestamp, CancellationToken CancellationToken); + /// + /// Thread pool work item that runs a request holding a permit. + /// + private sealed class DispatchItem(RequestQueue queue, QueuedRequest queued) +#if NETCOREAPP3_0_OR_GREATER + : IThreadPoolWorkItem +#endif + { + /// + /// Runs the request; the returned task never faults. + /// + public void Execute() + { + _ = queue.RunWithPermitAsync(queued); + } + } + /// /// Server that dispatches admitted requests. /// private readonly ServerBase m_server; /// - /// Number of workers started before any requests arrive. + /// Number of fair-queue workers started before any requests arrive. /// private readonly int m_minThreadCount; /// - /// Maximum number of simultaneously retained worker tasks. + /// Maximum number of requests executing at once: the permits of the shared + /// queue, or the retained worker tasks of the fair queue. /// private readonly int m_maxThreadCount; /// - /// Whether parked requests release their execution workers. + /// Maximum number of requests waiting for a permit. + /// + private readonly int m_maxRequestCount; + + /// + /// Whether parked requests release their permit or worker. /// private readonly bool m_decoupleHeldPublishRequests; /// - /// Compatibility FIFO used when no isolation provider is installed. + /// Requests waiting for a permit when no isolation provider is installed, + /// guarded by . /// - private readonly System.Threading.Channels.Channel m_queue; + private readonly Queue m_waiting; + + /// + /// Number of requests in , changed under + /// and read without it by the queued gauge. + /// + private int m_waitingCount; + + /// + /// Token of , read without touching the source, which + /// shutdown disposes once every permit and handler has finished. + /// + private readonly CancellationToken m_stopToken; // Ownership transfers to StopCoreAsync, which disposes these after asynchronous shutdown. // TODO: Remove the pragma when CA2213 recognizes deferred asynchronous ownership. #pragma warning disable CA2213 @@ -653,7 +816,15 @@ namespace Opc.Ua private int m_processingCount; /// - /// Number of workers currently executing rather than parked or idle. + /// Number of handlers that parked and released their permit or worker, + /// changed with . + /// + private int m_parkedCount; + + /// + /// Number of requests executing rather than parked or waiting: the permits in + /// use, changed under , or the busy fair-queue + /// workers, changed with . /// private int m_activeThreadCount; diff --git a/src/Opc.Ua.Core/Stack/Server/ServerBase.cs b/src/Opc.Ua.Core/Stack/Server/ServerBase.cs index 883f316feb..b2f8d2b98a 100644 --- a/src/Opc.Ua.Core/Stack/Server/ServerBase.cs +++ b/src/Opc.Ua.Core/Stack/Server/ServerBase.cs @@ -668,11 +668,6 @@ namespace Opc.Ua maxRequestThreadCount = minRequestThreadCount; } - if (maxRequestThreadCount < 100) - { - maxRequestThreadCount = 100; - } - if (maxQueuedRequestCount < 100) { maxQueuedRequestCount = 100; diff --git a/src/Opc.Ua.Server/ResourceIsolation/ServerResourceIsolationOptions.cs b/src/Opc.Ua.Server/ResourceIsolation/ServerResourceIsolationOptions.cs index c0ec83ab39..1c961b1845 100644 --- a/src/Opc.Ua.Server/ResourceIsolation/ServerResourceIsolationOptions.cs +++ b/src/Opc.Ua.Server/ResourceIsolation/ServerResourceIsolationOptions.cs @@ -281,7 +281,9 @@ namespace Opc.Ua.Server ? rateLimits.MaxConcurrentSessionEstablishment : ServerRateLimitOptions.DefaultMaxConcurrentSessionEstablishment; int queuedCapacity = Math.Max(100, server.MaxQueuedRequestCount); - int executionCapacity = Math.Max(100, Math.Max(server.MinRequestThreadCount, server.MaxRequestThreadCount)); + // the request queue starts at most this many workers, so the execution stage cannot + // promise protected classes more than the workers can run + int executionCapacity = Math.Max(1, Math.Max(server.MinRequestThreadCount, server.MaxRequestThreadCount)); long retainedRequestSlots = 2L * queuedCapacity + executionCapacity; if (retainedRequestSlots > long.MaxValue / quotas.MaxMessageSize) { @@ -382,6 +384,13 @@ namespace Opc.Ua.Server SubtractFloor(ref remaining, owner.Floors[ii]); trustedTotal += owner.Floors[ii]; } + if (remaining < unit && ii == (int)ResourceIsolationStage.RequestExecution && item?.Capacity == null) + { + throw new ArgumentException( + $"MaxRequestThreadCount ({executionCapacity}) must exceed the requests reserved for " + + "bootstrap, reconnect, control and trusted owners. Increase MaxRequestThreadCount or " + + "reduce the RequestExecution reservations."); + } if (remaining < unit || hardLimit < Math.Max(Math.Max(bootstrapFloor, reconnectFloor), controlFloor)) { throw new ArgumentException($"The {(ResourceIsolationStage)ii} pools cannot hold an operation."); diff --git a/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs index 4b4af639e0..232d373bcf 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs @@ -254,6 +254,22 @@ namespace Opc.Ua.Core.Tests.Stack.Server } } + /// + /// Creates the queue from configuration, as starting the server does. + /// + public void InitializeQueue(ApplicationConfiguration configuration) + { + InitializeRequestQueue(configuration); + } + + /// + /// Stops the queue, as stopping the server does. + /// + public ValueTask StopQueueAsync() + { + return StopRequestQueueAsync(); + } + /// /// Replaces the default queue with the requested worker and admission envelope. /// @@ -780,6 +796,179 @@ namespace Opc.Ua.Core.Tests.Stack.Server queued.ProcessingCompleted.TrySetResult(true); } + /// + /// Without an isolation provider the queue starts no workers: requests take a + /// permit and run on the thread pool. + /// + [Test] + public async Task SharedQueueStartsNoWorkersAsync() + { + using var server = new TestServer(); + server.ReplaceRequestQueue(10, 20, 100); + Assert.That(server.WorkerCount, Is.Zero); + + var request = new TestEndpointIncomingRequest(); + server.ScheduleIncomingRequest(request); + await AwaitOrFailAsync(request.ProcessingStarted.Task, "the request to start").ConfigureAwait(false); + Assert.That(server.WorkerCount, Is.Zero); + request.ProcessingCompleted.TrySetResult(true); + } + + /// + /// At most maxThreadCount requests execute at once; a finished request hands its + /// permit to the oldest waiting request. + /// + [Test] + public async Task PermitsLimitConcurrencyAndServeWaitersInOrderAsync() + { + using var server = new TestServer(); + server.ReplaceRequestQueue(1, 3, 100); + using var metrics = new QueueMetricsListener(server.QueueMeter); + var requests = new TestEndpointIncomingRequest[10]; + for (int ii = 0; ii < requests.Length; ii++) + { + requests[ii] = new TestEndpointIncomingRequest(); + server.ScheduleIncomingRequest(requests[ii]); + } + + for (int ii = 0; ii < 3; ii++) + { + await AwaitOrFailAsync(requests[ii].ProcessingStarted.Task, $"request {ii} to start") + .ConfigureAwait(false); + } + Assert.That(await CompletedWithinAsync(requests[3].ProcessingStarted.Task, 200).ConfigureAwait(false), + Is.False, "a fourth request must wait for a permit"); + Assert.That(metrics.Observe("opcua.server.request_queue.executing"), Is.EqualTo(3)); + Assert.That(metrics.Observe("opcua.server.request_queue.queued"), Is.EqualTo(7)); + + for (int ii = 3; ii < requests.Length; ii++) + { + requests[ii - 3].ProcessingCompleted.TrySetResult(true); + await AwaitOrFailAsync(requests[ii].ProcessingStarted.Task, $"request {ii} to start") + .ConfigureAwait(false); + if (ii + 1 < requests.Length) + { + Assert.That(requests[ii + 1].ProcessingStarted.Task.IsCompleted, Is.False, + "only the oldest waiting request receives the permit"); + } + } + for (int ii = requests.Length - 3; ii < requests.Length; ii++) + { + requests[ii].ProcessingCompleted.TrySetResult(true); + } + await AwaitOrFailAsync( + WaitUntilAsync(() => metrics.Observe("opcua.server.request_queue.executing") == 0), + "every permit to be returned").ConfigureAwait(false); + Assert.That(metrics.Observe("opcua.server.request_queue.queued"), Is.Zero); + } + + /// + /// A configured MaxRequestThreadCount below 100 is used as configured. + /// + [Test] + public async Task ConfiguredConcurrencyBelowHundredIsHonoredAsync() + { + using var server = new TestServer(); + server.InitializeQueue(new ApplicationConfiguration + { + ServerConfiguration = new ServerConfiguration + { + MinRequestThreadCount = 1, + MaxRequestThreadCount = 2, + MaxQueuedRequestCount = 100 + } + }); + var requests = new TestEndpointIncomingRequest[3]; + for (int ii = 0; ii < requests.Length; ii++) + { + requests[ii] = new TestEndpointIncomingRequest(); + server.ScheduleIncomingRequest(requests[ii]); + } + + await AwaitOrFailAsync(requests[0].ProcessingStarted.Task, "request 0 to start").ConfigureAwait(false); + await AwaitOrFailAsync(requests[1].ProcessingStarted.Task, "request 1 to start").ConfigureAwait(false); + Assert.That(await CompletedWithinAsync(requests[2].ProcessingStarted.Task, 200).ConfigureAwait(false), + Is.False, "the third request must wait for one of the two permits"); + + requests[0].ProcessingCompleted.TrySetResult(true); + await AwaitOrFailAsync(requests[2].ProcessingStarted.Task, "request 2 to start").ConfigureAwait(false); + requests[1].ProcessingCompleted.TrySetResult(true); + requests[2].ProcessingCompleted.TrySetResult(true); + } + + /// + /// Stopping rejects waiting requests and completes only after executing and + /// parked handlers have finished. + /// + [Test] + public async Task StopDrainsExecutingAndParkedRequestsAsync() + { + using var server = new TestServer(); + server.ReplaceRequestQueue(1, 1, 10, decoupleHeldPublishRequests: true); + var parked = new ParkableTestRequest(); + var executing = new TestEndpointIncomingRequest(); + var waiting = new TestEndpointIncomingRequest(); + + server.ScheduleIncomingRequest(parked); + await AwaitOrFailAsync(parked.ProcessingStarted.Task, "the request to park").ConfigureAwait(false); + server.ScheduleIncomingRequest(executing); + await AwaitOrFailAsync(executing.ProcessingStarted.Task, "the second request to start") + .ConfigureAwait(false); + server.ScheduleIncomingRequest(waiting); + + Task stop = server.StopQueueAsync().AsTask(); + await AwaitOrFailAsync(waiting.ProcessingCompleted.Task, "the waiting request to be rejected") + .ConfigureAwait(false); + Assert.That(waiting.CompletedStatusCode, Is.EqualTo(StatusCodes.BadServerHalted)); + Assert.That(await CompletedWithinAsync(stop, 200).ConfigureAwait(false), Is.False, + "stop waits for the executing request"); + + executing.ProcessingCompleted.TrySetResult(true); + Assert.That(await CompletedWithinAsync(stop, 200).ConfigureAwait(false), Is.False, + "stop waits for the parked request"); + + parked.ProcessingCompleted.TrySetResult(true); + await AwaitOrFailAsync(stop, "stop to complete").ConfigureAwait(false); + } + + /// + /// A request that holds a permit but has not started when the queue stops is + /// halted, and stop still waits for its permit. + /// + [Test] + public async Task ConcurrentStopHaltsOrRunsEveryRequestAsync() + { + using var server = new TestServer(); + server.ReplaceRequestQueue(1, 4, 1000); + var requests = new TestEndpointIncomingRequest[200]; + for (int ii = 0; ii < requests.Length; ii++) + { + requests[ii] = new TestEndpointIncomingRequest(); + TestEndpointIncomingRequest request = requests[ii]; + _ = request.ProcessingStarted.Task.ContinueWith( + _ => request.ProcessingCompleted.TrySetResult(true), TaskScheduler.Default); + } + + var schedule = Task.Run(() => + { + foreach (TestEndpointIncomingRequest request in requests) + { + server.ScheduleIncomingRequest(request); + } + }); + await Task.Delay(1).ConfigureAwait(false); + await server.StopQueueAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false); + await schedule.ConfigureAwait(false); + + Task all = Task.WhenAll(Array.ConvertAll(requests, r => r.ProcessingCompleted.Task)); + await AwaitOrFailAsync(all, "every request to complete").ConfigureAwait(false); + foreach (TestEndpointIncomingRequest request in requests) + { + Assert.That(request.CompletedStatusCode, + Is.Null.Or.EqualTo(StatusCodes.BadServerHalted)); + } + } + /// /// A request the isolation provider can classify. /// diff --git a/tests/Opc.Ua.Server.Tests/ResourceIsolation/IsolationConfigurationValidationTests.cs b/tests/Opc.Ua.Server.Tests/ResourceIsolation/IsolationConfigurationValidationTests.cs index 0c81d7a600..7e6ac891b8 100644 --- a/tests/Opc.Ua.Server.Tests/ResourceIsolation/IsolationConfigurationValidationTests.cs +++ b/tests/Opc.Ua.Server.Tests/ResourceIsolation/IsolationConfigurationValidationTests.cs @@ -60,6 +60,32 @@ namespace Opc.Ua.Server.Tests .And.Message.Contains("two minutes")); } + /// + /// The execution stage holds no more requests than the request queue has workers, + /// so reserved classes cannot be crowded out of the workers by ordinary requests. + /// + [Test] + public void RuntimePlanExecutionCapacityIsMaxRequestThreadCount() + { + ResourceIsolationStagePlan execution = Options() + .CreateRuntimePlan(Configuration(maxRequestThreadCount: 4), new ServerRateLimitOptions()) + .GetStage(ResourceIsolationStage.RequestExecution); + + Assert.That(execution.Capacity, Is.EqualTo(4)); + Assert.That(execution.SharedCapacity, Is.EqualTo(1)); + } + + /// + /// Too few workers for the reserved requests are named at startup. + /// + [Test] + public void RuntimePlanRejectsMaxRequestThreadCountBelowReservations() + { + Assert.That( + () => Options().CreateRuntimePlan(Configuration(maxRequestThreadCount: 3), new ServerRateLimitOptions()), + Throws.ArgumentException.With.Message.Contains("MaxRequestThreadCount (3)")); + } + [Test] public async Task StartupRejectsInvalidTimeoutWithoutInstallingOrDisposingProvidersAsync( [Values] ServerResourceIsolationMode mode, @@ -418,11 +444,17 @@ namespace Opc.Ua.Server.Tests /// /// Supplies finite totals with room for all protected classes and ordinary reconnect capacity. /// - private static ApplicationConfiguration Configuration() + private static ApplicationConfiguration Configuration(int maxRequestThreadCount = 100) { return new ApplicationConfiguration { - ServerConfiguration = new ServerConfiguration { MaxSessionCount = 2, MaxChannelCount = 100 }, + ServerConfiguration = new ServerConfiguration + { + MaxSessionCount = 2, + MaxChannelCount = 100, + MinRequestThreadCount = 1, + MaxRequestThreadCount = maxRequestThreadCount + }, TransportQuotas = new TransportQuotas { MaxMessageSize = 10, MaxBufferSize = 8192 } }; }