diff --git a/src/Opc.Ua.Core/Stack/Tcp/ChannelAsyncOperation.cs b/src/Opc.Ua.Core/Stack/Tcp/ChannelAsyncOperation.cs index 4c87a81f00..40275518e2 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/ChannelAsyncOperation.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/ChannelAsyncOperation.cs @@ -60,14 +60,8 @@ namespace Opc.Ua.Bindings object? asyncState, ILogger logger, TimeProvider? timeProvider) + : this(callback, asyncState, logger, timeProvider, long.MaxValue) { - m_callback = callback; - m_asyncState = asyncState; - m_synchronous = false; - m_completed = false; - m_logger = logger; - m_timeProvider = timeProvider ?? TimeProvider.System; - if (timeout is > 0 and not int.MaxValue) { m_timer = m_timeProvider.CreateTimer( @@ -78,6 +72,57 @@ namespace Opc.Ua.Bindings } } + /// + /// Initializes the object without a timer of its own: the owner times + /// the operation out with once + /// has passed. + /// + /// Called when the operation completes. + /// The state passed to the callback. + /// Logs errors of the callback. + /// The clock of the deadline. + /// + /// The at which the operation + /// times out, or for none. + /// + private protected ChannelAsyncOperation( + AsyncCallback? callback, + object? asyncState, + ILogger logger, + TimeProvider? timeProvider, + long timeoutDeadline) + { + m_callback = callback; + m_asyncState = asyncState; + m_synchronous = false; + m_completed = false; + m_logger = logger; + m_timeProvider = timeProvider ?? TimeProvider.System; + TimeoutDeadline = timeoutDeadline; + } + + /// + /// The timestamp at which the owner times the operation out, or + /// when the operation has a timer of its + /// own or no timeout. + /// + internal long TimeoutDeadline { get; } + + /// + /// Completes the operation with BadRequestTimeout if + /// is at or before . + /// + /// if the deadline has passed. + internal bool TimeOutIfExpired(long now) + { + if (now < TimeoutDeadline) + { + return false; + } + InternalComplete(false, new ServiceResult(StatusCodes.BadRequestTimeout)); + return true; + } + /// /// Frees any unmanaged resources. /// diff --git a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs index 2d4102d933..880d7a0af1 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs @@ -2606,6 +2606,20 @@ namespace Opc.Ua.Bindings { } + /// + /// Initializes the object without a timer of its own: the channel + /// times it out once has passed. + /// + internal WriteOperation( + AsyncCallback? callback, + object? asyncState, + ILogger logger, + TimeProvider timeProvider, + long timeoutDeadline) + : base(callback, asyncState, logger, timeProvider, timeoutDeadline) + { + } + /// /// The request id associated with the operation. /// diff --git a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs index 8f70d8f1d8..297e827beb 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryClientChannel.cs @@ -180,7 +180,15 @@ namespace Opc.Ua.Bindings m_handshakeTimer = null; m_requestedToken?.Dispose(); m_requestedToken = null; - m_requests?.Clear(); + DisposeTimeoutTimer(); + + // the operations have no timers of their own: without the + // channel's timer a pending operation would wait forever. + foreach (KeyValuePair operation in m_requests.ToArray()) + { + operation.Value.Fault(StatusCodes.BadSecureChannelClosed); + } + m_requests.Clear(); m_handshakeOperation?.Dispose(); m_handshakeOperation = null; ServerCertificate?.Dispose(); @@ -1674,7 +1682,13 @@ namespace Opc.Ua.Bindings { requestId = Utils.IncrementIdentifier(ref m_lastRequestId); } - var operation = new WriteOperation(timeout, callback, state, m_logger, TimeProvider) + long deadline = long.MaxValue; + if (timeout is > 0 and not int.MaxValue) + { + deadline = TimeProvider.GetTimestamp() + + (timeout * TimeProvider.TimestampFrequency / 1000); + } + var operation = new WriteOperation(callback, state, m_logger, TimeProvider, deadline) { RequestId = requestId }; @@ -1684,9 +1698,100 @@ namespace Opc.Ua.Bindings "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); + } return operation; } + /// + /// Arms the timeout timer for unless it is + /// already armed for an earlier or the same time. One timer per channel + /// times out all pending operations instead of a timer per operation; + /// with equal timeouts a new deadline is never the earliest, so a + /// request usually only reads the scheduled time. + /// + private void ScheduleTimeoutScan(long deadline) + { + lock (m_timeoutLock) + { + if (deadline >= m_nextTimeoutScan) + { + return; + } + m_nextTimeoutScan = 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; + + if (m_timeoutTimer == null) + { + m_timeoutTimer = TimeProvider.CreateTimer( + static state => ((UaSCUaBinaryClientChannel)state!).OnTimeoutScan(), + this, + dueTime, + Timeout.InfiniteTimeSpan); + } + else + { + m_timeoutTimer.Change(dueTime, Timeout.InfiniteTimeSpan); + } + } + } + + /// + /// Times out the pending operations whose deadline has passed and arms + /// the timer for the earliest remaining deadline. + /// + private void OnTimeoutScan() + { + lock (m_timeoutLock) + { + // disposed. + if (m_nextTimeoutScan == long.MinValue) + { + return; + } + // an operation added from here on schedules its own deadline. + m_nextTimeoutScan = long.MaxValue; + } + + long now = TimeProvider.GetTimestamp(); + long next = long.MaxValue; + foreach (KeyValuePair entry in m_requests) + { + WriteOperation operation = entry.Value; + if (!operation.TimeOutIfExpired(now) && operation.TimeoutDeadline < next) + { + next = operation.TimeoutDeadline; + } + } + + if (next != long.MaxValue) + { + ScheduleTimeoutScan(next); + } + } + + /// + /// Stops the timeout timer for good. + /// + private void DisposeTimeoutTimer() + { + lock (m_timeoutLock) + { + m_nextTimeoutScan = long.MinValue; + m_timeoutTimer?.Dispose(); + m_timeoutTimer = null; + } + } + /// /// Cleans up after an asychronous operation completes. /// @@ -2000,6 +2105,9 @@ namespace Opc.Ua.Bindings private WriteOperation? m_handshakeOperation; private ChannelToken? m_requestedToken; private ITimer? m_handshakeTimer; + private readonly Lock m_timeoutLock = new(); + private ITimer? m_timeoutTimer; + private long m_nextTimeoutScan = long.MaxValue; private bool m_reconnecting; private int m_waitBetweenReconnects; private readonly IUaSCByteTransportFactory m_transportFactory; diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryClientChannelDeterministicTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryClientChannelDeterministicTests.cs index dfc92ef660..3115733608 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryClientChannelDeterministicTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryClientChannelDeterministicTests.cs @@ -815,6 +815,66 @@ namespace Opc.Ua.Core.Tests.Stack.Transport Assert.That(pool.DuplicateReturns, Is.Zero); } + /// + /// The channel times out pending operations with one timer: each operation + /// fails with BadRequestTimeout once its own deadline passes, a shorter + /// timeout first, and not before. + /// + [Test] + public async Task PendingOperationsTimeOutAtTheirOwnDeadlinesAsync() + { + var timeProvider = new FakeTimeProvider(); + var transport = new RecordingByteTransport(); + var factory = new RecordingByteTransportFactory(transport); + using var channel = new TestClientChannel( + m_buffers, + factory, + m_quotas, + null, + BuildEndpoint(MessageSecurityMode.None, SecurityPolicies.None), + m_telemetry, + timeProvider); + channel.SetupReverseTransport(transport); + + Task connectTask = channel + .ConnectAsync(new Uri("opc.tcp://localhost:4840"), 5000, CancellationToken.None) + .AsTask(); + Assert.That( + await CompletesWithinAsync(transport.FirstSendTask, 30).ConfigureAwait(false), + Is.True, + "channel never sent the Hello message"); + + // queued until the channel is connected. + Task requestTask = channel + .SendRequestAsync(new GetEndpointsRequest(), 1000, CancellationToken.None) + .AsTask(); + + timeProvider.Advance(TimeSpan.FromMilliseconds(999)); + Assert.That(requestTask.IsCompleted, Is.False); + + timeProvider.Advance(TimeSpan.FromMilliseconds(2)); + Assert.That( + await CompletesWithinAsync(requestTask, 30).ConfigureAwait(false), + Is.True, + "the request did not time out"); + ServiceResultException requestError = Assert.ThrowsAsync( + async () => await requestTask.ConfigureAwait(false))!; + Assert.That(requestError.StatusCode, Is.EqualTo((uint)StatusCodes.BadRequestTimeout)); + Assert.That(connectTask.IsCompleted, Is.False); + + timeProvider.Advance(TimeSpan.FromMilliseconds(3998)); + Assert.That(connectTask.IsCompleted, Is.False); + + timeProvider.Advance(TimeSpan.FromMilliseconds(2)); + Assert.That( + await CompletesWithinAsync(connectTask, 30).ConfigureAwait(false), + Is.True, + "the connect did not time out"); + ServiceResultException connectError = Assert.ThrowsAsync( + async () => await connectTask.ConfigureAwait(false))!; + Assert.That(connectError.StatusCode, Is.EqualTo((uint)StatusCodes.BadRequestTimeout)); + } + private async Task RunHandshakeToFaultAsync( Func> feed) {