mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-29 15:11:53 +08:00
Server: leave the ThreadPool alone and add request queue metrics (#4917)
Q4 and Q5 of the request-queue comparison with ASP.NET Core/Kestrel (Q1 is #4914). Two commits, one per item. ## Q4: stop changing the process ThreadPool `RequestQueue` called `ThreadPool.SetMinThreads` / `SetMaxThreads` for the whole process every time a queue was created, including a throwaway 10-worker queue that the `ServerBase` constructor built and `InitializeRequestQueue` disposed again at start. - The calls are removed. Nothing in the repo (src, tests, samples) depends on them. Queue workers are asynchronous and do not hold a pool thread while they wait. `SetMaxThreads` never had an effect, because the default maximum (32767) is always above the configured counts. - The constructor no longer builds the throwaway queue. Until the server starts, `ServerBase.ScheduleIncomingRequest` completes a request with `BadServerHalted`. That request used to reach `ProcessRequestAsync` of a server that had not started yet. ### Release note (behaviour change) > The server no longer raises the process-wide thread pool minimum to `MinRequestThreadCount` (default 10). On machines with fewer logical processors than `MinRequestThreadCount`, the thread pool now starts with its own default minimum (the processor count). Hosts whose node managers block threads synchronously, and that relied on the old minimum, can set it themselves with `ThreadPool.SetMinThreads` or the `System.Threading.ThreadPool.MinThreads` runtime setting. ### Test hosts CI showed that test hosts depended on the old minimum. On the small linux runners, tests that block pool threads in parallel left a queued request waiting more than 5 s for a thread (`RequestQueueTests`, `RequestLifetimeTests`). `tests/TestThreadPoolSetup.cs` is an assembly-wide NUnit `SetUpFixture` that `tests/Directory.Build.targets` compiles into every NUnit test project. It restores the minimum of at least 10 worker and I/O threads, for the test hosts only. ## Q5: request queue metrics The SharedOnly path only logged `RequestQueueFull` (with a `// TODO: make a metric`). The queue now publishes instruments in both the shared and the isolation mode. They follow the style of `DefaultServerResourceIsolationProvider` and are modelled on Kestrel's `kestrel.queued_requests`, `http.server.active_requests` and `aspnetcore.rate_limiting.request.time_in_queue`. The meter comes from the server's `ITelemetryContext` (`CreateMeter()`, named after `Opc.Ua.Core`) and is disposed once shutdown has drained the queue. | Instrument | Type | |---|---| | `opcua.server.request_queue.queued` | observable gauge | | `opcua.server.request_queue.executing` | observable gauge | | `opcua.server.request_queue.parked` | observable gauge | | `opcua.server.request_queue.time_in_queue` | histogram, seconds | | `opcua.server.request_queue.rejections` | counter, tag `status` | Cost on the request path: - The gauges are observed from counters that already exist, so they add no work per request. - The time in queue uses a `Stopwatch` timestamp taken at enqueue, and only while the histogram has a listener, on both paths. The fair queue stamps an `Entry` only after its admission succeeds. Once the Q2 enqueue/received timestamp lands, it can replace this one. - Rejections are tagged with the interned symbolic id of the status code through the single-tag `Add` overload, so no tag array is allocated. - Rejections include queue full, halted, cancelled while queued, failed revalidation and every fair-admission rejection, because the fair queue's completion callback now goes through `RejectRequest`. Handler failures are not rejections. `docs/ServerScalability.md` lists the instruments. ## Tests - `RequestQueueTests`: request before start is halted, ThreadPool limits unchanged, shared queue metrics (queued, executing, parked, rejections by status, time in queue), parked gauge, fair queue metrics. - `Opc.Ua.Core.Tests` `Stack.Server`: 1946 passed, 0 failed (net10.0). - `Opc.Ua.Server.Tests`: 9653 passed. The 8 failures are symlink tests (`FileLinksRejectReadWriteAndCopy` and similar) that fail on this machine with "insufficient privileges" and are unrelated to this change. - net48 build of `Opc.Ua.Core.Tests` compiles. Conflicts to expect: the Q2/Q3 session (time in queue against TimeoutHint, retry-after) touches the dequeue and reject spots of `ServerBase.RequestQueue.cs`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Marc Schier <marcschier@users.noreply.github.com> Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: marcschier <11168470+marcschier@users.noreply.github.com>
This commit is contained in:
romanett
co-authored by
Claude Opus 5.5
Marc Schier
copilot-swe-agent[bot]
marcschier
Claude Opus 5.5
Marc Schier
copilot-swe-agent[bot]
marcschier
parent
c121cefde8
commit
70e9873328
+2
-1
@@ -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`.
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -71,7 +71,8 @@ namespace Opc.Ua
|
||||
int maxQueuedRequests,
|
||||
long requestCost,
|
||||
bool decoupleHeldPublishRequests,
|
||||
Action<IEndpointIncomingRequest, StatusCode> complete)
|
||||
Action<IEndpointIncomingRequest, StatusCode> complete,
|
||||
Func<long>? 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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the <see cref="Stopwatch"/> time the request was admitted, or 0
|
||||
/// when it was not measured. Set before the entry is linked.
|
||||
/// </summary>
|
||||
internal long EnqueueTimestamp { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Retained queue slot.
|
||||
/// </summary>
|
||||
@@ -1618,6 +1627,12 @@ namespace Opc.Ua
|
||||
/// </summary>
|
||||
private readonly Action<IEndpointIncomingRequest, StatusCode> m_complete;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private readonly Func<long>? m_enqueueTimestamp;
|
||||
|
||||
/// <summary>
|
||||
/// Scheduling FIFOs retained while nonempty, keyed by owner in weighted mode
|
||||
/// or by the empty string for global FIFO mode.
|
||||
|
||||
@@ -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<long>(
|
||||
"opcua.server.request_queue.rejections",
|
||||
description: "Requests the request queue completed without running them, by status code.");
|
||||
m_timeInQueue = m_meter.CreateHistogram<double>(
|
||||
"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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Takes the enqueue timestamp for the time-in-queue histogram, or 0 when
|
||||
/// nothing listens to it.
|
||||
/// </summary>
|
||||
private long GetEnqueueTimestamp()
|
||||
{
|
||||
return m_timeInQueue.Enabled ? Stopwatch.GetTimestamp() : 0;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Records how long a dequeued request waited, if its enqueue time was taken.
|
||||
/// </summary>
|
||||
private void RecordTimeInQueue(long enqueueTimestamp)
|
||||
{
|
||||
if (enqueueTimestamp != 0 && m_timeInQueue.Enabled)
|
||||
{
|
||||
m_timeInQueue.Record(
|
||||
(double)(Stopwatch.GetTimestamp() - enqueueTimestamp) / Stopwatch.Frequency);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts shared asynchronous disposal without blocking the caller.
|
||||
/// </summary>
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -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
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Couples a retained request with its caller-controlled lifetime.
|
||||
/// Counts a request the queue completes without running it, then completes it.
|
||||
/// </summary>
|
||||
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<string, object?>("status", status ?? "Bad"));
|
||||
}
|
||||
CompleteRequest(request, statusCode);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Couples a retained request with its caller-controlled lifetime and the
|
||||
/// <see cref="Stopwatch"/> time it was queued (0 when not measured).
|
||||
/// </summary>
|
||||
private readonly record struct QueuedRequest(
|
||||
IEndpointIncomingRequest Request,
|
||||
long EnqueueTimestamp,
|
||||
CancellationToken CancellationToken);
|
||||
|
||||
/// <summary>
|
||||
@@ -554,7 +609,23 @@ namespace Opc.Ua
|
||||
/// Worker cancellation source retained until shutdown has drained all handlers.
|
||||
/// </summary>
|
||||
private readonly CancellationTokenSource m_cts;
|
||||
|
||||
/// <summary>
|
||||
/// Request queue instruments, disposed once shutdown has drained all handlers.
|
||||
/// </summary>
|
||||
private readonly Meter m_meter;
|
||||
#pragma warning restore CA2213
|
||||
|
||||
/// <summary>
|
||||
/// Requests completed by the queue without running, tagged by status code.
|
||||
/// </summary>
|
||||
private readonly Counter<long> m_rejections;
|
||||
|
||||
/// <summary>
|
||||
/// Time from enqueue to dequeue, in seconds.
|
||||
/// </summary>
|
||||
private readonly Histogram<double> m_timeInQueue;
|
||||
|
||||
/// <summary>
|
||||
/// Worker tasks joined by asynchronous shutdown.
|
||||
/// </summary>
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -17,4 +17,16 @@
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" GeneratePathProperty="true" />
|
||||
</ItemGroup>
|
||||
|
||||
<!--
|
||||
Every NUnit test assembly gets TestThreadPoolSetup.cs, an assembly-wide SetUpFixture that raises
|
||||
the test host's thread pool minimum, which the server no longer changes for the process.
|
||||
-->
|
||||
<Target Name="_AddTestThreadPoolSetup"
|
||||
BeforeTargets="CoreCompile"
|
||||
Condition="'$(RestrictForLegacyTfm)' != 'true'">
|
||||
<ItemGroup Condition="'@(PackageReference->WithMetadataValue('Identity', 'NUnit'))' != ''">
|
||||
<Compile Include="$(MSBuildThisFileDirectory)TestThreadPoolSetup.cs" />
|
||||
</ItemGroup>
|
||||
</Target>
|
||||
|
||||
</Project>
|
||||
|
||||
@@ -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<KeepAliveEventArgs>();
|
||||
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));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the meter of the installed request queue.
|
||||
/// </summary>
|
||||
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)!;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Replaces the default queue with the requested worker and admission envelope.
|
||||
/// </summary>
|
||||
@@ -276,6 +295,73 @@ namespace Opc.Ua.Core.Tests.Stack.Server
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Before start a null request is rejected like the queue rejects it.
|
||||
/// </summary>
|
||||
[Test]
|
||||
public void NullRequestBeforeStartThrows()
|
||||
{
|
||||
using var server = new TestServer();
|
||||
|
||||
Assert.Throws<ArgumentNullException>(() => server.ScheduleIncomingRequest(null!));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A completion callback that throws while a request is halted before start does
|
||||
/// not escape to the transport that scheduled the request.
|
||||
/// </summary>
|
||||
[Test]
|
||||
public void FaultingCompletionBeforeStartIsContained()
|
||||
{
|
||||
using var server = new TestServer();
|
||||
var request = new Mock<IEndpointIncomingRequest>();
|
||||
request
|
||||
.Setup(value => value.OperationCompleted(It.IsAny<IServiceResponse?>(), It.IsAny<ServiceResult>()))
|
||||
.Throws(new InvalidOperationException("completion failed"));
|
||||
|
||||
Assert.DoesNotThrow(() => server.ScheduleIncomingRequest(request.Object));
|
||||
request.Verify(
|
||||
value => value.OperationCompleted(
|
||||
null, It.Is<ServiceResult>(error => error.StatusCode == StatusCodes.BadServerHalted)),
|
||||
Times.Once);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The queue's worker counts are not applied to the process-wide ThreadPool.
|
||||
/// </summary>
|
||||
[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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The shared queue reports queued and executing requests, rejections by status code
|
||||
/// and the time a request waited for a worker.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A parked request leaves the executing gauge and is reported as parked.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The isolation path reports the same instruments, including rejections from fair admission.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A request the isolation provider can classify.
|
||||
/// </summary>
|
||||
private sealed class FairTestRequest : IEndpointIncomingRequest
|
||||
{
|
||||
public TaskCompletionSource<bool> ProcessingStarted { get; } =
|
||||
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
public TaskCompletionSource<bool> 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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Listens to the instruments of one request queue meter.
|
||||
/// </summary>
|
||||
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<int>((instrument, value, tags, _) =>
|
||||
m_observed[instrument.Name] = value);
|
||||
m_listener.SetMeasurementEventCallback<long>((instrument, value, tags, _) =>
|
||||
{
|
||||
string status = string.Empty;
|
||||
foreach (KeyValuePair<string, object?> tag in tags)
|
||||
{
|
||||
if (tag.Key == "status")
|
||||
{
|
||||
status = (string)tag.Value!;
|
||||
}
|
||||
}
|
||||
m_rejections.AddOrUpdate(status, value, (_, total) => total + value);
|
||||
});
|
||||
m_listener.SetMeasurementEventCallback<double>((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<string, int> m_observed = new();
|
||||
private readonly ConcurrentDictionary<string, long> m_rejections = new();
|
||||
private readonly ConcurrentQueue<double> m_timesInQueue = new();
|
||||
}
|
||||
|
||||
private static async Task WaitUntilAsync(Func<bool> 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();
|
||||
|
||||
@@ -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.
|
||||
|
||||
/// <summary>
|
||||
/// Raises the thread pool minimum of the test host for every test assembly.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The server no longer changes the process-wide thread pool. Up to 2.0.1 every
|
||||
/// <c>ServerBase</c> raised the minimum to at least its <c>MinRequestThreadCount</c>
|
||||
/// (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.
|
||||
/// </remarks>
|
||||
[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));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user