diff --git a/src/Opc.Ua.Core/Stack/Bindings/ArrayPoolBufferManagerBase.cs b/src/Opc.Ua.Core/Stack/Bindings/ArrayPoolBufferManagerBase.cs index 9be9df6c50..1186fc6e31 100644 --- a/src/Opc.Ua.Core/Stack/Bindings/ArrayPoolBufferManagerBase.cs +++ b/src/Opc.Ua.Core/Stack/Bindings/ArrayPoolBufferManagerBase.cs @@ -108,18 +108,15 @@ public int GetExpectedBufferSize(int size) } int expectedSize = RoundUpToPoolBucket(checked(size + m_metadataByteCount)); - bool usesSharedPool = m_maxBufferSize <= kSharedPoolThreshold - m_metadataByteCount; -#if NET5_0_OR_GREATER - if (usesSharedPool) + if (m_maxBufferSize <= kSharedPoolThreshold - m_metadataByteCount) { + // the bounded pool allocates the exact bucket when it is exhausted return expectedSize; } -#endif - // Configurable pools, including Shared on .NET Framework, can fall - // back to the next bucket when the requested bucket is exhausted. - int largestBucket = usesSharedPool - ? kSharedPoolThreshold - : RoundUpToPoolBucket(checked(m_maxBufferSize + m_metadataByteCount)); + + // Configurable pools can fall back to the next bucket when the + // requested bucket is exhausted. + int largestBucket = RoundUpToPoolBucket(checked(m_maxBufferSize + m_metadataByteCount)); return expectedSize < largestBucket ? expectedSize * 2 : expectedSize; } @@ -220,13 +217,16 @@ protected virtual void OnBufferReturning(byte[] buffer, string owner) /// The selected array pool. internal static ArrayPool CreateArrayPool(int maxBufferSize, int metadataByteCount) { + int maximumArrayLength = RoundUpToPoolBucket( + checked(maxBufferSize + metadataByteCount)); if (maxBufferSize <= kSharedPoolThreshold - metadataByteCount) { - return ArrayPool.Shared; + // Not ArrayPool.Shared: it keeps a few arrays per processor and + // allocates a new chunk for every response once a deep pipeline on a + // host with few processors has exhausted it. + return BoundedArrayPool.GetShared(maximumArrayLength); } - int maximumArrayLength = RoundUpToPoolBucket( - checked(maxBufferSize + metadataByteCount)); return ArrayPool.Create(maximumArrayLength, 4); } diff --git a/src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs b/src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs new file mode 100644 index 0000000000..d7db0db85e --- /dev/null +++ b/src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs @@ -0,0 +1,231 @@ +/* ======================================================================== + * 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.Buffers; +using System.Collections.Concurrent; +using System.Threading; + +namespace Opc.Ua.Bindings +{ + /// + /// An array pool for the transport's chunk-sized buffers that keeps enough arrays for + /// deep request pipelines, independent of the processor count. + /// + /// + /// keeps a few arrays per bucket and processor. A server + /// with a thousand responses waiting to be sent exhausts it on a host with one or two + /// processors and then allocates a new chunk-sized array for every response. This pool + /// keeps up to arrays per power-of-two bucket in a + /// lock-free queue, so the retained memory is bounded by MaxArraysPerBucket times + /// the sum of the bucket sizes, which is less than + /// 2 * MaxArraysPerBucket * MaxArrayLength. Arrays are only retained once they were + /// returned, so a pool grows to the peak concurrency it has seen, up to that bound. + /// One instance per maximum array length is shared by all buffer managers of the process, + /// so client and server channels do not each retain their own arrays. + /// + internal sealed class BoundedArrayPool : ArrayPool + { + /// + /// Creates a pool for arrays up to bytes. + /// + /// The largest pooled array length, a power of two. + /// The number of arrays retained per bucket. + /// + internal BoundedArrayPool(int maxArrayLength, int maxArraysPerBucket) + { + if (maxArrayLength < kMinimumArrayLength || (maxArrayLength & (maxArrayLength - 1)) != 0) + { + throw new ArgumentOutOfRangeException(nameof(maxArrayLength)); + } + if (maxArraysPerBucket <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maxArraysPerBucket)); + } + + MaxArrayLength = maxArrayLength; + MaxArraysPerBucket = maxArraysPerBucket; + m_buckets = new Bucket[GetBucketIndex(maxArrayLength) + 1]; + for (int ii = 0; ii < m_buckets.Length; ii++) + { + m_buckets[ii] = new Bucket(kMinimumArrayLength << ii); + } + } + + /// + /// The largest pooled array length. Larger requests are allocated and dropped on return. + /// + public int MaxArrayLength { get; } + + /// + /// The number of arrays each bucket retains at most. + /// + public int MaxArraysPerBucket { get; } + + /// + /// Gets the process-wide pool for arrays up to bytes, + /// rounded up to a power of two. + /// + /// The largest array length the caller rents. + public static BoundedArrayPool GetShared(int maxArrayLength) + { + int length = kMinimumArrayLength; + while (length < maxArrayLength) + { + length <<= 1; + } + return s_pools.GetOrAdd( + length, + static key => new BoundedArrayPool(key, GetDefaultMaxArraysPerBucket(key))); + } + + /// + /// Retains at most 64 MiB in the largest bucket, at most 1024 and at least one array: 1024 arrays + /// for the default 64 KiB transport buffer, enough for about a thousand responses in + /// flight (for example 10 channels with 100 outstanding requests each). + /// + /// The largest pooled array length. + internal static int GetDefaultMaxArraysPerBucket(int maxArrayLength) + { + long arrays = kMaxRetainedBytesPerBucket / maxArrayLength; + return (int)Math.Min(kMaxArraysPerBucket, Math.Max(1, arrays)); + } + + /// + /// The number of arrays currently retained by the bucket holding + /// . + /// + /// An array length up to . + internal int GetRetainedCount(int arrayLength) + { + return Volatile.Read(ref m_buckets[GetBucketIndex(arrayLength)].Count); + } + + /// + public override byte[] Rent(int minimumLength) + { + if (minimumLength < 0) + { + throw new ArgumentOutOfRangeException(nameof(minimumLength)); + } + if (minimumLength == 0) + { + return []; + } + if (minimumLength > MaxArrayLength) + { + return new byte[minimumLength]; + } + + Bucket bucket = m_buckets[GetBucketIndex(minimumLength)]; + if (bucket.Arrays.TryDequeue(out byte[]? array)) + { + Interlocked.Decrement(ref bucket.Count); + return array; + } + return new byte[bucket.ArrayLength]; + } + + /// + public override void Return(byte[] array, bool clearArray = false) + { + if (array == null) + { + throw new ArgumentNullException(nameof(array)); + } + if (array.Length == 0 || array.Length > MaxArrayLength) + { + return; + } + + Bucket bucket = m_buckets[GetBucketIndex(array.Length)]; + if (array.Length != bucket.ArrayLength) + { + throw new ArgumentException( + "The buffer is not associated with this pool and may not be returned to it.", + nameof(array)); + } + + // count first, so concurrent returns cannot overshoot the bound + if (Interlocked.Increment(ref bucket.Count) > MaxArraysPerBucket) + { + Interlocked.Decrement(ref bucket.Count); + return; + } + if (clearArray) + { + Array.Clear(array, 0, array.Length); + } + bucket.Arrays.Enqueue(array); + } + + /// + /// Returns the index of the smallest bucket holding bytes. + /// + private static int GetBucketIndex(int length) + { + int index = 0; + int bucketLength = kMinimumArrayLength; + while (bucketLength < length) + { + bucketLength <<= 1; + index++; + } + return index; + } + + /// + /// The retained arrays of one power-of-two length. + /// + private sealed class Bucket(int arrayLength) + { + /// + /// The length of every array in this bucket. + /// + public readonly int ArrayLength = arrayLength; + + /// + /// The retained arrays. + /// + public readonly ConcurrentQueue Arrays = new(); + + /// + /// The retained arrays, including returns about to be enqueued. + /// + public int Count; + } + + private const int kMinimumArrayLength = 16; + private const int kMaxArraysPerBucket = 1024; + private const long kMaxRetainedBytesPerBucket = 64L * 1024 * 1024; + private static readonly ConcurrentDictionary s_pools = new(); + private readonly Bucket[] m_buckets; + } +} diff --git a/src/Opc.Ua.Core/Stack/Server/EndpointBase.EndpointIncomingRequest.cs b/src/Opc.Ua.Core/Stack/Server/EndpointBase.EndpointIncomingRequest.cs index 9ea37371f0..3aac4ef895 100644 --- a/src/Opc.Ua.Core/Stack/Server/EndpointBase.EndpointIncomingRequest.cs +++ b/src/Opc.Ua.Core/Stack/Server/EndpointBase.EndpointIncomingRequest.cs @@ -34,12 +34,26 @@ namespace Opc.Ua { + /// + /// An incoming request whose links the + /// cancellation token it was scheduled with into its request lifetime, so the request queue + /// need not link that token again. + /// + internal interface ILinksScheduledCancellation + { + /// + /// The token the request was scheduled with. + /// + CancellationToken ScheduledCancellationToken { get; } + } + public abstract partial class EndpointBase { /// /// An object that handles an incoming request for an endpoint. /// - protected struct EndpointIncomingRequest : IParkableIncomingRequest, IEquatable + protected struct EndpointIncomingRequest : IParkableIncomingRequest, IEquatable, + ILinksScheduledCancellation { /// /// Initialize the Object with a Request @@ -91,14 +105,12 @@ public ValueTask ProcessAsync(CancellationToken cancellationTo /// public readonly async ValueTask CallAsync(CancellationToken cancellationToken = default) { - using CancellationTokenSource? timeoutHintCts = (int)Request.RequestHeader.TimeoutHint > 0 ? - TimeProvider.System.CreateCancellationTokenSource( - TimeSpan.FromMilliseconds((int)Request.RequestHeader.TimeoutHint)) : null; - + // one token source per request for the caller, the transport and the timeout hint + int timeoutHint = (int)Request.RequestHeader.TimeoutHint; using var requestLifetime = new RequestLifetime( - timeoutHintCts != null ? - [cancellationToken, m_transportCancellationToken, timeoutHintCts.Token] : - [cancellationToken, m_transportCancellationToken]); + timeoutHint > 0 ? TimeSpan.FromMilliseconds(timeoutHint) : TimeSpan.Zero, + cancellationToken, + m_transportCancellationToken); // Flow the park sink so a handler that parks (e.g. a held Publish // waiting for notifications) can release the processing worker. @@ -206,6 +218,10 @@ public readonly bool Equals(EndpointIncomingRequest other) return Request.RequestHeader.Equals(other.Request.RequestHeader); } + /// + readonly CancellationToken ILinksScheduledCancellation.ScheduledCancellationToken => + m_transportCancellationToken; + private readonly EndpointBase m_endpoint; private readonly ServiceResponsePooledValueTaskSource m_vts; diff --git a/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs b/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs index a735900d5f..13e954746a 100644 --- a/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs +++ b/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs @@ -195,6 +195,7 @@ public bool TryEnqueue( m_owners.Add(schedulingKey, ownerQueue); } entry.OwnerQueue = ownerQueue; + entry.Sequence = ++m_sequence; entry.Node = ownerQueue.Requests.AddLast(entry); m_count++; admission.TransferToQueue(); @@ -217,6 +218,12 @@ public bool TryEnqueue( /// A blocked global FIFO head cannot be overtaken by another owner. /// Provider calls and completions are deliberately outside the queue gate. /// + /// + /// The selected head is unlinked under the gate, so a dispatch that is admitted needs no + /// second gate entry, and in weighted mode several workers select at once. A head whose + /// execution admission fails returns to the front of its owner's FIFO. In global FIFO + /// mode one worker selects at a time, so the blocked head keeps its place. + /// public bool TryDequeue([NotNullWhen(true)] out Entry? result) { int attempts = -1; @@ -225,7 +232,6 @@ public bool TryDequeue([NotNullWhen(true)] out Entry? result) IDisposable? executionLease = null; StatusCode error = StatusCodes.Good; bool acquired = false; - bool removed = false; try { if (candidate.CancellationToken.IsCancellationRequested) @@ -248,62 +254,78 @@ public bool TryDequeue([NotNullWhen(true)] out Entry? result) } catch { - lock (m_gate) - { - m_selecting = false; - removed = Remove(candidate); - } - Pulse(); + EndSelection(); executionLease?.Dispose(); - if (removed) - { - candidate.Dispose(); - m_complete(candidate.Request, StatusCodes.BadInternalError); - } + Pulse(); + candidate.Dispose(); + m_complete(candidate.Request, StatusCodes.BadInternalError); throw; } - lock (m_gate) + if (acquired) { - m_selecting = false; - if (candidate.Node != null) + candidate.SetExecutionLease(executionLease!); + executionLease = null; + EndSelection(); + Pulse(); + candidate.ReleaseQueueLease(); + if (Volatile.Read(ref m_stopped)) { - if (acquired || StatusCode.IsBad(error)) - { - removed = Remove(candidate, acquired); - if (acquired) - { - candidate.SetExecutionLease(executionLease!); - executionLease = null; - } - } - else - { - candidate.OwnerQueue!.LastAttempt = version; - Rotate(candidate.OwnerQueue); - } + // stopped while selecting: the request was no longer linked for Dispose + candidate.Dispose(); + m_complete(candidate.Request, StatusCodes.BadServerHalted); + break; + } + if (candidate.CancellationToken.IsCancellationRequested) + { + // cancelled while selecting: the callback found the request unlinked + candidate.Dispose(); + m_complete(candidate.Request, StatusCodes.BadRequestCancelledByClient); + continue; } + result = candidate; + return true; } executionLease?.Dispose(); - if (Volatile.Read(ref m_version) != version) + + if (StatusCode.IsBad(error)) { + EndSelection(); Pulse(); + candidate.Dispose(); + m_complete(candidate.Request, error); + continue; } - if (removed) + // Execution capacity is exhausted: the head returns to the front of its FIFO and + // the owner is not retried before capacity or membership changes. + StatusCode terminal = StatusCodes.Good; + lock (m_gate) { - candidate.ReleaseQueueLease(); - if (StatusCode.IsBad(error)) + m_selecting = false; + if (m_stopped) { - candidate.Dispose(); - m_complete(candidate.Request, error); + terminal = StatusCodes.BadServerHalted; + } + else if (candidate.CancellationToken.IsCancellationRequested) + { + // the cancellation callback found the request unlinked + terminal = StatusCodes.BadRequestCancelledByClient; } else { - result = candidate; - return true; + Relink(candidate, version); } } + if (StatusCode.IsBad(terminal)) + { + candidate.Dispose(); + m_complete(candidate.Request, terminal); + } + if (Volatile.Read(ref m_version) != version) + { + Pulse(); + } } result = null; return false; @@ -428,7 +450,8 @@ private static bool IsControlRequest(IServiceRequest request) } /// - /// Selects one eligible owner head while excluding concurrent provider admission attempts. + /// Selects and unlinks one eligible owner head. In global FIFO mode concurrent selection is + /// excluded, so the head cannot be overtaken while its admission is attempted. /// The first call of a dequeue pass ( negative) snapshots the /// number of scheduling FIFOs under the same gate acquisition; each call consumes one attempt. /// @@ -441,7 +464,7 @@ private bool TrySelect(ref int attempts, [NotNullWhen(true)] out Entry? entry, o { attempts = m_owners.Count; } - if (attempts-- > 0 && !m_stopped && !m_selecting) + if (attempts-- > 0 && !m_stopped && (m_useFairScheduling || !m_selecting)) { int remaining = m_round.Count; while (remaining-- > 0 && m_round.First != null) @@ -450,6 +473,7 @@ private bool TrySelect(ref int attempts, [NotNullWhen(true)] out Entry? entry, o if (owner.LastAttempt != version) { entry = owner.Requests.First!.Value; + Unlink(entry); m_selecting = true; return true; } @@ -461,6 +485,64 @@ private bool TrySelect(ref int attempts, [NotNullWhen(true)] out Entry? entry, o } } + /// + /// Unlinks a selected head under the queue gate and consumes one turn of its owner. + /// The version is not changed: a head that fails admission is relinked and its owner + /// must not be retried before capacity changes. + /// + private void Unlink(Entry entry) + { + OwnerQueue owner = entry.OwnerQueue!; + owner.Requests.Remove(entry.Node!); + entry.Node = null; + entry.OwnerQueue = null; + m_count--; + if (owner.Requests.Count == 0) + { + m_round.Remove(owner.RoundNode!); + m_owners.Remove(m_useFairScheduling ? entry.Owner.Key : string.Empty); + } + else if (--owner.Remaining == 0) + { + Rotate(owner); + } + } + + /// + /// Returns a head whose execution admission failed to the front of its owner's FIFO + /// under the queue gate, and marks the owner as attempted at . + /// + private void Relink(Entry entry, long version) + { + string schedulingKey = m_useFairScheduling ? entry.Owner.Key : string.Empty; + if (!m_owners.TryGetValue(schedulingKey, out OwnerQueue? owner)) + { + owner = new OwnerQueue(m_useFairScheduling ? entry.Owner.Weight : 1); + owner.RoundNode = m_round.AddLast(owner); + m_owners.Add(schedulingKey, owner); + } + entry.OwnerQueue = owner; + // concurrent selections of one owner may fail in any order; keep admission order + LinkedListNode? next = owner.Requests.First; + while (next != null && next.Value.Sequence < entry.Sequence) + { + next = next.Next; + } + entry.Node = next == null ? owner.Requests.AddLast(entry) : owner.Requests.AddBefore(next, entry); + m_count++; + owner.LastAttempt = version; + Rotate(owner); + } + + /// + /// Ends a selection that removed its head for good, outside the queue gate. + /// + private void EndSelection() + { + Changed(); + Volatile.Write(ref m_selecting, false); + } + /// /// Removes a linked request while the caller holds the queue gate. /// @@ -602,6 +684,11 @@ internal Entry( /// internal OwnerQueue? OwnerQueue { get; set; } + /// + /// Gets or sets the admission order, assigned under the queue gate. + /// + internal long Sequence { get; set; } + /// /// Gets or sets the FIFO node; null means no longer queued. /// @@ -886,7 +973,13 @@ public void Dispose() private long m_wakeSignalCount; /// - /// Whether an owner head is currently undergoing provider admission outside the gate. + /// Admission order of the last linked request. + /// + private long m_sequence; + + /// + /// Whether a selected head is undergoing provider admission outside the gate; only global + /// FIFO mode waits for it, weighted mode selects concurrently. /// private bool m_selecting; diff --git a/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs index 91b450ef5e..7c64202c32 100644 --- a/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs +++ b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs @@ -165,18 +165,27 @@ public void ScheduleIncomingRequest( bool serverStopped; bool queueFull = false; - lock (m_workerGate) + if (m_fairQueue != null) { - serverStopped = m_stopped; - if (!serverStopped && m_fairQueue == null) + // The fair queue rejects admissions itself once stopped, so the + // worker gate is not needed on this path. + serverStopped = Volatile.Read(ref m_stopped); + } + else + { + lock (m_workerGate) { - queueFull = !m_queue.Writer.TryWrite(new QueuedRequest(request, cancellationToken)); - if (!queueFull && - m_totalThreadCount < m_maxThreadCount && - m_activeThreadCount >= m_totalThreadCount) + serverStopped = m_stopped; + if (!serverStopped) { - Interlocked.Increment(ref m_totalThreadCount); - m_workers.Add(Task.Run(() => WorkerLoopAsync(m_cts.Token), CancellationToken.None)); + queueFull = !m_queue.Writer.TryWrite(new QueuedRequest(request, cancellationToken)); + if (!queueFull && + m_totalThreadCount < m_maxThreadCount && + m_activeThreadCount >= m_totalThreadCount) + { + Interlocked.Increment(ref m_totalThreadCount); + m_workers.Add(Task.Run(() => WorkerLoopAsync(m_cts.Token), CancellationToken.None)); + } } } } @@ -242,7 +251,11 @@ protected virtual void Dispose(bool disposing) /// private void GrowFairWorkers() { - if (m_fairQueue?.Count > 0) + // The counters are re-checked under the gate; the unlocked pre-check keeps the + // gate off the per-request path while a worker is idle or the pool is full. + if (m_fairQueue?.Count > 0 && + Volatile.Read(ref m_totalThreadCount) < m_maxThreadCount && + Volatile.Read(ref m_activeThreadCount) >= Volatile.Read(ref m_totalThreadCount)) { lock (m_workerGate) { @@ -277,15 +290,20 @@ private async Task StopCoreAsync(Task[] workers) m_server.m_logger.RequestQueueProcessingFailed(ex); } await Task.WhenAll(workers).ConfigureAwait(false); - Task? processing; + Task processing; lock (m_workerGate) { - processing = m_processingDrained?.Task; - } - if (processing != null) - { - await processing.ConfigureAwait(false); + // Parked handlers may still finish after the workers; the last one to + // finish completes the drain, unless none is left already. + m_processingDrained = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + if (Volatile.Read(ref m_processingCount) == 0) + { + m_processingDrained.TrySetResult(true); + } + processing = m_processingDrained.Task; } + await processing.ConfigureAwait(false); m_cts.Dispose(); } @@ -410,14 +428,7 @@ private async Task ProcessRequestSafeAsync( CancellationToken ct) { IEndpointIncomingRequest request = queued.Request; - lock (m_workerGate) - { - if (m_processingCount++ == 0) - { - m_processingDrained = - new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - } - } + Interlocked.Increment(ref m_processingCount); try { if (entry != null) @@ -429,7 +440,11 @@ private async Task ProcessRequestSafeAsync( return; } } - using CancellationTokenSource? linked = queued.CancellationToken.CanBeCanceled ? + // An endpoint request links the token it was scheduled with into its own + // request lifetime; linking it here as well would cost a second token source. + using CancellationTokenSource? linked = queued.CancellationToken.CanBeCanceled && + !(request is ILinksScheduledCancellation scheduled && + scheduled.ScheduledCancellationToken == queued.CancellationToken) ? CancellationTokenSource.CreateLinkedTokenSource(ct, queued.CancellationToken) : null; await m_server.ProcessRequestAsync(request, linked?.Token ?? ct).ConfigureAwait(false); } @@ -449,11 +464,13 @@ private async Task ProcessRequestSafeAsync( finally { entry?.Dispose(); - lock (m_workerGate) + // Only shutdown waits for the drain, and it starts once every worker + // has exited, so the gate is needed only after the stop. + if (Interlocked.Decrement(ref m_processingCount) == 0 && Volatile.Read(ref m_stopped)) { - if (--m_processingCount == 0) + lock (m_workerGate) { - m_processingDrained!.TrySetResult(true); + m_processingDrained?.TrySetResult(true); } } } @@ -539,7 +556,8 @@ private readonly record struct QueuedRequest( private Task? m_stopTask; /// - /// Number of handlers that still retain request accounting. + /// Number of handlers that still retain request accounting, changed with + /// . /// private int m_processingCount; diff --git a/src/Opc.Ua.Server/ResourceIsolation/DefaultServerResourceIsolationProvider.cs b/src/Opc.Ua.Server/ResourceIsolation/DefaultServerResourceIsolationProvider.cs index 60d2b7deba..e184b18bf6 100644 --- a/src/Opc.Ua.Server/ResourceIsolation/DefaultServerResourceIsolationProvider.cs +++ b/src/Opc.Ua.Server/ResourceIsolation/DefaultServerResourceIsolationProvider.cs @@ -28,6 +28,7 @@ * ======================================================================*/ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; using System.Diagnostics.Metrics; @@ -114,16 +115,7 @@ public DefaultServerResourceIsolationProvider( /// /// Number of keys with active leases. Released keys are removed immediately. /// - public int TrackedOwnerCount - { - get - { - lock (m_lock) - { - return m_owners.Count; - } - } - } + public int TrackedOwnerCount => Volatile.Read(ref m_ownerCount); /// public ResourceIsolationOwner ClassifyConnection(IPEndPoint? remoteEndpoint) @@ -323,12 +315,8 @@ public bool TryAcquire( { throw new ArgumentOutOfRangeException(nameof(amount)); } - ResourceIsolationFailureReason reason; - lock (m_lock) - { - ThrowIfDisposed(); - reason = AcquireUnderLock(stage, owner, amount, out lease); - } + ThrowIfDisposed(); + ResourceIsolationFailureReason reason = Acquire(stage, owner, amount, out lease); failure = new ResourceIsolationFailure( reason, reason == ResourceIsolationFailureReason.None ? TimeSpan.Zero : TimeSpan.FromSeconds(1)); if (reason != ResourceIsolationFailureReason.None) @@ -355,10 +343,7 @@ public long GetUsage(ResourceIsolationStage stage) { throw new ArgumentOutOfRangeException(nameof(stage)); } - lock (m_lock) - { - return m_stages[(int)stage].Used; - } + return Interlocked.Read(ref m_stages[(int)stage].Used); } /// @@ -390,19 +375,29 @@ internal static string GetTrustedOwnerKey(string key) private IEnumerable> ObserveUsage() { var values = new Measurement[kStageCount]; - lock (m_lock) + for (int ii = 0; ii < values.Length; ii++) { - for (int ii = 0; ii < values.Length; ii++) - { - values[ii] = new Measurement( - m_stages[ii].Used, - new KeyValuePair("stage", ((ResourceIsolationStage)ii).ToString())); - } + values[ii] = new Measurement( + Interlocked.Read(ref m_stages[ii].Used), + new KeyValuePair("stage", ((ResourceIsolationStage)ii).ToString())); } return values; } - private ResourceIsolationFailureReason AcquireUnderLock( + /// + /// Admits a lease without a server-wide lock when the owner already holds a lease of the + /// same accounting class. Owner-table changes (a new owner, a class appearing or + /// disappearing) take the table lock. + /// + /// + /// Each counter is reserved with a compare-and-swap that never passes its limit, and the + /// counters already reserved are returned when a later one is full. Two acquisitions + /// racing for the last units may therefore both be rejected where a serialized order would + /// admit one; no limit is ever exceeded. + /// A class count of an owner only crosses zero under the table lock, so the owner's + /// exclusive class and the protected table slots stay consistent with its leases. + /// + private ResourceIsolationFailureReason Acquire( ResourceIsolationStage stage, ResourceIsolationOwner owner, long amount, @@ -419,9 +414,76 @@ private ResourceIsolationFailureReason AcquireUnderLock( } int index = (int)stage; StageState state = m_stages[index]; - m_owners.TryGetValue(owner.Key, out OwnerState? accounting); - long ownerUsed = accounting?.Usage[index] ?? 0; - if (amount > owner.GetHardLimit(stage) - ownerUsed) + ResourceIsolationClass accountingClass = identity.OnlyStage.HasValue + ? ResourceIsolationClass.Established : owner.Class; + ResourceIsolationFailureReason reason; + long fromReserved; + + if (m_owners.TryGetValue(owner.Key, out OwnerState? accounting) && + TryAddClassLease(accounting, accountingClass)) + { + // The owner's set of active classes does not change, so neither does its + // exclusive class nor the number of protected table slots. + reason = Reserve(state, accounting, owner, identity, stage, amount, out fromReserved); + if (reason != ResourceIsolationFailureReason.None) + { + ReleaseClassLease(accounting, accountingClass); + return reason; + } + lease = new Lease(this, accounting, stage, owner.Class, accountingClass, amount, fromReserved); + return ResourceIsolationFailureReason.None; + } + + lock (m_tableLock) + { + m_owners.TryGetValue(owner.Key, out accounting); + OwnerState target = accounting ?? new OwnerState(owner.Key); + reason = Reserve(state, target, owner, identity, stage, amount, out fromReserved); + if (reason != ResourceIsolationFailureReason.None) + { + return reason; + } + ResourceIsolationClass? nextExclusiveClass = + accounting == null || accounting.ExclusiveClass == accountingClass ? accountingClass : null; + // A mixed-class key cannot fill a protected table slot: its ordinary leases + // may outlive every protected lease without releasing the entry. + int protectedSlots = GetReservedOwnerSlots(accounting?.ExclusiveClass, nextExclusiveClass); + long projectedOwnerCount = (long)m_ownerCount + (accounting == null ? 1 : 0); + if (projectedOwnerCount > Plan.MaxTrackedOwners - protectedSlots) + { + Unreserve(state, target, owner.Class, stage, amount, fromReserved); + return ResourceIsolationFailureReason.OwnerTableFull; + } + if (accounting == null) + { + m_owners[owner.Key] = target; + Volatile.Write(ref m_ownerCount, m_ownerCount + 1); + } + if (Interlocked.Increment(ref target.ClassLeases[(int)accountingClass]) == 1) + { + UpdateExclusiveClassLocked(target); + } + lease = new Lease(this, target, stage, owner.Class, accountingClass, amount, fromReserved); + return ResourceIsolationFailureReason.None; + } + } + + /// + /// Reserves the owner's ceiling, its protected floor and the stage capacity, + /// in the order the rejection reasons are reported. + /// + private static ResourceIsolationFailureReason Reserve( + StageState state, + OwnerState accounting, + ResourceIsolationOwner owner, + OwnerIdentity identity, + ResourceIsolationStage stage, + long amount, + out long fromReserved) + { + fromReserved = 0; + int index = (int)stage; + if (!TryAddBounded(ref accounting.Usage[index], amount, owner.GetHardLimit(stage))) { return ResourceIsolationFailureReason.OwnerLimit; } @@ -433,95 +495,175 @@ private ResourceIsolationFailureReason AcquireUnderLock( ResourceIsolationClass.Trusted => identity.Trusted!.Floors[index], _ => 0 }; - long reservedUsed = owner.Class switch + if (floor > 0) { - ResourceIsolationClass.Bootstrap => state.BootstrapUsed, - ResourceIsolationClass.Reconnect => state.ReconnectUsed, - ResourceIsolationClass.Control => state.ControlUsed, - ResourceIsolationClass.Trusted => accounting?.TrustedUsage[index] ?? 0, - _ => 0 - }; - long fromReserved = Math.Min(amount, floor - reservedUsed); + ref long reservedUsed = ref GetReservedCounter(state, accounting, owner.Class, index); + while (true) + { + long used = Interlocked.Read(ref reservedUsed); + long available = Math.Min(amount, floor - used); + if (available <= 0 || + Interlocked.CompareExchange(ref reservedUsed, used + available, used) == used) + { + fromReserved = Math.Max(available, 0); + break; + } + } + } long fromShared = amount - fromReserved; - if (amount > state.Plan.Capacity - state.Used || - fromShared > state.Plan.SharedCapacity - state.SharedUsed) + bool admitted = TryAddBounded(ref state.Used, amount, state.Plan.Capacity); + if (admitted && fromShared > 0 && + !TryAddBounded(ref state.SharedUsed, fromShared, state.Plan.SharedCapacity)) + { + Interlocked.Add(ref state.Used, -amount); + admitted = false; + } + if (!admitted) { + if (fromReserved > 0) + { + Interlocked.Add( + ref GetReservedCounter(state, accounting, owner.Class, index), -fromReserved); + } + Interlocked.Add(ref accounting.Usage[index], -amount); + fromReserved = 0; return ResourceIsolationFailureReason.Capacity; } - ResourceIsolationClass accountingClass = identity.OnlyStage.HasValue - ? ResourceIsolationClass.Established : owner.Class; - ResourceIsolationClass? nextExclusiveClass = - accounting == null || accounting.ExclusiveClass == accountingClass ? accountingClass : null; - // A mixed-class key cannot fill a protected table slot: its ordinary leases - // may outlive every protected lease without releasing the entry. - int protectedSlots = GetReservedOwnerSlots(accounting?.ExclusiveClass, nextExclusiveClass); - long projectedOwnerCount = (long)m_owners.Count + (accounting == null ? 1 : 0); - if (projectedOwnerCount > Plan.MaxTrackedOwners - protectedSlots) + return ResourceIsolationFailureReason.None; + } + + /// + /// Adds unless the counter would exceed ; + /// the counter never shows more than the limit, not even transiently. + /// + private static bool TryAddBounded(ref long counter, long amount, long limit) + { + while (true) { - return ResourceIsolationFailureReason.OwnerTableFull; + long current = Interlocked.Read(ref counter); + if (amount > limit - current) + { + return false; + } + if (Interlocked.CompareExchange(ref counter, current + amount, current) == current) + { + return true; + } } - if (accounting == null) + } + + /// + /// Returns the counters of a successful reservation. + /// + private static void Unreserve( + StageState state, + OwnerState accounting, + ResourceIsolationClass ownerClass, + ResourceIsolationStage stage, + long amount, + long fromReserved) + { + int index = (int)stage; + if (amount > fromReserved) { - accounting = new OwnerState(owner.Key); - m_owners.Add(owner.Key, accounting); + Interlocked.Add(ref state.SharedUsed, fromReserved - amount); } - var acquired = new Lease(this, accounting, stage, owner.Class, accountingClass, amount, fromReserved); - accounting.Usage[index] += amount; - accounting.ActiveLeases++; - UpdateOwnerClasses(accounting, accountingClass, 1); - state.Used += amount; - state.SharedUsed += fromShared; - switch (owner.Class) + Interlocked.Add(ref state.Used, -amount); + if (fromReserved > 0) + { + Interlocked.Add(ref GetReservedCounter(state, accounting, ownerClass, index), -fromReserved); + } + Interlocked.Add(ref accounting.Usage[index], -amount); + } + + /// + /// The protected-floor counter a class draws from; only called for classes with a floor. + /// + private static ref long GetReservedCounter( + StageState state, + OwnerState accounting, + ResourceIsolationClass ownerClass, + int index) + { + switch (ownerClass) { case ResourceIsolationClass.Bootstrap: - state.BootstrapUsed += fromReserved; - break; + return ref state.BootstrapUsed; case ResourceIsolationClass.Reconnect: - state.ReconnectUsed += fromReserved; - break; - case ResourceIsolationClass.Trusted: - accounting.TrustedUsage[index] += fromReserved; - break; + return ref state.ReconnectUsed; case ResourceIsolationClass.Control: - state.ControlUsed += fromReserved; - break; + return ref state.ControlUsed; + case ResourceIsolationClass.Trusted: + return ref accounting.TrustedUsage[index]; + default: + throw new ArgumentOutOfRangeException(nameof(ownerClass)); } - lease = acquired; - return ResourceIsolationFailureReason.None; } private void Release(Lease lease) { - lock (m_lock) + Unreserve( + m_stages[(int)lease.Stage], lease.Owner, lease.Class, lease.Stage, lease.Amount, lease.Reserved); + ReleaseClassLease(lease.Owner, lease.AccountingClass); + CapacityAvailable?.Invoke(lease.Stage); + } + + /// + /// Adds a lease to a class the owner already holds, without changing its set of classes. + /// + private static bool TryAddClassLease(OwnerState owner, ResourceIsolationClass ownerClass) + { + ref long count = ref owner.ClassLeases[(int)ownerClass]; + while (true) { - int index = (int)lease.Stage; - StageState state = m_stages[index]; - OwnerState owner = lease.Owner; - owner.Usage[index] -= lease.Amount; - state.Used -= lease.Amount; - state.SharedUsed -= lease.Amount - lease.Reserved; - switch (lease.Class) + long current = Interlocked.Read(ref count); + if (current <= 0) { - case ResourceIsolationClass.Bootstrap: - state.BootstrapUsed -= lease.Reserved; - break; - case ResourceIsolationClass.Reconnect: - state.ReconnectUsed -= lease.Reserved; - break; - case ResourceIsolationClass.Trusted: - owner.TrustedUsage[index] -= lease.Reserved; - break; - case ResourceIsolationClass.Control: - state.ControlUsed -= lease.Reserved; - break; + return false; } - UpdateOwnerClasses(owner, lease.AccountingClass, -1); - if (--owner.ActiveLeases == 0) + if (Interlocked.CompareExchange(ref count, current + 1, current) == current) { - m_owners.Remove(owner.Key); + return true; } } - CapacityAvailable?.Invoke(lease.Stage); + } + + /// + /// Removes a lease from an owner class; the last lease of a class is removed under the + /// table lock, which also drops the owner entry once no lease remains. + /// + private void ReleaseClassLease(OwnerState owner, ResourceIsolationClass ownerClass) + { + ref long count = ref owner.ClassLeases[(int)ownerClass]; + while (true) + { + long current = Interlocked.Read(ref count); + if (current <= 1) + { + break; + } + if (Interlocked.CompareExchange(ref count, current - 1, current) == current) + { + return; + } + } + lock (m_tableLock) + { + if (Interlocked.Decrement(ref count) != 0) + { + return; + } + UpdateExclusiveClassLocked(owner); + for (int ii = 0; ii < owner.ClassLeases.Length; ii++) + { + if (Interlocked.Read(ref owner.ClassLeases[ii]) != 0) + { + return; + } + } + m_owners.TryRemove(owner.Key, out _); + Volatile.Write(ref m_ownerCount, m_ownerCount - 1); + } } private int GetReservedOwnerSlots( @@ -557,17 +699,20 @@ private int GetProjectedExclusiveOwnerCount( (next == ownerClass ? 1 : 0); } - private void UpdateOwnerClasses(OwnerState owner, ResourceIsolationClass ownerClass, int change) + /// + /// Recomputes the owner's sole active class after one of its class counts crossed zero. + /// Concurrent lock-free changes never cross zero, so the result is stable under the lock. + /// + private void UpdateExclusiveClassLocked(OwnerState owner) { if (owner.ExclusiveClass is { } previous) { m_exclusiveOwners[(int)previous]--; } - owner.ClassLeases[(int)ownerClass] += change; ResourceIsolationClass? exclusive = null; for (int ii = 0; ii < owner.ClassLeases.Length; ii++) { - if (owner.ClassLeases[ii] == 0) + if (Interlocked.Read(ref owner.ClassLeases[ii]) == 0) { continue; } @@ -670,6 +815,7 @@ private static string Hash(byte[] value) /// /// Tracks total, shared and protected usage for one resource stage. + /// The counters are fields so they can be updated with . /// private sealed class StageState(ResourceIsolationStagePlan plan) { @@ -681,27 +827,27 @@ private sealed class StageState(ResourceIsolationStagePlan plan) /// /// All currently leased capacity. /// - public long Used { get; set; } + public long Used; /// /// Leased capacity drawn from the unreserved pool. /// - public long SharedUsed { get; set; } + public long SharedUsed; /// /// Leased bootstrap floor. /// - public long BootstrapUsed { get; set; } + public long BootstrapUsed; /// /// Leased reconnect floor. /// - public long ReconnectUsed { get; set; } + public long ReconnectUsed; /// /// Leased control-work floor. /// - public long ControlUsed { get; set; } + public long ControlUsed; } /// @@ -715,12 +861,13 @@ private sealed class OwnerState(string key) public string Key { get; } = key; /// - /// The sole active class, or null when leases span classes. + /// The sole active class, or null when leases span classes; changed under the table lock. /// public ResourceIsolationClass? ExclusiveClass { get; set; } /// - /// Counts outstanding leases by class for protected tracking-table capacity. + /// Counts outstanding leases by class for protected tracking-table capacity. A count + /// crosses zero only under the table lock; the entry is dropped when all are zero. /// public long[] ClassLeases { get; } = new long[kClassCount]; @@ -733,11 +880,6 @@ private sealed class OwnerState(string key) /// Amounts drawn from this owner's provisioned floors. /// public long[] TrustedUsage { get; } = new long[kStageCount]; - - /// - /// Keeps the owner entry alive until its last reservation is released. - /// - public long ActiveLeases { get; set; } } /// @@ -958,11 +1100,15 @@ public void Dispose() private DefaultServerResourceIsolationProvider? m_provider = provider; } - private readonly Lock m_lock = new(); + /// + /// Serializes owner-table changes: owner entries, exclusive classes and protected slots. + /// + private readonly Lock m_tableLock = new(); private const int kStageCount = (int)ResourceIsolationStage.ParkedRequest + 1; private const int kClassCount = (int)ResourceIsolationClass.Control + 1; private readonly StageState[] m_stages = new StageState[kStageCount]; - private readonly Dictionary m_owners = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary m_owners = new(StringComparer.Ordinal); + private int m_ownerCount; private readonly ConditionalWeakTable m_classifications = new(); private readonly ISessionBindingProvider? m_sessionBindings; private readonly IResourceIsolationClassifier? m_classifier; diff --git a/src/Opc.Ua.Server/Server/RequestManager.cs b/src/Opc.Ua.Server/Server/RequestManager.cs index 1e95a84e84..c966697614 100644 --- a/src/Opc.Ua.Server/Server/RequestManager.cs +++ b/src/Opc.Ua.Server/Server/RequestManager.cs @@ -28,6 +28,7 @@ * ======================================================================*/ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -63,7 +64,7 @@ public RequestManager(IServerInternal server, TimeProvider? timeProvider) { m_server = server ?? throw new ArgumentNullException(nameof(server)); m_logger = server.Telemetry.CreateLogger(); - m_requests = []; + m_requests = new ConcurrentDictionary(); m_requestTimer = null; m_timeProvider = timeProvider ?? (server as ITimeProviderProvider)?.TimeProvider @@ -100,6 +101,7 @@ protected virtual void Dispose(bool disposing) m_requests.Clear(); requestDrains = [.. m_requestDrains]; m_requestDrains.Clear(); + PublishDrainCountLocked(); m_activeValidationScopes.Clear(); m_lifecycleExtension?.DisposeLocked(); } @@ -157,31 +159,35 @@ public void RequestReceived(OperationContext context) throw new ArgumentNullException(nameof(context)); } - lock (m_requestsLock) + OperationContext existingContext = m_requests.GetOrAdd(context.RequestId, context); + if (!ReferenceEquals(existingContext, context)) + { + throw new InvalidOperationException( + $"A different request with id {context.RequestId} is already active."); + } + m_currentServiceDispatchScope.Value?.RegisterRequest(context.RequestId); + + // Registration takes no lock. A Session close marks the Session before it sweeps the + // registered requests, and a Cancel remembers itself before it scans them; the fence + // orders this registration before the caller checks either mark, so the sweep sees + // the request or the request sees the mark. + Interlocked.MemoryBarrier(); + + if (context.OperationDeadline < DateTime.MaxValue) { - if (m_requests.TryGetValue( - context.RequestId, - out OperationContext? existingContext)) + // Counted before the timer is checked, so a timer that is being stopped for + // lack of deadlines either sees the count change or is seen as stopped here. + Interlocked.Increment(ref m_deadlineRegistrations); + if (Volatile.Read(ref m_requestTimer) == null) { - if (ReferenceEquals(existingContext, context)) + lock (m_requestsLock) { - m_currentServiceDispatchScope.Value? - .RegisterRequest(context.RequestId); - return; + m_requestTimer ??= m_timeProvider.CreateTimer( + OnTimerExpired, + null, + TimeSpan.FromSeconds(1), + TimeSpan.FromSeconds(1)); } - throw new InvalidOperationException( - $"A different request with id {context.RequestId} is already active."); - } - m_requests.Add(context.RequestId, context); - m_currentServiceDispatchScope.Value?.RegisterRequest(context.RequestId); - - if (context.OperationDeadline < DateTime.MaxValue && m_requestTimer == null) - { - m_requestTimer = m_timeProvider.CreateTimer( - OnTimerExpired, - null, - TimeSpan.FromSeconds(1), - TimeSpan.FromSeconds(1)); } } } @@ -207,28 +213,25 @@ private void CompleteRequest(OperationContext context) throw new ArgumentNullException(nameof(context)); } - bool removed; - lock (m_requestsLock) + // remove the request, only when it is this context. + bool removed = ((ICollection>)m_requests).Remove( + new KeyValuePair(context.RequestId, context)); + if (removed) { - // remove the request. - removed = m_requests.TryGetValue( - context.RequestId, - out OperationContext? existingContext) && - ReferenceEquals(existingContext, context) && - m_requests.Remove(context.RequestId); - if (removed) + if (HasRequestDrains()) { - for (int ii = m_requestDrains.Count - 1; ii >= 0; ii--) + lock (m_requestsLock) { - if (m_requestDrains[ii].Complete(context.RequestId)) + for (int ii = m_requestDrains.Count - 1; ii >= 0; ii--) { - m_requestDrains.RemoveAt(ii); + if (m_requestDrains[ii].Complete(context.RequestId)) + { + m_requestDrains.RemoveAt(ii); + } } + PublishDrainCountLocked(); } } - } - if (removed) - { context.RequestLifetime?.MarkCompleted(); } } @@ -271,29 +274,18 @@ internal bool IsExecutingRequest(IOperationContext? context) return false; } - lock (m_requestsLock) - { - return m_requests.TryGetValue( - operation.RequestId, - out OperationContext? existingContext) && - ReferenceEquals(existingContext, operation); - } + return m_requests.TryGetValue( + operation.RequestId, + out OperationContext? existingContext) && + ReferenceEquals(existingContext, operation); } /// /// Gets the optional request lifecycle extension registered for server shutdown and /// NodeManager lifecycle coordination. /// - internal RequestManagerLifecycleExtension? LifecycleExtension - { - get - { - lock (m_requestsLock) - { - return m_lifecycleExtension; - } - } - } + internal RequestManagerLifecycleExtension? LifecycleExtension => + Volatile.Read(ref m_lifecycleExtension); /// /// Registers the optional request lifecycle extension. Calling this method more than once @@ -341,18 +333,12 @@ internal IDisposable EnterValidationScope() { long validationId = Interlocked.Increment( ref m_lastValidationScopeId); - lock (m_requestsLock) - { - m_activeValidationScopes.Add(validationId); - } + m_activeValidationScopes.TryAdd(validationId, 0); - RequestValidationScope? previousScope = m_currentValidationScope.Value; - var scope = new RequestValidationScope( - this, - validationId, - previousScope); - m_currentValidationScope.Value = scope; - return scope; + // A drain that misses this scope started first; the fence orders the registration + // before validation reads the routing tables the drain's caller has already changed. + Interlocked.MemoryBarrier(); + return new RequestValidationScope(this, validationId); } /// @@ -371,9 +357,16 @@ internal IDisposable EnterRequestScope(OperationContext context) } RequestReceived(context); + + // The dispatch scope already carries the id to the lifecycle extension; setting the + // ambient id as well would allocate an execution context to no effect. + if (m_currentServiceDispatchScope.Value?.RequestId == context.RequestId) + { + return new RequestExecutionScope(this, context, null, restoreRequestId: false); + } uint? previousRequestId = m_currentRequestId.Value; m_currentRequestId.Value = context.RequestId; - return new RequestExecutionScope(this, context, previousRequestId); + return new RequestExecutionScope(this, context, previousRequestId, restoreRequestId: true); } /// @@ -455,27 +448,46 @@ internal async ValueTask WaitForCurrentRequestsAsync( lock (m_requestsLock) { m_requestDrains.Remove(requestDrain); + PublishDrainCountLocked(); } } } private RequestDrain? CreateRequestDrainLocked(out TimeSpan budget) { + // Published before the snapshot: a request or validation scope completing without + // the lock either completed before the snapshot, or sees a drain and takes the lock. + Interlocked.Exchange(ref m_drainCount, m_requestDrains.Count + 1); List requestIds = CollectRequestsToAwait(out budget); + List validationIds = [.. m_activeValidationScopes.Keys]; - if (requestIds.Count == 0 && - m_activeValidationScopes.Count == 0) + RequestDrain? requestDrain = null; + if (requestIds.Count != 0 || validationIds.Count != 0) { - return null; + requestDrain = new RequestDrain(requestIds, validationIds); + m_requestDrains.Add(requestDrain); } - - var requestDrain = new RequestDrain( - requestIds, - m_activeValidationScopes); - m_requestDrains.Add(requestDrain); + PublishDrainCountLocked(); return requestDrain; } + /// + /// Publishes the number of registered drains for the lock-free completion paths. + /// + private void PublishDrainCountLocked() + { + Interlocked.Exchange(ref m_drainCount, m_requestDrains.Count); + } + + /// + /// Whether a completion has to update drains; a full fence after the completion's + /// removal, so a drain that is not seen here cannot have seen the request either. + /// + private bool HasRequestDrains() + { + return Interlocked.CompareExchange(ref m_drainCount, 0, 0) != 0; + } + internal uint? GetCurrentRequestIdForLifecycleExtension() { return m_currentServiceDispatchScope.Value?.RequestId ?? @@ -501,6 +513,7 @@ internal void MarkLifecycleWaiterWaiting( { EnsureLifecycleExtensionRegisteredLocked(extension); extension.MarkWaiterWaitingLocked(requestId, m_requestDrains); + PublishDrainCountLocked(); } } @@ -730,24 +743,26 @@ private uint CancelMatchingRequests(NodeId sessionId, uint requestHandle, DateTi // their cancellation callbacks never run while m_requests is being enumerated. lock (m_requestsLock) { - foreach (OperationContext request in m_requests.Values) - { - if (request.SessionId == sessionId && - request.ClientHandle == requestHandle) - { - matchingRequests.Add(request); - } - } - // OPC 10000-4 5.7.5.2: all outstanding requests with the handle are cancelled, // including those still waiting in the server request queue, which are not // registered yet. Remember the cancellation for a while so they are rejected // when they reach validation. A Cancel without a timestamp can never prove // that a queued request was sent before it, so it is not remembered. + // Remembered before the scan: a request registering concurrently is either + // found by the scan or finds the remembered Cancel in IsCancelledBeforeAdmission. if (!sessionId.IsNull && cancelTimestamp != DateTime.MinValue) { AddPendingCancelLocked(sessionId, requestHandle, cancelTimestamp); } + + foreach (OperationContext request in m_requests.Values) + { + if (request.SessionId == sessionId && + request.ClientHandle == requestHandle) + { + matchingRequests.Add(request); + } + } } // flag requests as cancelled. A request that another cancellation (a timeout, a @@ -794,9 +809,9 @@ internal bool IsCancelledBeforeAdmission(OperationContext context) } // Lock-free fast path for the common case that no Cancel is remembered. The request - // was registered under m_requestsLock before this call, and a Cancel scans the - // registered requests and remembers itself under the same lock: either the Cancel - // saw (and cancelled) this request, or its update of the count is visible here. + // was registered (followed by a full fence) before this call, and a Cancel remembers + // itself (with a full fence) before it scans the registered requests: either the + // Cancel saw (and cancelled) this request, or its update of the count is visible here. if (Volatile.Read(ref m_pendingCancelSessionCount) == 0) { return false; @@ -875,7 +890,7 @@ private void AddPendingCancelLocked(NodeId sessionId, uint requestHandle, DateTi } pendingCancels.Add(new PendingCancel(requestHandle, cancelTimestamp, now + PendingCancelWindow)); - Volatile.Write(ref m_pendingCancelSessionCount, m_pendingCancels.Count); + Interlocked.Exchange(ref m_pendingCancelSessionCount, m_pendingCancels.Count); } /// @@ -931,6 +946,8 @@ private void OnTimerExpired(object? state) // cancellation callbacks never run while m_requests is being enumerated. lock (m_requestsLock) { + long deadlineRegistrations = Interlocked.Read(ref m_deadlineRegistrations); + // find the completed request. bool deadlineExists = false; @@ -946,11 +963,19 @@ private void OnTimerExpired(object? state) } } - // check if the timer can be cancelled. + // check if the timer can be cancelled. A request registering a deadline + // concurrently either changed the count or finds the timer stopped and starts it. if (m_requestTimer != null && !deadlineExists) { - m_requestTimer.Dispose(); - m_requestTimer = null; + ITimer timer = Interlocked.Exchange(ref m_requestTimer, null)!; + if (Interlocked.Read(ref m_deadlineRegistrations) != deadlineRegistrations) + { + m_requestTimer = timer; + } + else + { + timer.Dispose(); + } } } @@ -974,11 +999,25 @@ private void OnTimerExpired(object? state) private readonly TimeProvider m_timeProvider; private readonly AsyncLocal m_currentRequestId = new(); private readonly AsyncLocal m_currentServiceDispatchScope = new(); - private readonly AsyncLocal m_currentValidationScope = new(); - private readonly Dictionary m_requests; + + /// + /// The executing requests. Registration and completion take no lock; enumeration, + /// drains, timeouts and Cancel run under . + /// + private readonly ConcurrentDictionary m_requests; private readonly List m_requestDrains = []; private readonly Lock m_requestsLock = new(); - private readonly HashSet m_activeValidationScopes = []; + private readonly ConcurrentDictionary m_activeValidationScopes = new(); + + /// + /// The number of registered drains (plus one while a drain takes its snapshot). + /// + private int m_drainCount; + + /// + /// The number of requests registered with a deadline, so a stopping timer notices them. + /// + private long m_deadlineRegistrations; private readonly Dictionary> m_pendingCancels = []; private int m_pendingCancelSessionCount; private DateTime m_nextPendingCancelSweep; @@ -1085,14 +1124,17 @@ private sealed class RequestExecutionScope : IDisposable /// The owning request manager. /// The context of the request being executed. /// The direct request id to restore on dispose. + /// Whether the scope set the ambient request id. public RequestExecutionScope( RequestManager requestManager, OperationContext context, - uint? previousRequestId) + uint? previousRequestId, + bool restoreRequestId) { m_requestManager = requestManager; m_context = context; m_previousRequestId = previousRequestId; + m_restoreRequestId = restoreRequestId; } /// @@ -1104,13 +1146,17 @@ public void Dispose() { m_disposed = true; m_requestManager.CompleteRequest(m_context); - m_requestManager.m_currentRequestId.Value = m_previousRequestId; + if (m_restoreRequestId) + { + m_requestManager.m_currentRequestId.Value = m_previousRequestId; + } } } private readonly RequestManager m_requestManager; private readonly OperationContext m_context; private readonly uint? m_previousRequestId; + private readonly bool m_restoreRequestId; private bool m_disposed; } @@ -1170,15 +1216,12 @@ private sealed class RequestValidationScope : IDisposable /// /// The owning request manager. /// The id the drain waits for. - /// The validation scope to restore on dispose. public RequestValidationScope( RequestManager requestManager, - long validationId, - RequestValidationScope? previousScope) + long validationId) { m_requestManager = requestManager; m_validationId = validationId; - m_previousScope = previousScope; } public long ValidationId => m_validationId; @@ -1191,10 +1234,13 @@ public void Dispose() if (!m_disposed) { m_disposed = true; + m_requestManager.m_activeValidationScopes.TryRemove(m_validationId, out _); + if (!m_requestManager.HasRequestDrains()) + { + return; + } lock (m_requestManager.m_requestsLock) { - m_requestManager.m_activeValidationScopes.Remove( - m_validationId); for (int ii = m_requestManager.m_requestDrains.Count - 1; ii >= 0; @@ -1206,14 +1252,13 @@ public void Dispose() m_requestManager.m_requestDrains.RemoveAt(ii); } } + m_requestManager.PublishDrainCountLocked(); } - m_requestManager.m_currentValidationScope.Value = m_previousScope; } } private readonly RequestManager m_requestManager; private readonly long m_validationId; - private readonly RequestValidationScope? m_previousScope; private bool m_disposed; } } diff --git a/src/Opc.Ua.Server/Server/StandardServer.cs b/src/Opc.Ua.Server/Server/StandardServer.cs index 8b67566052..04a9d46a41 100644 --- a/src/Opc.Ua.Server/Server/StandardServer.cs +++ b/src/Opc.Ua.Server/Server/StandardServer.cs @@ -3625,7 +3625,7 @@ protected async ValueTask ValidateRequestAsync( // request was registered, did not see it when it aborted the Session's outstanding // requests (OPC 10000-4 5.7.2.1). The close marks the Session closing before it // sweeps the registered requests under the request manager lock, and registration - // takes the same lock, so a request the sweep missed sees the mark here. + // ends with a full fence, so a request the sweep missed sees the mark here. if (context.Session is ISession admittedSession && SessionTermination.IsClosingOrClaimed(admittedSession)) { diff --git a/src/Opc.Ua.Types/BuiltIn/RequestLifetime.cs b/src/Opc.Ua.Types/BuiltIn/RequestLifetime.cs index e14da39b70..eb7599b645 100644 --- a/src/Opc.Ua.Types/BuiltIn/RequestLifetime.cs +++ b/src/Opc.Ua.Types/BuiltIn/RequestLifetime.cs @@ -92,6 +92,56 @@ public RequestLifetime(params CancellationToken[] externalTokens) m_statusCode = StatusCodes.Good; } + /// + /// Initializes a lifetime that is cancelled by either token or once + /// elapses, with a single token source for all three. + /// + /// + /// Equivalent to passing both tokens and the token of a separate timeout source to + /// , without the token array, the + /// timeout source and its registration. Tokens that cannot be cancelled are not linked. + /// + /// The time after which the request is cancelled, or + /// or for none. + /// A token that cancels the request. + /// Another token that cancels the request. + /// is negative + /// and not . + public RequestLifetime(TimeSpan timeout, CancellationToken first, CancellationToken second) + { + if (timeout < TimeSpan.Zero && timeout != Timeout.InfiniteTimeSpan) + { + throw new ArgumentOutOfRangeException(nameof(timeout)); + } + if (!first.CanBeCanceled || first == second) + { + first = second; + second = default; + } + if (!first.CanBeCanceled) + { + m_cts = new CancellationTokenSource(); + } + else if (!second.CanBeCanceled) + { +#if NET + m_cts = CancellationTokenSource.CreateLinkedTokenSource(first); +#else + m_cts = CancellationTokenSource.CreateLinkedTokenSource(first, second); +#endif + } + else + { + m_cts = CancellationTokenSource.CreateLinkedTokenSource(first, second); + } + if (timeout > TimeSpan.Zero) + { + m_cts.CancelAfter(timeout); + } + CancellationToken = m_cts.Token; + m_statusCode = StatusCodes.Good; + } + /// /// Attempts to cancel the request and assigns the corresponding status code. /// diff --git a/tests/Opc.Ua.Core.Tests/Stack/Server/FairRequestQueueTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Server/FairRequestQueueTests.cs index da4e1f6cf4..f2217d7579 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Server/FairRequestQueueTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Server/FairRequestQueueTests.cs @@ -434,6 +434,120 @@ public void CancellationDuringGrantCannotExecuteOrDoubleComplete(bool useFairSch Assert.That(provider.TotalUsed, Is.Zero); } + /// + /// In weighted mode a worker selects the next request while another worker's admission + /// is still running; global FIFO mode keeps one selection at a time. + /// + [TestCase(false)] + [TestCase(true)] + public void WeightedModeSelectsWhileAnotherAdmissionRuns(bool useFairScheduling) + { + var provider = new TestProvider(executionCapacity: 2) { UseFairScheduling = useFairScheduling }; + NodeId token = provider.AddOwner("A"); + using var queue = CreateQueue(provider); + Enqueue(queue, new TestRequest(token, 1)); + Enqueue(queue, new TestRequest(token, 2)); + bool nested = false; + FairRequestQueue.Entry? second = null; + provider.BeforeExecutionGrant = () => + { + provider.BeforeExecutionGrant = null!; + nested = queue.TryDequeue(out second); + }; + + using FairRequestQueue.Entry first = Dequeue(queue); + + Assert.That(first.Request.Request.RequestHeader.RequestHandle, Is.EqualTo(1)); + Assert.That(nested, Is.EqualTo(useFairScheduling)); + if (useFairScheduling) + { + Assert.That(second!.Request.Request.RequestHeader.RequestHandle, Is.EqualTo(2)); + second.Dispose(); + } + else + { + using FairRequestQueue.Entry next = Dequeue(queue); + Assert.That(next.Request.Request.RequestHeader.RequestHandle, Is.EqualTo(2)); + } + first.Dispose(); + Assert.That(queue.Count, Is.Zero); + Assert.That(queue.OwnerCount, Is.Zero); + Assert.That(provider.TotalUsed, Is.Zero); + } + + /// + /// Concurrent selections of one owner that both find execution capacity exhausted + /// return to the owner's FIFO in admission order, whichever fails last. + /// + [Test] + public async Task ConcurrentlyBlockedHeadsKeepAdmissionOrderAsync() + { + var provider = new TestProvider(executionCapacity: 1); + NodeId token = provider.AddOwner("A"); + using var queue = CreateQueue(provider); + Enqueue(queue, new TestRequest(token, 1)); + FairRequestQueue.Entry running = Dequeue(queue); + Enqueue(queue, new TestRequest(token, 2)); + Enqueue(queue, new TestRequest(token, 3)); + using var laterInGrant = new ManualResetEventSlim(); + using var releaseLater = new ManualResetEventSlim(); + Task? later = null; + int grants = 0; + provider.BeforeExecutionGrant = () => + { + if (Interlocked.Increment(ref grants) == 1) + { + // request 2 is selected; a second worker selects request 3 and holds its grant + later = Task.Run(() => queue.TryDequeue(out _)); + Assert.That(laterInGrant.Wait(TimeSpan.FromSeconds(5)), Is.True); + } + else + { + laterInGrant.Set(); + Assert.That(releaseLater.Wait(TimeSpan.FromSeconds(5)), Is.True); + } + }; + + // request 2 fails first and is relinked before request 3 fails + Assert.That(queue.TryDequeue(out _), Is.False); + releaseLater.Set(); + Assert.That(await later!.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false), Is.False); + provider.BeforeExecutionGrant = null!; + Assert.That(provider.ExecutionAttempts, Is.EqualTo(3)); + Assert.That(queue.Count, Is.EqualTo(2)); + + running.Dispose(); + using (FairRequestQueue.Entry next = Dequeue(queue)) + { + Assert.That(next.Request.Request.RequestHeader.RequestHandle, Is.EqualTo(2)); + } + using (FairRequestQueue.Entry last = Dequeue(queue)) + { + Assert.That(last.Request.Request.RequestHeader.RequestHandle, Is.EqualTo(3)); + } + Assert.That(provider.TotalUsed, Is.Zero); + } + /// + /// A queue stopped while a selected request's admission runs completes that request + /// once with Bad_ServerHalted and returns all of its accounting. + /// + [TestCase(false)] + [TestCase(true)] + public void StopDuringGrantCompletesTheSelectedRequestOnce(bool useFairScheduling) + { + var provider = new TestProvider { UseFairScheduling = useFairScheduling }; + NodeId token = provider.AddOwner("A"); + FairRequestQueue queue = CreateQueue(provider); + var request = new TestRequest(token); + Enqueue(queue, request); + provider.BeforeExecutionGrant = queue.Dispose; + + Assert.That(queue.TryDequeue(out _), Is.False); + Assert.That(request.CompletionCount, Is.EqualTo(1)); + Assert.That(request.Status, Is.EqualTo(StatusCodes.BadServerHalted)); + Assert.That(provider.TotalUsed, Is.Zero); + } + [TestCase(false)] [TestCase(true)] public void RunningCancellationKeepsAccountingUntilTheHandlerFinishes(bool useFairScheduling) diff --git a/tests/Opc.Ua.Core.Tests/Stack/Server/RequestLifetimeTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Server/RequestLifetimeTests.cs index ac7d3388f7..6568144d0f 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Server/RequestLifetimeTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Server/RequestLifetimeTests.cs @@ -30,6 +30,7 @@ // CA2000: test code; many disposables are ownership-transferred to test fixtures or short-lived, // making CA2000 noisy without a real leak risk. Disabled file-level for the suite. #pragma warning disable CA2000 +using System; using System.Threading; using NUnit.Framework; @@ -164,5 +165,91 @@ public void None_Property_ReturnsCompletedInstance() Assert.That(lifetime.CancellationToken.IsCancellationRequested, Is.False); } + + [TestCase(0)] + [TestCase(1)] + public void TimeoutConstructorIsCancelledByEitherToken(int cancelled) + { + using var first = new CancellationTokenSource(); + using var second = new CancellationTokenSource(); + using var lifetime = new RequestLifetime(TimeSpan.Zero, first.Token, second.Token); + + Assert.That(lifetime.CancellationToken.IsCancellationRequested, Is.False); + (cancelled == 0 ? first : second).Cancel(); + + Assert.That(lifetime.CancellationToken.IsCancellationRequested, Is.True); + Assert.That(lifetime.StatusCode, Is.EqualTo(StatusCodes.Good)); + } + + [Test] + public void TimeoutConstructorIsCancelledWhenTheTimeoutElapses() + { + using var lifetime = new RequestLifetime( + TimeSpan.FromMilliseconds(20), CancellationToken.None, CancellationToken.None); + + Assert.That( + lifetime.CancellationToken.WaitHandle.WaitOne(TimeSpan.FromSeconds(10)), Is.True); + Assert.That(lifetime.StatusCode, Is.EqualTo(StatusCodes.Good)); + } + + [Test] + public void TimeoutConstructorWithoutTimeoutOrTokensIsOnlyCancelledExplicitly() + { + using var first = new CancellationTokenSource(); + using var lifetime = new RequestLifetime(Timeout.InfiniteTimeSpan, first.Token, first.Token); + using var unlinked = new RequestLifetime(TimeSpan.Zero, default, default); + + Assert.That(lifetime.CancellationToken.WaitHandle.WaitOne(50), Is.False); + first.Cancel(); + Assert.That(lifetime.CancellationToken.IsCancellationRequested, Is.True); + + Assert.That(unlinked.CancellationToken.CanBeCanceled, Is.True); + Assert.That(unlinked.TryCancel(StatusCodes.BadRequestCancelledByClient), Is.True); + Assert.That(unlinked.StatusCode, Is.EqualTo(StatusCodes.BadRequestCancelledByClient)); + } + + [Test] + public void TimeoutConstructorRejectsNegativeTimeouts() + { + Assert.That( + () => new RequestLifetime(TimeSpan.FromMilliseconds(-2), default, default), + Throws.TypeOf()); + } + +#if NET + /// + /// One token source replaces the token array, the timeout source and its registration + /// that the params constructor needs for the same three inputs. + /// + [Test] + public void TimeoutConstructorAllocatesLessThanLinkingATimeoutSource() + { + using var request = new CancellationTokenSource(); + using var transport = new CancellationTokenSource(); + var timeout = TimeSpan.FromMinutes(5); + for (int ii = 0; ii < 2; ii++) + { + using var warm = new RequestLifetime(timeout, request.Token, transport.Token); + using var warmTimeout = new CancellationTokenSource(timeout); + using var warmOld = new RequestLifetime(request.Token, transport.Token, warmTimeout.Token); + } + + long before = GC.GetAllocatedBytesForCurrentThread(); + using (var timeoutSource = new CancellationTokenSource(timeout)) + using (new RequestLifetime(request.Token, transport.Token, timeoutSource.Token)) + { + } + long linked = GC.GetAllocatedBytesForCurrentThread() - before; + + before = GC.GetAllocatedBytesForCurrentThread(); + using (new RequestLifetime(timeout, request.Token, transport.Token)) + { + } + long single = GC.GetAllocatedBytesForCurrentThread() - before; + + TestContext.Out.WriteLine($"params + timeout source: {linked} B, timeout constructor: {single} B"); + Assert.That(single, Is.LessThan(linked)); + } +#endif } } diff --git a/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs index 6ff79691da..b2b24bc29c 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Server/RequestQueueTests.cs @@ -73,6 +73,61 @@ public void OperationCompleted(IServiceResponse? response, ServiceResult error) } } + /// + /// Records the token the queue passes to , optionally declaring that + /// it links its scheduled token itself, as endpoint requests do. + /// + private class TokenRecordingRequest : IEndpointIncomingRequest + { + public TaskCompletionSource Called { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource Release { get; } = new(); + public IServiceRequest Request => null!; + public SecureChannelContext SecureChannelContext => null!; + + public ValueTask CallAsync(CancellationToken cancellationToken = default) + { + Called.TrySetResult(cancellationToken); + return new ValueTask(Release.Task); + } + + public void OperationCompleted(IServiceResponse? response, ServiceResult error) + { + Release.TrySetResult(true); + } + } + + private sealed class SelfLinkingRequest(CancellationToken scheduled) : + TokenRecordingRequest, ILinksScheduledCancellation + { + public CancellationToken ScheduledCancellationToken { get; } = scheduled; + } + + /// + /// The queue links the scheduled token into the handler's token unless the request links + /// it itself, so endpoint requests do not pay for a second linked token source. + /// + [TestCase(false)] + [TestCase(true)] + public async Task ScheduledTokenIsLinkedOnlyWhenTheRequestDoesNotLinkItAsync(bool selfLinking) + { + using var server = new TestServer(); + server.ReplaceRequestQueue(1, 2, 10); + using var transport = new CancellationTokenSource(); + TokenRecordingRequest request = selfLinking + ? new SelfLinkingRequest(transport.Token) + : new TokenRecordingRequest(); + + server.ScheduleIncomingRequest(request, transport.Token); + CancellationToken passed = await request.Called.Task + .WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); + transport.Cancel(); + + Assert.That(passed.IsCancellationRequested, Is.EqualTo(!selfLinking)); + Assert.That(passed.CanBeCanceled, Is.True, "the server's stop token is always passed"); + request.Release.TrySetResult(true); + } + private sealed class ParkableTestRequest : IParkableIncomingRequest { public TaskCompletionSource ProcessingStarted { get; } = new TaskCompletionSource(); diff --git a/tests/Opc.Ua.Core.Tests/Types/Bindings/ArrayPoolBufferManagerBaseCoverageTests.cs b/tests/Opc.Ua.Core.Tests/Types/Bindings/ArrayPoolBufferManagerBaseCoverageTests.cs index 60b874304a..85a8530b30 100644 --- a/tests/Opc.Ua.Core.Tests/Types/Bindings/ArrayPoolBufferManagerBaseCoverageTests.cs +++ b/tests/Opc.Ua.Core.Tests/Types/Bindings/ArrayPoolBufferManagerBaseCoverageTests.cs @@ -263,24 +263,21 @@ public void SelectedPoolRoundsExpectedSizesToPoolBuckets() NUnitTelemetryContext.Create(), metadataByteCount: 0); -#if NET5_0_OR_GREATER + // the bounded pool never falls back to the next bucket, on any framework const int minimumExpectedSize = 16; -#else - const int minimumExpectedSize = 32; -#endif Assert.That(manager.GetExpectedBufferSize(0), Is.EqualTo(minimumExpectedSize)); Assert.That(manager.GetExpectedBufferSize(16), Is.EqualTo(minimumExpectedSize)); Assert.That(manager.GetExpectedBufferSize(17), Is.EqualTo(2 * minimumExpectedSize)); } /// - /// Verifies the inclusive shared-pool threshold with and without metadata. + /// Verifies the inclusive bounded-pool threshold with and without metadata. /// /// The maximum payload size. /// The metadata size. [TestCase(1024 * 1024, 0)] [TestCase((1024 * 1024) - 1, 1)] - public void PoolSelectionAtThresholdUsesSharedPool( + public void PoolSelectionAtThresholdUsesProcessWideBoundedPool( int maxBufferSize, int metadataByteCount) { @@ -288,7 +285,23 @@ public void PoolSelectionAtThresholdUsesSharedPool( maxBufferSize, metadataByteCount); - Assert.That(pool, Is.SameAs(ArrayPool.Shared)); + Assert.That(pool, Is.Not.SameAs(ArrayPool.Shared)); + Assert.That(pool, Is.SameAs(BoundedArrayPool.GetShared(1024 * 1024))); + } + + /// + /// Verifies that client and server managers of the default transport buffer size + /// share one bounded pool, whatever cookie they append. + /// + [Test] + public void DefaultTransportBufferManagersShareOneBoundedPool() + { + ArrayPool fast = ArrayPoolBufferManagerBase.CreateArrayPool(65535, 0); + ArrayPool cookie = ArrayPoolBufferManagerBase.CreateArrayPool(65535, 1); + + Assert.That(fast, Is.SameAs(cookie)); + Assert.That(((BoundedArrayPool)fast).MaxArrayLength, Is.EqualTo(65536)); + Assert.That(((BoundedArrayPool)fast).MaxArraysPerBucket, Is.EqualTo(1024)); } /// diff --git a/tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs b/tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs new file mode 100644 index 0000000000..274b10312d --- /dev/null +++ b/tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs @@ -0,0 +1,225 @@ +/* ======================================================================== + * 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.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using Opc.Ua.Bindings; +using Opc.Ua.Tests; + +namespace Opc.Ua.Core.Tests.Stack.Bindings +{ + /// + /// Tests the bounded transport buffer pool. + /// + [TestFixture] + [Category("BufferManager")] + [Parallelizable] + public sealed class BoundedArrayPoolTests + { + [Test] + public void RentReturnsExactPowerOfTwoBuckets() + { + var pool = new BoundedArrayPool(1024, 4); + + Assert.That(pool.Rent(1), Has.Length.EqualTo(16)); + Assert.That(pool.Rent(16), Has.Length.EqualTo(16)); + Assert.That(pool.Rent(17), Has.Length.EqualTo(32)); + Assert.That(pool.Rent(1000), Has.Length.EqualTo(1024)); + Assert.That(pool.Rent(0), Is.Empty); + } + + [Test] + public void ReturnedArrayIsReused() + { + var pool = new BoundedArrayPool(1024, 4); + byte[] first = pool.Rent(500); + pool.Return(first); + + Assert.That(pool.GetRetainedCount(512), Is.EqualTo(1)); + Assert.That(pool.Rent(300), Is.SameAs(first)); + Assert.That(pool.GetRetainedCount(512), Is.Zero); + } + + /// + /// A thousand outstanding buffers come back to the pool, which ArrayPool.Shared cannot + /// promise on a host with one processor: its per-processor partitions are much smaller. + /// + [Test] + public void ThousandOutstandingBuffersAreAllRetainedAndReused() + { + var pool = new BoundedArrayPool(65536, 1024); + byte[][] rented = [.. Enumerable.Range(0, 1000).Select(_ => pool.Rent(65535))]; + foreach (byte[] array in rented) + { + pool.Return(array); + } + + Assert.That(pool.GetRetainedCount(65536), Is.EqualTo(1000)); + var reused = new HashSet(rented); + for (int ii = 0; ii < rented.Length; ii++) + { + Assert.That(reused.Contains(pool.Rent(65535)), Is.True); + } + Assert.That(pool.GetRetainedCount(65536), Is.Zero); + } + + [Test] + public void ReturnBeyondTheBucketBoundDropsTheArray() + { + var pool = new BoundedArrayPool(64, 2); + byte[] a = pool.Rent(64); + byte[] b = pool.Rent(64); + byte[] c = pool.Rent(64); + pool.Return(a); + pool.Return(b); + pool.Return(c); + + Assert.That(pool.GetRetainedCount(64), Is.EqualTo(2)); + Assert.That(pool.Rent(64), Is.SameAs(a)); + Assert.That(pool.Rent(64), Is.SameAs(b)); + Assert.That(pool.Rent(64), Is.Not.SameAs(c)); + } + + [Test] + public async Task ConcurrentReturnsNeverExceedTheBoundAsync() + { + const int bound = 8; + var pool = new BoundedArrayPool(256, bound); + byte[][] arrays = [.. Enumerable.Range(0, 64).Select(_ => pool.Rent(256))]; + + await Task.WhenAll(arrays.Select(array => Task.Run(() => pool.Return(array)))) + .ConfigureAwait(false); + + Assert.That(pool.GetRetainedCount(256), Is.EqualTo(bound)); + var retained = new HashSet(); + for (int ii = 0; ii < bound; ii++) + { + retained.Add(pool.Rent(256)); + } + Assert.That(retained, Has.Count.EqualTo(bound)); + Assert.That(pool.GetRetainedCount(256), Is.Zero); + } + + [Test] + public void OversizedRentIsAllocatedExactlyAndNotRetained() + { + var pool = new BoundedArrayPool(64, 4); + byte[] large = pool.Rent(100); + + Assert.That(large, Has.Length.EqualTo(100)); + Assert.DoesNotThrow(() => pool.Return(large)); + Assert.That(pool.Rent(64), Is.Not.SameAs(large)); + } + + [Test] + public void ReturnOfForeignArrayLengthThrows() + { + var pool = new BoundedArrayPool(64, 4); + + Assert.That(() => pool.Return(new byte[48]), Throws.ArgumentException); + Assert.That(() => pool.Return(null!), Throws.ArgumentNullException); + Assert.DoesNotThrow(() => pool.Return([])); + } + + [Test] + public void ClearArrayZeroesTheReturnedArray() + { + var pool = new BoundedArrayPool(64, 4); + byte[] array = pool.Rent(64); + array.AsSpan().Fill(0xAB); + + pool.Return(array, clearArray: true); + + Assert.That(pool.Rent(64).All(b => b == 0), Is.True); + } + + [Test] + public void InvalidArgumentsThrow() + { + Assert.That(() => new BoundedArrayPool(100, 4), Throws.TypeOf()); + Assert.That(() => new BoundedArrayPool(8, 4), Throws.TypeOf()); + Assert.That(() => new BoundedArrayPool(64, 0), Throws.TypeOf()); + Assert.That(() => new BoundedArrayPool(64, 4).Rent(-1), Throws.TypeOf()); + } + + [TestCase(16, 1024)] + [TestCase(65536, 1024)] + [TestCase(131072, 512)] + [TestCase(1024 * 1024, 64)] + [TestCase(16 * 1024 * 1024, 4)] + [TestCase(128 * 1024 * 1024, 1)] + public void DefaultBoundKeepsAtMost64MiBInTheLargestBucket(int maxArrayLength, int expected) + { + Assert.That(BoundedArrayPool.GetDefaultMaxArraysPerBucket(maxArrayLength), Is.EqualTo(expected)); + } + + [Test] + public void SharedPoolIsKeyedByRoundedMaximumLength() + { + BoundedArrayPool pool = BoundedArrayPool.GetShared(40000); + + Assert.That(pool.MaxArrayLength, Is.EqualTo(65536)); + Assert.That(BoundedArrayPool.GetShared(65536), Is.SameAs(pool)); + Assert.That(BoundedArrayPool.GetShared(65537), Is.Not.SameAs(pool)); + } + + /// + /// A buffer manager over the bounded pool reuses every chunk buffer of a deep pipeline + /// instead of allocating one per response once more buffers are in flight than processors. + /// The pool is the default one's configuration but private to the test, so parallel tests + /// sharing the process-wide pool cannot take its arrays. + /// + [Test] + public void FastBufferManagerReusesBuffersOfADeepPipeline() + { + var pool = new BoundedArrayPool(65536, BoundedArrayPool.GetDefaultMaxArraysPerBucket(65536)); + var manager = new FastBufferManager("pipeline", 65535, NUnitTelemetryContext.Create(), pool); + var first = new HashSet(); + for (int ii = 0; ii < 200; ii++) + { + first.Add(manager.TakeBuffer(65535, "pipeline")); + } + foreach (byte[] buffer in first) + { + manager.ReturnBuffer(buffer, "pipeline"); + } + Assert.That(pool.GetRetainedCount(65536), Is.EqualTo(200)); + + for (int ii = 0; ii < 200; ii++) + { + Assert.That(first, Does.Contain(manager.TakeBuffer(65535, "pipeline"))); + } + Assert.That(pool.GetRetainedCount(65536), Is.Zero); + } + } +} diff --git a/tests/Opc.Ua.Server.Tests/RequestManagerTests.cs b/tests/Opc.Ua.Server.Tests/RequestManagerTests.cs index bdf06bbbc3..56a76d0385 100644 --- a/tests/Opc.Ua.Server.Tests/RequestManagerTests.cs +++ b/tests/Opc.Ua.Server.Tests/RequestManagerTests.cs @@ -1181,6 +1181,63 @@ public void NestedValidationScopesLeaveRegisteredRequestsToTheirOwnScope() m_requestManager.RequestCompleted(innerContext); } + /// + /// Requests and validation scopes are registered and completed without the request + /// manager lock. A drain that snapshots them concurrently must still be released by every + /// completion it waits for, or it would run into its two-minute timeout. + /// + [Test] + [Category("NodeManagerLifecycle")] + public async Task DrainsRacingLockFreeRegistrationAndCompletionAlwaysCompleteAsync() + { + using var stop = new CancellationTokenSource(); + int completed = 0; + var churn = new Task[4]; + for (int worker = 0; worker < churn.Length; worker++) + { + churn[worker] = Task.Run(() => + { + uint handle = 0; + while (!stop.IsCancellationRequested) + { + IDisposable validation = m_requestManager.EnterValidationScope(); + using var lifetime = new RequestLifetime(); + OperationContext context = CreateOperationContext(++handle, lifetime, 60000); + m_requestManager.RequestReceived(context); + validation.Dispose(); + Assert.That(m_requestManager.IsExecutingRequest(context), Is.True); + m_requestManager.RequestCompleted(context); + Assert.That(m_requestManager.IsExecutingRequest(context), Is.False); + Interlocked.Increment(ref completed); + } + }); + } + + try + { + // the drains must race the churn, so wait until it is running + using var started = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + while (Volatile.Read(ref completed) < 1000) + { + await Task.Delay(1, started.Token).ConfigureAwait(false); + } + for (int ii = 0; ii < 200; ii++) + { + await m_requestManager.WaitForCurrentRequestsAsync().AsTask() + .WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false); + } + } + finally + { + stop.Cancel(); + await Task.WhenAll(churn).ConfigureAwait(false); + } + + Assert.That(completed, Is.GreaterThan(0)); + await AssertCompletesWithinTimeoutAsync(m_requestManager.WaitForCurrentRequestsAsync().AsTask()) + .ConfigureAwait(false); + } + private static OperationContext CreateOperationContext( uint requestHandle, RequestLifetime requestLifetime) diff --git a/tests/Opc.Ua.Server.Tests/ResourceIsolation/RuntimeResourceIsolationTests.cs b/tests/Opc.Ua.Server.Tests/ResourceIsolation/RuntimeResourceIsolationTests.cs index 2968e11092..047200f0ac 100644 --- a/tests/Opc.Ua.Server.Tests/ResourceIsolation/RuntimeResourceIsolationTests.cs +++ b/tests/Opc.Ua.Server.Tests/ResourceIsolation/RuntimeResourceIsolationTests.cs @@ -1554,6 +1554,105 @@ public async Task RestartPreservesHostReplacementsAndDisposesOnlyOldOwnedProvide disposable.Verify(d => d.Dispose(), Times.Never); } + /// + /// Admission without the server-wide lock: concurrent leases of ordinary and protected + /// owners never exceed the stage capacity or an owner ceiling, the owner table never holds + /// more entries than allowed, and every counter and table entry is returned at the end. + /// + [Test] + public async Task ConcurrentLeasesStayWithinLimitsAndAreAllReturnedAsync() + { + ServerResourceIsolationOptions options = Options(); + options.MaxTrackedOwners = 4; + const ResourceIsolationStage stage = ResourceIsolationStage.Connection; + using DefaultServerResourceIsolationProvider provider = + CreateProvider(options, new TestClassifier()); + ResourceIsolationStagePlan plan = provider.Plan.GetStage(stage); + ResourceIsolationOwner[] owners = + [ + provider.ClassifyConnection(Peer(1)), + provider.ClassifyConnection(Peer(2)), + provider.ClassifyConnection(Peer(10)), + provider.ClassifyConnection(Peer(11)), + provider.ClassifyConnection(Peer(12)), + provider.ClassifyConnection(Peer(13)) + ]; + long[] ownerHeld = new long[owners.Length]; + long totalHeld = 0; + int violations = 0; + int admitted = 0; + int maxOwners = 0; + + async Task WorkerAsync(int seed) + { + // a deterministic xorshift per worker; the interleaving is what varies + uint random = (uint)seed + 0x9E3779B9u; + int Next(int max) + { + random ^= random << 13; + random ^= random >> 17; + random ^= random << 5; + return (int)(random % (uint)max); + } + await Task.Yield(); + var held = new List<(int Owner, long Amount, IDisposable Lease)>(); + for (int ii = 0; ii < 20000; ii++) + { + if (held.Count > 0 && (held.Count > 3 || Next(2) == 0)) + { + (int owner, long amount, IDisposable lease) = held[^1]; + held.RemoveAt(held.Count - 1); + Interlocked.Add(ref ownerHeld[owner], -amount); + Interlocked.Add(ref totalHeld, -amount); + lease.Dispose(); + lease.Dispose(); + continue; + } + int index = Next(owners.Length); + long take = 1 + Next(3); + if (provider.TryAcquire(stage, owners[index], take, out IDisposable? acquired, out _)) + { + Interlocked.Increment(ref admitted); + if (Interlocked.Add(ref ownerHeld[index], take) > owners[index].GetHardLimit(stage) || + Interlocked.Add(ref totalHeld, take) > plan.Capacity || + provider.GetUsage(stage) > plan.Capacity) + { + Interlocked.Increment(ref violations); + } + int tracked = provider.TrackedOwnerCount; + int observed; + while (tracked > (observed = Volatile.Read(ref maxOwners)) && + Interlocked.CompareExchange(ref maxOwners, tracked, observed) != observed) + { + } + held.Add((index, take, acquired!)); + } + } + foreach ((int owner, long amount, IDisposable lease) in held) + { + Interlocked.Add(ref ownerHeld[owner], -amount); + Interlocked.Add(ref totalHeld, -amount); + lease.Dispose(); + } + } + + await Task.WhenAll(Enumerable.Range(0, 8).Select(WorkerAsync)).ConfigureAwait(false); + + Assert.That(violations, Is.Zero); + Assert.That(admitted, Is.GreaterThan(1000)); + Assert.That(maxOwners, Is.LessThanOrEqualTo(options.MaxTrackedOwners)); + Assert.That(provider.GetUsage(stage), Is.Zero); + Assert.That(provider.TrackedOwnerCount, Is.Zero); + + // nothing leaked: the protected floors and the whole shared capacity are available again + Assert.That(plan.BootstrapReserved, Is.Positive); + Assert.That(plan.ReconnectReserved, Is.Positive); + using IDisposable bootstrap = Acquire(provider, stage, owners[0], plan.BootstrapReserved); + using IDisposable reconnect = Acquire(provider, stage, owners[1], plan.ReconnectReserved); + using IDisposable shared = Acquire(provider, stage, owners[2], plan.SharedCapacity); + AssertRejected(provider, stage, owners[3], 1, ResourceIsolationFailureReason.Capacity); + } + private static IDisposable Acquire( DefaultServerResourceIsolationProvider provider, ResourceIsolationStage stage,