diff --git a/docs/Benchmarks.md b/docs/Benchmarks.md index 050cd1170b..ee2a894b49 100644 --- a/docs/Benchmarks.md +++ b/docs/Benchmarks.md @@ -844,7 +844,8 @@ The net10.0 in-process short `BufferManagerBenchmarks` run completed 256 `Buffer once, not the total number of open sessions. In these measurements, 100–200 workers served thousands of sessions. A very large pool oversubscribes cores during connection bursts and slows establishment. - `MinRequestThreadCount` prewarms the pool to avoid thread-pool cold start. + `MinRequestThreadCount` workers are started with the server; the + server does not change the thread pool's limits. See [Request workers and the request queue](ServerScalability.md#request-workers-and-the-request-queue) for the current behaviour of these settings and of `MaxQueuedRequestCount`. diff --git a/docs/ServerScalability.md b/docs/ServerScalability.md index dd1ef98961..7c04c3c7a5 100644 --- a/docs/ServerScalability.md +++ b/docs/ServerScalability.md @@ -223,18 +223,28 @@ A decoded request waits in the server's request queue until a request worker tak | Setting | Configuration default | Current behaviour | |---|---|---| -| `MinRequestThreadCount` | 10 | Workers started with the server (at least 1). The server also raises the thread pool's minimum worker and I/O threads to this value, so up to this many threads are created without the thread pool's injection delay. | -| `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. The thread pool's maximum is raised to at least this value. | +| `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`. | Guidance: - **Size `MaxQueuedRequestCount` for the pipelining of all clients**: the sum over the clients of the requests each keeps outstanding. For example, 10 clients with 1,000 outstanding Reads each need 10,000. With the Balanced and TrustedReservations profiles the protected bootstrap, reconnect and control entries (one each by default) and the floors of trusted owners are added on top of the configured value, and held Publishes are counted in a separate stage of the same size; see [request and caller limits](ResourceIsolation.md#configuring-request-and-caller-limits). Releases up to 2.0.1 carved the three protected entries out of the configured value, so there size it at clients × outstanding requests + 3. - **`MaxRequestThreadCount` bounds concurrent request execution, not threads.** Request handling is asynchronous, so a few busy workers per core are enough for in-memory node managers; the default 100 rarely needs raising. Raise it when node managers block a worker for I/O or locks (synchronous `CustomNodeManager2` code waiting on a device), or when `DecoupleHeldPublishRequests` is off. -- **Keep `MinRequestThreadCount` moderate.** It pre-warms the pool for connect bursts, but because it also sets the thread pool's minimum, a value far above the processor count lets the pool run many more threads than processors under load. On shared or pinned processors, a preempted thread then delays every GC suspension until it runs again. +- **The thread pool is the host's to configure.** Up to 2.0.1 the server set the process-wide thread pool minimum to `MinRequestThreadCount`, which let the pool run many more threads than processors under load; on shared or pinned processors a preempted thread then delays every GC suspension until it runs again. A host whose node managers block threads and that relied on that minimum sets it itself, with `ThreadPool.SetMinThreads` or the `System.Threading.ThreadPool.MinThreads` runtime setting. - **Increasing the minimum worker count did not help the one-processor workload.** A `MinRequestThreadCount` of 4, 16 or 200 changed the number of ready threads but not CPU per request or latency; see [Small Windows machines and thread priority boost](#small-windows-machines-and-thread-priority-boost) for the measured scheduling change. With the FairShare, Balanced and TrustedReservations profiles, an idle worker spins for a few microseconds before it parks only when the runtime reports at least 8 processors (at most one spinning worker per 8 processors); on smaller machines it parks at once. - `BadServerTooBusy` responses to established sessions while the processors are not saturated mean that an admission limit, not the CPU, is the limit. With an isolation profile that is not necessarily the queue count: the retained request bytes, the parked-request stage or a per-owner limit reject requests the same way while queue entries are free. Check the `opcua.server.isolation.rejections` metric (its `stage` and `reason` labels) and `opcua.server.isolation.usage` before raising `MaxQueuedRequestCount`; see [transport behavior and overload signals](ResourceIsolation.md#transport-behavior-and-overload-signals). +The request queue publishes these metrics in every isolation profile, on the meter named after the `Opc.Ua.Core` assembly: + +| Instrument | Type | Meaning | +|---|---|---| +| `opcua.server.request_queue.queued` | gauge | Requests waiting for a worker. | +| `opcua.server.request_queue.executing` | gauge | Requests executing on a worker. | +| `opcua.server.request_queue.parked` | gauge | Requests (held `Publish` requests) that parked and released their worker. | +| `opcua.server.request_queue.time_in_queue` | histogram, s | Time from admission until a worker took the request. | +| `opcua.server.request_queue.rejections` | counter, `status` | Requests the queue completed without running them, by status code: `BadServerTooBusy` (queue full), `BadServerHalted` (stopping), `BadRequestCancelledByClient` (cancelled while queued) or the status of a failed admission or revalidation. | + ## Scaling out A single node's handshake throughput is bounded by its cores. To scale beyond one node, distribute connections across several server nodes behind a connection-distributing front end. The stack provides distributed mirroring of the address space, sessions, and subscriptions for exactly this high-availability and distributed topology. diff --git a/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs b/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs index 4b625879f5..6e9e82daa6 100644 --- a/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs +++ b/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs @@ -71,7 +71,8 @@ namespace Opc.Ua int maxQueuedRequests, long requestCost, bool decoupleHeldPublishRequests, - Action complete) + Action complete, + Func? enqueueTimestamp = null) { m_provider = provider ?? throw new ArgumentNullException(nameof(provider)); m_reservations = provider as IResourceIsolationReservations; @@ -91,6 +92,7 @@ namespace Opc.Ua m_useFairScheduling = provider.UseFairScheduling; m_decoupleHeldPublishRequests = decoupleHeldPublishRequests; m_complete = complete ?? throw new ArgumentNullException(nameof(complete)); + m_enqueueTimestamp = enqueueTimestamp; m_hot.SpinScore = kMaxSpinScore; m_provider.CapacityAvailable += OnCapacityAvailable; } @@ -196,6 +198,7 @@ namespace Opc.Ua error = StatusCodes.BadRequestCancelledByClient; return false; } + entry.EnqueueTimestamp = m_enqueueTimestamp?.Invoke() ?? 0; // Reserve the queue slot before linking: the count never exceeds the bound, // also while admissions of different owners run concurrently. if (Interlocked.Increment(ref m_hot.Count) > m_maxQueuedRequests) @@ -1122,6 +1125,12 @@ namespace Opc.Ua CancellationToken = cancellationToken; } + /// + /// Gets or sets the time the request was admitted, or 0 + /// when it was not measured. Set before the entry is linked. + /// + internal long EnqueueTimestamp { get; set; } + /// /// Retained queue slot. /// @@ -1618,6 +1627,12 @@ namespace Opc.Ua /// private readonly Action m_complete; + /// + /// Takes the admission timestamp of an entry, or returns 0 when time in queue + /// is not measured; null when the owner does not measure it. + /// + private readonly Func? m_enqueueTimestamp; + /// /// Scheduling FIFOs retained while nonempty, keyed by owner in weighted mode /// or by the empty string for global FIFO mode. diff --git a/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs index 0de7dd6fb6..0f08f9288a 100644 --- a/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs +++ b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs @@ -29,6 +29,8 @@ using System; using System.Collections.Generic; +using System.Diagnostics; +using System.Diagnostics.Metrics; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; @@ -95,28 +97,39 @@ namespace Opc.Ua maxRequestCount, maxRequestCost, decoupleHeldPublishRequests, - CompleteRequest); + RejectRequest, + GetEnqueueTimestamp); } + m_meter = server.m_telemetry.CreateMeter(); + m_rejections = m_meter.CreateCounter( + "opcua.server.request_queue.rejections", + description: "Requests the request queue completed without running them, by status code."); + m_timeInQueue = m_meter.CreateHistogram( + "opcua.server.request_queue.time_in_queue", + unit: "s", + 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, + 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."); + 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."); + m_workers = []; m_cts = new CancellationTokenSource(); m_activeThreadCount = 0; m_totalThreadCount = 0; m_stopped = false; - ThreadPool.GetMinThreads(out minThreadCount, out int minCompletionPortThreads); - - ThreadPool.SetMinThreads( - Math.Max(minThreadCount, m_minThreadCount), - Math.Max(minCompletionPortThreads, m_minThreadCount) - ); - - ThreadPool.GetMaxThreads(out maxThreadCount, out int maxCompletionPortThreads); - - ThreadPool.SetMaxThreads( - Math.Max(maxThreadCount, m_maxThreadCount), - Math.Max(maxCompletionPortThreads, m_maxThreadCount) - ); + // The process-wide ThreadPool limits are left to the host: workers are + // asynchronous and do not occupy a pool thread while they wait. // Start worker tasks. Increment m_totalThreadCount before Task.Run so that // ScheduleIncomingRequest sees the correct worker count immediately and does @@ -178,7 +191,8 @@ namespace Opc.Ua serverStopped = m_stopped; if (!serverStopped) { - queueFull = !m_queue.Writer.TryWrite(new QueuedRequest(request, cancellationToken)); + queueFull = !m_queue.Writer.TryWrite( + new QueuedRequest(request, GetEnqueueTimestamp(), cancellationToken)); if (!queueFull && m_totalThreadCount < m_maxThreadCount && m_activeThreadCount >= m_totalThreadCount) @@ -192,7 +206,7 @@ namespace Opc.Ua if (serverStopped) { - CompleteRequest(request, StatusCodes.BadServerHalted); + RejectRequest(request, StatusCodes.BadServerHalted); return; } @@ -202,7 +216,7 @@ namespace Opc.Ua { if (!m_fairQueue.TryEnqueue(request, cancellationToken, out StatusCode error)) { - CompleteRequest(request, error); + RejectRequest(request, error); } else { @@ -218,12 +232,32 @@ namespace Opc.Ua } if (queueFull) { - CompleteRequest(request, StatusCodes.BadServerTooBusy); - // TODO: make a metric + RejectRequest(request, StatusCodes.BadServerTooBusy); m_server.m_logger.RequestQueueFull(m_activeThreadCount); } } + /// + /// Takes the enqueue timestamp for the time-in-queue histogram, or 0 when + /// nothing listens to it. + /// + private long GetEnqueueTimestamp() + { + return m_timeInQueue.Enabled ? Stopwatch.GetTimestamp() : 0; + } + + /// + /// Records how long a dequeued request waited, if its enqueue time was taken. + /// + private void RecordTimeInQueue(long enqueueTimestamp) + { + if (enqueueTimestamp != 0 && m_timeInQueue.Enabled) + { + m_timeInQueue.Record( + (double)(Stopwatch.GetTimestamp() - enqueueTimestamp) / Stopwatch.Frequency); + } + } + /// /// Starts shared asynchronous disposal without blocking the caller. /// @@ -279,7 +313,7 @@ namespace Opc.Ua m_fairQueue?.Dispose(); while (m_queue.Reader.TryRead(out QueuedRequest request)) { - CompleteRequest(request.Request, StatusCodes.BadServerHalted); + RejectRequest(request.Request, StatusCodes.BadServerHalted); } try { @@ -305,6 +339,7 @@ namespace Opc.Ua } await processing.ConfigureAwait(false); m_cts.Dispose(); + m_meter.Dispose(); } /// @@ -334,17 +369,19 @@ namespace Opc.Ua m_server.m_logger.RequestQueueProcessingFailed(ex); continue; } - queued = new QueuedRequest(entry.Request, entry.CancellationToken); + queued = new QueuedRequest( + entry.Request, entry.EnqueueTimestamp, entry.CancellationToken); } else { queued = await m_queue.Reader.ReadAsync(ct).ConfigureAwait(false); } + RecordTimeInQueue(queued.EnqueueTimestamp); if (ct.IsCancellationRequested || Volatile.Read(ref m_stopped) || queued.CancellationToken.IsCancellationRequested) { entry?.Dispose(); - CompleteRequest( + RejectRequest( queued.Request, ct.IsCancellationRequested || Volatile.Read(ref m_stopped) ? StatusCodes.BadServerHalted : StatusCodes.BadRequestCancelledByClient); @@ -447,7 +484,7 @@ namespace Opc.Ua StatusCode status = entry.GetRevalidationStatus(); if (StatusCode.IsBad(status)) { - CompleteRequest(request, status); + RejectRequest(request, status); return; } } @@ -512,10 +549,28 @@ namespace Opc.Ua } /// - /// Couples a retained request with its caller-controlled lifetime. + /// Counts a request the queue completes without running it, then completes it. + /// + private void RejectRequest(IEndpointIncomingRequest request, StatusCode statusCode) + { + if (m_rejections.Enabled) + { + // Symbolic ids are interned, so the tag allocates nothing per rejection. + string? status = statusCode.SymbolicId ?? + (StatusCode.TryGetInternedStatusCode(statusCode.Code, out StatusCode interned) ? + interned.SymbolicId : null); + m_rejections.Add(1, new KeyValuePair("status", status ?? "Bad")); + } + CompleteRequest(request, statusCode); + } + + /// + /// Couples a retained request with its caller-controlled lifetime and the + /// time it was queued (0 when not measured). /// private readonly record struct QueuedRequest( IEndpointIncomingRequest Request, + long EnqueueTimestamp, CancellationToken CancellationToken); /// @@ -554,7 +609,23 @@ namespace Opc.Ua /// Worker cancellation source retained until shutdown has drained all handlers. /// private readonly CancellationTokenSource m_cts; + + /// + /// Request queue instruments, disposed once shutdown has drained all handlers. + /// + private readonly Meter m_meter; #pragma warning restore CA2213 + + /// + /// Requests completed by the queue without running, tagged by status code. + /// + private readonly Counter m_rejections; + + /// + /// Time from enqueue to dequeue, in seconds. + /// + private readonly Histogram m_timeInQueue; + /// /// Worker tasks joined by asynchronous shutdown. /// diff --git a/src/Opc.Ua.Core/Stack/Server/ServerBase.cs b/src/Opc.Ua.Core/Stack/Server/ServerBase.cs index c770d2a6b4..883f316feb 100644 --- a/src/Opc.Ua.Core/Stack/Server/ServerBase.cs +++ b/src/Opc.Ua.Core/Stack/Server/ServerBase.cs @@ -91,7 +91,6 @@ namespace Opc.Ua IRequestParkingPolicy? requestParkingPolicy) { ServerError = new ServiceResult(StatusCodes.BadServerHalted); - m_requestQueue = new RequestQueue(this, 10, 100, 1000); m_telemetry = telemetry; m_logger = m_telemetry.CreateLogger(this); m_transportBindings = transportBindings; @@ -247,7 +246,26 @@ namespace Opc.Ua IEndpointIncomingRequest request, CancellationToken cancellationToken = default) { - m_requestQueue.ScheduleIncomingRequest(request, cancellationToken); + RequestQueue? queue = m_requestQueue; + if (queue != null) + { + queue.ScheduleIncomingRequest(request, cancellationToken); + return; + } + + // The queue is created when the server starts; until then nothing can run the request. + if (request == null) + { + throw new ArgumentNullException(nameof(request)); + } + try + { + request.OperationCompleted(null, StatusCodes.BadServerHalted); + } + catch (Exception ex) + { + m_logger.RequestFaultDeliveryFailed(ex); + } } /// @@ -2191,7 +2209,7 @@ namespace Opc.Ua #pragma warning restore IDE1006 // Naming Styles private IServiceMessageContext? m_messageContext; - private RequestQueue m_requestQueue; + private RequestQueue? m_requestQueue; private readonly ITelemetryContext m_telemetry; private ITransportBindingRegistry? m_transportBindings; diff --git a/tests/Directory.Build.targets b/tests/Directory.Build.targets index 87ee65ce06..a69c7bbce3 100644 --- a/tests/Directory.Build.targets +++ b/tests/Directory.Build.targets @@ -17,4 +17,16 @@ + + + + + + + diff --git a/tests/Opc.Ua.Client.Tests/Session/SessionRecoveryAuditTests.cs b/tests/Opc.Ua.Client.Tests/Session/SessionRecoveryAuditTests.cs index 2ff653371d..b9c22df07b 100644 --- a/tests/Opc.Ua.Client.Tests/Session/SessionRecoveryAuditTests.cs +++ b/tests/Opc.Ua.Client.Tests/Session/SessionRecoveryAuditTests.cs @@ -238,6 +238,7 @@ namespace Opc.Ua.Client.Tests { var clock = new ObservableFakeTimeProvider(); using var keepAlive = new SemaphoreSlim(0); + var completedKeepAlives = System.Threading.Channels.Channel.CreateUnbounded(); var originalPublish = new AsyncOperationGate(); var activation = new AsyncOperationGate(); var resumedPublish = new AsyncOperationGate(); @@ -362,6 +363,14 @@ namespace Opc.Ua.Client.Tests Assert.That(session.Reconnecting, Is.True); Assert.That(session.GoodPublishRequestCount, Is.Zero, "The old Publish must finish while recovery owns admission, without refilling itself."); + session.KeepAlive += (current, args) => + { + if (args.Status == null && + current.TransportChannel.EndpointDescription.EndpointUrl == alternate.Description.EndpointUrl) + { + completedKeepAlives.Writer.TryWrite(args); + } + }; activation.Release(); await recovering.WaitAsync(s_timeout).ConfigureAwait(false); Assert.That(session.SessionId, Is.EqualTo(sessionId)); @@ -385,6 +394,11 @@ namespace Opc.Ua.Client.Tests clock.Advance(TimeSpan.FromMilliseconds(session.KeepAliveInterval)); Assert.That(await keepAlive.WaitAsync(s_timeout).ConfigureAwait(false), Is.True, "Reusing a session must restart ongoing keepalive reads."); + // Wait for response processing before advancing past the next keepalive deadline. + KeepAliveEventArgs completed = await completedKeepAlives.Reader.ReadAsync(cancellation.Token) + .AsTask().WaitAsync(s_timeout).ConfigureAwait(false); + Assert.That(completed.Status, Is.Null); + Assert.That(completed.CurrentState, Is.EqualTo(ServerState.Running)); } } } diff --git a/tests/Opc.Ua.Core.Tests/Stack/Server/EndpointIncomingRequestTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Server/EndpointIncomingRequestTests.cs index 0a238f1b4d..b071cdeb76 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Server/EndpointIncomingRequestTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Server/EndpointIncomingRequestTests.cs @@ -60,7 +60,6 @@ namespace Opc.Ua.Core.Tests.Stack.Server field!.SetValue(this, ServiceMessageContext.Create(NUnitTelemetryContext.Create(true))); FieldInfo queueField = typeof(ServerBase).GetField( "m_requestQueue", BindingFlags.NonPublic | BindingFlags.Instance)!; - ((IDisposable)queueField!.GetValue(this)!).Dispose(); m_queue = new RequestQueue(this, 1, 1, 10); queueField.SetValue(this, m_queue); } diff --git a/tests/Opc.Ua.Core.Tests/Stack/Server/FairRequestQueueTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Server/FairRequestQueueTests.cs index 33a3839736..fb0efbc1ab 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Server/FairRequestQueueTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Server/FairRequestQueueTests.cs @@ -2642,7 +2642,7 @@ namespace Opc.Ua.Core.Tests.Stack.Server private int m_duplicateReleases; } - private class TestProvider : IServerResourceIsolationProvider, IResourceIsolationRevalidationProvider, + internal class TestProvider : IServerResourceIsolationProvider, IResourceIsolationRevalidationProvider, IResourceIsolationAdmissionEvidence { public TestProvider( diff --git a/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs index 47f6466220..4b4af639e0 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs @@ -31,6 +31,9 @@ // making CA2000 noisy without a real leak risk. Disabled file-level for the suite. #pragma warning disable CA2000 using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics.Metrics; using System.Reflection; using System.Threading; using System.Threading.Tasks; @@ -235,6 +238,22 @@ namespace Opc.Ua.Core.Tests.Stack.Server } } + /// + /// Gets the meter of the installed request queue. + /// + public Meter QueueMeter + { + get + { + FieldInfo queueField = typeof(ServerBase).GetField( + "m_requestQueue", BindingFlags.NonPublic | BindingFlags.Instance)!; + var queue = (RequestQueue)queueField!.GetValue(this)!; + FieldInfo meterField = typeof(RequestQueue).GetField( + "m_meter", BindingFlags.NonPublic | BindingFlags.Instance)!; + return (Meter)meterField!.GetValue(queue)!; + } + } + /// /// Replaces the default queue with the requested worker and admission envelope. /// @@ -276,6 +295,73 @@ namespace Opc.Ua.Core.Tests.Stack.Server } } + /// + /// A server that has not started has no request queue, so a request is halted + /// instead of starting workers for a queue that is replaced at start. + /// + [Test] + public void RequestBeforeStartIsHalted() + { + using var server = new TestServer(); + var request = new TestEndpointIncomingRequest(); + + server.ScheduleIncomingRequest(request); + + Assert.That(request.ProcessingStarted.Task.IsCompleted, Is.False); + Assert.That(request.CompletedStatusCode, Is.EqualTo(StatusCodes.BadServerHalted)); + } + + /// + /// Before start a null request is rejected like the queue rejects it. + /// + [Test] + public void NullRequestBeforeStartThrows() + { + using var server = new TestServer(); + + Assert.Throws(() => server.ScheduleIncomingRequest(null!)); + } + + /// + /// A completion callback that throws while a request is halted before start does + /// not escape to the transport that scheduled the request. + /// + [Test] + public void FaultingCompletionBeforeStartIsContained() + { + using var server = new TestServer(); + var request = new Mock(); + request + .Setup(value => value.OperationCompleted(It.IsAny(), It.IsAny())) + .Throws(new InvalidOperationException("completion failed")); + + Assert.DoesNotThrow(() => server.ScheduleIncomingRequest(request.Object)); + request.Verify( + value => value.OperationCompleted( + null, It.Is(error => error.StatusCode == StatusCodes.BadServerHalted)), + Times.Once); + } + + /// + /// The queue's worker counts are not applied to the process-wide ThreadPool. + /// + [Test] + public void QueueLeavesThreadPoolLimitsUnchanged() + { + ThreadPool.GetMinThreads(out int minWorkers, out int minIo); + ThreadPool.GetMaxThreads(out int maxWorkers, out int maxIo); + using var server = new TestServer(); + + server.ReplaceRequestQueue(minWorkers + 1, minWorkers + 2, 10); + + ThreadPool.GetMinThreads(out int minWorkersAfter, out int minIoAfter); + ThreadPool.GetMaxThreads(out int maxWorkersAfter, out int maxIoAfter); + Assert.That(minWorkersAfter, Is.EqualTo(minWorkers)); + Assert.That(minIoAfter, Is.EqualTo(minIo)); + Assert.That(maxWorkersAfter, Is.EqualTo(maxWorkers)); + Assert.That(maxIoAfter, Is.EqualTo(maxIo)); + } + [Test] public async Task TestRequestProcessingAsync() { @@ -601,6 +687,196 @@ namespace Opc.Ua.Core.Tests.Stack.Server } } + /// + /// The shared queue reports queued and executing requests, rejections by status code + /// and the time a request waited for a worker. + /// + [Test] + public async Task SharedQueueReportsQueueMetricsAsync() + { + using var server = new TestServer(); + server.ReplaceRequestQueue(1, 1, 1); + using var metrics = new QueueMetricsListener(server.QueueMeter); + var running = new TestEndpointIncomingRequest(); + var queued = new TestEndpointIncomingRequest(); + var rejected = new TestEndpointIncomingRequest(); + + server.ScheduleIncomingRequest(running); + await AwaitOrFailAsync(running.ProcessingStarted.Task, "the first request to start").ConfigureAwait(false); + server.ScheduleIncomingRequest(queued); + server.ScheduleIncomingRequest(rejected); + + Assert.That(rejected.CompletedStatusCode, Is.EqualTo(StatusCodes.BadServerTooBusy)); + Assert.That(metrics.Observe("opcua.server.request_queue.queued"), Is.EqualTo(1)); + Assert.That(metrics.Observe("opcua.server.request_queue.executing"), Is.EqualTo(1)); + Assert.That(metrics.Observe("opcua.server.request_queue.parked"), Is.Zero); + Assert.That(metrics.Rejections("BadServerTooBusy"), Is.EqualTo(1)); + + await Task.Delay(20).ConfigureAwait(false); + running.ProcessingCompleted.TrySetResult(true); + await AwaitOrFailAsync(queued.ProcessingStarted.Task, "the queued request to start").ConfigureAwait(false); + + double[] waits = metrics.TimesInQueue; + Assert.That(waits, Has.Length.EqualTo(2), "both executed requests were measured"); + Assert.That(waits[1], Is.GreaterThanOrEqualTo(0.015)); + Assert.That(metrics.Observe("opcua.server.request_queue.queued"), Is.Zero); + queued.ProcessingCompleted.TrySetResult(true); + } + + /// + /// A parked request leaves the executing gauge and is reported as parked. + /// + [Test] + public async Task ParkedRequestIsReportedAsParkedAsync() + { + using var server = new TestServer(); + server.ReplaceRequestQueue(1, 1, 10, decoupleHeldPublishRequests: true); + using var metrics = new QueueMetricsListener(server.QueueMeter); + var parked = new ParkableTestRequest(); + + server.ScheduleIncomingRequest(parked); + await AwaitOrFailAsync(parked.ProcessingStarted.Task, "the request to park").ConfigureAwait(false); + + await AwaitOrFailAsync( + WaitUntilAsync(() => metrics.Observe("opcua.server.request_queue.parked") == 1), + "the parked gauge").ConfigureAwait(false); + Assert.That(metrics.Observe("opcua.server.request_queue.executing"), Is.Zero); + + parked.ProcessingCompleted.TrySetResult(true); + await AwaitOrFailAsync( + WaitUntilAsync(() => metrics.Observe("opcua.server.request_queue.parked") == 0), + "the parked request to finish").ConfigureAwait(false); + } + + /// + /// The isolation path reports the same instruments, including rejections from fair admission. + /// + [Test] + public async Task FairQueueReportsQueueMetricsAsync() + { + var provider = new FairRequestQueueTests.TestProvider(executionCapacity: 1); + using var server = new TestServer(); + server.ReplaceRequestQueue(1, 1, 10, provider: provider); + using var metrics = new QueueMetricsListener(server.QueueMeter); + var running = new FairTestRequest(); + var queued = new FairTestRequest(); + var cancelled = new FairTestRequest(); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + + server.ScheduleIncomingRequest(running); + await AwaitOrFailAsync(running.ProcessingStarted.Task, "the first request to start").ConfigureAwait(false); + server.ScheduleIncomingRequest(queued); + server.ScheduleIncomingRequest(cancelled, cancellation.Token); + + Assert.That(cancelled.CompletedStatusCode, Is.EqualTo(StatusCodes.BadRequestCancelledByClient)); + Assert.That(metrics.Rejections("BadRequestCancelledByClient"), Is.EqualTo(1)); + Assert.That(metrics.Observe("opcua.server.request_queue.queued"), Is.EqualTo(1)); + Assert.That(metrics.Observe("opcua.server.request_queue.executing"), Is.EqualTo(1)); + + running.ProcessingCompleted.TrySetResult(true); + await AwaitOrFailAsync(queued.ProcessingStarted.Task, "the queued request to start").ConfigureAwait(false); + Assert.That(metrics.TimesInQueue, Has.Length.EqualTo(2)); + queued.ProcessingCompleted.TrySetResult(true); + } + + /// + /// A request the isolation provider can classify. + /// + private sealed class FairTestRequest : IEndpointIncomingRequest + { + public TaskCompletionSource ProcessingStarted { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource ProcessingCompleted { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + public StatusCode? CompletedStatusCode { get; private set; } + + public IServiceRequest Request { get; } = new ReadRequest { RequestHeader = new RequestHeader() }; + + public SecureChannelContext SecureChannelContext { get; } = + new("channel", null, RequestEncoding.Binary); + + public ValueTask CallAsync(CancellationToken cancellationToken = default) + { + ProcessingStarted.TrySetResult(true); + return new ValueTask(ProcessingCompleted.Task); + } + + public void OperationCompleted(IServiceResponse? response, ServiceResult error) + { + CompletedStatusCode = error?.StatusCode; + ProcessingCompleted.TrySetResult(true); + } + } + + /// + /// Listens to the instruments of one request queue meter. + /// + private sealed class QueueMetricsListener : IDisposable + { + public QueueMetricsListener(Meter meter) + { + m_listener = new MeterListener + { + InstrumentPublished = (instrument, listener) => + { + if (ReferenceEquals(instrument.Meter, meter)) + { + listener.EnableMeasurementEvents(instrument); + } + } + }; + m_listener.SetMeasurementEventCallback((instrument, value, tags, _) => + m_observed[instrument.Name] = value); + m_listener.SetMeasurementEventCallback((instrument, value, tags, _) => + { + string status = string.Empty; + foreach (KeyValuePair tag in tags) + { + if (tag.Key == "status") + { + status = (string)tag.Value!; + } + } + m_rejections.AddOrUpdate(status, value, (_, total) => total + value); + }); + m_listener.SetMeasurementEventCallback((instrument, value, tags, _) => + m_timesInQueue.Enqueue(value)); + m_listener.Start(); + } + + public double[] TimesInQueue => [.. m_timesInQueue]; + + public long Rejections(string status) + { + return m_rejections.TryGetValue(status, out long count) ? count : 0; + } + + public int Observe(string name) + { + m_listener.RecordObservableInstruments(); + return m_observed.TryGetValue(name, out int value) ? value : -1; + } + + public void Dispose() + { + m_listener.Dispose(); + } + + private readonly MeterListener m_listener; + private readonly ConcurrentDictionary m_observed = new(); + private readonly ConcurrentDictionary m_rejections = new(); + private readonly ConcurrentQueue m_timesInQueue = new(); + } + + private static async Task WaitUntilAsync(Func condition) + { + while (!condition()) + { + await Task.Delay(10).ConfigureAwait(false); + } + } + private static async Task AwaitOrFailAsync(Task task, string what, int timeoutMs = 5000) { using var cts = new CancellationTokenSource(); diff --git a/tests/TestThreadPoolSetup.cs b/tests/TestThreadPoolSetup.cs new file mode 100644 index 0000000000..84a5459670 --- /dev/null +++ b/tests/TestThreadPoolSetup.cs @@ -0,0 +1,59 @@ +/* ======================================================================== + * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. + * + * OPC Foundation MIT License 1.00 + * + * Permission is hereby granted, free of charge, to any person + * obtaining a copy of this software and associated documentation + * files (the "Software"), to deal in the Software without + * restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following + * conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES + * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR + * OTHER DEALINGS IN THE SOFTWARE. + * + * The complete license agreement can be found here: + * http://opcfoundation.org/License/MIT/1.00/ + * ======================================================================*/ + +using System; +using System.Threading; +using NUnit.Framework; + +#pragma warning disable CA1050 // A SetUpFixture outside any namespace applies to the whole test assembly. + +/// +/// Raises the thread pool minimum of the test host for every test assembly. +/// +/// +/// The server no longer changes the process-wide thread pool. Up to 2.0.1 every +/// ServerBase raised the minimum to at least its MinRequestThreadCount +/// (10 by default), and test hosts on small CI runners depend on that minimum: tests +/// running in parallel block pool threads, and a queued request could otherwise wait +/// seconds for the thread pool to inject a thread. +/// +[SetUpFixture] +public sealed class TestThreadPoolSetup +{ + private const int kMinThreads = 10; + + [OneTimeSetUp] + public void RaiseThreadPoolMinimum() + { + ThreadPool.GetMinThreads(out int workerThreads, out int completionPortThreads); + ThreadPool.SetMinThreads( + Math.Max(workerThreads, kMinThreads), + Math.Max(completionPortThreads, kMinThreads)); + } +}