Skip to content

Commit 84161c1

Browse files
AnatolyKochnevAnatolyKochnev
authored
[Client] Fixed excessive tasks spawning during session connection loss (OPCFoundation#3707)
Added checks for Session.KeepAliveStopped before trying to execute Publish for subscriptions.
1 parent 94653f1 commit 84161c1

3 files changed

Lines changed: 97 additions & 22 deletions

File tree

‎Libraries/Opc.Ua.Client/Session/Session.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3554,6 +3554,12 @@ public bool BeginPublish(int timeout)
35543554
return false;
35553555
}
35563556

3557+
if (KeepAliveStopped)
3558+
{
3559+
m_logger.LogWarning("Publish skipped due to session lost connection. Last successfull keepalive: {LastKeepAlive}", LastKeepAliveTime);
3560+
return false;
3561+
}
3562+
35573563
// get event handler to modify ack list
35583564
PublishSequenceNumbersToAcknowledgeEventHandler? callback
35593565
= m_PublishSequenceNumbersToAcknowledge;

‎Libraries/Opc.Ua.Client/Subscription/Subscription.cs‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,6 @@ namespace Opc.Ua.Client
4444
/// </summary>
4545
public class Subscription : ISnapshotRestore<SubscriptionState>, IDisposable, ICloneable
4646
{
47-
private const int kMinKeepAliveTimerInterval = 1000;
4847
private const int kKeepAliveTimerMargin = 1000;
4948
private const int kRepublishMessageExpiredTimeout = 10000;
5049

@@ -53,6 +52,11 @@ public class Subscription : ISnapshotRestore<SubscriptionState>, IDisposable, IC
5352
/// </summary>
5453
public const int RepublishMessageTimeout = 2500;
5554

55+
/// <summary>
56+
/// Minimum keep alive interval
57+
/// </summary>
58+
public const int MinKeepAliveTimerInterval = 1000;
59+
5660
/// <summary>
5761
/// Create subscription
5862
/// </summary>
@@ -2104,7 +2108,9 @@ private void HandleOnKeepAliveStopped()
21042108

21052109
if (session != null &&
21062110
session.Connected &&
2107-
!session.Reconnecting)
2111+
!session.Reconnecting &&
2112+
!session.KeepAliveStopped
2113+
)
21082114
{
21092115
TraceState("PUBLISHING STOPPED");
21102116

@@ -2198,7 +2204,7 @@ private int BeginPublishTimeout()
21982204
{
21992205
return Math.Max(
22002206
Math.Min(m_keepAliveInterval * 3, int.MaxValue),
2201-
kMinKeepAliveTimerInterval);
2207+
MinKeepAliveTimerInterval);
22022208
}
22032209

22042210
/// <summary>
@@ -2312,12 +2318,12 @@ private int CalculateKeepAliveInterval()
23122318
{
23132319
int keepAliveInterval = (int)
23142320
Math.Min(CurrentPublishingInterval * (CurrentKeepAliveCount + 1), int.MaxValue);
2315-
if (keepAliveInterval < kMinKeepAliveTimerInterval)
2321+
if (keepAliveInterval < MinKeepAliveTimerInterval)
23162322
{
23172323
keepAliveInterval = (int)Math.Min(
23182324
PublishingInterval * (KeepAliveCount + 1),
23192325
int.MaxValue);
2320-
keepAliveInterval = Math.Max(kMinKeepAliveTimerInterval, keepAliveInterval);
2326+
keepAliveInterval = Math.Max(MinKeepAliveTimerInterval, keepAliveInterval);
23212327
}
23222328
return keepAliveInterval;
23232329
}

‎Tests/Opc.Ua.Client.Tests/Subscription/SubscriptionUnitTests.cs‎

