diff --git a/src/Opc.Ua.Core/Stack/Tcp/ChannelAsyncOperation.cs b/src/Opc.Ua.Core/Stack/Tcp/ChannelAsyncOperation.cs index 32b4c4c1b1..af3ab0d4e4 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/ChannelAsyncOperation.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/ChannelAsyncOperation.cs @@ -34,10 +34,6 @@ using System.Threading.Tasks; using System.Threading.Tasks.Sources; using Microsoft.Extensions.Logging; -// An operation locks itself: a lock object of its own would be one more allocation -// per request, and nothing else locks an operation. -#pragma warning disable CA2002 // Do not lock on objects with weak identity - namespace Opc.Ua.Bindings { /// @@ -144,7 +140,7 @@ namespace Opc.Ua.Bindings { if (disposing) { - lock (this) + lock (m_lock) { m_timer?.Dispose(); m_timer = null; @@ -265,7 +261,7 @@ namespace Opc.Ua.Bindings { // check if the request has already completed. bool mustWait; - lock (this) + lock (m_lock) { mustWait = !m_completed; @@ -287,7 +283,7 @@ namespace Opc.Ua.Bindings } finally { - lock (this) + lock (m_lock) { // Dispose the event m_event?.Dispose(); @@ -297,7 +293,7 @@ namespace Opc.Ua.Bindings } // return the response. - lock (this) + lock (m_lock) { if (m_error != null && throwOnError) { @@ -319,7 +315,7 @@ namespace Opc.Ua.Bindings { // check if the request has already completed. bool mustWait; - lock (this) + lock (m_lock) { mustWait = !m_completed; @@ -369,7 +365,7 @@ namespace Opc.Ua.Bindings finally { cancellation.Dispose(); - lock (this) + lock (m_lock) { m_tcs = null; } @@ -382,7 +378,7 @@ namespace Opc.Ua.Bindings } // return the response. - lock (this) + lock (m_lock) { if (m_error != null && throwOnError) { @@ -410,7 +406,7 @@ namespace Opc.Ua.Bindings /// internal ValueTask WaitForCompletionAsync(CancellationToken ct) { - lock (this) + lock (m_lock) { if (m_completed) { @@ -456,7 +452,7 @@ namespace Opc.Ua.Bindings finally { m_waitRegistration.Dispose(); - lock (this) + lock (m_lock) { m_waiting = false; } @@ -467,7 +463,7 @@ namespace Opc.Ua.Bindings throw new ServiceResultException(StatusCodes.BadRequestInterrupted); } - lock (this) + lock (m_lock) { if (m_error != null) { @@ -499,7 +495,7 @@ namespace Opc.Ua.Bindings { get { - lock (this) + lock (m_lock) { m_properties ??= []; @@ -518,7 +514,7 @@ namespace Opc.Ua.Bindings { get { - lock (this) + lock (m_lock) { return m_asyncState; } @@ -530,7 +526,7 @@ namespace Opc.Ua.Bindings { get { - lock (this) + lock (m_lock) { m_event ??= new ManualResetEvent(m_completed); @@ -544,7 +540,7 @@ namespace Opc.Ua.Bindings { get { - lock (this) + lock (m_lock) { return m_synchronous; } @@ -556,7 +552,7 @@ namespace Opc.Ua.Bindings { get { - lock (this) + lock (m_lock) { return m_completed; } @@ -579,7 +575,7 @@ namespace Opc.Ua.Bindings /// protected virtual bool InternalComplete(bool doNotBlock, object? result) { - lock (this) + lock (m_lock) { // ignore multiple calls (i.e. a timeout after a response or vise versa). if (m_completed) @@ -667,6 +663,7 @@ namespace Opc.Ua.Bindings private static readonly TimerCallback s_onTimeout = static state => ((ChannelAsyncOperation)state!).OnTimeout(null); private static readonly object? s_boxedDefault = default(T); + private readonly Lock m_lock = new(); private readonly AsyncCallback? m_callback; private readonly object? m_asyncState; private readonly bool m_synchronous; diff --git a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs index 0f989e2679..eb54ff1a02 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs @@ -1700,17 +1700,28 @@ namespace Opc.Ua.Bindings { RequestId = requestId }; - if (!m_requests.TryAdd(operation.RequestId, operation)) + lock (m_timeoutLock) { - throw ServiceResultException.Unexpected( - "Could not add request {0} to list of pending operations.", - operation.RequestId); - } - // added first, so that a scan that starts after the deadline is - // scheduled finds the operation. - if (deadline != long.MaxValue) - { - ScheduleTimeoutScan(deadline); + // Dispose fails the operations it finds after it stopped the timer; + // an operation added later would neither be failed nor time out. + if (m_nextTimeoutScan == long.MinValue) + { + throw ServiceResultException.Create( + StatusCodes.BadSecureChannelClosed, + "The channel is disposed."); + } + if (!m_requests.TryAdd(operation.RequestId, operation)) + { + throw ServiceResultException.Unexpected( + "Could not add request {0} to list of pending operations.", + operation.RequestId); + } + // added first, so that a scan that starts after the deadline is + // scheduled finds the operation. + if (deadline != long.MaxValue) + { + ScheduleTimeoutScanLocked(deadline); + } } return operation; } @@ -1726,30 +1737,38 @@ namespace Opc.Ua.Bindings { lock (m_timeoutLock) { - if (deadline >= m_nextTimeoutScan) - { - return; - } - m_nextTimeoutScan = deadline; + ScheduleTimeoutScanLocked(deadline); + } + } - // rounded up: a timer that fires early finds nothing expired. - TimeSpan dueTime = TimeProvider.GetElapsedTime(TimeProvider.GetTimestamp(), deadline); - dueTime = dueTime > TimeSpan.Zero - ? TimeSpan.FromMilliseconds(Math.Ceiling(dueTime.TotalMilliseconds)) - : TimeSpan.Zero; + /// + /// with held. + /// + private void ScheduleTimeoutScanLocked(long deadline) + { + if (deadline >= m_nextTimeoutScan) + { + return; + } + m_nextTimeoutScan = deadline; - if (m_timeoutTimer == null) - { - m_timeoutTimer = TimeProvider.CreateTimer( - static state => ((UaSCUaBinaryClientChannel)state!).OnTimeoutScan(), - this, - dueTime, - Timeout.InfiniteTimeSpan); - } - else - { - m_timeoutTimer.Change(dueTime, Timeout.InfiniteTimeSpan); - } + // rounded up: a timer that fires early finds nothing expired. + TimeSpan dueTime = TimeProvider.GetElapsedTime(TimeProvider.GetTimestamp(), deadline); + dueTime = dueTime > TimeSpan.Zero + ? TimeSpan.FromMilliseconds(Math.Ceiling(dueTime.TotalMilliseconds)) + : TimeSpan.Zero; + + if (m_timeoutTimer == null) + { + m_timeoutTimer = TimeProvider.CreateTimer( + static state => ((UaSCUaBinaryClientChannel)state!).OnTimeoutScan(), + this, + dueTime, + Timeout.InfiniteTimeSpan); + } + else + { + m_timeoutTimer.Change(dueTime, Timeout.InfiniteTimeSpan); } } @@ -1788,7 +1807,8 @@ namespace Opc.Ua.Bindings } /// - /// Stops the timeout timer for good. + /// Stops the timeout timer for good. Operations are no longer added from + /// here on, so the ones pending afterwards are all the channel holds. /// private void DisposeTimeoutTimer() { diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryClientChannelDeterministicTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryClientChannelDeterministicTests.cs index 3115733608..c9871e6af9 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryClientChannelDeterministicTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryClientChannelDeterministicTests.cs @@ -875,6 +875,81 @@ namespace Opc.Ua.Core.Tests.Stack.Transport Assert.That(connectError.StatusCode, Is.EqualTo((uint)StatusCodes.BadRequestTimeout)); } + /// + /// A disposed channel has no timer to time requests out, so it must not + /// admit one: a request that reaches the channel after Dispose fails at + /// once instead of waiting for a response that never comes. + /// + [Test] + [CancelAfter(30000)] + public async Task RequestAfterDisposeFailsInsteadOfWaitingAsync() + { + var channel = new TestClientChannel( + m_buffers, + new RecordingByteTransportFactory(), + m_quotas, + null, + BuildEndpoint(MessageSecurityMode.None, SecurityPolicies.None), + m_telemetry, + new FakeTimeProvider()); + channel.Dispose(); + // a connecting channel queues the request instead of sending it. + channel.CurrentState = TcpChannelState.Connecting; + + Task request = channel + .SendRequestAsync(new GetEndpointsRequest(), 60000, CancellationToken.None) + .AsTask(); + + Assert.That( + await CompletesWithinAsync(request, 10).ConfigureAwait(false), + Is.True, + "the request was admitted by the disposed channel"); + Assert.That(request.IsFaulted, Is.True); + } + + /// + /// Every request that races Dispose completes: it is either rejected or + /// failed by Dispose, never left without a timer. + /// + [Test] + [CancelAfter(60000)] + public async Task RequestsRacingDisposeAllCompleteAsync() + { + for (int ii = 0; ii < 200; ii++) + { + var channel = new TestClientChannel( + m_buffers, + new RecordingByteTransportFactory(), + m_quotas, + null, + BuildEndpoint(MessageSecurityMode.None, SecurityPolicies.None), + m_telemetry, + new FakeTimeProvider()); + channel.CurrentState = TcpChannelState.Connecting; + + using var start = new Barrier(2); + // Task.Run unwraps the request task. + Task request = Task.Run(() => + { + start.SignalAndWait(); + return channel + .SendRequestAsync(new GetEndpointsRequest(), 60000, CancellationToken.None) + .AsTask(); + }); + Task dispose = Task.Run(() => + { + start.SignalAndWait(); + channel.Dispose(); + }); + await dispose.ConfigureAwait(false); + + Assert.That( + await CompletesWithinAsync(request, 10).ConfigureAwait(false), + Is.True, + $"request {ii} still waits after Dispose"); + } + } + private async Task RunHandshakeToFaultAsync( Func> feed) {