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 8a2c7409..c1f84c3a 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 @@ -117,7 +117,7 @@ public sealed class RequestScheduler : IAsyncDisposable /// /// Queue one CIP request. The returned reply carries the service's own general status; a non-zero /// status is not an exception here. Faults with when the channel - /// or the MSP envelope fails; cancels when fires. + /// or the MSP envelope fails, or the scheduler is disposed; cancels only when fires. /// public Task SendAsync(byte[] request, CancellationToken ct) { @@ -154,9 +154,10 @@ public sealed class RequestScheduler : IAsyncDisposable return p.Tcs.Task; } - private List TakeBatch() + /// Moves the next batch from the queue into . Filling a caller-owned + /// list means a throw part-way through still leaves every taken waiter reachable for faulting. + private void TakeBatch(List batch) { - var batch = new List(); lock (_lock) { var sizes = new List(); @@ -178,7 +179,6 @@ public sealed class RequestScheduler : IAsyncDisposable _queue.RemoveFirst(); } } - return batch; } private async Task PumpAsync() @@ -192,13 +192,40 @@ public sealed class RequestScheduler : IAsyncDisposable // so a surplus signal (e.g. from a cancelled waiter) is just one empty pass. while (!_stop.IsCancellationRequested) { - var batch = TakeBatch(); - if (batch.Count == 0) break; - await ExecuteAsync(batch).ConfigureAwait(false); + var batch = new List(); + try + { + TakeBatch(batch); + if (batch.Count == 0) break; + await ExecuteAsync(batch).ConfigureAwait(false); + } + catch (Exception ex) + { + // Last resort: ExecuteAsync already handles every exchange failure, so reaching here + // means the scheduler itself broke (e.g. MaxRequestBytes threw). Fault everything + // pending rather than let the pump die and leave callers hanging, then keep pumping + // so later requests still get an answer. + FaultAll(batch, ex); + break; + } } } } + private void FaultAll(List batch, Exception ex) + { + var error = ex as AbTransportException + ?? new AbTransportException(AbStatus.BadConnection, $"Request scheduler failed: {ex.Message}", ex); + List queued; + lock (_lock) + { + queued = [.. _queue]; + _queue.Clear(); + } + foreach (var b in batch) b.Fail(error, this); + foreach (var q in queued) q.Fail(error, this); + } + private async Task ExecuteAsync(List batch) { try @@ -224,7 +251,9 @@ public sealed class RequestScheduler : IAsyncDisposable } catch (OperationCanceledException) when (_stop.IsCancellationRequested) { - foreach (var b in batch) b.Cancel(_stop.Token, this); + // The scheduler is being disposed: not the caller's cancellation, so not an OCE. + var error = DisposedError(); + foreach (var b in batch) b.Fail(error, this); } catch (AbTransportException ex) { @@ -237,6 +266,17 @@ public sealed class RequestScheduler : IAsyncDisposable } } + /// + /// The error a waiter sees when the scheduler (i.e. its connection) goes away under it. Deliberately + /// NOT an : upstream treats OCE as "the caller unsubscribed" + /// and swallows it, whereas this must surface as a communication failure. A waiter cancelled by its + /// own token still gets an OCE. + /// + private static AbTransportException DisposedError() => + new(AbStatus.BadConnection, "Scheduler disposed."); + + /// Stops the pump. In-flight and queued waiters fault with + /// (). public async ValueTask DisposeAsync() { lock (_lock) @@ -253,7 +293,8 @@ public sealed class RequestScheduler : IAsyncDisposable remaining = [.. _queue]; _queue.Clear(); } - foreach (var p in remaining) p.Cancel(_stop.Token, this); + var error = DisposedError(); + foreach (var p in remaining) p.Fail(error, this); _signal.Dispose(); _stop.Dispose(); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs index db02b97b..4c8504aa 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs @@ -19,6 +19,14 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann /// size, type, struct handle, element count, offset. private const int FragmentedWriteOverhead = 12; + /// Upper bound on one read's accumulated data, so a controller that keeps answering + /// Partial (0x06) cannot grow memory without limit. Far above any real Logix tag. + internal const int MaxReadBytes = 2 * 1024 * 1024; + + /// Chunk alignment when one element is wider than a frame (Logix lays structures out on + /// at most 8-byte boundaries). + private const int WideElementAlignment = 8; + private readonly RequestScheduler _scheduler = scheduler ?? throw new ArgumentNullException(nameof(scheduler)); private readonly ICipChannel _channel = channel ?? throw new ArgumentNullException(nameof(channel)); @@ -52,6 +60,9 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann $"Read fragment type 0x{fragmentType:X4} differs from the first reply's 0x{type:X4}."); data.AddRange(chunk.Span); lastChunk = chunk.Length; + if (data.Count > MaxReadBytes) + throw new AbTransportException(AbStatus.TooLarge, + $"Fragmented read exceeded {MaxReadBytes} bytes; the controller kept answering Partial."); } return (type, handle, data.ToArray()); @@ -60,12 +71,23 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann /// /// Write to the tag at . One Write Tag when the /// request fits ; otherwise Write Tag Fragmented chunks of - /// MaxRequestBytes - 12 - path.Length bytes aligned down to the element width. + /// MaxRequestBytes - 12 - path.Length bytes aligned down to the element width (or to 8 bytes + /// when a single element is wider than a frame). /// + /// A fragmented write is NOT atomic across fragments — the controller applies each + /// fragment as it arrives, so a reader can observe a partly-written value. libplctag behaves the + /// same way. + /// For an atomic type, is not exactly + /// × the type's width. public async Task WriteAsync(byte[] path, ushort type, ushort? handle, ushort elements, ReadOnlyMemory data, CancellationToken ct) { ArgumentNullException.ThrowIfNull(path); + var atomicWidth = LogixTypeCode.ByteWidth(type); + if (atomicWidth > 0 && data.Length != elements * atomicWidth) + throw new ArgumentException( + $"Write of {elements} element(s) of type 0x{type:X4} needs {elements * atomicWidth} bytes, got {data.Length}.", + nameof(data)); var max = _channel.MaxRequestBytes; var isStruct = type == LogixTypeCode.Struct; var singleSize = 2 + path.Length + 2 + (isStruct ? 2 : 0) + 2 + data.Length; @@ -95,9 +117,10 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann private static int AlignedChunkSize(int room, int width) { - // A structure element can be wider than one frame; then align to the 4-byte boundary Logix - // lays structures out on rather than refusing the write. - var alignment = width <= room ? width : 4; + // A structure element can be wider than one frame. Write Tag Fragmented's offset is a byte + // offset and the controller reassembles, so this is allowed: align to 8 bytes instead of + // refusing the write. + var alignment = width <= room ? width : WideElementAlignment; var chunk = room / alignment * alignment; if (chunk <= 0) throw new AbTransportException(AbStatus.TooLarge, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/RequestSchedulerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/RequestSchedulerTests.cs index c96f4a4b..5d40cfb5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/RequestSchedulerTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/RequestSchedulerTests.cs @@ -242,6 +242,35 @@ public sealed class RequestSchedulerTests ch.Sent[1][0].ShouldBe((byte)0x4C); // c went bare: the cancelled b was not packed with it } + [Fact] + public async Task A_scheduler_fault_outside_the_exchange_fails_every_pending_waiter_and_keeps_pumping() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 1)); + var b = s.SendAsync(Req(8, 2), CancellationToken.None); + var c = s.SendAsync(Req(8, 3), CancellationToken.None); + var d = s.SendAsync(Req(8, 4), CancellationToken.None); + var broken = new InvalidOperationException("channel torn"); + ch.MaxRequestBytesThrows = broken; // TakeBatch reads it when packing c behind b + ch.Gate!.SetResult(); + + (await a.WaitAsync(Timeout, Ct)).GeneralStatus.ShouldBe((byte)0); + var exB = await Should.ThrowAsync(() => b.WaitAsync(Timeout, Ct)); // already taken + var exC = await Should.ThrowAsync(() => c.WaitAsync(Timeout, Ct)); // still queued + var exD = await Should.ThrowAsync(() => d.WaitAsync(Timeout, Ct)); + exB.Status.ShouldBe(AbStatus.BadConnection); + exB.InnerException.ShouldBeSameAs(broken); + exC.ShouldBeSameAs(exB); + exD.ShouldBeSameAs(exB); + s.QueuedCount.ShouldBe(0); + ch.SentCount.ShouldBe(1); + + // The pump survived: a later request is answered, not left hanging. + ch.MaxRequestBytesThrows = null; + (await s.SendAsync(Req(8, 5), CancellationToken.None).WaitAsync(Timeout, Ct)).GeneralStatus.ShouldBe((byte)0); + } + [Fact] public async Task A_token_already_cancelled_is_never_queued() { @@ -340,17 +369,24 @@ public sealed class RequestSchedulerTests } [Fact] - public async Task Dispose_cancels_queued_and_in_flight_waiters() + public async Task Dispose_faults_queued_and_in_flight_waiters_as_a_communication_failure_not_a_cancellation() { var ch = new FakeCipChannel(); var s = new RequestScheduler(ch, allowPacking: true); - var a = await HoldFirstAsync(s, ch, Req(8, 1)); - var b = s.SendAsync(Req(8, 2), CancellationToken.None); + var a = await HoldFirstAsync(s, ch, Req(8, 1)); // in flight + using var own = new CancellationTokenSource(); + var b = s.SendAsync(Req(8, 2), own.Token); // queued, with a live token await s.DisposeAsync(); - await Should.ThrowAsync(() => a.WaitAsync(Timeout, Ct)); - await Should.ThrowAsync(() => b.WaitAsync(Timeout, Ct)); + // Not OperationCanceledException: upstream would swallow that as "unsubscribed". + var inFlight = await Should.ThrowAsync(() => a.WaitAsync(Timeout, Ct)); + inFlight.Status.ShouldBe(AbStatus.BadConnection); + var queued = await Should.ThrowAsync(() => b.WaitAsync(Timeout, Ct)); + queued.Status.ShouldBe(AbStatus.BadConnection); + a.IsCanceled.ShouldBeFalse(); + b.IsCanceled.ShouldBeFalse(); + s.LiveRegistrations.ShouldBe(0); ch.SentCount.ShouldBe(1); Should.Throw(() => s.SendAsync(Req(8), CancellationToken.None)); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipChannel.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipChannel.cs index e2b07a3f..7e0179ac 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipChannel.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipChannel.cs @@ -18,7 +18,16 @@ internal sealed class FakeCipChannel : ICipChannel Respond = DefaultRespond; } - public int MaxRequestBytes { get; set; } + private int _maxRequestBytes; + + public int MaxRequestBytes + { + get => MaxRequestBytesThrows is { } ex ? throw ex : _maxRequestBytes; + set => _maxRequestBytes = value; + } + + /// When set, reading throws it (simulates a broken channel). + public Exception? MaxRequestBytesThrows { get; set; } /// Every request, in send order. public List Sent { get; } = []; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Logix/LogixTagClientTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Logix/LogixTagClientTests.cs index ef2a1672..1faee861 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Logix/LogixTagClientTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Logix/LogixTagClientTests.cs @@ -200,6 +200,62 @@ public sealed class LogixTagClientTests } } + [Fact] + public async Task A_struct_element_wider_than_one_frame_is_written_in_8_byte_aligned_contiguous_chunks() + { + const int room = 45; // aligns down to 40 + var (ch, client, s) = Create(max: room + 12 + Path.Length); + await using var _ = s; + var data = Bytes(200, 9); // 2 elements of a 100-byte struct (> room) + + await client.WriteAsync(Path, LogixTypeCode.Struct, 0x0ABC, 2, data, Ct); + + var reassembled = new List(); + foreach (var r in ch.Sent) + { + r[0].ShouldBe((byte)0x53); + r.Length.ShouldBeLessThanOrEqualTo(ch.MaxRequestBytes); + var body = r.AsSpan(2 + Path.Length); + BinaryPrimitives.ReadUInt16LittleEndian(body).ShouldBe(LogixTypeCode.Struct); + BinaryPrimitives.ReadUInt16LittleEndian(body[2..]).ShouldBe((ushort)0x0ABC); // handle + BinaryPrimitives.ReadUInt16LittleEndian(body[4..]).ShouldBe((ushort)2); // total elements + BinaryPrimitives.ReadUInt32LittleEndian(body[6..]).ShouldBe((uint)reassembled.Count); // contiguous + var chunk = body[10..]; + (chunk.Length % 8).ShouldBe(0); + reassembled.AddRange(chunk.ToArray()); + } + ch.Sent.Count.ShouldBe(5); + reassembled.ShouldBe(data); + } + + [Fact] + public async Task A_read_that_never_stops_answering_partial_is_capped() + { + var (ch, client, s) = Create(); + await using var _ = s; + ch.RespondPerService = r => FakeCipChannel.Reply(r[0], 0x06, TypedData(LogixTypeCode.DInt, new byte[4000])); + + var ex = await Should.ThrowAsync(() => client.ReadAsync(Path, 1000, Ct)); + + ex.Status.ShouldBe(AbStatus.TooLarge); + ch.Sent.Count.ShouldBe(LogixTagClient.MaxReadBytes / 4000 + 1); // stops at the first chunk past the cap + } + + [Theory] + [InlineData(LogixTypeCode.DInt, 4, 15)] + [InlineData(LogixTypeCode.DInt, 4, 17)] + [InlineData(LogixTypeCode.LReal, 2, 8)] + public async Task An_atomic_write_whose_data_does_not_match_its_element_count_is_rejected_before_the_wire( + ushort type, int elements, int length) + { + var (ch, client, s) = Create(); + await using var _ = s; + + await Should.ThrowAsync( + () => client.WriteAsync(Path, type, null, (ushort)elements, Bytes(length, 0), Ct)); + ch.SentCount.ShouldBe(0); + } + [Fact] public async Task A_failed_write_throws_with_the_reply_status() {