mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-28 14:09:56 +08:00
Client channel: one timeout timer per channel instead of one per request
Every request created an ITimer for its timeout (TimerQueueTimer + SystemTimeProviderTimer), ~113 B of the ~1.98 KB a Read-1 request allocates on the client. The operation now records its deadline and the client channel keeps one timer armed for the earliest deadline among its pending requests; the scan times out the expired ones with BadRequestTimeout and re-arms for the next. With equal timeouts a new request never moves the earliest deadline, so it only reads it under a lock. Dispose faults the requests still pending with BadSecureChannelClosed, since without the channel's timer they would no longer time out. ServerLoadHarness read --nodes 1 (20 sessions, 4 alternating rounds): client allocB/op 1983 -> 1871. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
484596c2a8
commit
77e2ca0b5a
@@ -60,14 +60,8 @@ namespace Opc.Ua.Bindings
|
|||||||
object? asyncState,
|
object? asyncState,
|
||||||
ILogger logger,
|
ILogger logger,
|
||||||
TimeProvider? timeProvider)
|
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)
|
if (timeout is > 0 and not int.MaxValue)
|
||||||
{
|
{
|
||||||
m_timer = m_timeProvider.CreateTimer(
|
m_timer = m_timeProvider.CreateTimer(
|
||||||
@@ -78,6 +72,57 @@ namespace Opc.Ua.Bindings
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes the object without a timer of its own: the owner times
|
||||||
|
/// the operation out with <see cref="TimeOutIfExpired"/> once
|
||||||
|
/// <paramref name="timeoutDeadline"/> has passed.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="callback">Called when the operation completes.</param>
|
||||||
|
/// <param name="asyncState">The state passed to the callback.</param>
|
||||||
|
/// <param name="logger">Logs errors of the callback.</param>
|
||||||
|
/// <param name="timeProvider">The clock of the deadline.</param>
|
||||||
|
/// <param name="timeoutDeadline">
|
||||||
|
/// The <see cref="TimeProvider.GetTimestamp"/> at which the operation
|
||||||
|
/// times out, or <see cref="long.MaxValue"/> for none.
|
||||||
|
/// </param>
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The timestamp at which the owner times the operation out, or
|
||||||
|
/// <see cref="long.MaxValue"/> when the operation has a timer of its
|
||||||
|
/// own or no timeout.
|
||||||
|
/// </summary>
|
||||||
|
internal long TimeoutDeadline { get; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Completes the operation with BadRequestTimeout if
|
||||||
|
/// <see cref="TimeoutDeadline"/> is at or before <paramref name="now"/>.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns><see langword="true"/> if the deadline has passed.</returns>
|
||||||
|
internal bool TimeOutIfExpired(long now)
|
||||||
|
{
|
||||||
|
if (now < TimeoutDeadline)
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
InternalComplete(false, new ServiceResult(StatusCodes.BadRequestTimeout));
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Frees any unmanaged resources.
|
/// Frees any unmanaged resources.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
|||||||
@@ -2606,6 +2606,20 @@ namespace Opc.Ua.Bindings
|
|||||||
{
|
{
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes the object without a timer of its own: the channel
|
||||||
|
/// times it out once <paramref name="timeoutDeadline"/> has passed.
|
||||||
|
/// </summary>
|
||||||
|
internal WriteOperation(
|
||||||
|
AsyncCallback? callback,
|
||||||
|
object? asyncState,
|
||||||
|
ILogger logger,
|
||||||
|
TimeProvider timeProvider,
|
||||||
|
long timeoutDeadline)
|
||||||
|
: base(callback, asyncState, logger, timeProvider, timeoutDeadline)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The request id associated with the operation.
|
/// The request id associated with the operation.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
|||||||
@@ -180,7 +180,15 @@ namespace Opc.Ua.Bindings
|
|||||||
m_handshakeTimer = null;
|
m_handshakeTimer = null;
|
||||||
m_requestedToken?.Dispose();
|
m_requestedToken?.Dispose();
|
||||||
m_requestedToken = null;
|
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<uint, WriteOperation> operation in m_requests.ToArray())
|
||||||
|
{
|
||||||
|
operation.Value.Fault(StatusCodes.BadSecureChannelClosed);
|
||||||
|
}
|
||||||
|
m_requests.Clear();
|
||||||
m_handshakeOperation?.Dispose();
|
m_handshakeOperation?.Dispose();
|
||||||
m_handshakeOperation = null;
|
m_handshakeOperation = null;
|
||||||
ServerCertificate?.Dispose();
|
ServerCertificate?.Dispose();
|
||||||
@@ -1674,7 +1682,13 @@ namespace Opc.Ua.Bindings
|
|||||||
{
|
{
|
||||||
requestId = Utils.IncrementIdentifier(ref m_lastRequestId);
|
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
|
RequestId = requestId
|
||||||
};
|
};
|
||||||
@@ -1684,9 +1698,100 @@ namespace Opc.Ua.Bindings
|
|||||||
"Could not add request {0} to list of pending operations.",
|
"Could not add request {0} to list of pending operations.",
|
||||||
operation.RequestId);
|
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;
|
return operation;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Arms the timeout timer for <paramref name="deadline"/> 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.
|
||||||
|
/// </summary>
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Times out the pending operations whose deadline has passed and arms
|
||||||
|
/// the timer for the earliest remaining deadline.
|
||||||
|
/// </summary>
|
||||||
|
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<uint, WriteOperation> entry in m_requests)
|
||||||
|
{
|
||||||
|
WriteOperation operation = entry.Value;
|
||||||
|
if (!operation.TimeOutIfExpired(now) && operation.TimeoutDeadline < next)
|
||||||
|
{
|
||||||
|
next = operation.TimeoutDeadline;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (next != long.MaxValue)
|
||||||
|
{
|
||||||
|
ScheduleTimeoutScan(next);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Stops the timeout timer for good.
|
||||||
|
/// </summary>
|
||||||
|
private void DisposeTimeoutTimer()
|
||||||
|
{
|
||||||
|
lock (m_timeoutLock)
|
||||||
|
{
|
||||||
|
m_nextTimeoutScan = long.MinValue;
|
||||||
|
m_timeoutTimer?.Dispose();
|
||||||
|
m_timeoutTimer = null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Cleans up after an asychronous operation completes.
|
/// Cleans up after an asychronous operation completes.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -2000,6 +2105,9 @@ namespace Opc.Ua.Bindings
|
|||||||
private WriteOperation? m_handshakeOperation;
|
private WriteOperation? m_handshakeOperation;
|
||||||
private ChannelToken? m_requestedToken;
|
private ChannelToken? m_requestedToken;
|
||||||
private ITimer? m_handshakeTimer;
|
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 bool m_reconnecting;
|
||||||
private int m_waitBetweenReconnects;
|
private int m_waitBetweenReconnects;
|
||||||
private readonly IUaSCByteTransportFactory m_transportFactory;
|
private readonly IUaSCByteTransportFactory m_transportFactory;
|
||||||
|
|||||||
@@ -815,6 +815,66 @@ namespace Opc.Ua.Core.Tests.Stack.Transport
|
|||||||
Assert.That(pool.DuplicateReturns, Is.Zero);
|
Assert.That(pool.DuplicateReturns, Is.Zero);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
[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<ServiceResultException>(
|
||||||
|
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<ServiceResultException>(
|
||||||
|
async () => await connectTask.ConfigureAwait(false))!;
|
||||||
|
Assert.That(connectError.StatusCode, Is.EqualTo((uint)StatusCodes.BadRequestTimeout));
|
||||||
|
}
|
||||||
|
|
||||||
private async Task<ServiceResultException> RunHandshakeToFaultAsync(
|
private async Task<ServiceResultException> RunHandshakeToFaultAsync(
|
||||||
Func<TestClientChannel, ValueTask<bool>> feed)
|
Func<TestClientChannel, ValueTask<bool>> feed)
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user