[Server] Make EndpointIncomingRequest a ReadonlyStruct & pool ValueTaks

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

@@ -13,6 +13,7 @@ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Diagnostics.CodeAnalysis;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;
@@ -76,7 +77,7 @@ protected EndpointBase(ServerBase server)
         }
 
         /// <inheritdoc/>
-        public Task<IServiceResponse> ProcessRequestAsync(
+        public ValueTask<IServiceResponse> ProcessRequestAsync(
             SecureChannelContext secureChannelContext,
             IServiceRequest request,
             CancellationToken cancellationToken = default)
@@ -670,7 +671,7 @@ public IServiceResponse Invoke(IServiceRequest request, SecureChannelContext sec
                 {
                     logger.LogWarning(
                         "Async Service invoced sychronously. Prefer using InvokeAsync for best performance.");
-                    return InvokeAsync(request, null).GetAwaiter().GetResult();
+                    return InvokeAsync(request, secureChannelContext).GetAwaiter().GetResult();
                 }
                 return m_invokeService?.Invoke(request, secureChannelContext);
             }
@@ -777,7 +778,7 @@ public void CallSynchronously()
             /// thread that calls IServerBase.ScheduleIncomingRequest().
             /// This method always traps any exceptions and reports them to the client as a fault.
             /// </remarks>
-            public async Task CallAsync(CancellationToken cancellationToken = default)
+            public async ValueTask CallAsync(CancellationToken cancellationToken = default)
             {
                 await OnProcessRequestAsync(null, cancellationToken).ConfigureAwait(false);
             }
@@ -1042,7 +1043,7 @@ .Body is AdditionalParametersType parameters &&
                         else
                         {
                             // call the service even when there is no trace information
-                            m_response = await m_service.InvokeAsync(Request,SecureChannelContext, cancellationToken)
+                            m_response = await m_service.InvokeAsync(Request, SecureChannelContext, cancellationToken)
                                 .ConfigureAwait(false);
                         }
                     }
@@ -1074,26 +1075,25 @@ .Body is AdditionalParametersType parameters &&
         /// <summary>
         /// An object that handles an incoming request for an endpoint.
         /// </summary>
-        protected class EndpointIncomingRequest : IEndpointIncomingRequest
+        protected readonly struct EndpointIncomingRequest : IEndpointIncomingRequest, IEquatable<EndpointIncomingRequest>
         {
             /// <summary>
             /// Initialize the Object with a Request
             /// </summary>
             public EndpointIncomingRequest(
                 EndpointBase endpoint,
                 SecureChannelContext context,
-                IServiceRequest request)
+                IServiceRequest request,
+                CancellationToken cancellationToken = default)
             {
                 m_endpoint = endpoint;
                 SecureChannelContext = context;
                 Request = request;
-                m_tcs = new TaskCompletionSource<IServiceResponse>(
-                    TaskCreationOptions.RunContinuationsAsynchronously);
+                m_vts = ServiceResponsePooledValueTaskSource.Create();
+                m_service = m_endpoint.FindService(Request.TypeId);
+                m_cancellationToken = cancellationToken;
             }
 
-            /// <inheritdoc/>
-            public object Calldata { get; set; }
-
             /// <inheritdoc/>
             public SecureChannelContext SecureChannelContext { get; }
 
@@ -1104,25 +1104,22 @@ public EndpointIncomingRequest(
             /// Process an incoming request
             /// </summary>
             /// <returns></returns>
-            public Task<IServiceResponse> ProcessAsync(CancellationToken cancellationToken = default)
+            public ValueTask<IServiceResponse> ProcessAsync(CancellationToken cancellationToken = default)
             {
                 try
                 {
-                    m_cancellationToken = cancellationToken;
-                    m_cancellationToken.Register(() => m_tcs.TrySetCanceled());
-                    m_service = m_endpoint.FindService(Request.TypeId);
-                    m_endpoint.ServerForContext.ScheduleIncomingRequest(this, m_cancellationToken);
+                    m_endpoint.ServerForContext.ScheduleIncomingRequest(this, cancellationToken);
                 }
                 catch (Exception e)
                 {
-                    m_tcs.TrySetResult(m_endpoint.CreateFault(Request, e));
+                    m_vts.SetResult(m_endpoint.CreateFault(Request, e));
                 }
 
-                return m_tcs.Task;
+                return m_vts.Task;
             }
 
             /// <inheritdoc/>
-            public async Task CallAsync(CancellationToken cancellationToken = default)
+            public async ValueTask CallAsync(CancellationToken cancellationToken = default)
             {
                 using CancellationTokenSource timeoutHintCts = (int)Request.RequestHeader.TimeoutHint > 0 ?
                     new CancellationTokenSource((int)Request.RequestHeader.TimeoutHint) : null;
@@ -1157,7 +1154,7 @@ .Body is AdditionalParametersType parameters &&
                     using (activity)
                     {
                         IServiceResponse response = await m_service.InvokeAsync(Request, SecureChannelContext, linkedCts.Token).ConfigureAwait(false);
-                        m_tcs.TrySetResult(response);
+                        m_vts.SetResult(response);
                     }
                 }
                 catch (Exception e)
@@ -1166,8 +1163,7 @@ .Body is AdditionalParametersType parameters &&
                     {
                         e = new ServiceResultException(StatusCodes.BadTimeout);
                     }
-
-                    m_tcs.TrySetResult(m_endpoint.CreateFault(Request, e));
+                    m_vts.SetResult(m_endpoint.CreateFault(Request, e));
                 }
             }
 