Lines changed: 80 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,22 @@ namespace Opc.Ua.Client.Tests
4242
[Parallelizable]
4343
public class SubscriptionUnitTests
4444
{
45+
private const PublishStateChangedMask kSessionNotConnected = PublishStateChangedMask.Stopped | PublishStateChangedMask.SessionNotConnected;
46+
47+
public record KeepAliveTestDataProvider(PublishStateChangedMask ExpectedPublishState) : IFormattable
48+
{
49+
public bool SessionConnected { get; init; }
50+
public bool SessionReconnecting { get; init; }
51+
public bool SessionKeepAliveStopped { get; init; }
52+
public string ToString(string format, IFormatProvider formatProvider)
53+
{
54+
return $"Connected={SessionConnected}, " +
55+
$"reconnecting={SessionReconnecting}, " +
56+
$"keepAlive={SessionKeepAliveStopped}. " +
57+
$"Expected status:{ExpectedPublishState}";
58+
}
59+
}
60+
4561
private sealed class SubscriptionContainer : IDisposable
4662
{
4763
private readonly CancellationTokenRegistration m_tokedCancellation;
@@ -73,28 +89,31 @@ private static Task AwaitForRepublishTimeout(CancellationToken ct)
7389
return Task.Delay(Subscription.RepublishMessageTimeout + 100, ct);
7490
}
7591

76-
private static ISession BuildSessionMock(Func<uint, uint, bool> republishHandler)
92+
private static ISession BuildSessionMock(Func<uint, uint, bool> republishHandler = null, Action<Mock<ISession>> setup = null)
7793
{
7894
uint subscriptionIdSeed = 0u;
7995

8096
var session = new Mock<ISession>();
81-
session
82-
.Setup(x => x.RepublishAsync(It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<CancellationToken>()))
83-
.ReturnsAsync<uint, uint, CancellationToken, ISession, (bool, ServiceResult)>((
84-
subscriptionId,
85-
sequenceNumber,
86-
ct) =>
87-
{
88-
if (subscriptionId > subscriptionIdSeed)
89-
{
90-
return (true, StatusCodes.BadSubscriptionIdInvalid);
91-
}
92-
if (republishHandler(subscriptionId, sequenceNumber))
97+
if (republishHandler is not null)
98+
{
99+
session
100+
.Setup(x => x.RepublishAsync(It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<CancellationToken>()))
101+
.ReturnsAsync<uint, uint, CancellationToken, ISession, (bool, ServiceResult)>((
102+
subscriptionId,
103+
sequenceNumber,
104+
ct) =>
93105
{
94-
return (true, ServiceResult.Good);
95-
}
96-
return (true, StatusCodes.BadMessageNotAvailable);
97-
});
106+
if (subscriptionId > subscriptionIdSeed)
107+
{
108+
return (true, StatusCodes.BadSubscriptionIdInvalid);
109+
}
110+
if (republishHandler(subscriptionId, sequenceNumber))
111+
{
112+
return (true, ServiceResult.Good);
113+
}
114+
return (true, StatusCodes.BadMessageNotAvailable);
115+
});
116+
}
98117
session
99118
.Setup(x => x
100119
.CreateSubscriptionAsync(
@@ -149,6 +168,7 @@ private static ISession BuildSessionMock(Func<uint, uint, bool> republishHandler
149168
StatusCodes.Good)],
150169
DiagnosticInfos = [.. subscriptionIds.ConvertAll(_ => new DiagnosticInfo())]
151170
});
171+
setup?.Invoke(session);
152172
return session.Object;
153173
}
154174

@@ -350,5 +370,48 @@ public async Task WillRepublishIfMissedMessagesInBetweenOfPublishesAsync(
350370
Assert.That(subscription.Notifications, Is.EquivalentTo(messages.Skip(1)));
351371
}
352372
}
373+
374+
[DatapointSource]
375+
public IEnumerable<KeepAliveTestDataProvider> SubscriptionKeepAliveValues()
376+
{
377+
yield return new(PublishStateChangedMask.Stopped) { SessionConnected = true, SessionReconnecting = false, SessionKeepAliveStopped = false };
378+
379+
yield return new(kSessionNotConnected) { SessionConnected = false, SessionReconnecting = false, SessionKeepAliveStopped = false };
380+
yield return new(kSessionNotConnected) { SessionConnected = false, SessionReconnecting = true, SessionKeepAliveStopped = false };
381+
yield return new(kSessionNotConnected) { SessionConnected = false, SessionReconnecting = false, SessionKeepAliveStopped = true };
382+
yield return new(kSessionNotConnected) { SessionConnected = false, SessionReconnecting = true, SessionKeepAliveStopped = true };
383+
384+
yield return new(kSessionNotConnected) { SessionConnected = true, SessionReconnecting = true, SessionKeepAliveStopped = false };
385+
yield return new(kSessionNotConnected) { SessionConnected = true, SessionReconnecting = true, SessionKeepAliveStopped = true };
386+
387+
yield return new(kSessionNotConnected) { SessionConnected = true, SessionReconnecting = false, SessionKeepAliveStopped = true };
388+
}
389+
390+
[Theory]
391+
[CancelAfter(Subscription.MinKeepAliveTimerInterval * 10)]
392+
public async Task RespectsStateOfSessionDuringKeepAliveCalls(KeepAliveTestDataProvider testData, CancellationToken ct)
393+
{
394+
var keepAliveCompleted = new TaskCompletionSource<PublishStateChangedMask>(TaskCreationOptions.RunContinuationsAsynchronously);
395+
void KeepAliveHasTriggered(Subscription x, PublishStateChangedEventArgs y) => keepAliveCompleted.TrySetResult(y.Status);
396+
ISession session = BuildSessionMock(
397+
setup: mock =>
398+
{
399+
mock.Setup(x => x.Connected).Returns(testData.SessionConnected);
400+
mock.Setup(x => x.Reconnecting).Returns(testData.SessionReconnecting);
401+
mock.Setup(x => x.KeepAliveStopped).Returns(testData.SessionKeepAliveStopped);
402+
});
403+
404+
using var subscription = new Subscription(
405+
NUnitTelemetryContext.Create(),
406+
new() { PublishingEnabled = true })
407+
{
408+
Session = session
409+
};
410+
subscription.PublishStatusChanged += KeepAliveHasTriggered;
411+
await subscription.CreateAsync(ct).ConfigureAwait(false);
412+
await Task.WhenAny(keepAliveCompleted.Task, Task.Delay(-1, ct)).ConfigureAwait(false);
413+
414+
Assert.That(keepAliveCompleted.Task.Result, Is.EqualTo(testData.ExpectedPublishState));
415+
}
353416
}
354417
}

0 commit comments

Comments
 (0)