dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge #566/#581/#583: per-pool transport diagnostics + per-device open-attempt limit, CIP-level close recycle, positive-control absence tests

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-07 23:31:17 -04:00
parent 35152fcaf7 fe94a3bdb8
commit e5fe2e5547
27 changed files with 915 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs
+205 -16
View File
@@ -2,11 +2,14 @@ using System.Buffers.Binary;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Cip;
/// <summary>Serializes every test class that asserts on <see cref="CipConnection"/>'s process-wide
/// diagnostic counters, so no parallel test can move them mid-assertion.</summary>
/// <summary>Serializes the loopback-socket test classes. Their diagnostic counters are per connection / per
/// pool since Gitea #581 (no longer process-wide statics), but <see cref="DeviceSessionPoolTests"/> still
/// reconfigures <see cref="DeviceSessionPool.Shared"/>'s logger, and the timing-sensitive storm test is kept
/// off a thread pool its neighbours are saturating.</summary>
[CollectionDefinition(Name, DisableParallelization = true)]
public sealed class CipConnectionStatics
{
@@ -66,7 +69,7 @@ public sealed class CipConnectionTests
target.ForwardOpenReply = fo => fo[0] == 0x5B ? FakeCipTarget.Refuse(0x08)(fo) : FakeCipTarget.Accept(fo);
var time = new ManualTimeProvider();
await using var conn = Create(target, Options(), time);
var before = Volatile.Read(ref CipConnection.ForwardOpenCount);
var before = conn.Diagnostics.ForwardOpens;
await conn.SendAsync(ReadTag, TestCt);
@@ -76,7 +79,7 @@ public sealed class CipConnectionTests
fos[1][0].ShouldBe((byte)0x54);
RegularOTParams(fos[1]).ShouldBe((ushort)(0x4200 | 504));
conn.MaxRequestBytes.ShouldBe(502);
(Volatile.Read(ref CipConnection.ForwardOpenCount) - before).ShouldBe(2);
(conn.Diagnostics.ForwardOpens - before).ShouldBe(2);
target.ConnectedRequests.ShouldBe(1);
}
@@ -109,14 +112,12 @@ public sealed class CipConnectionTests
[Fact]
public async Task A_spent_open_attempt_budget_refuses_the_reconnect_before_any_io()
{
// Task 18 review r1 (C1): the differential harness caps the process at one open attempt, so a press
// Task 18 review r1 (C1): the differential harness caps the press at one open attempt, so a press
// that refuses the Forward Open sees one TCP connect and one Forward Open, never one per remaining tag.
await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0100) };
var time = new ManualTimeProvider();
await using var conn = Create(target, Options(size: 504), time);
var before = Volatile.Read(ref CipConnection.OpenAttempts);
Volatile.Write(ref CipConnection.OpenAttemptBudget, before + 1);
try
using (conn.Diagnostics.LimitOpenAttempts(Key(target), 1))
{
await Should.ThrowAsync<AbTransportException>(() => conn.SendAsync(ReadTag, TestCt));
var retry = conn.SendAsync(ReadTag, TestCt);
@@ -128,13 +129,43 @@ public sealed class CipConnectionTests
refused.Message.ShouldContain("budget");
target.Server.Connections.ShouldBe(1);
target.ForwardOpens.ShouldBe(1);
}
finally
{
Volatile.Write(ref CipConnection.OpenAttemptBudget, -1);
conn.Diagnostics.OpenAttempts.ShouldBe(2); // the refused one is counted too
}
}
[Fact]
public async Task An_open_attempt_limit_is_scoped_to_one_device_and_lifted_by_its_scope()
{
// Gitea #566 / #581: the limit is per DeviceKey on one pool's diagnostics, not a process-wide static, so
// limiting the press under test can never refuse an open to any other device.
await using var limited = new FakeCipTarget();
await using var other = new FakeCipTarget();
var time = new ManualTimeProvider();
var pool = new DeviceSessionPool(time);
var limitedKey = Key(limited);
using (pool.Diagnostics.LimitOpenAttempts(limitedKey, 0))
{
await using var otherLease = pool.Acquire(Key(other), Options());
(await otherLease.Scheduler.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0);
await using var limitedLease = pool.Acquire(limitedKey, Options());
var refused = await Should.ThrowAsync<AbTransportException>(() => limitedLease.Scheduler.SendAsync(ReadTag, TestCt));
refused.Status.ShouldBe(AbStatus.Abort);
refused.Message.ShouldContain("budget");
Should.Throw<InvalidOperationException>(() => pool.Diagnostics.LimitOpenAttempts(limitedKey, 5)); // one at a time
}
pool.Diagnostics.HasOpenAttemptLimit(limitedKey).ShouldBeFalse();
time.Advance(TimeSpan.FromSeconds(5)); // the refused attempt still started the device's throttle interval
await using var lifted = pool.Acquire(limitedKey, Options());
(await lifted.Scheduler.SendAsync(ReadTag, TestCt).WaitAsync(Guard, TestCt)).GeneralStatus.ShouldBe((byte)0);
limited.ForwardOpens.ShouldBe(1); // the refused attempt never reached the wire; the lifted one did
other.ForwardOpens.ShouldBe(1);
}
private static DeviceKey Key(FakeCipTarget target) => new("127.0.0.1", target.Port, "1,0", "ControlLogix");
[Fact]
public async Task Connection_in_use_does_not_fall_back()
{
@@ -245,7 +276,7 @@ public sealed class CipConnectionTests
};
var time = new ManualTimeProvider();
await using var conn = Create(target, Options(), time);
var before = Volatile.Read(ref CipConnection.ForwardOpenCount);
var before = conn.Diagnostics.ForwardOpens;
using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestCt);
var pending = conn.SendAsync(ReadTag, cts.Token);
@@ -256,7 +287,7 @@ public sealed class CipConnectionTests
await conn.SendAsync(ReadTag, TestCt);
(Volatile.Read(ref CipConnection.ForwardOpenCount) - before).ShouldBe(1);
(conn.Diagnostics.ForwardOpens - before).ShouldBe(1);
target.ForwardOpens.ShouldBe(1);
target.Server.Connections.ShouldBe(1);
target.ConnectedRequests.ShouldBe(2);
@@ -321,7 +352,7 @@ public sealed class CipConnectionTests
await using var target = new FakeCipTarget();
var time = new ManualTimeProvider();
await using var conn = Create(target, Options(), time);
var before = Volatile.Read(ref CipConnection.RequestsSent);
var before = conn.Diagnostics.RequestsSent;
await conn.SendAsync(ReadTag, TestCt);
time.Advance(TimeSpan.FromSeconds(8));
@@ -336,7 +367,7 @@ public sealed class CipConnectionTests
time.Advance(TimeSpan.FromSeconds(1));
conn.KeepAlivesSent.ShouldBe(2);
await FakeCipTarget.Until(() => target.ConnectedRequests == 3, Guard, "the second keep-alive");
(Volatile.Read(ref CipConnection.RequestsSent) - before).ShouldBe(3); // keep-alives count
(conn.Diagnostics.RequestsSent - before).ShouldBe(3); // keep-alives count
target.ForwardOpens.ShouldBe(1);
}
@@ -368,6 +399,9 @@ public sealed class CipConnectionTests
time.Advance(TimeSpan.FromSeconds(30));
// Positive control (#583): the keep-alive timer really ticked (at 8, 16 and 24 s) and re-armed itself, so
// the zero below is the timer declining to send, not a timer that never ran.
time.HasTimerDueAt(time.Start + TimeSpan.FromSeconds(32)).ShouldBeTrue();
conn.KeepAlivesSent.ShouldBe(0);
target.ForwardOpens.ShouldBe(0);
target.Server.Connections.ShouldBe(0);
@@ -411,4 +445,159 @@ public sealed class CipConnectionTests
await dispose.WaitAsync(Guard, TestCt);
target.ForwardOpens.ShouldBe(1); // nothing is opened once disposal has begun
}
[Fact]
public async Task Disposing_during_the_session_registration_never_puts_a_Forward_Open_on_the_wire()
{
// Gitea #566: disposal that begins while the TCP connect / RegisterSession is on the wire lets that step
// finish (it is bounded), then releases the session instead of sending a Forward Open nobody would close.
await using var target = new FakeCipTarget();
var registerHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
target.Server.RegisterSessionGate = () => registerHeld.Task;
var time = new ManualTimeProvider();
var conn = Create(target, Options(), time);
var send = conn.SendAsync(ReadTag, TestCt);
await target.Server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.RegisterSession, Guard);
var dispose = conn.DisposeAsync().AsTask();
conn.OpenInProgress.ShouldBeTrue(); // disposal is waiting out the attempt on the wire...
dispose.IsCompleted.ShouldBeFalse(); // ...rather than abandoning it
registerHeld.SetResult();
(await Should.ThrowAsync<AbTransportException>(() => send.WaitAsync(Guard, TestCt))).Status.ShouldBe(AbStatus.Abort);
await dispose.WaitAsync(Guard, TestCt);
// The session is released (positive control); the server records frames in arrival order, so a Forward
// Open sent before the UnRegisterSession would already be counted.
await target.Server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession, Guard);
target.ForwardOpens.ShouldBe(0);
target.ForwardCloses.ShouldBe(0);
conn.Diagnostics.OpenAttempts.ShouldBe(1);
}
private const ushort AbandonedOpenRefusal = 0x7A11; // an extended status no other test uses
[Fact]
public async Task An_open_every_waiter_abandoned_leaves_no_unobserved_task_exception()
{
// Gitea #566: the shared open fails AFTER its only waiter gave up (its own token). Nothing awaits the
// open's task then; without the observer it would surface as UnobservedTaskException at finalization.
var unobserved = await UnobservedTaskProbe.CollectAsync(AbandonAFailingOpenAsync, Guard);
unobserved.ShouldNotContain(e => e.Message.Contains($"0x{AbandonedOpenRefusal:X4}"));
}
// Kept out of the test body so no local in the test's own state machine roots the abandoned open.
[System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.NoInlining)]
private static async Task AbandonAFailingOpenAsync()
{
await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, AbandonedOpenRefusal) };
var openHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
target.ForwardOpenGate = () => openHeld.Task;
var time = new ManualTimeProvider();
await using var conn = Create(target, Options(size: 504), time);
using (var cts = new CancellationTokenSource())
{
var waiter = conn.SendAsync(ReadTag, cts.Token);
await FakeCipTarget.Until(() => target.ForwardOpens == 1, Guard, "the Forward Open");
await cts.CancelAsync();
await Should.ThrowAsync<OperationCanceledException>(() => waiter);
}
openHeld.SetResult(); // the target refuses, after every waiter has gone
await FakeCipTarget.Until(() => !conn.OpenInProgress, Guard, "the abandoned open to fail");
}
// ---- CIP-level close (Gitea #566) ------------------------------------------------------------
private static byte[] Ok(byte[] cip) => [(byte)(cip[0] | 0x80), 0, 0, 0];
private static byte[] Status(byte[] cip, byte general, ushort extended) =>
[(byte)(cip[0] | 0x80), 0, general, 1, (byte)extended, (byte)(extended >> 8)];
[Theory]
[InlineData(ConnectionManagerServices.TargetConnectionNotFound)]
[InlineData(ConnectionManagerServices.ConnectionTimedOut)]
public async Task A_reply_saying_the_CIP_connection_is_gone_retires_it_and_the_next_send_reopens(ushort extended)
{
// The target dropped the class-3 connection but the TCP session is still up: every later connected request
// would get the same answer with the session looking healthy, so the connection is recycled at once.
await using var target = new FakeCipTarget();
var gone = 0;
target.ConnectedHandler = cip => Task.FromResult<byte[]?>(
Interlocked.Exchange(ref gone, 0) == 1 ? Status(cip, 0x01, extended) : Ok(cip));
var time = new ManualTimeProvider();
await using var conn = Create(target, Options(), time);
await conn.SendAsync(ReadTag, TestCt);
Volatile.Write(ref gone, 1);
var ex = await Should.ThrowAsync<AbTransportException>(() => conn.SendAsync(ReadTag, TestCt));
ex.Status.ShouldBe(AbStatus.BadConnection);
ex.Message.ShouldContain($"0x{extended:X4}");
conn.Diagnostics.ConnectionsClosedByTarget.ShouldBe(1);
await target.Server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession, Guard);
await FakeCipTarget.Until(() => target.Server.OpenConnections == 0, Guard, "the session to close");
target.ForwardCloses.ShouldBe(0); // the target just said there is nothing to close
var reopen = conn.SendAsync(ReadTag, TestCt);
await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(1), Guard); // through the throttle
time.Advance(TimeSpan.FromSeconds(1));
CipReply.Parse(await reopen.WaitAsync(Guard, TestCt)).GeneralStatus.ShouldBe((byte)0);
target.ForwardOpens.ShouldBe(2);
target.Server.Connections.ShouldBe(2);
}
[Fact]
public async Task A_keepalive_told_the_connection_is_gone_recycles_it_before_any_request_fails()
{
await using var target = new FakeCipTarget();
target.ConnectedHandler = cip => Task.FromResult<byte[]?>(
cip.SequenceEqual(CipConnection.KeepAliveRequest) ? Status(cip, 0x01, 0x0107) : Ok(cip));
var time = new ManualTimeProvider();
await using var conn = Create(target, Options(), time);
await conn.SendAsync(ReadTag, TestCt);
time.Advance(TimeSpan.FromSeconds(8)); // idle: the keep-alive goes out and is told the connection is gone
await FakeCipTarget.Until(() => conn.Diagnostics.ConnectionsClosedByTarget == 1, Guard, "the retirement");
await FakeCipTarget.Until(() => target.Server.OpenConnections == 0, Guard, "the session to close");
// The next request reopens (the throttle interval is long past) instead of failing on a dead connection.
CipReply.Parse(await conn.SendAsync(ReadTag, TestCt).WaitAsync(Guard, TestCt)).GeneralStatus.ShouldBe((byte)0);
target.ForwardOpens.ShouldBe(2);
}
[Fact]
public async Task Other_error_statuses_describe_the_request_and_leave_the_connection_alone()
{
// Negative control for the recycle: a tag-level error and a connection-failure status that is not about
// this connection come back as ordinary replies, and the next request rides the same connection.
await using var target = new FakeCipTarget();
var replies = new Queue<Func<byte[], byte[]>>(
[
cip => Status(cip, 0x04, 0x0000), // path segment error: the tag
cip => Status(cip, 0x01, 0x0204), // unconnected request timed out: not this connection
Ok,
]);
target.ConnectedHandler = cip => Task.FromResult<byte[]?>(replies.Dequeue()(cip));
var time = new ManualTimeProvider();
await using var conn = Create(target, Options(), time);
CipReply.Parse(await conn.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0x04);
CipReply.Parse(await conn.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0x01);
CipReply.Parse(await conn.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0); // same connection
conn.Diagnostics.ConnectionsClosedByTarget.ShouldBe(0);
target.ForwardOpens.ShouldBe(1);
target.Server.Connections.ShouldBe(1);
}
[Fact]
public void Only_connection_failure_0107_and_0203_mean_the_connection_is_gone()
{
ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x01, 1, 0x07, 0x01]).ShouldBe((ushort)0x0107);
ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x01, 1, 0x03, 0x02]).ShouldBe((ushort)0x0203);
ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x01, 1, 0x04, 0x02]).ShouldBeNull(); // 0x0204
ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x01, 0]).ShouldBeNull(); // no extended word
ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x05, 1, 0x07, 0x01]).ShouldBeNull(); // not 0x01
ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0, 0]).ShouldBeNull();
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs
+8 -4
View File
@@ -69,7 +69,8 @@ public sealed class ForwardOpenThrottleTests
var first = throttle.RunAsync(Attempt, TestCt);
time.Advance(TimeSpan.FromSeconds(5)); // the interval is no obstacle; only the in-flight one is
var others = Enumerable.Range(0, 4).Select(_ => throttle.RunAsync(Attempt, TestCt)).ToArray();
await Task.Delay(50, TestCt);
// Positive control (#583): all four have reached the throttle and are parked behind the in-flight attempt.
await FakeCipTarget.Until(() => throttle.Queued == 4, Guard, "four callers queued behind the first");
Volatile.Read(ref inFlight).ShouldBe(1);
others.ShouldAllBe(t => !t.IsCompleted);
@@ -131,8 +132,8 @@ public sealed class ForwardOpenThrottleTests
await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0113) };
var time = new ManualTimeProvider();
var options = new ConnectionOptions { ConnectionSize = 504, Timeout = TimeSpan.FromSeconds(2) };
var conn = new CipConnection(new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"), options,
new ForwardOpenThrottle(options.MinForwardOpenInterval, time), time);
var throttle = new ForwardOpenThrottle(options.MinForwardOpenInterval, time);
var conn = new CipConnection(new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"), options, throttle, time);
using var stop = CancellationTokenSource.CreateLinkedTokenSource(TestCt);
var failures = 0;
@@ -151,7 +152,10 @@ public sealed class ForwardOpenThrottleTests
await Task.Delay(10, TestCt);
time.Advance(TimeSpan.FromMilliseconds(100));
}
await Task.Delay(100, TestCt); // let any attempt released by the last step reach the wire
// Positive control (#583), instead of a fixed settle: wait until the next attempt is parked on the interval.
// Then no attempt is in flight and none can start until the clock moves, so every attempt the last step
// released has already reached the wire and been counted.
await FakeCipTarget.Until(() => throttle.IsWaitingOutInterval, Guard, "the next attempt to park on the interval");
await stop.CancelAsync();
await Task.WhenAll(callers).WaitAsync(Guard, TestCt);
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs
+70 -2
View File
@@ -203,7 +203,9 @@ public sealed class DeviceSessionPoolTests
await using var c = pool.Acquire(key, Options());
var cSend = c.Scheduler.SendAsync(ReadTag, TestCt);
await Task.Delay(300, TestCt);
// Positive control (#583): C's open has reached the predecessor wait and parked there, so the absence of
// its Forward Open below is the wait holding it, not C merely not having got that far yet.
await FakeCipTarget.Until(() => pool.Diagnostics.PredecessorWaits == 1, Guard, "C to park behind A's close");
target.ForwardOpens.ShouldBe(1); // C is held back behind A
cSend.IsCompleted.ShouldBeFalse();
@@ -292,7 +294,7 @@ public sealed class DeviceSessionPoolTests
time.Advance(TimeSpan.FromSeconds(5)); // the throttle is no obstacle
await using var b = pool.Acquire(key, Options());
var bSend = b.Scheduler.SendAsync(ReadTag, TestCt);
await Task.Delay(300, TestCt);
await FakeCipTarget.Until(() => pool.Diagnostics.PredecessorWaits == 1, Guard, "B to park behind A's close");
target.ForwardOpens.ShouldBe(1); // held behind A's close
bSend.IsCompleted.ShouldBeFalse();
@@ -304,6 +306,72 @@ public sealed class DeviceSessionPoolTests
pool.EntryCount.ShouldBe(1);
}
/// <summary>
/// Gitea #566: releasing a lease while the device's shared open is still on the wire does not wait for that
/// open (up to its whole open budget) — <see cref="ConnectionLease.ReleaseInBackground"/> (#579) returns at
/// once, and the teardown it hands back still Forward Closes what the open produced once it lands.
/// </summary>
[Fact]
public async Task Release_in_background_does_not_wait_behind_an_in_flight_shared_open()
{
await using var target = new FakeCipTarget();
var openHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
target.ForwardOpenGate = () => openHeld.Task;
var pool = new DeviceSessionPool(new ManualTimeProvider());
var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix");
var lease = pool.Acquire(key, Options());
var send = lease.Scheduler.SendAsync(ReadTag, TestCt);
await FakeCipTarget.Until(() => target.ForwardOpens == 1, Guard, "the Forward Open");
((CipConnection)lease.Channel).OpenInProgress.ShouldBeTrue();
var teardown = lease.ReleaseInBackground(); // returns synchronously: the open is still held
pool.EntryCount.ShouldBe(0);
teardown.IsCompleted.ShouldBeFalse("the teardown waits for the open on the wire; the release call did not");
(await Should.ThrowAsync<AbTransportException>(() => send.WaitAsync(Guard, TestCt))).Status.ShouldBe(AbStatus.BadConnection);
openHeld.SetResult();
await teardown.WaitAsync(Guard, TestCt);
target.ForwardCloses.ShouldBe(1); // what the late-accepted open created is still closed
}
/// <summary>Gitea #566: the close path disposes the connection even when the scheduler's dispose throws —
/// otherwise the connection's keep-alive would hold the press connection open forever.</summary>
[Fact]
public async Task Close_disposes_the_connection_even_when_the_scheduler_dispose_throws()
{
var logger = new CapturingLogger();
var scheduler = new Disposable(() => throw new InvalidOperationException("scheduler dispose failed"));
var connection = new Disposable(() => { });
var key = new DeviceKey("plc1", 44818, "1,0", "ControlLogix");
await DeviceSessionPool.CloseAsync(key, scheduler, connection, logger);
scheduler.Disposed.ShouldBe(1);
connection.Disposed.ShouldBe(1);
logger.Warnings.ShouldBe(1);
// And a throwing connection dispose is logged, never propagated to the release.
var failing = new Disposable(() => throw new InvalidOperationException("close failed"));
await DeviceSessionPool.CloseAsync(key, new Disposable(() => { }), failing, logger);
failing.Disposed.ShouldBe(1);
logger.Warnings.ShouldBe(2);
}
private sealed class Disposable(Action onDispose) : IAsyncDisposable
{
private int _disposed;
public int Disposed => Volatile.Read(ref _disposed);
public ValueTask DisposeAsync()
{
Interlocked.Increment(ref _disposed);
onDispose();
return ValueTask.CompletedTask;
}
}
[Fact]
public void Shared_is_one_process_wide_instance() =>
DeviceSessionPool.Shared.ShouldBeSameAs(DeviceSessionPool.Shared);
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs
+4 -22
View File
@@ -271,28 +271,10 @@ public sealed class EncapsulationClientTests
[Fact]
public async Task An_abandoned_exchange_that_times_out_leaves_no_unobserved_task_exception()
{
var unobserved = new List<Exception>();
EventHandler<UnobservedTaskExceptionEventArgs> onUnobserved = (_, e) =>
{
if (e.Exception.Flatten().InnerExceptions.Any(x => x.Message.Contains("within 317 ms")))
lock (unobserved) unobserved.Add(e.Exception);
};
TaskScheduler.UnobservedTaskException += onUnobserved;
try
{
await AbandonAnExchangeThatTimesOut();
for (var i = 0; i < 5; i++)
{
GC.Collect();
GC.WaitForPendingFinalizers();
await Task.Delay(20, TestCt);
}
lock (unobserved) unobserved.ShouldBeEmpty();
}
finally
{
TaskScheduler.UnobservedTaskException -= onUnobserved;
}
// Gitea #583: collected against a positive control (a deliberately unobserved fault that MUST be reported),
// not after a fixed number of GC cycles.
var unobserved = await UnobservedTaskProbe.CollectAsync(AbandonAnExchangeThatTimesOut, Guard);
unobserved.ShouldNotContain(x => x.Message.Contains("within 317 ms"));
}
// Kept out of the test body so no local in the test's own state machine roots the abandoned task.
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeEipServer.cs
+4
View File
@@ -61,6 +61,9 @@ internal sealed class FakeEipServer : IAsyncDisposable
/// <summary>Synchronous convenience setter for <see cref="HandlerAsync"/>.</summary>
public Func<byte[], byte[]?> Handler { set => HandlerAsync = req => Task.FromResult(value(req)); }
/// <summary>Awaited before a RegisterSession is answered (the request is already recorded).</summary>
public Func<Task>? RegisterSessionGate { get; set; }
/// <summary>Never answer non-session requests.</summary>
public bool Stall { get; set; }
@@ -160,6 +163,7 @@ internal sealed class FakeEipServer : IAsyncDisposable
switch (Command(frame))
{
case EncapCommand.RegisterSession:
if (RegisterSessionGate is { } gate) await gate();
reply = (byte[])frame.Clone();
BinaryPrimitives.WriteUInt32LittleEndian(reply.AsSpan(4), Session);
break;
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/UnobservedTaskProbe.cs
+54
View File
@@ -0,0 +1,54 @@
using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
/// <summary>
/// Collects <see cref="TaskScheduler.UnobservedTaskException"/>s raised after some code abandons work, with a
/// positive control (Gitea #583): after the abandon, a task is deliberately faulted and dropped, and garbage is
/// collected until THAT task's exception has been reported. Only then is the collection returned — so "nothing
/// of ours was reported" means the finalizer pass that would have reported it really ran, not merely that a
/// fixed number of GC cycles went by before the runtime got round to it.
/// </summary>
internal static class UnobservedTaskProbe
{
public static async Task<IReadOnlyList<Exception>> CollectAsync(Func<Task> abandon, TimeSpan timeout)
{
var marker = $"unobserved-task-probe-control-{Guid.NewGuid():N}";
var seen = new ConcurrentQueue<Exception>();
var controlReported = 0;
EventHandler<UnobservedTaskExceptionEventArgs> handler = (_, e) =>
{
foreach (var ex in e.Exception.Flatten().InnerExceptions)
{
if (ex.Message == marker) Volatile.Write(ref controlReported, 1);
else seen.Enqueue(ex);
}
};
TaskScheduler.UnobservedTaskException += handler;
try
{
await abandon();
AbandonControl(marker);
var giveUp = DateTime.UtcNow + timeout;
while (Volatile.Read(ref controlReported) == 0)
{
if (DateTime.UtcNow > giveUp)
throw new TimeoutException("The positive control's unobserved exception was never reported.");
GC.Collect();
GC.WaitForPendingFinalizers();
await Task.Delay(10);
}
// One more full pass, so anything that became garbage in the same cycle as the control is reported too.
GC.Collect();
GC.WaitForPendingFinalizers();
}
finally
{
TaskScheduler.UnobservedTaskException -= handler;
}
return [.. seen];
}
[MethodImpl(MethodImplOptions.NoInlining)]
private static void AbandonControl(string marker) => _ = Task.FromException(new InvalidOperationException(marker));
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs
+14 -19
View File
@@ -92,9 +92,9 @@ internal sealed class DifferentialAbortedException(string message, Exception? in
/// <para><b>Stop on first failure (review r1 C1).</b> The first connection-level failure on either side —
/// connect / RegisterSession / Forward Open refused, timeout, session lost — aborts the whole run with the
/// operation and status; nothing is retried. The native side is additionally capped at ONE open attempt
/// process-wide (<see cref="CipConnection.OpenAttemptBudget"/>), so a reconnect cannot reach the wire even
/// to the press under test (<see cref="ConnectionDiagnostics.LimitOpenAttempts"/>), so a reconnect cannot reach the wire even
/// from inside an enumerator that swallows a per-program failure, and the run aborts the moment
/// <see cref="CipConnection.ForwardOpenCount"/> passes 1.</para>
/// <see cref="ConnectionDiagnostics.ForwardOpens"/> passes 1.</para>
/// </summary>
internal static class DifferentialRun
{
@@ -222,9 +222,9 @@ internal static class DifferentialRun
/// <summary>
/// Runs <paramref name="body"/> with every native operation sharing ONE connection, held open by an anchor
/// lease: the process-wide Forward Open counter is reset, the open-attempt budget is set to exactly one
/// more attempt, and <paramref name="body"/> receives a check to call after each operation that aborts the
/// moment a second Forward Open reaches the wire. The anchor is released and the budget restored in
/// lease: the shared pool's Forward Open counters are reset, the press is limited to exactly one
/// open attempt, and <paramref name="body"/> receives a check to call after each operation that aborts the
/// moment a second Forward Open reaches the wire. The anchor is released and the limit lifted in
/// <c>finally</c>. Shared by the differential and the latency comparison.
/// </summary>
/// <returns>The body's result, the Forward Opens counted between the reset and the anchor release, and the
@@ -233,43 +233,38 @@ internal static class DifferentialRun
string phase, DifferentialTarget target, Func<Action<string>, Task<T>> body)
{
var anchorParams = target.Params("@tags");
var key = NativeAbCipDevice.Key(anchorParams);
var diagnostics = DeviceSessionPool.Shared.Diagnostics;
void CheckOpens(string operation)
{
var opens = Volatile.Read(ref CipConnection.ForwardOpenCount);
var opens = diagnostics.ForwardOpens;
if (opens > 1)
throw new DifferentialAbortedException(
$"Differential ABORTED in the {phase} phase after {operation}: {opens} Forward Open frames on the wire, " +
"expected exactly one (a reconnect, or a Large→regular fallback). Nothing was retried.");
}
Interlocked.Exchange(ref CipConnection.ForwardOpenCount, 0);
Interlocked.Exchange(ref CipConnection.ForwardClosesSent, 0);
Interlocked.Exchange(ref CipConnection.ForwardClosesAcked, 0);
Volatile.Write(ref CipConnection.OpenAttemptBudget, Volatile.Read(ref CipConnection.OpenAttempts) + 1);
diagnostics.ResetForwardCounts();
T result;
int forwardOpens;
NativeCloses forwardCloses;
try
// Exactly one open attempt to the press under test; lifted when the scope ends, whatever happens.
using (diagnostics.LimitOpenAttempts(key, 1))
{
// Acquire does no I/O; the key and options carry no tag-specific field, so every lease shares this entry.
var anchor = DeviceSessionPool.Shared.Acquire(NativeAbCipDevice.Key(anchorParams), NativeAbCipDevice.Options(anchorParams));
var anchor = DeviceSessionPool.Shared.Acquire(key, NativeAbCipDevice.Options(anchorParams));
try
{
result = await body(CheckOpens).ConfigureAwait(false);
}
finally
{
forwardOpens = Volatile.Read(ref CipConnection.ForwardOpenCount);
forwardOpens = diagnostics.ForwardOpens;
await anchor.DisposeAsync().ConfigureAwait(false);
forwardCloses = new NativeCloses(
Volatile.Read(ref CipConnection.ForwardClosesSent), Volatile.Read(ref CipConnection.ForwardClosesAcked));
forwardCloses = new NativeCloses(diagnostics.ForwardClosesSent, diagnostics.ForwardClosesAcked);
}
}
finally
{
Volatile.Write(ref CipConnection.OpenAttemptBudget, -1);
}
return (result, forwardOpens, forwardCloses);
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs
+28 -11
View File
@@ -30,12 +30,28 @@ public sealed class DifferentialSafetyTests
() => DifferentialRun.RunNativePhaseAsync(target, selection, TestContext.Current.CancellationToken));
ex.Message.ShouldContain("native phase at browse");
// Long enough for a throttled (≥ 1 s) reconnect to have reached the listener, had one been allowed.
await Task.Delay(TimeSpan.FromSeconds(2.5), TestContext.Current.CancellationToken);
endpoint.Connections.ShouldBe(1);
Volatile.Read(ref CipConnection.OpenAttemptBudget).ShouldBe(-1, "the budget is restored after the phase");
DeviceSessionPool.Shared.Diagnostics.HasOpenAttemptLimit(Key(target)).ShouldBeFalse("the limit is lifted after the phase");
await ControlOpenAsync(target);
endpoint.Connections.ShouldBe(2, "the phase's one connect, then the control's — no reconnect from the phase");
}
/// <summary>
/// Positive control (Gitea #583) for "no reconnect reached the listener": one more open to the same device,
/// made after the phase, once its limit is lifted. It goes through the same per-device Forward Open throttle,
/// whose waiters are served in arrival order, so by the time it has reached the listener (it is counted on
/// accept, before the hang-up that fails it) any reconnect the phase had queued would already have got there
/// — and, the limit being gone, nothing would have refused it.
/// </summary>
private static async Task ControlOpenAsync(DifferentialTarget target)
{
// A native tag initialisation (an existence read), like the harness's own: same pool, same device key.
using var runtime = AbCipTransportFactories.For(AbCipTransport.Native).Tags
.Create(target.Params("A", 1, false, AbCipDataType.DInt));
await Should.ThrowAsync<AbTransportException>(() => runtime.InitializeAsync(TestContext.Current.CancellationToken));
}
private static DeviceKey Key(DifferentialTarget target) => NativeAbCipDevice.Key(target.Params("@tags"));
[Fact]
public async Task A_failing_endpoint_aborts_the_libplctag_phase_before_any_tag_is_read()
{
@@ -146,13 +162,13 @@ public sealed class DifferentialSafetyTests
public async Task The_open_attempt_budget_alone_stops_a_second_open_even_when_nothing_aborts_the_phase()
{
// Unlike the abort test above, the body here SWALLOWS every failure and keeps going, so the harness's
// abort-on-first-failure can never end the phase: only CipConnection.OpenAttemptBudget stands between
// abort-on-first-failure can never end the phase: only the per-device open-attempt limit stands between
// the second request and the wire.
await using var endpoint = HangUpListener.Start();
var target = Target(endpoint.Port);
var statuses = new List<AbStatus>();
string? secondMessage = null;
var attemptsBefore = Volatile.Read(ref CipConnection.OpenAttempts);
var attemptsBefore = DeviceSessionPool.Shared.Diagnostics.OpenAttempts;
var tags = AbCipTransportFactories.For(AbCipTransport.Native).Tags;
await DifferentialRun.WithNativeAnchorAsync("native", target, async _ =>
@@ -178,11 +194,12 @@ public sealed class DifferentialSafetyTests
statuses.Count.ShouldBe(2, "both requests failed; neither aborted the phase");
statuses[1].ShouldBe(AbStatus.Abort);
secondMessage.ShouldNotBeNull().ShouldContain("budget");
Volatile.Read(ref CipConnection.OpenAttempts).ShouldBe(attemptsBefore + 2, "the second attempt was counted, then refused");
// Long enough for a throttled reconnect to have reached the listener had the budget let one through.
await Task.Delay(TimeSpan.FromSeconds(1.5), TestContext.Current.CancellationToken);
endpoint.Connections.ShouldBe(1, "exactly the one budgeted open attempt reached the wire");
Volatile.Read(ref CipConnection.OpenAttemptBudget).ShouldBe(-1, "the budget is restored afterwards");
// The pool is process-wide, so another test's opens could move this counter; at least these two are in it.
(DeviceSessionPool.Shared.Diagnostics.OpenAttempts - attemptsBefore).ShouldBeGreaterThanOrEqualTo(2,
"the second attempt was counted, then refused");
DeviceSessionPool.Shared.Diagnostics.HasOpenAttemptLimit(Key(target)).ShouldBeFalse("the limit is lifted afterwards");
await ControlOpenAsync(target);
endpoint.Connections.ShouldBe(2, "exactly the one budgeted open attempt reached the wire, then the control's");
}
[Fact]
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs
+1 -1
View File
@@ -126,7 +126,7 @@ public sealed class PollLatencyComparisonTests(DifferentialRunFixture fixture)
$"{passes.Sum(p => p.TagFailures)} |";
}
private static string Rtt(ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip.CipConnection.ExchangeStats? s) =>
private static string Rtt(ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip.ConnectionDiagnostics.ExchangeStats? s) =>
s is not { Count: > 0 } x ? "-" : $"{Ms(x.MinMs)}/{Ms(x.MeanMs)}/{Ms(x.MaxMs)}";
private static string Ms(double ms) => ms.ToString("0.0", CultureInfo.InvariantCulture);
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs
+8 -8
View File
@@ -5,13 +5,13 @@ using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential;
/// <summary>One timed burst: every tag read once, concurrently.</summary>
/// <param name="Requests">Native only: <see cref="CipConnection.RequestsSent"/> delta over the pass (keep-alives included).</param>
/// <param name="Requests">Native only: <see cref="ConnectionDiagnostics.RequestsSent"/> delta over the pass (keep-alives included).</param>
/// <param name="TagFailures">Reads that ended with a tag-level status (a connection-level one aborts instead).</param>
/// <param name="Gc">Collections that ran during the pass, per generation (T21 review r1 M8: a client-side stall shows here).</param>
/// <param name="Rtt">Native only: round trip of each CIP exchange in the pass, from <see cref="CipConnection"/> (a slow
/// controller shows here; a slow pass with fast exchanges points at the client).</param>
internal sealed record LatencyPass(
int Index, TimeSpan Wall, int? Requests, int TagFailures, GcDelta Gc = default, CipConnection.ExchangeStats? Rtt = null);
int Index, TimeSpan Wall, int? Requests, int TagFailures, GcDelta Gc = default, ConnectionDiagnostics.ExchangeStats? Rtt = null);
/// <summary>GC collections per generation over a window (gen0 counts include gen1/gen2 collections, as the runtime reports).</summary>
internal readonly record struct GcDelta(int Gen0, int Gen1, int Gen2)
@@ -137,7 +137,7 @@ internal static class PollLatencyRun
var runtimes = new List<(SelectedTag Tag, IAbCipTagRuntime Runtime)>();
try
{
var before = Volatile.Read(ref CipConnection.RequestsSent);
var before = DeviceSessionPool.Shared.Diagnostics.RequestsSent;
foreach (var tag in tags)
{
var rt = await CreateAsync("latency native", target, nativeFactory, tag, ct).ConfigureAwait(false);
@@ -145,7 +145,7 @@ internal static class PollLatencyRun
if (rt is null) { problems.Add($"native could not read '{tag.Name}', which libplctag read"); continue; }
runtimes.Add((tag, rt));
}
var setup = Volatile.Read(ref CipConnection.RequestsSent) - before;
var setup = DeviceSessionPool.Shared.Diagnostics.RequestsSent - before;
var passes = await RunPassesAsync("latency native", target, runtimes, countRequests: true, checkOpens, ct)
.ConfigureAwait(false);
return (passes, setup);
@@ -214,15 +214,15 @@ internal static class PollLatencyRun
{
if (p > 1 && target.LatencyPassGapMs > 0) await Task.Delay(target.LatencyPassGapMs, ct).ConfigureAwait(false);
var requestsBefore = Volatile.Read(ref CipConnection.RequestsSent);
CipConnection.ResetExchangeStats();
var requestsBefore = DeviceSessionPool.Shared.Diagnostics.RequestsSent;
DeviceSessionPool.Shared.Diagnostics.ResetExchangeStats();
var gcBefore = GcDelta.Now();
var clock = Stopwatch.StartNew();
var outcomes = await Task.WhenAll(runtimes.Select(r => ReadOnceAsync(r.Runtime, ct))).ConfigureAwait(false);
clock.Stop();
var gc = GcDelta.Now() - gcBefore;
var rtt = countRequests ? CipConnection.ReadExchangeStats() : (CipConnection.ExchangeStats?)null;
var requests = Volatile.Read(ref CipConnection.RequestsSent) - requestsBefore;
var rtt = countRequests ? DeviceSessionPool.Shared.Diagnostics.ReadExchangeStats() : (ConnectionDiagnostics.ExchangeStats?)null;
var requests = DeviceSessionPool.Shared.Diagnostics.RequestsSent - requestsBefore;
var failures = 0;
for (var i = 0; i < outcomes.Length; i++)
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs
+75 -80
View File
@@ -48,26 +48,33 @@ public sealed class AbCipAlarmProjectionTests
AbCipAlarmProjection.MapSeverity(900).ShouldBe(AlarmSeverity.Critical);
}
private static AbCipDriverOptions Options(bool enabled) => new()
{
Devices = [new AbCipDeviceOptions(Device)],
RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
EnableAlarmProjection = enabled,
AlarmPollInterval = TimeSpan.FromMilliseconds(20),
// The ALMD projection here drives the parent-UDT runtime via offset-keyed values,
// so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003).
EnableDeclarationOnlyUdtGrouping = true,
};
/// <summary>Verifies that disabled alarm projection returns a valid handle but does not poll.</summary>
[Fact]
public async Task FeatureFlag_Off_SubscribeAlarms_Returns_Handle_But_Never_Polls()
{
var factory = new FakeAbCipTagFactory();
var opts = new AbCipDriverOptions
{
Devices = [new AbCipDeviceOptions(Device)],
RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
EnableAlarmProjection = false, // explicit; also the default
};
var drv = new AbCipDriver(opts, "drv-1", factory);
var drv = new AbCipDriver(Options(enabled: false), "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
handle.ShouldNotBeNull();
handle.DiagnosticId.ShouldContain("abcip-alarm-sub-");
// Wait a touch — if polling were active, a fake member-read would be triggered.
await Task.Delay(100);
// Positive control (#583): the same options with the flag ON poll three times — so a loop this
// subscription had started would have read (and so created) the ALMD runtime by now.
await PolledThreeTimesWithTheFlagOnAsync();
factory.Tags.ShouldNotContainKey("HighTemp");
factory.Tags.ShouldNotContainKey("HighTemp.InFaulted");
factory.Tags.ShouldNotContainKey("HighTemp.Severity");
@@ -80,17 +87,7 @@ public sealed class AbCipAlarmProjectionTests
public async Task FeatureFlag_On_Subscribe_Starts_Polling_And_Fires_Raise_On_0_to_1()
{
var factory = new FakeAbCipTagFactory();
var opts = new AbCipDriverOptions
{
Devices = [new AbCipDeviceOptions(Device)],
RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
EnableAlarmProjection = true,
AlarmPollInterval = TimeSpan.FromMilliseconds(20),
// The ALMD projection here drives the parent-UDT runtime via offset-keyed values,
// so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003).
EnableDeclarationOnlyUdtGrouping = true,
};
var drv = new AbCipDriver(opts, "drv-1", factory);
var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
var events = new List<AlarmEventArgs>();
@@ -99,21 +96,16 @@ public sealed class AbCipAlarmProjectionTests
var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
// The ALMD UDT is declared so whole-UDT grouping kicks in; the parent HighTemp runtime
// gets created + polled. Set InFaulted offset-value to 0 first (clear), wait a tick,
// then flip to 1 (fault) + wait for the raise event.
// gets created + polled. Set InFaulted offset-value to 0 first (clear), let polls seed the
// "last-seen false" state, then flip to 1 (fault) + wait for the raise event.
await WaitForTagCreation(factory, "HighTemp");
factory.Tags["HighTemp"].ValuesByOffset[0] = 0; // InFaulted=false at offset 0
factory.Tags["HighTemp"].ValuesByOffset[8] = 500; // Severity at offset 8 (after InFaulted+Acked)
await Task.Delay(80); // let a tick seed the "last-seen false" state
await WaitForMoreReadsAsync(factory, "HighTemp", 2);
factory.Tags["HighTemp"].ValuesByOffset[0] = 1; // flip to faulted
await Task.Delay(200); // allow several polls to be safe
lock (events)
{
events.ShouldContain(e => e.SourceNodeId == "HighTemp" && e.AlarmType == "ALMD"
&& e.Message.Contains("raised"));
}
await WaitForEventAsync(events, e => e.SourceNodeId == "HighTemp" && e.AlarmType == "ALMD"
&& e.Message.Contains("raised"));
await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None);
await drv.ShutdownAsync(CancellationToken.None);
@@ -135,15 +127,7 @@ public sealed class AbCipAlarmProjectionTests
public async Task Resubscribe_Collapses_To_Single_Loop_No_Duplicate_Raise()
{
var factory = new FakeAbCipTagFactory();
var opts = new AbCipDriverOptions
{
Devices = [new AbCipDeviceOptions(Device)],
RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
EnableAlarmProjection = true,
AlarmPollInterval = TimeSpan.FromMilliseconds(20),
EnableDeclarationOnlyUdtGrouping = true,
};
var drv = new AbCipDriver(opts, "drv-1", factory);
var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
var raises = new List<AlarmEventArgs>();
@@ -158,17 +142,18 @@ public sealed class AbCipAlarmProjectionTests
await WaitForTagCreation(factory, "HighTemp");
factory.Tags["HighTemp"].ValuesByOffset[0] = 0; // InFaulted=false at offset 0
factory.Tags["HighTemp"].ValuesByOffset[8] = 500; // Severity at offset 8
await Task.Delay(80); // let sub1's loop seed its "last-seen false" baseline
await WaitForMoreReadsAsync(factory, "HighTemp", 2); // sub1's loop has seeded its baseline
// Second subscribe = the actor re-subscribing across a reconnect. Its loop reads the same
// shared HighTemp runtime; give it time to seed its own "last-seen false" baseline too.
// shared HighTemp runtime; let it poll (and so seed its own "last-seen false" baseline) too.
var sub2 = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
await Task.Delay(80);
await WaitForMoreReadsAsync(factory, "HighTemp", 3);
factory.Tags["HighTemp"].ValuesByOffset[0] = 1; // one 0->1 transition
// Wait past several 20ms poll intervals so any leaked second loop has ample time to also
// fire its duplicate raise before we assert.
await Task.Delay(250);
await WaitForEventAsync(raises, _ => true);
// Processed-count wait (#583), not a fixed delay: six more reads of the shared runtime after the raise,
// so a leaked second loop — which reads it too — has had polls in which to fire its duplicate.
await WaitForMoreReadsAsync(factory, "HighTemp", 6);
await drv.UnsubscribeAlarmsAsync(sub1, CancellationToken.None);
await drv.UnsubscribeAlarmsAsync(sub2, CancellationToken.None);
@@ -186,17 +171,7 @@ public sealed class AbCipAlarmProjectionTests
public async Task Clear_Event_Fires_On_1_to_0_Transition()
{
var factory = new FakeAbCipTagFactory();
var opts = new AbCipDriverOptions
{
Devices = [new AbCipDeviceOptions(Device)],
RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
EnableAlarmProjection = true,
AlarmPollInterval = TimeSpan.FromMilliseconds(20),
// The ALMD projection here drives the parent-UDT runtime via offset-keyed values,
// so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003).
EnableDeclarationOnlyUdtGrouping = true,
};
var drv = new AbCipDriver(opts, "drv-1", factory);
var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
var events = new List<AlarmEventArgs>();
@@ -207,16 +182,10 @@ public sealed class AbCipAlarmProjectionTests
factory.Tags["HighTemp"].ValuesByOffset[0] = 1;
factory.Tags["HighTemp"].ValuesByOffset[8] = 500;
await Task.Delay(80); // observe raise
await WaitForEventAsync(events, e => e.Message.Contains("raised"));
factory.Tags["HighTemp"].ValuesByOffset[0] = 0;
await Task.Delay(200);
lock (events)
{
events.ShouldContain(e => e.Message.Contains("raised"));
events.ShouldContain(e => e.Message.Contains("cleared"));
}
await WaitForEventAsync(events, e => e.Message.Contains("cleared"));
await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None);
await drv.ShutdownAsync(CancellationToken.None);
@@ -227,33 +196,59 @@ public sealed class AbCipAlarmProjectionTests
public async Task Unsubscribe_Stops_The_Poll_Loop()
{
var factory = new FakeAbCipTagFactory();
var opts = new AbCipDriverOptions
{
Devices = [new AbCipDeviceOptions(Device)],
RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
EnableAlarmProjection = true,
AlarmPollInterval = TimeSpan.FromMilliseconds(20),
// The ALMD projection here drives the parent-UDT runtime via offset-keyed values,
// so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003).
EnableDeclarationOnlyUdtGrouping = true,
};
var drv = new AbCipDriver(opts, "drv-1", factory);
var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
await WaitForTagCreation(factory, "HighTemp");
var preUnsubReadCount = factory.Tags["HighTemp"].ReadCount;
await WaitForMoreReadsAsync(factory, "HighTemp", 1); // the loop is running
// Unsubscribe awaits the loop's exit, so no read can land after it returns.
await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None);
await Task.Delay(100); // well past several poll intervals if the loop were still alive
var postUnsubReadCount = factory.Tags["HighTemp"].ReadCount;
var postDelayReadCount = factory.Tags["HighTemp"].ReadCount;
// Allow at most one straggler read between the unsubscribe-cancel + the loop exit.
(postDelayReadCount - preUnsubReadCount).ShouldBeLessThanOrEqualTo(1);
// Positive control (#583): an identical, still-subscribed driver polls three more times on the same
// interval — so the unsubscribed loop, had it survived, would have read again by now.
await PolledThreeTimesWithTheFlagOnAsync();
factory.Tags["HighTemp"].ReadCount.ShouldBe(postUnsubReadCount);
await drv.ShutdownAsync(CancellationToken.None);
}
/// <summary>A control driver with the projection ON polls its ALMD runtime three times.</summary>
private static async Task PolledThreeTimesWithTheFlagOnAsync()
{
var factory = new FakeAbCipTagFactory();
var control = new AbCipDriver(Options(enabled: true), "drv-control", factory);
await control.InitializeAsync("{}", CancellationToken.None);
var handle = await control.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
await WaitForTagCreation(factory, "HighTemp");
await WaitForMoreReadsAsync(factory, "HighTemp", 3);
await control.UnsubscribeAlarmsAsync(handle, CancellationToken.None);
await control.ShutdownAsync(CancellationToken.None);
}
/// <summary>Waits for <paramref name="more"/> further reads of the fake runtime (a processed-count wait).</summary>
private static async Task WaitForMoreReadsAsync(FakeAbCipTagFactory factory, string tagName, int more)
{
var tag = factory.Tags[tagName];
var target = tag.ReadCount + more;
var deadline = DateTime.UtcNow.AddSeconds(5);
while (tag.ReadCount < target && DateTime.UtcNow < deadline) await Task.Delay(5);
tag.ReadCount.ShouldBeGreaterThanOrEqualTo(target, $"'{tagName}' should have been read {more} more times");
}
private static async Task WaitForEventAsync(List<AlarmEventArgs> events, Func<AlarmEventArgs, bool> match)
{
var deadline = DateTime.UtcNow.AddSeconds(5);
while (DateTime.UtcNow < deadline)
{
lock (events) if (events.Any(match)) return;
await Task.Delay(5);
}
throw new TimeoutException("The expected alarm event was never raised.");
}
private static async Task WaitForTagCreation(FakeAbCipTagFactory factory, string tagName)
{
var deadline = DateTime.UtcNow.AddSeconds(2);
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipDriverCodeReviewRegressionTests.cs
+5 -2
View File
@@ -152,8 +152,11 @@ public sealed class AbCipDriverCodeReviewRegressionTests
}, "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
// Give the probe loop a moment to actually start spinning.
await Task.Delay(60);
// Wait until the probe loop is actually spinning (#583: a processed count, not a fixed moment).
var deadline = DateTime.UtcNow.AddSeconds(5);
while (!(factory.Tags.TryGetValue("ProbeTag", out var probe) && probe.ReadCount >= 2) && DateTime.UtcNow < deadline)
await Task.Delay(5, TestContext.Current.CancellationToken);
factory.Tags["ProbeTag"].ReadCount.ShouldBeGreaterThanOrEqualTo(2);
// Must complete cleanly — no ObjectDisposedException from a loop racing a disposed CTS.
await Should.NotThrowAsync(() => drv.ShutdownAsync(CancellationToken.None));
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs
+28 -2
View File
@@ -100,11 +100,36 @@ public sealed class AbCipHostProbeTests
drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
await drv.InitializeAsync("{}", CancellationToken.None);
await Task.Delay(300);
// #583: structural, not a fixed wait — no probe loop was ever started, so no transition can follow.
drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask.ShouldBeNull();
factory.Tags.ShouldBeEmpty();
transitions.ShouldBeEmpty();
drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown);
await drv.ShutdownAsync(CancellationToken.None);
await ProbeLoopIsObservableAsync();
}
/// <summary>
/// Positive control for the "no probe loop" assertions above: the same observation (the device's
/// <c>ProbeTask</c>) on a driver whose probe IS configured is non-null, and that loop does produce a
/// transition — so a null <c>ProbeTask</c> really means "no loop", not "a loop the hook cannot see".
/// </summary>
private static async Task ProbeLoopIsObservableAsync()
{
var factory = new FakeAbCipTagFactory { Customise = p => new FakeAbCipTag(p) { Status = 0 } };
var transitions = new ConcurrentQueue<HostStatusChangedEventArgs>();
var control = new AbCipDriver(new AbCipDriverOptions
{
Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")],
Probe = new AbCipProbeOptions { Enabled = true, ProbeTagPath = "@raw_cpu_type", Interval = TimeSpan.FromMilliseconds(100) },
}, "drv-control", factory);
control.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
await control.InitializeAsync("{}", CancellationToken.None);
(control.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask is not null).ShouldBeTrue();
await WaitForAsync(() => !transitions.IsEmpty, TimeSpan.FromSeconds(2));
transitions.ShouldNotBeEmpty();
await control.ShutdownAsync(CancellationToken.None);
}
/// <summary>Verifies that the probe is skipped when ProbeTagPath is null.</summary>
@@ -118,10 +143,11 @@ public sealed class AbCipHostProbeTests
}, "drv-1");
await drv.InitializeAsync("{}", CancellationToken.None);
await Task.Delay(200);
drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask.ShouldBeNull(); // #583: no loop, so no transition
drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown);
await drv.ShutdownAsync(CancellationToken.None);
await ProbeLoopIsObservableAsync();
}
/// <summary>Verifies that the probe loops across multiple devices independently.</summary>
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLoggingTests.cs
+11 -5
View File
@@ -65,8 +65,12 @@ public sealed class AbCipLoggingTests
await drv.InitializeAsync("{}", CancellationToken.None);
// Give the probe loop a couple of ticks to log.
await Task.Delay(200);
// Wait (bounded) for the entry itself rather than a fixed couple of ticks (#583).
var deadline = DateTime.UtcNow.AddSeconds(5);
while (!logger.Entries.Any(e => e.Message.Contains("probe", StringComparison.OrdinalIgnoreCase)
|| (e.Exception?.Message.Contains("simulated probe init failure") ?? false))
&& DateTime.UtcNow < deadline)
await Task.Delay(5, TestContext.Current.CancellationToken);
await drv.ShutdownAsync(CancellationToken.None);
@@ -186,8 +190,10 @@ public sealed class AbCipLoggingTests
/// <summary>Test logger that captures all log entries.</summary>
internal sealed class CapturingLogger<T> : ILogger<T>
{
/// <summary>Gets the captured log entries.</summary>
public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = new();
private readonly List<(LogLevel Level, string Message, Exception? Exception)> _entries = new();
/// <summary>Gets a snapshot of the captured log entries (the probe loop logs from another thread).</summary>
public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get { lock (_entries) return [.. _entries]; } }
/// <summary>Begins a scope (stub implementation).</summary>
/// <typeparam name="TState">The type of the scope state.</typeparam>
@@ -206,7 +212,7 @@ public sealed class AbCipLoggingTests
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
Entries.Add((logLevel, formatter(state, exception), exception));
lock (_entries) _entries.Add((logLevel, formatter(state, exception), exception));
}
private sealed class NullScope : IDisposable
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs
+47 -11
View File
@@ -67,7 +67,8 @@ public sealed class AbCipSubscriptionTests
drv.OnDataChange += (_, e) => events.Enqueue(e);
var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
await Task.Delay(500);
// Processed-count wait (#583): five polls have read the unchanged value, not "500 ms went by".
await WaitForReadsAsync(factory, "Speed", 5);
await drv.UnsubscribeAsync(handle, CancellationToken.None);
events.Count.ShouldBe(1);
@@ -101,22 +102,30 @@ public sealed class AbCipSubscriptionTests
public async Task Unsubscribe_halts_polling()
{
var (drv, factory) = NewDriver(
new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt));
new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt),
new AbCipTagDefinition("Temp", "ab://10.0.0.5/1,0", "Temp", AbCipDataType.DInt));
await drv.InitializeAsync("{}", CancellationToken.None);
var tagRef = new FakeAbCipTag(new AbCipTagCreateParams("10.0.0.5", 44818, "1,0", "controllogix", "Speed", TimeSpan.FromSeconds(2))) { Value = 1 };
factory.Customise = _ => tagRef;
factory.Customise = p => p.TagName == "Speed" ? tagRef : new FakeAbCipTag(p) { Value = 1 };
var events = new ConcurrentQueue<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => events.Enqueue(e);
var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1));
var controlHandle = await drv.SubscribeAsync(["Temp"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
await WaitForAsync(() => events.Any(e => e.FullReference == "Speed"), TimeSpan.FromSeconds(1));
await drv.UnsubscribeAsync(handle, CancellationToken.None);
var afterUnsub = events.Count;
var afterUnsub = events.Count(e => e.FullReference == "Speed");
var readsAfterUnsub = tagRef.ReadCount;
tagRef.Value = 999;
await Task.Delay(400);
events.Count.ShouldBe(afterUnsub);
// Positive control (#583): the still-subscribed sibling, on the same interval, polls three more times —
// so Speed's loop, had it survived the unsubscribe, has had three chances to read the change.
await WaitForReadsAsync(factory, "Temp", factory.Tags["Temp"].ReadCount + 3);
events.Count(e => e.FullReference == "Speed").ShouldBe(afterUnsub);
tagRef.ReadCount.ShouldBe(readsAfterUnsub);
await drv.UnsubscribeAsync(controlHandle, CancellationToken.None);
}
/// <summary>Verifies that polling intervals below 100ms are floored to the minimum.</summary>
@@ -132,12 +141,17 @@ public sealed class AbCipSubscriptionTests
drv.OnDataChange += (_, e) => events.Enqueue(e);
var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(5), CancellationToken.None);
await Task.Delay(300);
// Processed-count wait (#583): three polls, then measure the spacing they actually ran at.
var started = System.Diagnostics.Stopwatch.StartNew();
await WaitForReadsAsync(factory, "Speed", 3);
var elapsed = started.Elapsed;
await drv.UnsubscribeAsync(handle, CancellationToken.None);
// Value is stable → only the initial-data push fires; the 100 ms floor keeps polls sparse enough
// that no extra event is produced against a stable value.
// Value is stable → only the initial-data push fires.
events.Count.ShouldBe(1);
// The floor: the initial read is immediate, so the two later ones took at least two 100 ms intervals.
// At the requested 5 ms they would have taken ~10 ms.
elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(180));
}
/// <summary>Verifies that ShutdownAsync cancels all active subscriptions.</summary>
@@ -154,11 +168,25 @@ public sealed class AbCipSubscriptionTests
_ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1));
var speed = factory.Tags["Speed"];
await drv.ShutdownAsync(CancellationToken.None);
var afterShutdown = events.Count;
await Task.Delay(300);
var readsAfterShutdown = speed.ReadCount;
speed.Value = 2; // a change any surviving loop would raise
// Positive control (#583): a second, independent driver polling on the same interval reads three more
// times — so the shut-down driver's loop, had it survived, has had three chances to raise the change.
var (control, controlFactory) = NewDriver(
new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt));
await control.InitializeAsync("{}", CancellationToken.None);
controlFactory.Customise = p => new FakeAbCipTag(p) { Value = 1 };
_ = await control.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
await WaitForReadsAsync(controlFactory, "Speed", 3);
await control.ShutdownAsync(CancellationToken.None);
events.Count.ShouldBe(afterShutdown);
speed.ReadCount.ShouldBe(readsAfterShutdown);
}
/// <summary>Verifies that subscriptions on UDT members use the synthesized full reference.</summary>
@@ -195,4 +223,12 @@ public sealed class AbCipSubscriptionTests
while (!condition() && DateTime.UtcNow < deadline)
await Task.Delay(20);
}
/// <summary>Waits until the fake runtime for <paramref name="tag"/> has served at least <paramref name="reads"/>
/// reads (a processed-count wait), failing — never passing vacuously — if it does not get there.</summary>
private static async Task WaitForReadsAsync(FakeAbCipTagFactory factory, string tag, int reads)
{
await WaitForAsync(() => factory.Tags.TryGetValue(tag, out var t) && t.ReadCount >= reads, TimeSpan.FromSeconds(5));
factory.Tags[tag].ReadCount.ShouldBeGreaterThanOrEqualTo(reads, $"'{tag}' should have been polled {reads} times");
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs
+3 -2
View File
@@ -25,7 +25,8 @@ internal class FakeAbCipTag : IAbCipTagRuntime
/// <summary>Gets the count of <see cref="InitializeAsync"/> invocations.</summary>
public int InitializeCount { get; private set; }
/// <summary>Gets the count of <see cref="ReadAsync"/> invocations.</summary>
public int ReadCount { get; private set; }
public int ReadCount => Volatile.Read(ref _readCount);
private int _readCount; // interlocked: poll loops read it from another thread (#583 processed-count waits)
/// <summary>Gets the count of <see cref="WriteAsync"/> invocations.</summary>
public int WriteCount { get; private set; }
/// <summary>Gets a value indicating whether the tag has been disposed.</summary>
@@ -48,7 +49,7 @@ internal class FakeAbCipTag : IAbCipTagRuntime
/// <param name="cancellationToken">The cancellation token for the operation.</param>
public virtual Task ReadAsync(CancellationToken cancellationToken)
{
ReadCount++;
Interlocked.Increment(ref _readCount);
if (ThrowOnRead) throw Exception ?? new InvalidOperationException("fake read failure");
return Task.CompletedTask;
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/LibplctagOrphanReleaseTests.cs
+13 -7
View File
@@ -18,7 +18,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests;
[CollectionDefinition(Name, DisableParallelization = true)]
public sealed class LibplctagOrphanReleaseCollection
{
/// <summary>The release counters and the rooting seam are process-wide: nothing else may run alongside.</summary>
/// <summary>The release counters are process-wide: nothing else may run alongside.</summary>
public const string Name = "libplctag orphan release";
}
@@ -90,23 +90,29 @@ public sealed class LibplctagOrphanReleaseTests
[Fact]
public async Task Without_a_readable_handle_the_failed_tag_is_rooted_instead()
{
// Gitea #581: the "unreadable handle" decision is passed in, not flipped through a process-wide switch.
var rooted = LibplctagOrphanRelease.RootedCount;
var released = LibplctagOrphanRelease.Released;
var rt = new LibplctagTagRuntime(Refused());
LibplctagOrphanRelease.ForceRootingForTest = true;
var tag = LibplctagTagRuntime.BuildTag(Refused());
try
{
await Should.ThrowAsync<AbTransportException>(() => rt.InitializeAsync(CancellationToken.None));
await Should.ThrowAsync<Exception>(() => tag.InitializeAsync(CancellationToken.None));
var handle = LibplctagOrphanRelease.HandleOf(tag);
handle.ShouldBeGreaterThan(0, "plc_tag_create_ex returned a handle before the failure");
LibplctagOrphanRelease.ReleaseIfOrphaned(tag, canReadHandles: false).ShouldBeNull();
LibplctagOrphanRelease.RootedCount.ShouldBe(rooted + 1);
LibplctagOrphanRelease.Released.ShouldBe(released);
NativeTagExists(handle).ShouldBeTrue("the rooting fallback destroys nothing");
LibplctagOrphanRelease.HandleOf(tag).ShouldBe(handle);
}
finally
{
LibplctagOrphanRelease.ForceRootingForTest = false;
// Clean up the native tag this test deliberately leaked (its callback was unregistered by the timeout).
var handle = rt.NativeHandle;
var handle = LibplctagOrphanRelease.HandleOf(tag);
if (handle > 0) plctag.plc_tag_destroy(handle);
rt.Dispose();
tag.Dispose();
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipLeaseReleaseTests.cs
+3 -1
View File
@@ -72,7 +72,9 @@ public sealed class NativeAbCipLeaseReleaseTests
// the first one's Forward Close is in flight (review finding 22) — the pool holds it behind the close.
var second = (NativeAbCipTagRuntime)factory.Create(Params(target.Port));
var init = second.InitializeAsync(Ct);
await Task.Delay(300, Ct);
// Positive control (#583): the second open has reached the pool's predecessor wait and parked there, so
// the missing Forward Open below is that wait holding it — not the open simply not having got that far.
await FakeCipTarget.Until(() => pool.Diagnostics.PredecessorWaits == 1, Guard, "the re-create to park behind the close");
target.ForwardOpens.ShouldBe(1);
init.IsCompleted.ShouldBeFalse();
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyCapabilityTests.cs
+42 -7
View File
@@ -71,11 +71,17 @@ public sealed class AbLegacyCapabilityTests
{
var tagRef = new FakeAbLegacyTag(
new AbLegacyTagCreateParams("10.0.0.5", 44818, "1,0", "slc500", "N7:0", TimeSpan.FromSeconds(2))) { Value = 1 };
var factory = new FakeAbLegacyTagFactory { Customise = _ => tagRef };
var control = new FakeAbLegacyTag(
new AbLegacyTagCreateParams("10.0.0.5", 44818, "1,0", "slc500", "N7:1", TimeSpan.FromSeconds(2))) { Value = 1 };
var factory = new FakeAbLegacyTagFactory { Customise = p => p.TagName == "N7:1" ? control : tagRef };
var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0")],
RawTags = AbLegacyRawTags.Entries([new AbLegacyTagDefinition("X", "ab://10.0.0.5/1,0", "N7:0", AbLegacyDataType.Int)]),
RawTags = AbLegacyRawTags.Entries(
[
new AbLegacyTagDefinition("X", "ab://10.0.0.5/1,0", "N7:0", AbLegacyDataType.Int),
new AbLegacyTagDefinition("Y", "ab://10.0.0.5/1,0", "N7:1", AbLegacyDataType.Int),
]),
Probe = new AbLegacyProbeOptions { Enabled = false },
}, "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
@@ -84,13 +90,22 @@ public sealed class AbLegacyCapabilityTests
drv.OnDataChange += (_, e) => events.Enqueue(e);
var handle = await drv.SubscribeAsync(["X"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1));
var controlHandle = await drv.SubscribeAsync(["Y"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
await WaitForAsync(() => events.Count(e => e.FullReference == "X") >= 1, TimeSpan.FromSeconds(1));
await drv.UnsubscribeAsync(handle, CancellationToken.None);
var afterUnsub = events.Count;
var afterUnsub = events.Count(e => e.FullReference == "X");
var readsAfterUnsub = tagRef.ReadCount;
tagRef.Value = 999;
await Task.Delay(300, TestContext.Current.CancellationToken);
events.Count.ShouldBe(afterUnsub);
// Positive control (#583): the still-subscribed sibling, on the same interval, polls three more times —
// so X's loop, had it survived, has had three chances to read the change and raise it.
var controlReads = control.ReadCount;
await WaitForAsync(() => control.ReadCount >= controlReads + 3, TimeSpan.FromSeconds(3));
control.ReadCount.ShouldBeGreaterThanOrEqualTo(controlReads + 3);
events.Count(e => e.FullReference == "X").ShouldBe(afterUnsub);
tagRef.ReadCount.ShouldBe(readsAfterUnsub);
await drv.UnsubscribeAsync(controlHandle, CancellationToken.None);
}
// ---- IHostConnectivityProbe ----
@@ -171,10 +186,30 @@ public sealed class AbLegacyCapabilityTests
Probe = new AbLegacyProbeOptions { Enabled = true, ProbeAddress = null },
}, "drv-1");
await drv.InitializeAsync("{}", CancellationToken.None);
await Task.Delay(200, TestContext.Current.CancellationToken);
// #583: structural, not a fixed wait — no probe loop was ever started, so no transition can follow.
drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeCts.ShouldBeNull();
drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown);
await drv.ShutdownAsync(CancellationToken.None);
// Positive control: the same observation on a driver whose probe IS configured sees the loop, and that loop
// produces a transition — so a null ProbeCts really means "no loop".
var transitions = new ConcurrentQueue<HostStatusChangedEventArgs>();
var control = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0")],
Probe = new AbLegacyProbeOptions
{
Enabled = true, Interval = TimeSpan.FromMilliseconds(100),
Timeout = TimeSpan.FromMilliseconds(50), ProbeAddress = "S:0",
},
}, "drv-control", new FakeAbLegacyTagFactory());
control.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
await control.InitializeAsync("{}", CancellationToken.None);
control.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeCts.ShouldNotBeNull();
await WaitForAsync(() => !transitions.IsEmpty, TimeSpan.FromSeconds(2));
transitions.ShouldNotBeEmpty();
await control.ShutdownAsync(CancellationToken.None);
}
// ---- IPerCallHostResolver ----
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/FakeAbLegacyTag.cs
+3 -2
View File
@@ -43,7 +43,8 @@ internal class FakeAbLegacyTag : IAbLegacyTagRuntime
public int InitializeCount { get; private set; }
/// <summary>Gets the count of read calls.</summary>
public int ReadCount { get; private set; }
public int ReadCount => Volatile.Read(ref _readCount);
private int _readCount; // interlocked: poll loops read it from another thread (#583 processed-count waits)
/// <summary>Gets the count of write calls.</summary>
public int WriteCount { get; private set; }
@@ -70,7 +71,7 @@ internal class FakeAbLegacyTag : IAbLegacyTagRuntime
/// <returns>A task representing the asynchronous operation.</returns>
public virtual Task ReadAsync(CancellationToken ct)
{
ReadCount++;
Interlocked.Increment(ref _readCount);
if (ThrowOnRead) throw Exception ?? new InvalidOperationException();
return Task.CompletedTask;
}