@@ -1176,18 +1172,52 @@ public void OperationCompleted(IServiceResponse response, ServiceResult error)
             {
                 if (ServiceResult.IsBad(error))
                 {
-                    m_tcs.TrySetResult(m_endpoint.CreateFault(Request, new ServiceResultException(error)));
+                    m_vts.SetResult(m_endpoint.CreateFault(Request, new ServiceResultException(error)));
                 }
                 else
                 {
-                    m_tcs.TrySetResult(response);
+                    m_vts.SetResult(response);
                 }
             }
 
+            /// <inheritdoc/>
+            public override bool Equals(object obj)
+            {
+                if (obj is EndpointIncomingRequest other)
+                {
+                    return Request.RequestHeader.Equals(other.Request.RequestHeader);
+                }
+                return false;
+            }
+
+            /// <inheritdoc/>
+            public override int GetHashCode()
+            {
+                return Request.RequestHeader.GetHashCode();
+            }
+
+            /// <inheritdoc/>
+            public static bool operator ==(EndpointIncomingRequest left, EndpointIncomingRequest right)
+            {
+                return left.Equals(right);
+            }
+
+            /// <inheritdoc/>
+            public static bool operator !=(EndpointIncomingRequest left, EndpointIncomingRequest right)
+            {
+                return !(left == right);
+            }
+
+            /// <inheritdoc/>
+            public bool Equals(EndpointIncomingRequest other)
+            {
+                return Request.RequestHeader.Equals(other.Request.RequestHeader);
+            }
+
             private readonly EndpointBase m_endpoint;
-            private CancellationToken m_cancellationToken;
-            private ServiceDefinition m_service;
-            private readonly TaskCompletionSource<IServiceResponse> m_tcs;
+            private readonly ServiceDefinition m_service;
+            private readonly ServiceResponsePooledValueTaskSource m_vts;
+            private readonly CancellationToken m_cancellationToken;
         }
 
         /// <summary>

Stack/Opc.Ua.Core/Stack/Server/IServerBase.cs

@@ -105,12 +105,6 @@ public interface IEndpointIncomingRequest
         /// <value>The secure channel context.</value>
         SecureChannelContext SecureChannelContext { get; }
 
-        /// <summary>
-        /// Gets or sets the call data associated with the request.
-        /// </summary>
-        /// <value>The call data.</value>
-        object Calldata { get; set; }
-
         /// <summary>
         /// Used to call the default asynchronous handler.
         /// </summary>
@@ -119,7 +113,7 @@ public interface IEndpointIncomingRequest
         /// thread that calls IServerBase.ScheduleIncomingRequest().
         /// This method always traps any exceptions and reports them to the client as a fault.
         /// </remarks>
-        Task CallAsync(CancellationToken cancellationToken = default);
+        ValueTask CallAsync(CancellationToken cancellationToken = default);
 
         /// <summary>
         /// Used to indicate that the asynchronous operation has completed.

Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs

@@ -1611,7 +1611,7 @@ protected virtual void Dispose(bool disposing)
 
                     foreach (IEndpointIncomingRequest request in m_queue.ToList())
                     {
-                        Utils.SilentDispose(request);
+                        request.OperationCompleted(null, StatusCodes.BadServerHalted);
                     }
 #if NETSTANDARD2_1_OR_GREATER
                     m_queue.Clear();

Stack/Opc.Ua.Core/Stack/Transport/ITransportListenerCallback.cs

@@ -28,7 +28,7 @@ public interface ITransportListenerCallback : IAuditEventCallback
         /// <param name="request">The incoming request.</param>
         /// <param name="cancellationToken">The cancellation token.</param>
         /// <returns>The response to return over the secure channel.</returns>
