From 12c82cb0edca3acb30cbcb442377422c8bf4f78d Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Mon, 12 Oct 2026 06:57:09 +0200 Subject: [PATCH 1/7] Rent transport buffers from a bounded process-wide pool ArrayPool.Shared keeps a few arrays per bucket and processor. A server with about a thousand responses waiting to be sent exhausts it on a host with one or two processors (or a container limited to them), and then allocates a new 64 KiB chunk for every response: 54 KB per Read instead of 7.6 KB with 10 channels x 100 outstanding requests at DOTNET_PROCESSOR_COUNT=1. Buffer managers up to 1 MiB now rent from a BoundedArrayPool: power-of-two buckets in lock-free queues, up to 1024 arrays (64 MiB) in the largest bucket, one instance per maximum array length shared by all client and server managers of the process. Retained memory stays below 2 x 64 MiB. Co-Authored-By: Claude Opus 5.5 --- .../Bindings/ArrayPoolBufferManagerBase.cs | 24 +- .../Stack/Bindings/BoundedArrayPool.cs | 232 ++++++++++++++++++ ...ArrayPoolBufferManagerBaseCoverageTests.cs | 27 +- .../Types/Bindings/BoundedArrayPoolTests.cs | 232 ++++++++++++++++++ 4 files changed, 496 insertions(+), 19 deletions(-) create mode 100644 src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs create mode 100644 tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs 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..b207b85359 --- /dev/null +++ b/src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs @@ -0,0 +1,232 @@ +/* ======================================================================== + * 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 up to 64 MiB in the largest bucket, between 16 and 1024 arrays: 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(kMinArraysPerBucket, 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 kMinArraysPerBucket = 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/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..d81d616ed4 --- /dev/null +++ b/tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs @@ -0,0 +1,232 @@ +/* ======================================================================== + * 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, 16)] + 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)); + } + + /// + /// The default buffer manager reuses the chunk buffers of a deep pipeline instead of + /// allocating one per response once more buffers are in flight than processors. + /// +#if NET + [Test] + public void FastBufferManagerReusesBuffersOfADeepPipeline() + { + var manager = new FastBufferManager("pipeline", 65535, NUnitTelemetryContext.Create()); + var first = new List(); + for (int ii = 0; ii < 200; ii++) + { + first.Add(manager.TakeBuffer(65535, "pipeline")); + } + foreach (byte[] buffer in first) + { + manager.ReturnBuffer(buffer, "pipeline"); + } + + // Other tests share the process-wide pool, so count allocated bytes rather than + // identities: the second pass takes retained arrays instead of allocating. + var second = new List(200); + long before = GC.GetAllocatedBytesForCurrentThread(); + for (int ii = 0; ii < 200; ii++) + { + second.Add(manager.TakeBuffer(65535, "pipeline")); + } + long allocated = GC.GetAllocatedBytesForCurrentThread() - before; + foreach (byte[] buffer in second) + { + manager.ReturnBuffer(buffer, "pipeline"); + } + + Assert.That(allocated, Is.LessThan(10L * 65536)); + } +#endif + } +} From 25fa6198b6e553d57ccefb9fa31eddca6ca963fb Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Mon, 12 Oct 2026 07:06:34 +0200 Subject: [PATCH 2/7] Admit isolation leases without the provider-wide lock Every request takes three leases (RequestQueue, RequestQueueBytes, RequestExecution) and returns them, and each of the six calls took the provider's single lock: about 18 % of the server's on-CPU time spun on it with 10 channels x 100 outstanding Reads on 24 cores. Stage counters, owner usage and the protected floors are now reserved with bounded compare-and-swap loops that never pass a limit, and the counters already reserved are returned when a later one is full. The owner table is a ConcurrentDictionary. An owner that already holds a lease of the same accounting class is admitted without any lock; a class count of an owner crosses zero only under a table lock, which keeps the exclusive class, the protected table slots and entry removal consistent with the leases. Lease release takes the table lock only for the owner's last lease of a class. Co-Authored-By: Claude Opus 5.5 --- .../DefaultServerResourceIsolationProvider.cs | 372 ++++++++++++------ .../RuntimeResourceIsolationTests.cs | 99 +++++ 2 files changed, 358 insertions(+), 113 deletions(-) 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/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, From 777d8b0370198bee5872fe754964c720ad8812bd Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Mon, 12 Oct 2026 07:14:04 +0200 Subject: [PATCH 3/7] Select queued requests concurrently with one gate entry per dispatch FairRequestQueue took its server-wide gate three times per request (enqueue, select, finish) and let only one worker select at a time: the others went back to wait for a wake signal while the selector ran the revalidation and the execution admission outside the gate. The selected head is now unlinked when it is selected, so an admitted dispatch needs no second gate entry, and in weighted mode several workers select at once. A head that finds execution capacity exhausted is relinked in admission order (concurrent selections of one owner may fail in any order) and its owner is not retried before capacity changes, as before. A request cancelled or a queue stopped while its admission runs is completed once by the selecting worker. Global FIFO mode still selects one request at a time, so its blocked head cannot be overtaken. Co-Authored-By: Claude Opus 5.5 --- .../Stack/Server/FairRequestQueue.cs | 173 ++++++++++++++---- .../Stack/Server/FairRequestQueueTests.cs | 114 ++++++++++++ 2 files changed, 247 insertions(+), 40 deletions(-) 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/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) From 53c854d0be2c0e9f5aef29d296144ecef88a4abb Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Mon, 12 Oct 2026 07:27:26 +0200 Subject: [PATCH 4/7] Keep the request queue's worker gate off the per-request path ServerBase.RequestQueue took its server-wide worker gate three to five times per request: in ScheduleIncomingRequest even in fair mode, where the locked block does nothing, in GrowFairWorkers at enqueue and at worker start, and twice in ProcessRequestSafeAsync to count running handlers. Fair mode now reads the stop flag without the gate (the fair queue rejects admissions itself once stopped). GrowFairWorkers checks the counters before it takes the gate, which it then needs only when a worker is to be added. Running handlers are counted with Interlocked; shutdown creates the drain completion after the workers have exited and the last handler completes it. Co-Authored-By: Claude Opus 5.5 --- .../Stack/Server/ServerBase.RequestQueue.cs | 70 +++++++++++-------- 1 file changed, 42 insertions(+), 28 deletions(-) diff --git a/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs index 91b450ef5e..00b00d9b60 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) @@ -449,11 +460,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 +552,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; From 982bb11090f9c5e22b0e7f7fe388f47bd8c9f985 Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Mon, 12 Oct 2026 07:27:26 +0200 Subject: [PATCH 5/7] Register and complete requests without the request manager lock RequestManager took its server-wide lock four times per request (validation scope enter and exit, registration, completion) and wrote three AsyncLocal scopes, each set and restored. Executing requests and validation scopes are ConcurrentDictionaries, so registration and completion take no lock. A drain publishes a count before it snapshots them, and a completion takes the lock only when a drain is registered, so every completion a drain waits for still reaches it. The request timer is stopped only when no deadline was registered meanwhile. A Cancel remembers itself before it scans the registered requests, and registration ends with a full fence, so a request registering concurrently with a Cancel or a Session close sweep is either seen or sees the mark. The validation scope's AsyncLocal was never read and is gone; the request scope skips the ambient id when the dispatch scope already carries it. Enumeration, drains, timeouts and Cancel still run under the lock. Co-Authored-By: Claude Opus 5.5 --- src/Opc.Ua.Server/Server/RequestManager.cs | 251 +++++++++++------- src/Opc.Ua.Server/Server/StandardServer.cs | 2 +- .../RequestManagerTests.cs | 57 ++++ 3 files changed, 206 insertions(+), 104 deletions(-) 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/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) From 125ee35afbf8c090871ce515cb94862f59ef0704 Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Mon, 12 Oct 2026 09:10:34 +0200 Subject: [PATCH 6/7] Create one cancellation source per request and link the transport token once Per request, EndpointIncomingRequest.CallAsync built a CancellationToken[] and a separate timeout CancellationTokenSource for the TimeoutHint, and linked both with the caller's and the transport's token (528 bytes for the lifetime alone). Before that, the request queue had already linked the worker's stop token with the same transport token. The new RequestLifetime(TimeSpan timeout, CancellationToken first, CancellationToken second) constructor links the two tokens (only those that can be cancelled, a duplicate once) and applies the timeout with CancelAfter on the same source: 240 bytes. A timeout still leaves the status Good, which CallAsync reports as Bad_Timeout. Requests that link their scheduled token themselves say so through the internal ILinksScheduledCancellation, and the queue then passes the stop token alone instead of a second linked source. Other IEndpointIncomingRequest implementations keep the linked token. A ServerBase.ProcessRequestAsync override now sees the server's stop token for endpoint requests; the request lifetime the services observe is still cancelled by the transport, the client and the timeout hint. Co-Authored-By: Claude Opus 5.5 --- .../EndpointBase.EndpointIncomingRequest.cs | 32 +++++-- .../Stack/Server/ServerBase.RequestQueue.cs | 6 +- src/Opc.Ua.Types/BuiltIn/RequestLifetime.cs | 50 +++++++++++ .../Stack/Server/RequestLifetimeTests.cs | 87 +++++++++++++++++++ .../Stack/Server/RequestQueueTests.cs | 55 ++++++++++++ 5 files changed, 221 insertions(+), 9 deletions(-) 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/ServerBase.RequestQueue.cs b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs index 00b00d9b60..7c64202c32 100644 --- a/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs +++ b/src/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs @@ -440,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); } 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/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(); From 77523e67fdd4966cb179d761a54eb16dd367cd6c Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Mon, 12 Oct 2026 09:53:44 +0200 Subject: [PATCH 7/7] Bound the transport pool strictly and isolate the pipeline test Review feedback: the default pool kept at least 16 arrays, so a pool for arrays above 4 MiB could retain more than the documented 64 MiB per bucket. The floor is now one array. The transport only uses the pool up to 1 MiB, where the bound was already 64 arrays, so nothing changes for it. The deep-pipeline test measured allocations through the process-wide pool, which parallel tests share; it now uses a private pool with the default configuration and checks that every buffer is reused. Co-Authored-By: Claude Opus 5.5 --- .../Stack/Bindings/BoundedArrayPool.cs | 5 ++- .../Types/Bindings/BoundedArrayPoolTests.cs | 31 +++++++------------ 2 files changed, 14 insertions(+), 22 deletions(-) diff --git a/src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs b/src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs index b207b85359..d7db0db85e 100644 --- a/src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs +++ b/src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs @@ -107,7 +107,7 @@ public static BoundedArrayPool GetShared(int maxArrayLength) } /// - /// Retains up to 64 MiB in the largest bucket, between 16 and 1024 arrays: 1024 arrays + /// 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). /// @@ -115,7 +115,7 @@ public static BoundedArrayPool GetShared(int maxArrayLength) internal static int GetDefaultMaxArraysPerBucket(int maxArrayLength) { long arrays = kMaxRetainedBytesPerBucket / maxArrayLength; - return (int)Math.Min(kMaxArraysPerBucket, Math.Max(kMinArraysPerBucket, arrays)); + return (int)Math.Min(kMaxArraysPerBucket, Math.Max(1, arrays)); } /// @@ -223,7 +223,6 @@ private sealed class Bucket(int arrayLength) } private const int kMinimumArrayLength = 16; - private const int kMinArraysPerBucket = 16; private const int kMaxArraysPerBucket = 1024; private const long kMaxRetainedBytesPerBucket = 64L * 1024 * 1024; private static readonly ConcurrentDictionary s_pools = new(); diff --git a/tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs b/tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs index d81d616ed4..274b10312d 100644 --- a/tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs +++ b/tests/Opc.Ua.Core.Tests/Types/Bindings/BoundedArrayPoolTests.cs @@ -176,7 +176,8 @@ public void InvalidArgumentsThrow() [TestCase(65536, 1024)] [TestCase(131072, 512)] [TestCase(1024 * 1024, 64)] - [TestCase(16 * 1024 * 1024, 16)] + [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)); @@ -193,15 +194,17 @@ public void SharedPoolIsKeyedByRoundedMaximumLength() } /// - /// The default buffer manager reuses the chunk buffers of a deep pipeline instead of - /// allocating one per response once more buffers are in flight than processors. + /// 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. /// -#if NET [Test] public void FastBufferManagerReusesBuffersOfADeepPipeline() { - var manager = new FastBufferManager("pipeline", 65535, NUnitTelemetryContext.Create()); - var first = new List(); + 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")); @@ -210,23 +213,13 @@ public void FastBufferManagerReusesBuffersOfADeepPipeline() { manager.ReturnBuffer(buffer, "pipeline"); } + Assert.That(pool.GetRetainedCount(65536), Is.EqualTo(200)); - // Other tests share the process-wide pool, so count allocated bytes rather than - // identities: the second pass takes retained arrays instead of allocating. - var second = new List(200); - long before = GC.GetAllocatedBytesForCurrentThread(); for (int ii = 0; ii < 200; ii++) { - second.Add(manager.TakeBuffer(65535, "pipeline")); + Assert.That(first, Does.Contain(manager.TakeBuffer(65535, "pipeline"))); } - long allocated = GC.GetAllocatedBytesForCurrentThread() - before; - foreach (byte[] buffer in second) - { - manager.ReturnBuffer(buffer, "pipeline"); - } - - Assert.That(allocated, Is.LessThan(10L * 65536)); + Assert.That(pool.GetRetainedCount(65536), Is.Zero); } -#endif } }