mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-29 07:07:24 +08:00
Server: run SharedOnly requests on concurrency permits instead of queue workers (#4919)
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<QueuedRequest>` 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 <noreply@anthropic.com>
This commit is contained in:
romanett
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
70e9873328
commit
8b1cc6b7e3
@@ -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.
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
|
||||
@@ -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 <file>` | | 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). |
|
||||
|
||||
@@ -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<string, string>(StringComparer.Ordinal) { ["DOTNET_gcServer"] = "1" });
|
||||
string ready = handle.m_lines.Take();
|
||||
|
||||
@@ -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<string> 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<string> serverLines = startedLines;
|
||||
|
||||
@@ -1017,21 +1017,33 @@ namespace Opc.Ua
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The minimum number of threads assigned to processing requests.
|
||||
/// The lower bound of the request concurrency limit; values below 1 are raised to 1.
|
||||
/// </summary>
|
||||
/// <value>The minimum request thread count.</value>
|
||||
/// <remarks>
|
||||
/// Requests do not run on dedicated threads. With a resource isolation profile
|
||||
/// other than SharedOnly, this many request workers are started with the server.
|
||||
/// <see cref="MaxRequestThreadCount"/> is raised to this value.
|
||||
/// </remarks>
|
||||
/// <value>The minimum request concurrency.</value>
|
||||
[DataTypeField(Order = 3)]
|
||||
public int MinRequestThreadCount { get; set; } = 10;
|
||||
|
||||
/// <summary>
|
||||
/// The maximum number of threads assigned to processing requests.
|
||||
/// The maximum number of requests that execute at once (the request concurrency limit).
|
||||
/// </summary>
|
||||
/// <value>The maximum request thread count.</value>
|
||||
/// <remarks>
|
||||
/// Requests run on the .NET thread pool, not on dedicated threads. A request that
|
||||
/// finds the limit reached waits in the request queue (see
|
||||
/// <see cref="MaxQueuedRequestCount"/>). With <see cref="DecoupleHeldPublishRequests"/>
|
||||
/// a held request, such as a <c>Publish</c> waiting for notifications, does not
|
||||
/// count against the limit while it waits.
|
||||
/// </remarks>
|
||||
/// <value>The maximum request concurrency.</value>
|
||||
[DataTypeField(Order = 4)]
|
||||
public int MaxRequestThreadCount { get; set; } = 100;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <value>The maximum queued request count.</value>
|
||||
[DataTypeField(Order = 5)]
|
||||
|
||||
@@ -40,22 +40,34 @@ namespace Opc.Ua
|
||||
public partial class ServerBase
|
||||
{
|
||||
/// <summary>
|
||||
/// Asynchronously manages a queue of requests.
|
||||
/// Limits how many requests execute at once and queues the requests that wait.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Without an isolation provider a request takes one of
|
||||
/// <c>maxThreadCount</c> 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
|
||||
/// <see cref="FairRequestQueue"/>.
|
||||
/// </remarks>
|
||||
protected class RequestQueue : IDisposable
|
||||
{
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="RequestQueue"/> class.
|
||||
/// </summary>
|
||||
/// <param name="server">The server.</param>
|
||||
/// <param name="minThreadCount">The minimum number of threads in the pool.</param>
|
||||
/// <param name="maxThreadCount">The maximum number of threads in the pool.</param>
|
||||
/// <param name="minThreadCount">
|
||||
/// The number of fair-queue workers started up front. Without an isolation
|
||||
/// provider no workers are started, and it only has to be positive.
|
||||
/// </param>
|
||||
/// <param name="maxThreadCount">
|
||||
/// The maximum number of requests executing at once.
|
||||
/// </param>
|
||||
/// <param name="maxRequestCount">The maximum number of requests that will placed in the queue.</param>
|
||||
/// <param name="decoupleHeldPublishRequests">
|
||||
/// When <c>true</c> (the default) a request that parks (for example a held
|
||||
/// <c>Publish</c> 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.
|
||||
/// <c>Publish</c> 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.
|
||||
/// </param>
|
||||
/// <param name="resourceIsolationProvider">
|
||||
/// 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<QueuedRequest>(options);
|
||||
m_maxRequestCount = maxRequestCount;
|
||||
m_waiting = new Queue<QueuedRequest>();
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs requests while this permit is handed from one request to the next
|
||||
/// waiting one, then returns the permit.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Never throws: <see cref="ProcessRequestSafeAsync"/> completes every request.
|
||||
/// </remarks>
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts shared asynchronous disposal without blocking the caller.
|
||||
/// </summary>
|
||||
@@ -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<bool>(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
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Ran by the worker threads to process requests.
|
||||
/// Ran by the fair-queue workers to process requests.
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
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.
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Counts a request that released its permit or worker at the park point as
|
||||
/// parked until its handler finishes.
|
||||
/// </summary>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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);
|
||||
|
||||
/// <summary>
|
||||
/// Thread pool work item that runs a request holding a permit.
|
||||
/// </summary>
|
||||
private sealed class DispatchItem(RequestQueue queue, QueuedRequest queued)
|
||||
#if NETCOREAPP3_0_OR_GREATER
|
||||
: IThreadPoolWorkItem
|
||||
#endif
|
||||
{
|
||||
/// <summary>
|
||||
/// Runs the request; the returned task never faults.
|
||||
/// </summary>
|
||||
public void Execute()
|
||||
{
|
||||
_ = queue.RunWithPermitAsync(queued);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Server that dispatches admitted requests.
|
||||
/// </summary>
|
||||
private readonly ServerBase m_server;
|
||||
|
||||
/// <summary>
|
||||
/// Number of workers started before any requests arrive.
|
||||
/// Number of fair-queue workers started before any requests arrive.
|
||||
/// </summary>
|
||||
private readonly int m_minThreadCount;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private readonly int m_maxThreadCount;
|
||||
|
||||
/// <summary>
|
||||
/// Whether parked requests release their execution workers.
|
||||
/// Maximum number of requests waiting for a permit.
|
||||
/// </summary>
|
||||
private readonly int m_maxRequestCount;
|
||||
|
||||
/// <summary>
|
||||
/// Whether parked requests release their permit or worker.
|
||||
/// </summary>
|
||||
private readonly bool m_decoupleHeldPublishRequests;
|
||||
|
||||
/// <summary>
|
||||
/// Compatibility FIFO used when no isolation provider is installed.
|
||||
/// Requests waiting for a permit when no isolation provider is installed,
|
||||
/// guarded by <see cref="m_workerGate"/>.
|
||||
/// </summary>
|
||||
private readonly System.Threading.Channels.Channel<QueuedRequest> m_queue;
|
||||
private readonly Queue<QueuedRequest> m_waiting;
|
||||
|
||||
/// <summary>
|
||||
/// Number of requests in <see cref="m_waiting"/>, changed under
|
||||
/// <see cref="m_workerGate"/> and read without it by the queued gauge.
|
||||
/// </summary>
|
||||
private int m_waitingCount;
|
||||
|
||||
/// <summary>
|
||||
/// Token of <see cref="m_cts"/>, read without touching the source, which
|
||||
/// shutdown disposes once every permit and handler has finished.
|
||||
/// </summary>
|
||||
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;
|
||||
|
||||
/// <summary>
|
||||
/// Number of workers currently executing rather than parked or idle.
|
||||
/// Number of handlers that parked and released their permit or worker,
|
||||
/// changed with <see cref="Interlocked"/>.
|
||||
/// </summary>
|
||||
private int m_parkedCount;
|
||||
|
||||
/// <summary>
|
||||
/// Number of requests executing rather than parked or waiting: the permits in
|
||||
/// use, changed under <see cref="m_workerGate"/>, or the busy fair-queue
|
||||
/// workers, changed with <see cref="Interlocked"/>.
|
||||
/// </summary>
|
||||
private int m_activeThreadCount;
|
||||
|
||||
|
||||
@@ -668,11 +668,6 @@ namespace Opc.Ua
|
||||
maxRequestThreadCount = minRequestThreadCount;
|
||||
}
|
||||
|
||||
if (maxRequestThreadCount < 100)
|
||||
{
|
||||
maxRequestThreadCount = 100;
|
||||
}
|
||||
|
||||
if (maxQueuedRequestCount < 100)
|
||||
{
|
||||
maxQueuedRequestCount = 100;
|
||||
|
||||
@@ -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.");
|
||||
|
||||
@@ -254,6 +254,22 @@ namespace Opc.Ua.Core.Tests.Stack.Server
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates the queue from configuration, as starting the server does.
|
||||
/// </summary>
|
||||
public void InitializeQueue(ApplicationConfiguration configuration)
|
||||
{
|
||||
InitializeRequestQueue(configuration);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stops the queue, as stopping the server does.
|
||||
/// </summary>
|
||||
public ValueTask StopQueueAsync()
|
||||
{
|
||||
return StopRequestQueueAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Replaces the default queue with the requested worker and admission envelope.
|
||||
/// </summary>
|
||||
@@ -780,6 +796,179 @@ namespace Opc.Ua.Core.Tests.Stack.Server
|
||||
queued.ProcessingCompleted.TrySetResult(true);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Without an isolation provider the queue starts no workers: requests take a
|
||||
/// permit and run on the thread pool.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// At most maxThreadCount requests execute at once; a finished request hands its
|
||||
/// permit to the oldest waiting request.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A configured MaxRequestThreadCount below 100 is used as configured.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stopping rejects waiting requests and completes only after executing and
|
||||
/// parked handlers have finished.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A request that holds a permit but has not started when the queue stops is
|
||||
/// halted, and stop still waits for its permit.
|
||||
/// </summary>
|
||||
[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));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A request the isolation provider can classify.
|
||||
/// </summary>
|
||||
|
||||
+34
-2
@@ -60,6 +60,32 @@ namespace Opc.Ua.Server.Tests
|
||||
.And.Message.Contains("two minutes"));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Too few workers for the reserved requests are named at startup.
|
||||
/// </summary>
|
||||
[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
|
||||
/// <summary>
|
||||
/// Supplies finite totals with room for all protected classes and ordinary reconnect capacity.
|
||||
/// </summary>
|
||||
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 }
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user