diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/RequestScheduler.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/RequestScheduler.cs index c1f84c3a..f9102c21 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/RequestScheduler.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/RequestScheduler.cs @@ -19,13 +19,16 @@ public sealed class RequestScheduler : IAsyncDisposable private const byte MultipleServiceCode = 0x0A; private const byte EmbeddedServiceError = 0x1E; - private sealed class Pending(byte[] request) + private sealed class Pending(byte[] request, bool exclusive) { private bool _completed; private CancellationTokenRegistration _registration; private bool _hasRegistration; public byte[] Request { get; } = request; + + /// True when this request must never share a Multiple Service Packet with another. + public bool Exclusive { get; } = exclusive; public TaskCompletionSource Tcs { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); public LinkedListNode? Node { get; set; } @@ -119,12 +122,16 @@ public sealed class RequestScheduler : IAsyncDisposable /// status is not an exception here. Faults with when the channel /// or the MSP envelope fails, or the scheduler is disposed; cancels only when fires. /// - public Task SendAsync(byte[] request, CancellationToken ct) + /// The CIP request. + /// Cancels this waiter only. + /// When true the request is sent alone, never packed into a Multiple Service + /// Packet: for requests whose reply can fill the whole connection (the packer budgets request bytes only). + public Task SendAsync(byte[] request, CancellationToken ct, bool exclusive = false) { ArgumentNullException.ThrowIfNull(request); if (ct.IsCancellationRequested) return Task.FromCanceled(ct); - var p = new Pending(request); + var p = new Pending(request, exclusive); lock (_lock) { ObjectDisposedException.ThrowIf(_disposed, this); @@ -165,7 +172,7 @@ public sealed class RequestScheduler : IAsyncDisposable { if (batch.Count > 0) { - if (!_allowPacking) break; + if (!_allowPacking || n.Value.Exclusive) break; sizes.Add(n.Value.Request.Length); if (MultipleServicePacket.BuildSize(sizes) > _channel.MaxRequestBytes) break; } @@ -177,6 +184,7 @@ public sealed class RequestScheduler : IAsyncDisposable } batch.Add(n.Value); _queue.RemoveFirst(); + if (n.Value.Exclusive) break; } } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagEnumerator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagEnumerator.cs index d9028a74..c05d3259 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagEnumerator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagEnumerator.cs @@ -1,4 +1,6 @@ using System.Runtime.CompilerServices; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Logix; @@ -14,6 +16,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip; internal sealed class NativeAbCipTagEnumerator : IAbCipTagEnumerator { private readonly Func Release)> _acquire; + private readonly ILogger _logger; /// Production constructor: leases from . /// The connection pool (the factory passes ). @@ -28,9 +31,14 @@ internal sealed class NativeAbCipTagEnumerator : IAbCipTagEnumerator /// Test seam: supplies the scheduler and its release directly. /// Acquires the scheduler for a device, with the callback that releases it. + /// Receives release failures; null discards them. internal NativeAbCipTagEnumerator( - Func Release)> acquire) => + Func Release)> acquire, + ILogger? logger = null) + { _acquire = acquire; + _logger = logger ?? NullLogger.Instance; + } /// public async IAsyncEnumerable EnumerateAsync( @@ -47,7 +55,7 @@ internal sealed class NativeAbCipTagEnumerator : IAbCipTagEnumerator } finally { - NativeLease.ReleaseBounded(release); + NativeLease.ReleaseBounded(release, _logger); } } @@ -61,6 +69,12 @@ internal static class NativeListingReader private const string TagsSuffix = ".@tags"; private const string ProgramPrefix = "Program:"; + /// Page cap for one listing; a real controller needs a small fraction of this. + internal const int MaxPages = 10_000; + + /// Byte cap for one listing (64 MiB). + internal const long MaxListingBytes = 64L * 1024 * 1024; + /// Reads @tags or Program:X.@tags in full. /// The device's shared scheduler. /// The listing pseudo-tag name. @@ -80,15 +94,26 @@ internal static class NativeListingReader using var all = new MemoryStream(); uint start = 0; - while (true) + for (var page = 0; page < MaxPages; page++) { var request = SymbolObjectServices.GetInstanceAttributeList(program, start); - var reply = await scheduler.SendAsync(request, cancellationToken).ConfigureAwait(false); + // Exclusive: the reply can fill the connection, and the packer budgets request bytes only. + var reply = await scheduler.SendAsync(request, cancellationToken, exclusive: true).ConfigureAwait(false); var (entries, last, more) = SymbolObjectServices.ParseListingReply(reply); all.Write(entries); + if (all.Length > MaxListingBytes) + throw new AbTransportException(AbStatus.BadReply, + $"Symbol listing '{listingName}' exceeded {MaxListingBytes} bytes."); if (!more) return all.ToArray(); + // Instances ascend; a page that does not advance past its start, or a last id that would + // wrap, is a misbehaving reply and would otherwise loop forever against the controller. + if (last < start || last == uint.MaxValue) + throw new AbTransportException(AbStatus.BadReply, + $"Symbol listing '{listingName}' made no progress (start {start}, last {last})."); start = last + 1; } + throw new AbTransportException(AbStatus.BadReply, + $"Symbol listing '{listingName}' exceeded {MaxPages} pages."); } } @@ -105,10 +130,13 @@ internal static class NativeLease /// Upper bound on waiting for the lease release (and its Forward Close). private static readonly TimeSpan DisposeWait = TimeSpan.FromSeconds(2); - /// Releases, waiting at most 2 s; never throws. - public static void ReleaseBounded(Func release) + /// Releases, waiting at most 2 s; never throws, whatever the release delegate does. + public static void ReleaseBounded(Func release, ILogger logger) { try { release().AsTask().Wait(DisposeWait); } - catch (AggregateException) { /* a faulted close is the pool's to log */ } + catch (Exception ex) + { + logger.LogWarning(ex, "Releasing the native AbCip device lease failed."); + } } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTemplateReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTemplateReader.cs index 35847805..0a9d5bbe 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTemplateReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTemplateReader.cs @@ -1,3 +1,5 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Logix; @@ -17,7 +19,11 @@ internal sealed class NativeAbCipTemplateReader : IAbCipTemplateReader /// Bytes of the template definition header not counted in the Read Template length. private const int DefinitionHeaderBytes = 23; + /// Upper bound on a template's Read Template length; Logix templates are far smaller. + internal const long MaxTemplateBytes = 64 * 1024; + private readonly Func Release)> _acquire; + private readonly ILogger _logger; /// Production constructor: leases from . /// The connection pool (the factory passes ). @@ -32,9 +38,14 @@ internal sealed class NativeAbCipTemplateReader : IAbCipTemplateReader /// Test seam: supplies the scheduler, channel and release directly. /// Acquires the connection parts for a device. + /// Receives release failures; null discards them. internal NativeAbCipTemplateReader( - Func Release)> acquire) => + Func Release)> acquire, + ILogger? logger = null) + { _acquire = acquire; + _logger = logger ?? NullLogger.Instance; + } /// public async Task ReadAsync( @@ -53,7 +64,7 @@ internal sealed class NativeAbCipTemplateReader : IAbCipTemplateReader // The CIP-documented Read Template length: definition words * 4 - 23. var total = checked((long)attrs.DefinitionWords * 4 - DefinitionHeaderBytes); - if (total < 0) + if (total < 0 || total > MaxTemplateBytes) throw new AbTransportException(AbStatus.BadReply, $"Template {id} reports an impossible definition size of {attrs.DefinitionWords} words."); @@ -63,7 +74,7 @@ internal sealed class NativeAbCipTemplateReader : IAbCipTemplateReader { var chunk = (ushort)Math.Min(total - offset, channel.MaxRequestBytes - RequestOverhead); var reply = await scheduler - .SendAsync(TemplateObjectServices.ReadTemplate(id, (uint)offset, chunk), cancellationToken) + .SendAsync(TemplateObjectServices.ReadTemplate(id, (uint)offset, chunk), cancellationToken, exclusive: true) .ConfigureAwait(false); if (reply.GeneralStatus is not (0x00 or 0x06)) throw new AbTransportException(reply.Status, @@ -80,7 +91,7 @@ internal sealed class NativeAbCipTemplateReader : IAbCipTemplateReader } finally { - NativeLease.ReleaseBounded(release); + NativeLease.ReleaseBounded(release, _logger); } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTagEnumeratorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTagEnumeratorTests.cs index 87c802e5..c2f2a445 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTagEnumeratorTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTagEnumeratorTests.cs @@ -135,4 +135,78 @@ public sealed class NativeAbCipTagEnumeratorTests bytes.ShouldBe(Concat(Entries(ControllerEntries))); } + + private static FakeCipChannel RepeatingPartialPage(uint lastInstance) + { + var ch = new FakeCipChannel(); + // One entry (instance = lastInstance, name "A") replied with "more" every time. + var entry = new byte[22 + 1]; + BinaryPrimitives.WriteUInt32LittleEndian(entry, lastInstance); + BinaryPrimitives.WriteUInt16LittleEndian(entry.AsSpan(20), 1); + entry[22] = (byte)'A'; + ch.RespondPerService = _ => FakeCipChannel.Reply(0x55, 0x06, entry); + return ch; + } + + [Fact] + public async Task A_partial_page_that_never_advances_fails_fast_instead_of_looping() + { + // Instance 5 is repeated forever: page 2 starts at 6 but answers last = 5 (< start). + var ch = RepeatingPartialPage(5); + await using var sched = new RequestScheduler(ch, allowPacking: false); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + cts.CancelAfter(TimeSpan.FromSeconds(10)); + + var ex = await Should.ThrowAsync( + () => NativeListingReader.ReadAsync(sched, "@tags", cts.Token)); + + ex.Status.ShouldBe(AbStatus.BadReply); + ch.SentCount.ShouldBeLessThan(5); + } + + [Fact] + public async Task A_partial_page_ending_at_the_maximum_instance_is_rejected_rather_than_wrapping() + { + var ch = RepeatingPartialPage(uint.MaxValue); + await using var sched = new RequestScheduler(ch, allowPacking: false); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + cts.CancelAfter(TimeSpan.FromSeconds(10)); + + var ex = await Should.ThrowAsync( + () => NativeListingReader.ReadAsync(sched, "@tags", cts.Token)); + + ex.Status.ShouldBe(AbStatus.BadReply); + ch.SentCount.ShouldBe(1); + } + + [Fact] + public async Task Cancelling_mid_enumeration_releases_the_lease() + { + var ch = Serve(new() { [""] = Entries(ControllerEntries) }); + ch.Gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var sched = new RequestScheduler(ch, allowPacking: false); + var released = 0; + using var enumerator = new NativeAbCipTagEnumerator(_ => (sched, () => { released++; return ValueTask.CompletedTask; })); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + + var run = Task.Run(async () => + { + await foreach (var _ in enumerator.EnumerateAsync(Params, cts.Token)) { } + }, TestContext.Current.CancellationToken); + await ch.WaitForSendsAsync(1); + cts.Cancel(); + + await Should.ThrowAsync(() => run); + released.ShouldBe(1); + ch.Gate.TrySetResult(); + } + + [Fact] + public void A_throwing_release_delegate_never_escapes_ReleaseBounded() + { + NativeLease.ReleaseBounded(() => throw new InvalidOperationException("boom"), + Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance); + NativeLease.ReleaseBounded(() => ValueTask.FromException(new ObjectDisposedException("x")), + Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTemplateReaderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTemplateReaderTests.cs index e7463982..b68aab94 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTemplateReaderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTemplateReaderTests.cs @@ -128,4 +128,50 @@ public sealed class NativeAbCipTemplateReaderTests reader.ReadAsync(Params, 0x0FCE, TestContext.Current.CancellationToken)); released.ShouldBe(1); } + + [Fact] + public async Task A_template_read_queued_behind_a_small_read_on_a_packing_scheduler_goes_out_alone() + { + var captured = Convert.FromHexString(CipTemplateObjectWireCaptureTests.LookupMemberTemplate); + var ch = ServeTemplate(captured, 504); + ch.Gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var served = ch.RespondPerService; + ch.RespondPerService = r => r[0] == 0x0E ? FakeCipChannel.OkReply(r) : served(r); + await using var sched = new RequestScheduler(ch, allowPacking: true); + var ct = TestContext.Current.CancellationToken; + using var reader = new NativeAbCipTemplateReader(_ => (sched, ch, () => ValueTask.CompletedTask)); + + // A is in flight (gated); B queues behind it; the template read then queues behind B. + var small = new byte[] { 0x0E, 0x02, 0x20, 0x01, 0x24, 0x01 }; // Get Attribute Single + var a = sched.SendAsync(small, ct); + await ch.WaitForSendsAsync(1); + var b = sched.SendAsync(small, ct); + var read = reader.ReadAsync(Params, 3386, ct); + await Task.Delay(200, ct); + ch.Gate.SetResult(); + await Task.WhenAll(a, b); + await read; + + // Packing was live (the attribute query shared an MSP with B), yet no Read Template rode in one. + ch.Sent.ShouldContain(r => r[0] == 0x0A); + foreach (var msp in ch.Sent.Where(r => r[0] == 0x0A)) + FakeCipChannel.SplitMsp(msp).ShouldNotContain(e => e[0] == 0x4C); + ch.Sent.ShouldContain(r => r[0] == 0x4C); // the Read Template requests went out bare + } + + [Fact] + public async Task A_template_size_over_the_bound_is_rejected() + { + var captured = Convert.FromHexString(CipTemplateObjectWireCaptureTests.StringTemplate); + BinaryPrimitives.WriteUInt32LittleEndian(captured.AsSpan(2), 0x10000000); + var ch = ServeTemplate(captured, 504); + await using var sched = new RequestScheduler(ch, allowPacking: false); + using var reader = new NativeAbCipTemplateReader(_ => (sched, ch, () => ValueTask.CompletedTask)); + + var ex = await Should.ThrowAsync( + () => reader.ReadAsync(Params, 0x0FCE, TestContext.Current.CancellationToken)); + + ex.Status.ShouldBe(AbStatus.BadReply); + ch.Sent.ShouldNotContain(r => r[0] == 0x4C); + } }