diff --git a/Stack/Opc.Ua.Core/Stack/Server/EndpointBase.cs b/Stack/Opc.Ua.Core/Stack/Server/EndpointBase.cs index 8eb69d3ec5..1ca8b81dc9 100644 --- a/Stack/Opc.Ua.Core/Stack/Server/EndpointBase.cs +++ b/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) } /// - public Task ProcessRequestAsync( + public ValueTask 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. /// - 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,7 +1075,7 @@ .Body is AdditionalParametersType parameters && /// /// An object that handles an incoming request for an endpoint. /// - protected class EndpointIncomingRequest : IEndpointIncomingRequest + protected readonly struct EndpointIncomingRequest : IEndpointIncomingRequest, IEquatable { /// /// Initialize the Object with a Request @@ -1082,18 +1083,17 @@ protected class EndpointIncomingRequest : IEndpointIncomingRequest public EndpointIncomingRequest( EndpointBase endpoint, SecureChannelContext context, - IServiceRequest request) + IServiceRequest request, + CancellationToken cancellationToken = default) { m_endpoint = endpoint; SecureChannelContext = context; Request = request; - m_tcs = new TaskCompletionSource( - TaskCreationOptions.RunContinuationsAsynchronously); + m_vts = ServiceResponsePooledValueTaskSource.Create(); + m_service = m_endpoint.FindService(Request.TypeId); + m_cancellationToken = cancellationToken; } - /// - public object Calldata { get; set; } - /// public SecureChannelContext SecureChannelContext { get; } @@ -1104,25 +1104,22 @@ public EndpointIncomingRequest( /// Process an incoming request /// /// - public Task ProcessAsync(CancellationToken cancellationToken = default) + public ValueTask 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; } /// - 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); } } + /// + public override bool Equals(object obj) + { + if (obj is EndpointIncomingRequest other) + { + return Request.RequestHeader.Equals(other.Request.RequestHeader); + } + return false; + } + + /// + public override int GetHashCode() + { + return Request.RequestHeader.GetHashCode(); + } + + /// + public static bool operator ==(EndpointIncomingRequest left, EndpointIncomingRequest right) + { + return left.Equals(right); + } + + /// + public static bool operator !=(EndpointIncomingRequest left, EndpointIncomingRequest right) + { + return !(left == right); + } + + /// + 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 m_tcs; + private readonly ServiceDefinition m_service; + private readonly ServiceResponsePooledValueTaskSource m_vts; + private readonly CancellationToken m_cancellationToken; } /// diff --git a/Stack/Opc.Ua.Core/Stack/Server/IServerBase.cs b/Stack/Opc.Ua.Core/Stack/Server/IServerBase.cs index e8ff28d566..71e97eeb99 100644 --- a/Stack/Opc.Ua.Core/Stack/Server/IServerBase.cs +++ b/Stack/Opc.Ua.Core/Stack/Server/IServerBase.cs @@ -105,12 +105,6 @@ public interface IEndpointIncomingRequest /// The secure channel context. SecureChannelContext SecureChannelContext { get; } - /// - /// Gets or sets the call data associated with the request. - /// - /// The call data. - object Calldata { get; set; } - /// /// Used to call the default asynchronous handler. /// @@ -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. /// - Task CallAsync(CancellationToken cancellationToken = default); + ValueTask CallAsync(CancellationToken cancellationToken = default); /// /// Used to indicate that the asynchronous operation has completed. diff --git a/Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs b/Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs index ed27873a2c..483585a62b 100644 --- a/Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs +++ b/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(); diff --git a/Stack/Opc.Ua.Core/Stack/Transport/ITransportListenerCallback.cs b/Stack/Opc.Ua.Core/Stack/Transport/ITransportListenerCallback.cs index 8c4a7bf658..0311ee60b4 100644 --- a/Stack/Opc.Ua.Core/Stack/Transport/ITransportListenerCallback.cs +++ b/Stack/Opc.Ua.Core/Stack/Transport/ITransportListenerCallback.cs @@ -28,7 +28,7 @@ public interface ITransportListenerCallback : IAuditEventCallback /// The incoming request. /// The cancellation token. /// The response to return over the secure channel. - Task ProcessRequestAsync( + ValueTask ProcessRequestAsync( SecureChannelContext secureChannelContext, IServiceRequest request, CancellationToken cancellationToken = default); diff --git a/Stack/Opc.Ua.Core/Types/Utils/ValueTask/ManualResetValueTaskSource.cs b/Stack/Opc.Ua.Core/Types/Utils/ValueTask/ManualResetValueTaskSource.cs new file mode 100644 index 0000000000..dc621bd2b7 --- /dev/null +++ b/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 +{ + /// + /// A reusable value task source. + /// + /// + internal class ManualResetValueTaskSource : IValueTaskSource, IValueTaskSource + { + private ManualResetValueTaskSourceCore 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 continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags) + { + m_core.OnCompleted(continuation, state, token, flags); + } + + public ValueTask Task => new(this, m_core.Version); + public ValueTask SourceTask => new(this, m_core.Version); + } +} diff --git a/Stack/Opc.Ua.Core/Types/Utils/ValueTask/ObjectPool.cs b/Stack/Opc.Ua.Core/Types/Utils/ValueTask/ObjectPool.cs new file mode 100644 index 0000000000..70edec9004 --- /dev/null +++ b/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 +{ + /// + /// A simple object pool implementation. + /// + /// The type of object to pool. + internal class ObjectPool where T : class + { + private readonly ConcurrentBag m_objects; + private readonly Func m_objectGenerator; + private readonly int m_maxSize; + + /// + /// Initializes a new instance of the class. + /// + /// The function to generate new objects. + /// The maximum size of the pool. + public ObjectPool(Func 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(); + } + + /// + /// Gets an object from the pool. + /// + /// An object from the pool or a new one if the pool is empty. + public T Get() + { + if (m_objects.TryTake(out T item)) + { + return item; + } + + return m_objectGenerator(); + } + + /// + /// Returns an object to the pool. + /// + /// The object to return. + public void Return(T item) + { + if (m_objects.Count < m_maxSize) + { + m_objects.Add(item); + } + } + } +} diff --git a/Stack/Opc.Ua.Core/Types/Utils/ValueTask/ServiceResponsePooledValueTaskSource.cs b/Stack/Opc.Ua.Core/Types/Utils/ValueTask/ServiceResponsePooledValueTaskSource.cs new file mode 100644 index 0000000000..68f85b8724 --- /dev/null +++ b/Stack/Opc.Ua.Core/Types/Utils/ValueTask/ServiceResponsePooledValueTaskSource.cs @@ -0,0 +1,128 @@ +/* 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; +using System.Threading.Tasks; +using System.Threading.Tasks.Sources; + +namespace Opc.Ua +{ + /// + /// A pooled value task source for IServiceResponse. + /// + internal sealed class ServiceResponsePooledValueTaskSource : IValueTaskSource, IValueTaskSource + { + private static readonly ObjectPool s_pool = + new(() => new ServiceResponsePooledValueTaskSource(), 1024); + + private readonly ManualResetValueTaskSource m_source; + private int m_resultRetrieved; + + /// + /// Private constructor to enforce pooling. + /// + private ServiceResponsePooledValueTaskSource() + { + m_source = new ManualResetValueTaskSource(); + } + + /// + /// Creates or gets a pooled instance. + /// + public static ServiceResponsePooledValueTaskSource Create() + { + ServiceResponsePooledValueTaskSource source = s_pool.Get(); + source.m_resultRetrieved = 0; + return source; + } + + /// + /// Returns the object to the pool. + /// + private void ReturnToPool() + { + if (Interlocked.CompareExchange(ref m_resultRetrieved, 1, 0) == 0) + { + m_source.Reset(); + s_pool.Return(this); + } + } + + /// + /// The value task to await. + /// + public ValueTask Task => new(this, Version); + + /// + /// The value task to await. + /// + public ValueTask SourceTask => new(this, Version); + + /// + public short Version => m_source.Version; + + /// + /// Set the result of the task. + /// + public void SetResult(IServiceResponse result) + { + m_source.SetResult(result); + } + + /// + /// Set an exception for the task. + /// + public void SetException(Exception error) + { + m_source.SetException(error); + } + + /// + public IServiceResponse GetResult(short token) + { + try + { + return m_source.GetResult(token); + } + finally + { + ReturnToPool(); + } + } + + /// + void IValueTaskSource.GetResult(short token) + { + try + { + ((IValueTaskSource)m_source).GetResult(token); + } + finally + { + ReturnToPool(); + } + } + + /// + public ValueTaskSourceStatus GetStatus(short token) + { + return m_source.GetStatus(token); + } + + /// + public void OnCompleted(Action continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags) + { + m_source.OnCompleted(continuation, state, token, flags); + } + } +} diff --git a/Tests/Opc.Ua.Client.Tests/LoadTest.cs b/Tests/Opc.Ua.Client.Tests/LoadTest.cs index 56eb7742fd..b0117f6af4 100644 --- a/Tests/Opc.Ua.Client.Tests/LoadTest.cs +++ b/Tests/Opc.Ua.Client.Tests/LoadTest.cs @@ -236,8 +236,9 @@ public async Task ServerLoadTestAsync() /* expected */ } - // Wait for notifications to be processed - await Task.Delay(publishingInterval * 10).ConfigureAwait(false); + // Wait for server to process last write (testDurationSeconds / writeCount -> time for a single write) + // + some publishing intervals for notifications to be processed + await Task.Delay((testDurationSeconds / (writeCount - 1)) + (publishingInterval * 10)).ConfigureAwait(false); // Verification TestContext.Out.WriteLine($"Writer task wrote {writeCount} times.");