Server: remove the server-wide locks from the request pipeline, bounded transport buffer pool

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)
         /// <returns>The selected array pool.</returns>
         internal static ArrayPool<byte> CreateArrayPool(int maxBufferSize, int metadataByteCount)
         {
+            int maximumArrayLength = RoundUpToPoolBucket(
+                checked(maxBufferSize + metadataByteCount));
             if (maxBufferSize <= kSharedPoolThreshold - metadataByteCount)
             {
-                return ArrayPool<byte>.Shared;
+                // Not ArrayPool<byte>.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<byte>.Create(maximumArrayLength, 4);
         }
 

src/Opc.Ua.Core/Stack/Bindings/BoundedArrayPool.cs

@@ -0,0 +1,231 @@
+/* ========================================================================
+ * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
+ *
+ * OPC Foundation MIT License 1.00
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following
+ * conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
+ * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+ * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+ * OTHER DEALINGS IN THE SOFTWARE.
+ *
+ * The complete license agreement can be found here:
+ * http://opcfoundation.org/License/MIT/1.00/
+ * ======================================================================*/
+
+using System;
+using System.Buffers;
+using System.Collections.Concurrent;
+using System.Threading;
+
+namespace Opc.Ua.Bindings
+{
+    /// <summary>
+    /// An array pool for the transport's chunk-sized buffers that keeps enough arrays for
+    /// deep request pipelines, independent of the processor count.
+    /// </summary>
+    /// <remarks>
+    /// <see cref="ArrayPool{T}.Shared"/> 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 <see cref="MaxArraysPerBucket"/> arrays per power-of-two bucket in a
+    /// lock-free queue, so the retained memory is bounded by <c>MaxArraysPerBucket</c> times
+    /// the sum of the bucket sizes, which is less than
+    /// <c>2 * MaxArraysPerBucket * MaxArrayLength</c>. 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.
+    /// </remarks>
+    internal sealed class BoundedArrayPool : ArrayPool<byte>
+    {
+        /// <summary>
+        /// Creates a pool for arrays up to <paramref name="maxArrayLength"/> bytes.
+        /// </summary>
+        /// <param name="maxArrayLength">The largest pooled array length, a power of two.</param>
+        /// <param name="maxArraysPerBucket">The number of arrays retained per bucket.</param>
+        /// <exception cref="ArgumentOutOfRangeException"></exception>
+        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);
+            }
+        }
+
+        /// <summary>
+        /// The largest pooled array length. Larger requests are allocated and dropped on return.
+        /// </summary>
+        public int MaxArrayLength { get; }
+
+        /// <summary>
+        /// The number of arrays each bucket retains at most.
+        /// </summary>
+        public int MaxArraysPerBucket { get; }
+
+        /// <summary>
+        /// Gets the process-wide pool for arrays up to <paramref name="maxArrayLength"/> bytes,
+        /// rounded up to a power of two.
+        /// </summary>
+        /// <param name="maxArrayLength">The largest array length the caller rents.</param>
+        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)));
+        }
+
+        /// <summary>
+        /// 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).
+        /// </summary>
+        /// <param name="maxArrayLength">The largest pooled array length.</param>
+        internal static int GetDefaultMaxArraysPerBucket(int maxArrayLength)
+        {
+            long arrays = kMaxRetainedBytesPerBucket / maxArrayLength;
+            return (int)Math.Min(kMaxArraysPerBucket, Math.Max(1, arrays));
+        }
+
+        /// <summary>
+        /// The number of arrays currently retained by the bucket holding
+        /// <paramref name="arrayLength"/>.
+        /// </summary>
+        /// <param name="arrayLength">An array length up to <see cref="MaxArrayLength"/>.</param>
+        internal int GetRetainedCount(int arrayLength)
+        {
+            return Volatile.Read(ref m_buckets[GetBucketIndex(arrayLength)].Count);
+        }
+
+        /// <inheritdoc/>
+        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];
+        }
+
+        /// <inheritdoc/>
+        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);
+        }
+
+        /// <summary>
+        /// Returns the index of the smallest bucket holding <paramref name="length"/> bytes.
+        /// </summary>
+        private static int GetBucketIndex(int length)
+        {
+            int index = 0;
+            int bucketLength = kMinimumArrayLength;
+            while (bucketLength < length)
+            {
+                bucketLength <<= 1;
+                index++;
+            }
+            return index;
+        }
+
+        /// <summary>
+        /// The retained arrays of one power-of-two length.
+        /// </summary>
+        private sealed class Bucket(int arrayLength)
+        {
+            /// <summary>
+            /// The length of every array in this bucket.
+            /// </summary>
+            public readonly int ArrayLength = arrayLength;
+
+            /// <summary>
+            /// The retained arrays.
+            /// </summary>
+            public readonly ConcurrentQueue<byte[]> Arrays = new();
+
+            /// <summary>
+            /// The retained arrays, including returns about to be enqueued.
+            /// </summary>
+            public int Count;
+        }
+
+        private const int kMinimumArrayLength = 16;
+        private const int kMaxArraysPerBucket = 1024;
+        private const long kMaxRetainedBytesPerBucket = 64L * 1024 * 1024;
+        private static readonly ConcurrentDictionary<int, BoundedArrayPool> s_pools = new();
+        private readonly Bucket[] m_buckets;
+    }
+}

src/Opc.Ua.Core/Stack/Server/EndpointBase.EndpointIncomingRequest.cs

@@ -34,12 +34,26 @@
 
 namespace Opc.Ua
 {
+    /// <summary>
+    /// An incoming request whose <see cref="IEndpointIncomingRequest.CallAsync"/> links the
+    /// cancellation token it was scheduled with into its request lifetime, so the request queue
+    /// need not link that token again.
+    /// </summary>
+    internal interface ILinksScheduledCancellation
+    {
+        /// <summary>
+        /// The token the request was scheduled with.
+        /// </summary>
+        CancellationToken ScheduledCancellationToken { get; }
+    }
+
     public abstract partial class EndpointBase
     {
         /// <summary>
         /// An object that handles an incoming request for an endpoint.
         /// </summary>
-        protected struct EndpointIncomingRequest : IParkableIncomingRequest, IEquatable<EndpointIncomingRequest>
+        protected struct EndpointIncomingRequest : IParkableIncomingRequest, IEquatable<EndpointIncomingRequest>,
+            ILinksScheduledCancellation
         {
             /// <summary>
             /// Initialize the Object with a Request
@@ -91,14 +105,12 @@ public ValueTask<IServiceResponse> ProcessAsync(CancellationToken cancellationTo
             /// <inheritdoc/>
             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);
             }
 
+            /// <inheritdoc/>
+            readonly CancellationToken ILinksScheduledCancellation.ScheduledCancellationToken =>
+                m_transportCancellationToken;
+
             private readonly EndpointBase m_endpoint;
             private readonly ServiceResponsePooledValueTaskSource m_vts;