diff --git a/docs/drivers/AbCip.md b/docs/drivers/AbCip.md index 6d7b7193..10d40b5e 100644 --- a/docs/drivers/AbCip.md +++ b/docs/drivers/AbCip.md @@ -157,6 +157,17 @@ The driver has two interchangeable wire transports, selected by one `DriverConfi - **One CIP connection and one request scheduler per device** (`DeviceSessionPool`), shared by reads, writes, browse and template reads. The Multiple Service Packet scheduler coalesces requests with one request in flight; large reads (listing pages, Read Template) are sent unpacked and exclusive. + **Packing budgets the expected reply as well as the request (#582):** each queued request declares an + upper bound on its reply (`expectedReplyBytes`; reads: elements x element width, STRING = 88 bytes per + element, BOOL counted as 4) and a batch is closed when the summed MSP reply (`6 + 2n + sum`) would exceed + the connection size. A read whose size cannot be bounded (structure / unknown authored type) is sent + alone; writes and status-only replies count their 4-byte header. Pinned by + `RequestSchedulerTests.Mixed_scalar_string_and_array_replies_are_packed_within_the_reply_budget`. + ⚠️ RESIDUAL: mixed STRING/array packing has not been exercised against a real controller (the real-press + gate predates the budget); the budget is unit-tested against the scheduler and fake channel only. +- **Fragmented-read integrity (#567):** `Read Tag Fragmented` replies are compared against the first reply; + a changed type code *or* structure handle mid-read fails the read with `BadReply` rather than stitching + two types into one value. - **Forward Open throttle:** at most one attempt in flight, at least 1 s between starts, bounded by the open budget. - `ConnectionSize` / `Timeout` / `AllowPacking` are honoured per pooled connection, and the **first** @@ -194,7 +205,9 @@ The driver has two interchangeable wire transports, selected by one `DriverConfi (2026-10-08) and nothing else. Results: [native-ab-transport-gate-results](../plans/2026-10-07-native-ab-transport-gate-results.md). - **libplctag cannot be deleted** until a bench controller allows write verification. -- Wide-struct fragmented writes (8-byte alignment) are unit-tested only, never proven on a controller (#567). +- ⚠️ RESIDUAL (#567): wide-struct **fragmented writes** (8-byte alignment) are unit-tested only. Production + presses are read-only and there is no bench controller, so this cannot be proven; it stays open until a + bench controller exists. - Native browse: listing pages are unpaced, and template reads outside an enumeration open a connection per call (#572). - Micro800, routed and `logixpccc` paths are covered by `ab_server` only. The Micro800 profile passes 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 f9102c21..c386600f 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,8 +19,21 @@ public sealed class RequestScheduler : IAsyncDisposable private const byte MultipleServiceCode = 0x0A; private const byte EmbeddedServiceError = 0x1E; - private sealed class Pending(byte[] request, bool exclusive) + /// Smallest possible CIP reply: service, reserved, general status, additional-status size. + public const int ReplyHeaderBytes = 4; + + /// The expectedReplyBytes default: the reply carries no data beyond its header (a + /// write, a status-only service). Every caller whose reply carries data must pass its real bound. + public const int StatusOnlyReply = ReplyHeaderBytes; + + /// Pass as expectedReplyBytes when the reply size cannot be bounded: the request is + /// then sent alone, never packed, because its reply could fill the whole connection. + public const int UnknownReply = -1; + + private sealed class Pending(byte[] request, bool exclusive, int expectedReplyBytes) { + // Private lock object: lock(this) would let any holder of the Pending reference contend. + private readonly object _gate = new(); private bool _completed; private CancellationTokenRegistration _registration; private bool _hasRegistration; @@ -29,13 +42,16 @@ public sealed class RequestScheduler : IAsyncDisposable /// True when this request must never share a Multiple Service Packet with another. public bool Exclusive { get; } = exclusive; + + /// Expected whole-reply size (header included), or . + public int ExpectedReplyBytes { get; } = expectedReplyBytes; public TaskCompletionSource Tcs { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); public LinkedListNode? Node { get; set; } /// Stores the waiter's registration; disposes it at once if the waiter already completed. public void SetRegistration(CancellationTokenRegistration registration, RequestScheduler owner) { - lock (this) + lock (_gate) { if (!_completed) { @@ -54,7 +70,7 @@ public sealed class RequestScheduler : IAsyncDisposable { CancellationTokenRegistration registration; bool hasRegistration; - lock (this) + lock (_gate) { if (_completed) return false; _completed = true; @@ -125,13 +141,19 @@ public sealed class RequestScheduler : IAsyncDisposable /// 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) + /// Packet: for requests whose reply can fill the whole connection. + /// Upper bound on this request's whole reply (header included) as + /// it would sit inside a Multiple Service Packet reply. The packer budgets this against the + /// connection size as well as the request bytes (#582), so a batch of small requests with large + /// replies (STRINGs, arrays) is split instead of overflowing. Defaults to ; + /// sends the request alone. + public Task SendAsync(byte[] request, CancellationToken ct, bool exclusive = false, + int expectedReplyBytes = StatusOnlyReply) { ArgumentNullException.ThrowIfNull(request); if (ct.IsCancellationRequested) return Task.FromCanceled(ct); - var p = new Pending(request, exclusive); + var p = new Pending(request, exclusive || expectedReplyBytes < 0, Math.Max(expectedReplyBytes, UnknownReply)); lock (_lock) { ObjectDisposedException.ThrowIf(_disposed, this); @@ -168,19 +190,27 @@ public sealed class RequestScheduler : IAsyncDisposable lock (_lock) { var sizes = new List(); + var replySizes = new List(); + var replyBudget = _channel.MaxRequestBytes; // connections are symmetric: one size bounds both directions while (_queue.First is { } n) { if (batch.Count > 0) { if (!_allowPacking || n.Value.Exclusive) break; sizes.Add(n.Value.Request.Length); - if (MultipleServicePacket.BuildSize(sizes) > _channel.MaxRequestBytes) break; + replySizes.Add(n.Value.ExpectedReplyBytes); + if (MultipleServicePacket.BuildSize(sizes) > _channel.MaxRequestBytes + || ReplyEnvelopeSize(replySizes) > replyBudget) + { + break; + } } else { // The first request always goes, even when it alone exceeds the limit: it is sent // bare, and callers that must fit (LogixTagClient) fragment before they get here. sizes.Add(n.Value.Request.Length); + replySizes.Add(n.Value.ExpectedReplyBytes); } batch.Add(n.Value); _queue.RemoveFirst(); @@ -189,6 +219,15 @@ public sealed class RequestScheduler : IAsyncDisposable } } + /// Expected size of the MSP reply for the given embedded reply sizes: reply header, count, + /// one offset per reply, then the replies themselves. + internal static int ReplyEnvelopeSize(IReadOnlyList replySizes) + { + var total = ReplyHeaderBytes + 2 + 2 * replySizes.Count; + foreach (var r in replySizes) total += r; + return total; + } + private async Task PumpAsync() { while (true) 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 4c8504aa..87f4da25 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 @@ -38,11 +38,18 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann /// /// Packing bounds only the request size, so a packed batch of large reads can come back /// with 0x06 per service; that is expected, and this loop recovers it. + /// The encoded symbolic path. + /// Element count to read. + /// Cancels this read. + /// Upper bound on one element's data bytes ( + /// for a STRING), or 0 when unknown. The scheduler budgets the expected reply against the connection + /// size when packing (#582); unknown sends the read alone. public async Task<(ushort TypeCode, ushort? StructHandle, byte[] Data)> ReadAsync( - byte[] path, ushort elements, CancellationToken ct) + byte[] path, ushort elements, CancellationToken ct, int elementBytes = 0) { ArgumentNullException.ThrowIfNull(path); - var reply = await _scheduler.SendAsync(LogixServices.ReadTagFragmented(path, elements, 0), ct).ConfigureAwait(false); + var reply = await _scheduler.SendAsync(LogixServices.ReadTagFragmented(path, elements, 0), ct, + expectedReplyBytes: ExpectedReadReplyBytes(elements, elementBytes, 0)).ConfigureAwait(false); var (type, handle, first) = LogixServices.ParseReadReply(reply); // throws on any status but 0x00/0x06 var data = new List(first.Length); data.AddRange(first.Span); @@ -53,11 +60,17 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann if (lastChunk == 0) throw new AbTransportException(AbStatus.BadReply, "Partial read reply carried no data; the fragment loop cannot progress."); reply = await _scheduler.SendAsync( - LogixServices.ReadTagFragmented(path, elements, (uint)data.Count), ct).ConfigureAwait(false); - var (fragmentType, _, chunk) = LogixServices.ParseReadReply(reply); + LogixServices.ReadTagFragmented(path, elements, (uint)data.Count), ct, + expectedReplyBytes: ExpectedReadReplyBytes(elements, elementBytes, data.Count)).ConfigureAwait(false); + var (fragmentType, fragmentHandle, chunk) = LogixServices.ParseReadReply(reply); if (fragmentType != type) throw new AbTransportException(AbStatus.BadReply, $"Read fragment type 0x{fragmentType:X4} differs from the first reply's 0x{type:X4}."); + // A mid-read change of the structure handle means the tag's type changed under us (a + // download / online edit); stitching the halves would yield a value of neither type. + if (fragmentHandle != handle) + throw new AbTransportException(AbStatus.BadReply, + $"Read fragment structure handle {FormatHandle(fragmentHandle)} differs from the first reply's {FormatHandle(handle)}; the tag changed mid-read."); data.AddRange(chunk.Span); lastChunk = chunk.Length; if (data.Count > MaxReadBytes) @@ -68,6 +81,21 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann return (type, handle, data.ToArray()); } + private static string FormatHandle(ushort? handle) => handle is { } h ? $"0x{h:X4}" : "(none)"; + + /// Whole-reply bound for a Read Tag Fragmented at bytes in: + /// reply header, type (+ struct handle), then the remaining data. Unknown element size, or an + /// unrepresentable total, is . + internal static int ExpectedReadReplyBytes(ushort elements, int elementBytes, int alreadyRead) + { + if (elementBytes <= 0) return RequestScheduler.UnknownReply; + var remaining = (long)elements * elementBytes - alreadyRead; + if (remaining < 0) remaining = 0; + // 4 header + 2 type + 2 struct handle (counted for every type: a safe over-estimate). + var total = RequestScheduler.ReplyHeaderBytes + 4 + remaining; + return total > int.MaxValue ? RequestScheduler.UnknownReply : (int)total; + } + /// /// Write to the tag at . One Write Tag when the /// request fits ; otherwise Write Tag Fragmented chunks of diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTypeCode.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTypeCode.cs index eeb9cfb9..6c25961b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTypeCode.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTypeCode.cs @@ -5,6 +5,9 @@ public static class LogixTypeCode public const ushort Bool = 0xC1, SInt = 0xC2, Int = 0xC3, DInt = 0xC4, LInt = 0xC5, USInt = 0xC6, UInt = 0xC7, UDInt = 0xC8, ULInt = 0xC9, Real = 0xCA, LReal = 0xCB, BoolArray = 0xD3, Struct = 0x02A0; + /// Bytes of one Logix STRING element on the wire: 4-byte length + 82 characters + 2 pad. + public const int StringElementBytes = 88; + /// Byte width of an atomic type, or -1 when the code is not atomic (e.g. ). public static int ByteWidth(ushort code) => code switch { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagRuntime.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagRuntime.cs index 60910133..1fa3b11d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagRuntime.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagRuntime.cs @@ -122,7 +122,7 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime var path = Path(); try { - var (type, handle, data) = await client.ReadAsync(path,_elements, cancellationToken).ConfigureAwait(false); + var (type, handle, data) = await client.ReadAsync(path, _elements, cancellationToken, ExpectedElementBytes(_p.AuthoredType)).ConfigureAwait(false); _data = data; _typeCode = type; _structHandle = handle; @@ -136,6 +136,21 @@ internal sealed class NativeAbCipTagRuntime : IAbCipTagRuntime } } + /// + /// Upper bound on one element's wire bytes for the scheduler's reply budget (#582); 0 when the + /// authored type is unknown or a structure, which sends the read alone rather than guess. + /// BOOL counts 4 (a packed BOOL array reads back as DWORDs). + /// + internal static int ExpectedElementBytes(AbCipDataType? type) => type switch + { + AbCipDataType.Bool or AbCipDataType.DInt or AbCipDataType.UDInt or AbCipDataType.Real or AbCipDataType.Dt => 4, + AbCipDataType.SInt or AbCipDataType.USInt => 1, + AbCipDataType.Int or AbCipDataType.UInt => 2, + AbCipDataType.LInt or AbCipDataType.ULInt or AbCipDataType.LReal => 8, + AbCipDataType.String => LogixTypeCode.StringElementBytes, + _ => 0, + }; + /// public async Task WriteAsync(CancellationToken cancellationToken) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/NativeAbLegacyTagRuntime.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/NativeAbLegacyTagRuntime.cs index b11d3a82..576efc10 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/NativeAbLegacyTagRuntime.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/NativeAbLegacyTagRuntime.cs @@ -175,11 +175,15 @@ internal sealed class NativeAbLegacyTagRuntime : IAbLegacyTagRuntime return (Math.Max(1, PcccServices.MaxDataBytes / _elementSize), address); } - private async Task> ExchangeAsync(Func pccc, CancellationToken ct) + /// Bytes of an Execute PCCC reply beyond its data: CIP header, requestor id, PCCC command / status / TNS + /// (rounded up) — the scheduler budgets PcccReplyOverhead + data against the connection (#582). + private const int PcccReplyOverhead = 32; + + private async Task> ExchangeAsync(Func pccc, CancellationToken ct, int replyDataBytes = 0) { var tns = unchecked((ushort)Interlocked.Increment(ref s_tns)); var request = PcccServices.ExecutePccc(_vendor, _serial, pccc(tns)); - var reply = await _scheduler.SendAsync(request, ct).ConfigureAwait(false); + var reply = await _scheduler.SendAsync(request, ct, expectedReplyBytes: PcccReplyOverhead + replyDataBytes).ConfigureAwait(false); return PcccServices.ParseReply(reply, tns, _requestorId); } @@ -200,7 +204,7 @@ internal sealed class NativeAbLegacyTagRuntime : IAbLegacyTagRuntime tns => plc5 ? PcccServices.Plc5RangeRead(tns, at, (ushort)(bytes / 2), bytes) : PcccServices.SlcTypedRead(tns, at, bytes), - ct).ConfigureAwait(false); + ct, bytes).ConfigureAwait(false); if (data.Length < bytes) throw new AbTransportException(AbStatus.BadReply, $"PCCC read returned {data.Length} of {bytes} bytes."); if (data.Length > bytes) 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 8518f296..ea137c48 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 @@ -437,4 +437,48 @@ public sealed class RequestSchedulerTests ch.Sent[1][^1].ShouldBe((byte)2); FakeCipChannel.SplitMsp(ch.Sent[2]).Select(e => e[^1]).ShouldBe(new byte[] { 3, 4 }); } + + [Fact] + public async Task Mixed_scalar_string_and_array_replies_are_packed_within_the_reply_budget() + { + // Every request is 8 bytes, so the REQUEST budget would pack all of them into one MSP; only the + // expected-reply budget (#582) splits the batch. Reply envelope = 6 + 2n + sum(replies) <= 504. + const int scalar = 8, str = 96, array = 200; + var ch = new FakeCipChannel(maxRequestBytes: 504); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 1)); + + var sizes = new[] { scalar, str, str, str, str, array, scalar, RequestScheduler.UnknownReply }; + var tasks = sizes.Select((size, i) => s.SendAsync(Req(8, (byte)(10 + i)), Ct, expectedReplyBytes: size)).ToArray(); + ch.Gate!.SetResult(); + await Task.WhenAll(tasks.Append(a)).WaitAsync(Timeout, Ct); + + // [sc, S, S, S, S] = 6+10+392 = 408; the array (+200) would make 608 > 504, so it starts batch 2 + // with the trailing scalar; the unknown-size read goes alone. + ch.Sent.Count.ShouldBe(4); + FakeCipChannel.SplitMsp(ch.Sent[1]).Select(e => e[^1]).ShouldBe(new byte[] { 10, 11, 12, 13, 14 }); + FakeCipChannel.SplitMsp(ch.Sent[2]).Select(e => e[^1]).ShouldBe(new byte[] { 15, 16 }); + IsMsp(ch.Sent[3]).ShouldBeFalse(); + ch.Sent[3][^1].ShouldBe((byte)17); + ReplyEnvelopeFor(scalar, str, str, str, str).ShouldBeLessThanOrEqualTo(504); + ReplyEnvelopeFor(scalar, str, str, str, str, array).ShouldBeGreaterThan(504); + } + + [Fact] + public async Task Status_only_replies_still_pack_to_the_request_limit() + { + // Positive control for the reply budget: the default (status-only) reply never splits a batch + // the request budget allows, so existing write / probe coalescing is unchanged. + var ch = new FakeCipChannel(maxRequestBytes: 504); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 1)); + var rest = Enumerable.Range(0, 20).Select(i => s.SendAsync(Req(8, (byte)(2 + i)), Ct)).ToArray(); + ch.Gate!.SetResult(); + await Task.WhenAll(rest.Append(a)).WaitAsync(Timeout, Ct); + + ch.Sent.Count.ShouldBe(2); + FakeCipChannel.SplitMsp(ch.Sent[1]).Count.ShouldBe(20); + } + + private static int ReplyEnvelopeFor(params int[] replies) => RequestScheduler.ReplyEnvelopeSize(replies); } 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 1faee861..d54f270c 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 @@ -279,4 +279,29 @@ public sealed class LogixTagClientTests () => client.WriteAsync(Path, LogixTypeCode.DInt, null, 40, Bytes(160, 0), Ct)); ch.Sent.Count.ShouldBe(1); } + + [Fact] + public async Task A_struct_handle_that_changes_between_fragments_fails_the_read() + { + var (ch, client, s) = Create(); + await using var _ = s; + ch.RespondPerService = r => ReadOffset(r) == 0 + ? FakeCipChannel.Reply(r[0], 0x06, TypedData(LogixTypeCode.Struct, Bytes(8, 0), handle: 0x0FCE)) + : FakeCipChannel.Reply(r[0], 0x00, TypedData(LogixTypeCode.Struct, Bytes(4, 8), handle: 0x1234)); + + var ex = await Should.ThrowAsync(() => client.ReadAsync(Path, 1, Ct)); + + ex.Status.ShouldBe(AbStatus.BadReply); + ex.Message.ShouldContain("0x0FCE"); + ex.Message.ShouldContain("0x1234"); + } + + [Fact] + public void Expected_read_reply_bytes_budgets_elements_times_width_and_is_unknown_without_a_width() + { + // header 4 + type/handle 4 + 3 STRING elements. + LogixTagClient.ExpectedReadReplyBytes(3, LogixTypeCode.StringElementBytes, 0).ShouldBe(4 + 4 + 3 * 88); + LogixTagClient.ExpectedReadReplyBytes(3, LogixTypeCode.StringElementBytes, 100).ShouldBe(4 + 4 + 3 * 88 - 100); + LogixTagClient.ExpectedReadReplyBytes(3, 0, 0).ShouldBe(RequestScheduler.UnknownReply); + } }