diff --git a/src/Opc.Ua.Client/Session/Subscription/ClassicSubscriptionEngine.cs b/src/Opc.Ua.Client/Session/Subscription/ClassicSubscriptionEngine.cs
index d01fdd1005..76de95d470 100644
--- a/src/Opc.Ua.Client/Session/Subscription/ClassicSubscriptionEngine.cs
+++ b/src/Opc.Ua.Client/Session/Subscription/ClassicSubscriptionEngine.cs
@@ -134,14 +134,11 @@ public void StartPublishing(int timeout, bool fullQueue)
return;
}
- // Refill the pipeline up to the desired count. Every request is
- // bounded by the in flight reservation: StartPublishing runs once
- // per subscription as the subscriptions are created, so an
- // unbounded send stacks a pipeline worth of requests on every call
- // and the outstanding count grows far past the desired count. A
- // drained pipeline always has room, so this still refills a
- // pipeline whose requests are outstanding but are no longer
- // expected to return.
+ // Refill the pipeline up to the desired count. The reservation is
+ // the real bound: it counts the requests the session still expects
+ // plus the ones sent but not yet recorded, so this both refuses to
+ // overshoot and still refills a pipeline whose requests the session
+ // has written off.
int startCount = fullQueue
? 0
: GoodPublishRequestCount;
@@ -154,7 +151,7 @@ public void StartPublishing(int timeout, bool fullQueue)
if (!BeginPublishCore(timeout))
{
- Interlocked.Decrement(ref m_publishRequestsInFlight);
+ Interlocked.Decrement(ref m_unrecordedPublishRequests);
break;
}
}
@@ -225,38 +222,44 @@ protected virtual void Dispose(bool disposing)
///
/// Sends a publish request to the server.
///
+ ///
+ /// This is the recovery nudge behind
+ /// : a subscription
+ /// that has seen no notification asks for one more publish so the
+ /// server has a request to answer. It is bounded by the same
+ /// reservation as the automatic top up, because sending past the
+ /// desired count cannot help - the server already holds that many
+ /// requests and answers the surplus with
+ /// - while every
+ /// subscription firing this nudge at once would otherwise multiply the
+ /// pipeline by the number of subscriptions. A drained pipeline always
+ /// has room, so the nudge still gets through when it is the outstanding
+ /// requests that have stopped coming back.
+ ///
/// The timeout for publish requests
/// in milliseconds.
- /// True if the request was sent successfully.
+ ///
+ /// True if the pipeline holds the requested publish request when this
+ /// returns, whether it was sent here or was already outstanding.
+ ///
internal bool BeginPublish(int timeout)
{
- // An explicit request to send a publish (e.g. keep-alive recovery).
- // Respect the server-side limit when it has been learned from a
- // BadTooManyPublishRequests response: if the pipeline is already at
- // or above the desired count, delegate to the ordinary top-up path
- // so the reservation check prevents overshooting.
- int desiredCount = GetDesiredPublishRequestCount(false);
-
- if (desiredCount > 0 &&
- !TryReservePublishRequest(desiredCount, out _))
- {
- // Pipeline is already at capacity; queue a top-up instead so
- // the next completed request will refill the slot naturally.
- QueueBeginPublish();
- return true;
- }
+ // At least one, so an empty pipeline is always refillable even
+ // when no subscription has been created yet.
+ int limit = Math.Max(1, GetDesiredPublishRequestCount(false));
- // Either no desired count yet (engine not yet started) or we
- // successfully reserved a slot. Proceed unconditionally so that
- // a stalled pipeline can always be kick-started.
- if (desiredCount == 0)
+ if (!TryReservePublishRequest(limit, out _))
{
- Interlocked.Increment(ref m_publishRequestsInFlight);
+ // The pipeline already holds the requests this nudge asks for,
+ // so the caller's intent is met without sending. Sending anyway
+ // cannot help: the server holds that many requests already and
+ // answers the surplus with BadTooManyPublishRequests.
+ return true;
}
if (!BeginPublishCore(timeout))
{
- Interlocked.Decrement(ref m_publishRequestsInFlight);
+ Interlocked.Decrement(ref m_unrecordedPublishRequests);
return false;
}
@@ -264,9 +267,10 @@ internal bool BeginPublish(int timeout)
}
///
- /// Sends a publish request. The caller must have accounted for the
- /// request in and must release
- /// that reservation if this returns false.
+ /// Sends a publish request. The caller must have reserved the request
+ /// in . The reservation is
+ /// released here once the session has recorded the request, or by the
+ /// caller if this returns false.
///
private bool BeginPublishCore(int timeout)
{
@@ -349,6 +353,13 @@ private bool BeginPublishCore(int timeout)
activity,
requestHeader.RequestHandle,
DataTypes.PublishRequest);
+
+ // The session now counts this request, so the reservation that
+ // stood in for it is released. Releasing after the record is
+ // published keeps the total conservative: the request is
+ // briefly counted twice rather than not at all.
+ Interlocked.Decrement(ref m_unrecordedPublishRequests);
+
task.ConfigureAwait(false)
.GetAwaiter()
.OnCompleted(() => OnPublishComplete(
@@ -383,14 +394,6 @@ private void OnPublishComplete(
requestHeader.RequestHandle,
DataTypes.PublishRequest);
- // Release the reservation only once the session has retired the
- // request from its outstanding list. Releasing it when the task
- // completed would open a window in which a concurrent top up sees
- // a free slot while GoodPublishRequestCount still counts this
- // request, so the pipeline drifts above the limit by the number of
- // completions that overlap.
- Interlocked.Decrement(ref m_publishRequestsInFlight);
-
m_eventLogger.ClientEventPublishStop(
(int)requestHeader.RequestHandle,
sessionId);
@@ -877,39 +880,56 @@ private void QueueBeginPublish()
if (!BeginPublishCore(m_context.OperationTimeout))
{
- Interlocked.Decrement(ref m_publishRequestsInFlight);
+ Interlocked.Decrement(ref m_unrecordedPublishRequests);
}
}
///
/// Atomically reserves capacity for one more publish request if fewer
- /// than are in flight.
+ /// than are outstanding.
///
- /// The maximum number of requests in flight.
- /// The number observed in flight.
+ ///
+ /// Outstanding means the requests the session still expects to return
+ /// ()
+ /// plus the requests this engine has sent that the session has not
+ /// recorded yet. The session's count is authoritative: it drops a
+ /// request as soon as the session writes it off, which is what
+ /// does to the whole pipeline when
+ /// keep alives recover. Counting only what this engine has sent would
+ /// hold those write offs forever and the pipeline would never refill.
+ /// The unrecorded count closes the opposite gap: the session only
+ /// counts a request once AsyncRequestStarted has recorded it,
+ /// which happens after the request was issued, so concurrent callers
+ /// would otherwise all read the same lagging value and each send.
+ ///
+ /// The maximum number of outstanding requests.
+ /// The number observed outstanding.
/// True if a slot was reserved.
- private bool TryReservePublishRequest(int limit, out int inFlight)
+ private bool TryReservePublishRequest(int limit, out int outstanding)
{
- int current = Volatile.Read(ref m_publishRequestsInFlight);
-
- while (current < limit)
+ while (true)
{
- int prior = Interlocked.CompareExchange(
- ref m_publishRequestsInFlight,
- current + 1,
- current);
+ // Read the unrecorded count first. A request that moves from
+ // unrecorded to recorded between the two reads is then counted
+ // twice rather than missed, which errs towards sending less.
+ int unrecorded = Volatile.Read(ref m_unrecordedPublishRequests);
+ int current = m_context.GoodPublishRequestCount + unrecorded;
- if (prior == current)
+ if (current >= limit)
{
- inFlight = current;
- return true;
+ outstanding = current;
+ return false;
}
- current = prior;
+ if (Interlocked.CompareExchange(
+ ref m_unrecordedPublishRequests,
+ unrecorded + 1,
+ unrecorded) == unrecorded)
+ {
+ outstanding = current;
+ return true;
+ }
}
-
- inFlight = current;
- return false;
}
///
@@ -1150,7 +1170,7 @@ private bool BelowPublishRequestLimit(int requestCount)
private readonly Lock m_acknowledgementsToSendLock = new();
private List m_acknowledgementsToSend = [];
internal uint PublishCounter;
- private int m_publishRequestsInFlight;
+ private int m_unrecordedPublishRequests;
private int m_tooManyPublishRequests;
private int m_minPublishRequestCount;
private int m_maxPublishRequestCount;
diff --git a/tests/Opc.Ua.Client.Tests/Session/ClassicSubscriptionEngineTests.cs b/tests/Opc.Ua.Client.Tests/Session/ClassicSubscriptionEngineTests.cs
index 3dd235c32b..b36c3c65c1 100644
--- a/tests/Opc.Ua.Client.Tests/Session/ClassicSubscriptionEngineTests.cs
+++ b/tests/Opc.Ua.Client.Tests/Session/ClassicSubscriptionEngineTests.cs
@@ -622,7 +622,11 @@ public void ConcurrentPublishReEvaluationDoesNotExceedDesiredRequestCount()
.Callback(() => Interlocked.Increment(ref outstanding));
// Requests never complete, so nothing drains the outstanding count.
- var pending = new TaskCompletionSource();
+ // RunContinuationsAsynchronously keeps the completion below off
+ // this thread, so cancelling cannot re-enter the engine inline and
+ // issue further publishes while the assertion is evaluated.
+ var pending = new TaskCompletionSource(
+ TaskCreationOptions.RunContinuationsAsynchronously);
int issued = 0;
m_mockContext.Setup(c => c.PublishAsync(
It.IsAny(),
@@ -652,15 +656,183 @@ public void ConcurrentPublishReEvaluationDoesNotExceedDesiredRequestCount()
}
Task.WaitAll(threads);
+
+ // Sample before completing the pending requests so the result is
+ // fixed by the re-evaluation alone.
+ int issuedByReEvaluation = Volatile.Read(ref issued);
pending.TrySetCanceled();
Assert.That(
- Volatile.Read(ref issued),
+ issuedByReEvaluation,
Is.LessThanOrEqualTo(subscriptionCount),
"Concurrent publish re-evaluation must not issue more requests " +
"than the desired publish request count.");
}
+ [Test]
+ public void StartPublishingRefillsPipelineAfterRequestsAreWrittenOff()
+ {
+ const int subscriptionCount = 5;
+
+ var subscriptions = new List();
+ for (int ii = 0; ii < subscriptionCount; ii++)
+ {
+ var subscription = new Subscription(m_telemetry)
+ {
+ CurrentPublishingInterval = 100,
+ CurrentLifetimeCount = 10
+ };
+
+ // Created is Id != 0, and the server assigns the id, so the
+ // desired publish count only counts subscriptions the server
+ // has acknowledged.
+ typeof(Subscription).GetProperty(
+ nameof(Subscription.Id),
+ BindingFlags.Instance | BindingFlags.Public)!
+ .SetValue(subscription, (uint)(ii + 1));
+
+ subscriptions.Add(subscription);
+ }
+
+ m_mockContext.Setup(c => c.Connected).Returns(true);
+ m_mockContext.Setup(c => c.Subscriptions).Returns(subscriptions);
+ m_mockContext.Setup(c => c.PrepareAcknowledgementsToSend(
+ It.IsAny>()))
+ .Returns(([], []));
+
+ // Mirrors Session.GoodPublishRequestCount: a request is recorded by
+ // AsyncRequestStarted and stops counting once it is written off.
+ int outstanding = 0;
+ int writtenOff = 0;
+ m_mockContext.Setup(c => c.GoodPublishRequestCount)
+ .Returns(() => Volatile.Read(ref outstanding) - Volatile.Read(ref writtenOff));
+ m_mockContext.Setup(c => c.AsyncRequestStarted(
+ It.IsAny(),
+ It.IsAny(),
+ It.IsAny(),
+ It.IsAny()))
+ .Callback(() => Interlocked.Increment(ref outstanding));
+
+ // The requests never complete: this is the state Session.OnKeepAlive
+ // recovers from, where the responses are outstanding but are no
+ // longer expected to return. RunContinuationsAsynchronously keeps
+ // the completion below off the test thread: the engine hangs its
+ // continuation on this task with OnCompleted, so an inline
+ // completion would re-enter the engine synchronously and issue
+ // further publishes while the assertion is being evaluated.
+ var pending = new TaskCompletionSource(
+ TaskCreationOptions.RunContinuationsAsynchronously);
+ int issued = 0;
+ m_mockContext.Setup(c => c.PublishAsync(
+ It.IsAny(),
+ It.IsAny>(),
+ It.IsAny()))
+ .Returns(() =>
+ {
+ Interlocked.Increment(ref issued);
+ return new ValueTask(pending.Task);
+ });
+
+ using var engine = new ClassicSubscriptionEngine(m_mockContext.Object);
+
+ engine.StartPublishing(timeout: 5000, fullQueue: true);
+ int afterFill = Volatile.Read(ref issued);
+ Assert.That(afterFill, Is.EqualTo(subscriptionCount));
+
+ // Session.OnKeepAlive marks every outstanding publish request
+ // defunct and then calls StartPublishing to refill the pipeline.
+ Volatile.Write(ref writtenOff, afterFill);
+ Assert.That(m_mockContext.Object.GoodPublishRequestCount, Is.Zero);
+
+ engine.StartPublishing(timeout: 5000, fullQueue: false);
+
+ // Sample before completing the pending requests so the result is
+ // fixed by StartPublishing alone.
+ int afterRefill = Volatile.Read(ref issued);
+ pending.TrySetCanceled();
+
+ Assert.That(
+ afterRefill - afterFill,
+ Is.EqualTo(subscriptionCount),
+ "StartPublishing must refill a pipeline whose requests were " +
+ "written off, otherwise the session never publishes again.");
+ }
+
+ [Test]
+ public void BeginPublishIsBoundedByTheDesiredRequestCount()
+ {
+ const int subscriptionCount = 3;
+
+ var subscriptions = new List();
+ for (int ii = 0; ii < subscriptionCount; ii++)
+ {
+ subscriptions.Add(
+ new Subscription(m_telemetry)
+ {
+ CurrentPublishingInterval = 100,
+ CurrentLifetimeCount = 10
+ });
+ }
+
+ m_mockContext.Setup(c => c.Connected).Returns(true);
+ m_mockContext.Setup(c => c.Subscriptions).Returns(subscriptions);
+ m_mockContext.Setup(c => c.PrepareAcknowledgementsToSend(
+ It.IsAny>()))
+ .Returns(([], []));
+
+ int outstanding = 0;
+ int writtenOff = 0;
+ m_mockContext.Setup(c => c.GoodPublishRequestCount)
+ .Returns(() => Volatile.Read(ref outstanding) - Volatile.Read(ref writtenOff));
+ m_mockContext.Setup(c => c.AsyncRequestStarted(
+ It.IsAny(),
+ It.IsAny(),
+ It.IsAny(),
+ It.IsAny()))
+ .Callback(() => Interlocked.Increment(ref outstanding));
+
+ var pending = new TaskCompletionSource(
+ TaskCreationOptions.RunContinuationsAsynchronously);
+ int issued = 0;
+ m_mockContext.Setup(c => c.PublishAsync(
+ It.IsAny(),
+ It.IsAny>(),
+ It.IsAny()))
+ .Returns(() =>
+ {
+ Interlocked.Increment(ref issued);
+ return new ValueTask(pending.Task);
+ });
+
+ using var engine = new ClassicSubscriptionEngine(m_mockContext.Object);
+
+ // Every subscription nudges the pipeline when it sees no
+ // notification, which is what Subscription.HandleOnKeepAliveStopped
+ // does. The nudges must not multiply the pipeline.
+ for (int ii = 0; ii < subscriptionCount * 4; ii++)
+ {
+ engine.BeginPublish(timeout: 5000);
+ }
+
+ int afterNudges = Volatile.Read(ref issued);
+ Assert.That(
+ afterNudges,
+ Is.EqualTo(subscriptionCount),
+ "The keep alive nudge must not send past the desired publish " +
+ "request count.");
+
+ // Once the outstanding requests are written off the pipeline is
+ // drained, so the nudge has to get through again.
+ Volatile.Write(ref writtenOff, afterNudges);
+
+ Assert.That(engine.BeginPublish(timeout: 5000), Is.True);
+
+ int afterRecovery = Volatile.Read(ref issued);
+ pending.TrySetCanceled();
+
+ Assert.That(afterRecovery, Is.EqualTo(afterNudges + 1));
+ }
+
private static void InvokeOnPublishComplete(
ClassicSubscriptionEngine engine,
Task task,