-        Task<IServiceResponse> ProcessRequestAsync(
+        ValueTask<IServiceResponse> ProcessRequestAsync(
             SecureChannelContext secureChannelContext,
             IServiceRequest request,
             CancellationToken cancellationToken = default);

Stack/Opc.Ua.Core/Types/Utils/ValueTask/ManualResetValueTaskSource.cs

@@ -0,0 +1,73 @@
+/* Copyright (c) 1996-2022 The OPC Foundation. All rights reserved.
+   The source code in this file is covered under a dual-license scenario:
+     - RCL: for OPC Foundation Corporate Members in good-standing
+     - GPL V2: everybody else
+   RCL license terms accompanied with this source code. See http://opcfoundation.org/License/RCL/1.00/
+   GNU General Public License as published by the Free Software Foundation;
+   version 2 of the License are accompanied with this source code. See http://opcfoundation.org/License/GPLv2
+   This source code is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+*/
+
+using System;
+using System.Threading.Tasks;
+using System.Threading.Tasks.Sources;
+
+namespace Opc.Ua
+{
+    /// <summary>
+    /// A reusable value task source.
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    internal class ManualResetValueTaskSource<T> : IValueTaskSource<T>, IValueTaskSource
+    {
+        private ManualResetValueTaskSourceCore<T> m_core;
+
+        public bool RunContinuationsAsynchronously
+        {
+            get => m_core.RunContinuationsAsynchronously;
+            set => m_core.RunContinuationsAsynchronously = value;
+        }
+
+        public short Version => m_core.Version;
+
+        public void Reset()
+        {
+            m_core.Reset();
+        }
+
+        public void SetResult(T result)
+        {
+            m_core.SetResult(result);
+        }
+
+        public void SetException(Exception error)
+        {
+            m_core.SetException(error);
+        }
+
+        public T GetResult(short token)
+        {
+            return m_core.GetResult(token);
+        }
+
+        void IValueTaskSource.GetResult(short token)
+        {
+            m_core.GetResult(token);
+        }
+
+        public ValueTaskSourceStatus GetStatus(short token)
+        {
+            return m_core.GetStatus(token);
+        }
+
+        public void OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags)
+        {
+            m_core.OnCompleted(continuation, state, token, flags);
+        }
+
+        public ValueTask<T> Task => new(this, m_core.Version);
+        public ValueTask SourceTask => new(this, m_core.Version);
+    }
+}

Stack/Opc.Ua.Core/Types/Utils/ValueTask/ObjectPool.cs

@@ -0,0 +1,66 @@
+/* Copyright (c) 1996-2022 The OPC Foundation. All rights reserved.
+   The source code in this file is covered under a dual-license scenario:
+     - RCL: for OPC Foundation Corporate Members in good-standing
+     - GPL V2: everybody else
+   RCL license terms accompanied with this source code. See http://opcfoundation.org/License/RCL/1.00/
+   GNU General Public License as published by the Free Software Foundation;
+   version 2 of the License are accompanied with this source code. See http://opcfoundation.org/License/GPLv2
+   This source code is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+*/
+
+using System;
+using System.Collections.Concurrent;
+
+namespace Opc.Ua
+{
+    /// <summary>
+    /// A simple object pool implementation.
+    /// </summary>
+    /// <typeparam name="T">The type of object to pool.</typeparam>
+    internal class ObjectPool<T> where T : class
+    {
+        private readonly ConcurrentBag<T> m_objects;
+        private readonly Func<T> m_objectGenerator;
+        private readonly int m_maxSize;
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ObjectPool{T}"/> class.
+        /// </summary>
+        /// <param name="objectGenerator">The function to generate new objects.</param>
+        /// <param name="maxSize">The maximum size of the pool.</param>
+        public ObjectPool(Func<T> objectGenerator, int maxSize)
+        {
+            m_objectGenerator = objectGenerator ?? throw new ArgumentNullException(nameof(objectGenerator));
+            m_maxSize = maxSize > 0 ? maxSize : throw new ArgumentOutOfRangeException(nameof(maxSize));
+            m_objects = new ConcurrentBag<T>();
+        }
+
+        /// <summary>
+        /// Gets an object from the pool.
+        /// </summary>
+        /// <returns>An object from the pool or a new one if the pool is empty.</returns>
+        public T Get()
+        {
+            if (m_objects.TryTake(out T item))
+            {
+                return item;
+            }
+
+            return m_objectGenerator();
+        }
+
+        /// <summary>
+        /// Returns an object to the pool.
+        /// </summary>
+        /// <param name="item">The object to return.</param>
+        public void Return(T item)
+        {
+            if (m_objects.Count < m_maxSize)
+            {
+                m_objects.Add(item);
+            }
+        }
+    }
+}