mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-27 13:10:37 +08:00
Client channel: no request admitted after Dispose; operation keeps a private Lock
Review feedback on #4849: - Dispose stopped the timeout timer and failed the requests it found, but a request registered concurrently could be added after that snapshot and, with the timer gone, wait forever. Registration now happens under the timeout lock and is rejected with BadSecureChannelClosed once the channel is disposed, so every admitted request is in the drain. Tests for a request after Dispose and for requests racing Dispose (both hang without the fix). - ChannelAsyncOperation is public and inheritable, so lock (this) exposed its lock (docs/DeveloperGuide.md: never expose locks, use System.Threading.Lock). The private Lock is back; this costs ~49 B per request (read --nodes 1: 1423 -> 1472 B). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
41859b64a3
commit
b3d09802d0
@@ -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
|
||||
{
|
||||
/// <summary>
|
||||
@@ -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
|
||||
/// </exception>
|
||||
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
|
||||
/// </summary>
|
||||
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<T>)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;
|
||||
|
||||
@@ -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;
|
||||
/// <summary>
|
||||
/// <see cref="ScheduleTimeoutScan"/> with <see cref="m_timeoutLock"/> held.
|
||||
/// </summary>
|
||||
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
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private void DisposeTimeoutTimer()
|
||||
{
|
||||
|
||||
@@ -875,6 +875,81 @@ namespace Opc.Ua.Core.Tests.Stack.Transport
|
||||
Assert.That(connectError.StatusCode, Is.EqualTo((uint)StatusCodes.BadRequestTimeout));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Every request that races Dispose completes: it is either rejected or
|
||||
/// failed by Dispose, never left without a timer.
|
||||
/// </summary>
|
||||
[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<ServiceResultException> RunHandshakeToFaultAsync(
|
||||
Func<TestClientChannel, ValueTask<bool>> feed)
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user