From 1c7fd5d5dec56a432e1dfb19f8f744884b6ded09 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 7 Oct 2026 16:15:05 -0400 Subject: [PATCH] feat(ab-transport): MSP-coalescing request scheduler and fragmented Logix read/write --- .../Cip/RequestScheduler.cs | 260 +++++++++++++ .../Logix/LogixTagClient.cs | 113 ++++++ .../Cip/RequestSchedulerTests.cs | 357 ++++++++++++++++++ .../Fakes/FakeCipChannel.cs | 119 ++++++ .../Logix/LogixTagClientTests.cs | 226 +++++++++++ 5 files changed, 1075 insertions(+) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/RequestScheduler.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/RequestSchedulerTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipChannel.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Logix/LogixTagClientTests.cs 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 new file mode 100644 index 00000000..8a2c7409 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/RequestScheduler.cs @@ -0,0 +1,260 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; + +/// +/// Per-connection CIP request scheduler. Keeps exactly one request in flight on its +/// ; requests that arrive while one is in flight queue, and when the channel +/// frees the scheduler packs as many of them (FIFO) as fit in +/// into one Multiple Service Packet. Each waiter receives its own embedded . +/// +/// +/// One scheduler per device connection, never per tag (review finding B1). The channel is always +/// called with the scheduler's own stop token, never a waiter's, so one caller's cancellation can +/// never abort or fault a shared exchange (B2). A waiter cancelled while queued is removed and never +/// sent; one cancelled while its batch is in flight is abandoned and the batch's other waiters still +/// complete. There is deliberately no batching delay window: packing falls out of the single pump +/// being busy on the channel while requests accumulate. +/// +public sealed class RequestScheduler : IAsyncDisposable +{ + private const byte MultipleServiceCode = 0x0A; + private const byte EmbeddedServiceError = 0x1E; + + private sealed class Pending(byte[] request) + { + private bool _completed; + private CancellationTokenRegistration _registration; + private bool _hasRegistration; + + public byte[] Request { get; } = request; + 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) + { + if (!_completed) + { + _registration = registration; + _hasRegistration = true; + return; + } + } + registration.Dispose(); + Interlocked.Decrement(ref owner._liveRegistrations); + } + + /// Marks the waiter complete exactly once and releases its registration (finding 10). + /// Returns false if it was already completed. + private bool Finish(RequestScheduler owner) + { + CancellationTokenRegistration registration; + bool hasRegistration; + lock (this) + { + if (_completed) return false; + _completed = true; + registration = _registration; + hasRegistration = _hasRegistration; + _registration = default; + _hasRegistration = false; + } + if (hasRegistration) + { + // Safe from inside the registration's own callback: Dispose does not wait on the + // callback that is currently executing on this thread. + registration.Dispose(); + Interlocked.Decrement(ref owner._liveRegistrations); + } + return true; + } + + public void Succeed(CipReply reply, RequestScheduler owner) + { + if (Finish(owner)) Tcs.TrySetResult(reply); + } + + public void Fail(Exception ex, RequestScheduler owner) + { + if (Finish(owner)) Tcs.TrySetException(ex); + } + + public void Cancel(CancellationToken ct, RequestScheduler owner) + { + if (Finish(owner)) Tcs.TrySetCanceled(ct); + } + } + + private readonly ICipChannel _channel; + private readonly bool _allowPacking; + private readonly LinkedList _queue = new(); + private readonly object _lock = new(); + private readonly SemaphoreSlim _signal = new(0); + private readonly CancellationTokenSource _stop = new(); + private readonly Task _pump; + private bool _disposed; + private int _liveRegistrations; + private int _cancellationCallbacksRun; + + public RequestScheduler(ICipChannel channel, bool allowPacking) + { + _channel = channel ?? throw new ArgumentNullException(nameof(channel)); + _allowPacking = allowPacking; + _pump = Task.Run(PumpAsync); + } + + /// Test hook: waiter cancellation registrations that have not yet been disposed. + internal int LiveRegistrations => Volatile.Read(ref _liveRegistrations); + + /// Test hook: waiter cancellation callbacks that have actually run. A disposed + /// registration never runs, so this stays flat when a long-lived token is cancelled later. + internal int CancellationCallbacksRun => Volatile.Read(ref _cancellationCallbacksRun); + + /// Test hook: requests still queued (not yet taken into a batch). + internal int QueuedCount { get { lock (_lock) return _queue.Count; } } + + /// + /// 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. + /// + public Task SendAsync(byte[] request, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(request); + if (ct.IsCancellationRequested) return Task.FromCanceled(ct); + + var p = new Pending(request); + lock (_lock) + { + ObjectDisposedException.ThrowIf(_disposed, this); + p.Node = _queue.AddLast(p); + } + + if (ct.CanBeCanceled) + { + Interlocked.Increment(ref _liveRegistrations); + var registration = ct.Register(static state => + { + var (self, pending, token) = ((RequestScheduler, Pending, CancellationToken))state!; + Interlocked.Increment(ref self._cancellationCallbacksRun); + lock (self._lock) + { + if (pending.Node?.List is not null) self._queue.Remove(pending.Node); + } + pending.Cancel(token, self); + }, (this, p, ct)); + p.SetRegistration(registration, this); + } + + lock (_lock) + { + if (!_disposed) _signal.Release(); + } + return p.Tcs.Task; + } + + private List TakeBatch() + { + var batch = new List(); + lock (_lock) + { + var sizes = new List(); + while (_queue.First is { } n) + { + if (batch.Count > 0) + { + if (!_allowPacking) break; + sizes.Add(n.Value.Request.Length); + if (MultipleServicePacket.BuildSize(sizes) > _channel.MaxRequestBytes) 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); + } + batch.Add(n.Value); + _queue.RemoveFirst(); + } + } + return batch; + } + + private async Task PumpAsync() + { + while (true) + { + try { await _signal.WaitAsync(_stop.Token).ConfigureAwait(false); } + catch (OperationCanceledException) { return; } + + // A signal only means "there may be work"; drain the queue fully before waiting again, + // 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); + } + } + } + + private async Task ExecuteAsync(List batch) + { + try + { + if (batch.Count == 1) + { + // Never a waiter's token (B2). + var raw = await _channel.SendAsync(batch[0].Request, _stop.Token).ConfigureAwait(false); + batch[0].Succeed(CipReply.Parse(raw), this); + return; + } + + var requests = new List(batch.Count); + foreach (var b in batch) requests.Add(b.Request); + var envelopeRaw = await _channel.SendAsync(MultipleServicePacket.Build(requests), _stop.Token).ConfigureAwait(false); + var envelope = CipReply.Parse(envelopeRaw); + if (envelope.Service != MultipleServiceCode) + throw new AbTransportException(AbStatus.BadReply, $"MSP reply carries service 0x{envelope.Service:X2}."); + if (envelope.GeneralStatus is not (0x00 or EmbeddedServiceError)) + throw new AbTransportException(envelope.Status, $"Multiple Service Packet failed: general 0x{envelope.GeneralStatus:X2}."); + var replies = MultipleServicePacket.Parse(envelope, batch.Count); + for (var i = 0; i < batch.Count; i++) batch[i].Succeed(replies[i], this); + } + catch (OperationCanceledException) when (_stop.IsCancellationRequested) + { + foreach (var b in batch) b.Cancel(_stop.Token, this); + } + catch (AbTransportException ex) + { + foreach (var b in batch) b.Fail(ex, this); + } + catch (Exception ex) + { + var wrapped = new AbTransportException(AbStatus.BadConnection, $"CIP exchange failed: {ex.Message}", ex); + foreach (var b in batch) b.Fail(wrapped, this); + } + } + + public async ValueTask DisposeAsync() + { + lock (_lock) + { + if (_disposed) return; + _disposed = true; + } + _stop.Cancel(); + try { await _pump.ConfigureAwait(false); } catch { /* stopping */ } + + List remaining; + lock (_lock) + { + remaining = [.. _queue]; + _queue.Clear(); + } + foreach (var p in remaining) p.Cancel(_stop.Token, 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 new file mode 100644 index 00000000..db02b97b --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs @@ -0,0 +1,113 @@ +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Logix; + +/// +/// Logix tag read/write over a connection's , including the fragment +/// loops for values that do not fit in one frame. +/// +/// +/// Reads are issued as Read Tag Fragmented (0x52) starting at offset 0 — exactly what libplctag puts +/// on the wire (golden captures, Task 4) — so the native and libplctag frames stay byte-comparable, +/// and the continuation requests are the same service at a non-zero offset. +/// +public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel channel) +{ + private const byte StatusSuccess = 0x00; + private const byte StatusPartial = 0x06; + + /// Fixed bytes of a Write Tag Fragmented request besides path and data: service, path + /// size, type, struct handle, element count, offset. + private const int FragmentedWriteOverhead = 12; + + private readonly RequestScheduler _scheduler = scheduler ?? throw new ArgumentNullException(nameof(scheduler)); + private readonly ICipChannel _channel = channel ?? throw new ArgumentNullException(nameof(channel)); + + /// + /// Read elements of the tag at the encoded . + /// A Partial (0x06) reply keeps its data and the read continues at offset = bytes received so + /// far until a reply reports success. The type code and struct handle come from the first + /// reply; each fragment reply's repeated type header is stripped before its data is appended. + /// + /// 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. + public async Task<(ushort TypeCode, ushort? StructHandle, byte[] Data)> ReadAsync( + byte[] path, ushort elements, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(path); + var reply = await _scheduler.SendAsync(LogixServices.ReadTagFragmented(path, elements, 0), ct).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); + var lastChunk = first.Length; + + while (reply.GeneralStatus == StatusPartial) + { + 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); + if (fragmentType != type) + throw new AbTransportException(AbStatus.BadReply, + $"Read fragment type 0x{fragmentType:X4} differs from the first reply's 0x{type:X4}."); + data.AddRange(chunk.Span); + lastChunk = chunk.Length; + } + + return (type, handle, data.ToArray()); + } + + /// + /// 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. + /// + public async Task WriteAsync(byte[] path, ushort type, ushort? handle, ushort elements, + ReadOnlyMemory data, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(path); + var max = _channel.MaxRequestBytes; + var isStruct = type == LogixTypeCode.Struct; + var singleSize = 2 + path.Length + 2 + (isStruct ? 2 : 0) + 2 + data.Length; + if (singleSize <= max) + { + var reply = await _scheduler.SendAsync(LogixServices.WriteTag(path, type, handle, elements, data.Span), ct).ConfigureAwait(false); + EnsureSuccess(reply); + return; + } + + var chunkSize = AlignedChunkSize(max - FragmentedWriteOverhead - path.Length, ElementWidth(type, elements, data.Length)); + for (var offset = 0; offset < data.Length; offset += chunkSize) + { + var length = Math.Min(chunkSize, data.Length - offset); + var request = LogixServices.WriteTagFragmented(path, type, handle, elements, (uint)offset, data.Span.Slice(offset, length)); + var reply = await _scheduler.SendAsync(request, ct).ConfigureAwait(false); + EnsureSuccess(reply); + } + } + + private static int ElementWidth(ushort type, ushort elements, int dataLength) + { + var width = LogixTypeCode.ByteWidth(type); + if (width > 0) return width; + return elements > 0 ? Math.Max(1, dataLength / elements) : 1; + } + + 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; + var chunk = room / alignment * alignment; + if (chunk <= 0) + throw new AbTransportException(AbStatus.TooLarge, + $"No room for write data: MaxRequestBytes leaves {room} byte(s) after the path."); + return chunk; + } + + private static void EnsureSuccess(CipReply reply) + { + if (reply.GeneralStatus != StatusSuccess) + throw new AbTransportException(reply.Status, $"Write failed: general 0x{reply.GeneralStatus:X2}."); + } +} 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 new file mode 100644 index 00000000..c96f4a4b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/RequestSchedulerTests.cs @@ -0,0 +1,357 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Cip; + +/// +/// Deterministic scheduler tests: requests are held in flight with +/// (or a per-call hold) and the queue is observed through test hooks, never with sleeps. +/// +public sealed class RequestSchedulerTests +{ + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10); + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + /// A request of exactly bytes, service 0x4C, tagged by . + private static byte[] Req(int size, byte tag = 0) + { + var r = new byte[size]; + r[0] = 0x4C; + if (size > 1) r[^1] = tag; + return r; + } + + /// Puts in flight behind the channel gate and returns its task. + private static async Task> HoldFirstAsync(RequestScheduler s, FakeCipChannel ch, byte[] first) + { + var expected = ch.SentCount + 1; + ch.Gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var t = s.SendAsync(first, CancellationToken.None); + await ch.WaitForSendsAsync(expected); + return t; + } + + [Fact] + public async Task Requests_queued_behind_an_inflight_one_are_packed_into_one_msp() + { + 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); + s.QueuedCount.ShouldBe(3); + + ch.Gate!.SetResult(); + await Task.WhenAll(a, b, c, d).WaitAsync(Timeout, Ct); + + ch.Sent.Count.ShouldBe(2); + ch.Sent[0].ShouldBe(Req(8, 1)); + ch.Sent[1][0].ShouldBe((byte)0x0A); + FakeCipChannel.SplitMsp(ch.Sent[1]).ShouldBe([Req(8, 2), Req(8, 3), Req(8, 4)]); // FIFO + foreach (var t in new[] { a, b, c, d }) (await t).GeneralStatus.ShouldBe((byte)0); + } + + [Fact] + public async Task Msp_exactly_at_limit_packs() + { + // MSP of two 10-byte requests = 6 + 2 + 2*2 + 20 = 32. + MultipleServicePacket.BuildSize([10, 10]).ShouldBe(32); + var ch = new FakeCipChannel(maxRequestBytes: 32); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(4)); + var x = s.SendAsync(Req(10, 1), CancellationToken.None); + var y = s.SendAsync(Req(10, 2), CancellationToken.None); + ch.Gate!.SetResult(); + await Task.WhenAll(a, x, y).WaitAsync(Timeout, Ct); + + ch.Sent.Count.ShouldBe(2); + ch.Sent[1].Length.ShouldBe(32); + FakeCipChannel.SplitMsp(ch.Sent[1]).ShouldBe([Req(10, 1), Req(10, 2)]); + } + + [Fact] + public async Task Msp_one_byte_over_limit_splits() + { + var ch = new FakeCipChannel(maxRequestBytes: 31); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(4)); + var x = s.SendAsync(Req(10, 1), CancellationToken.None); + var y = s.SendAsync(Req(10, 2), CancellationToken.None); + ch.Gate!.SetResult(); + await Task.WhenAll(a, x, y).WaitAsync(Timeout, Ct); + + ch.Sent.Count.ShouldBe(3); + ch.Sent[1].ShouldBe(Req(10, 1)); + ch.Sent[2].ShouldBe(Req(10, 2)); + } + + [Fact] + public async Task A_request_that_does_not_fit_never_lets_a_later_one_overtake_it() + { + var ch = new FakeCipChannel(maxRequestBytes: 100); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(4, 1)); + var b = s.SendAsync(Req(10, 2), CancellationToken.None); + var big = s.SendAsync(Req(400, 3), CancellationToken.None); + var d = s.SendAsync(Req(10, 4), CancellationToken.None); + ch.Gate!.SetResult(); + await Task.WhenAll(a, b, big, d).WaitAsync(Timeout, Ct); + + // b + big does not fit, so b goes bare; big exceeds the limit alone and goes bare; then d. + ch.Sent.Select(r => r[^1]).ShouldBe(new byte[] { 1, 2, 3, 4 }); + ch.Sent.ShouldAllBe(r => r[0] == 0x4C); + } + + [Fact] + public async Task A_single_request_larger_than_the_limit_is_sent_bare() + { + var ch = new FakeCipChannel(maxRequestBytes: 20); + await using var s = new RequestScheduler(ch, allowPacking: true); + var reply = await s.SendAsync(Req(30, 9), CancellationToken.None).WaitAsync(Timeout, Ct); + + reply.GeneralStatus.ShouldBe((byte)0); + ch.Sent.Count.ShouldBe(1); + ch.Sent[0].ShouldBe(Req(30, 9)); + } + + [Fact] + public async Task Packing_disabled_sends_each_bare() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: false); + 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); + ch.Gate!.SetResult(); + await Task.WhenAll(a, b, c).WaitAsync(Timeout, Ct); + + ch.Sent.Count.ShouldBe(3); + ch.Sent.Select(r => r[^1]).ShouldBe(new byte[] { 1, 2, 3 }); + ch.Sent.ShouldAllBe(r => r[0] == 0x4C); + } + + [Fact] + public async Task Per_service_error_fails_only_its_waiter() + { + var ch = new FakeCipChannel + { + RespondPerService = r => r[^1] == 2 + ? FakeCipChannel.Reply(r[0], 0x04, [], 0x0000) + : FakeCipChannel.Reply(r[0], 0x00, [0xAB]), + }; + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 0)); + var ok = s.SendAsync(Req(8, 1), CancellationToken.None); + var bad = s.SendAsync(Req(8, 2), CancellationToken.None); + ch.Gate!.SetResult(); + await Task.WhenAll(a, ok, bad).WaitAsync(Timeout, Ct); + + ch.Sent.Count.ShouldBe(2); + var okReply = await ok; + okReply.GeneralStatus.ShouldBe((byte)0x00); + okReply.Status.ShouldBe(AbStatus.Ok); + okReply.Data.ToArray().ShouldBe(new byte[] { 0xAB }); + var badReply = await bad; + badReply.GeneralStatus.ShouldBe((byte)0x04); + badReply.ExtendedStatus.ShouldBe(new ushort[] { 0x0000 }); + } + + [Fact] + public async Task Msp_envelope_failure_faults_every_waiter_in_the_batch_with_the_same_exception() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 0)); + ch.Respond = r => Task.FromResult(r[0] == 0x0A + ? FakeCipChannel.Reply(0x0A, 0x08, []) // service not supported on the envelope + : FakeCipChannel.OkReply(r)); + var x = s.SendAsync(Req(8, 1), CancellationToken.None); + var y = s.SendAsync(Req(8, 2), CancellationToken.None); + ch.Gate!.SetResult(); + + (await a.WaitAsync(Timeout, Ct)).GeneralStatus.ShouldBe((byte)0); + var ex1 = await Should.ThrowAsync(() => x.WaitAsync(Timeout, Ct)); + var ex2 = await Should.ThrowAsync(() => y.WaitAsync(Timeout, Ct)); + ex1.ShouldBeSameAs(ex2); + ex1.Status.ShouldBe(AbStatusCodes.FromCip(0x08, [])); + } + + [Fact] + public async Task Channel_failure_faults_every_waiter_in_the_batch() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 0)); + var boom = new AbTransportException(AbStatus.Timeout, "reply timeout"); + ch.Respond = _ => Task.FromException(boom); + var x = s.SendAsync(Req(8, 1), CancellationToken.None); + var y = s.SendAsync(Req(8, 2), CancellationToken.None); + ch.Gate!.SetResult(); + + // a's own exchange also failed (the responder was swapped before release). + (await Should.ThrowAsync(() => a.WaitAsync(Timeout, Ct))).ShouldBeSameAs(boom); + (await Should.ThrowAsync(() => x.WaitAsync(Timeout, Ct))).ShouldBeSameAs(boom); + (await Should.ThrowAsync(() => y.WaitAsync(Timeout, Ct))).ShouldBeSameAs(boom); + + // The scheduler survives a failed exchange. + ch.Respond = ch.DefaultRespond; + (await s.SendAsync(Req(8, 3), CancellationToken.None).WaitAsync(Timeout, Ct)).GeneralStatus.ShouldBe((byte)0); + } + + [Fact] + public async Task A_non_transport_channel_exception_is_wrapped_and_shared_by_the_batch() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 0)); + ch.Respond = _ => Task.FromException(new IOException("socket reset")); + var x = s.SendAsync(Req(8, 1), CancellationToken.None); + var y = s.SendAsync(Req(8, 2), CancellationToken.None); + ch.Gate!.SetResult(); + + await Should.ThrowAsync(() => a.WaitAsync(Timeout, Ct)); + var ex1 = await Should.ThrowAsync(() => x.WaitAsync(Timeout, Ct)); + var ex2 = await Should.ThrowAsync(() => y.WaitAsync(Timeout, Ct)); + ex1.ShouldBeSameAs(ex2); + ex1.InnerException.ShouldBeOfType(); + } + + [Fact] + public async Task Queued_waiter_cancelled_before_send_is_never_sent() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 1)); + using var cts = new CancellationTokenSource(); + var b = s.SendAsync(Req(8, 2), cts.Token); + s.QueuedCount.ShouldBe(1); + + await cts.CancelAsync(); + await Should.ThrowAsync(() => b.WaitAsync(Timeout, Ct)); + b.IsCanceled.ShouldBeTrue(); + s.QueuedCount.ShouldBe(0); + + var c = s.SendAsync(Req(8, 3), CancellationToken.None); + ch.Gate!.SetResult(); + await Task.WhenAll(a, c).WaitAsync(Timeout, Ct); + + ch.Sent.Count.ShouldBe(2); + ch.Sent.Select(r => r[^1]).ShouldBe(new byte[] { 1, 3 }); + ch.Sent[1][0].ShouldBe((byte)0x4C); // c went bare: the cancelled b was not packed with it + } + + [Fact] + public async Task A_token_already_cancelled_is_never_queued() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: true); + var t = s.SendAsync(Req(8), new CancellationToken(canceled: true)); + t.IsCanceled.ShouldBeTrue(); + s.QueuedCount.ShouldBe(0); + s.LiveRegistrations.ShouldBe(0); + ch.SentCount.ShouldBe(0); + } + + [Fact] + public async Task A_waiter_cancelled_while_its_batch_is_in_flight_is_abandoned_and_the_others_complete() + { + var ch = new FakeCipChannel(); + var mspHold = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + ch.Respond = async r => + { + if (r[0] == 0x0A) await mspHold.Task; + return await ch.DefaultRespond(r); + }; + await using var s = new RequestScheduler(ch, allowPacking: true); + var a = await HoldFirstAsync(s, ch, Req(8, 1)); + using var cts = new CancellationTokenSource(); + var b = s.SendAsync(Req(8, 2), cts.Token); + var c = s.SendAsync(Req(8, 3), CancellationToken.None); + ch.Gate!.SetResult(); + await ch.WaitForSendsAsync(2); // the MSP carrying b and c is now in flight + s.QueuedCount.ShouldBe(0); + + await cts.CancelAsync(); + await Should.ThrowAsync(() => b.WaitAsync(Timeout, Ct)); + c.IsCompleted.ShouldBeFalse(); + ch.Tokens[1].IsCancellationRequested.ShouldBeFalse(); // the shared exchange was not aborted + + mspHold.SetResult(); + (await c.WaitAsync(Timeout, Ct)).GeneralStatus.ShouldBe((byte)0); + (await a.WaitAsync(Timeout, Ct)).GeneralStatus.ShouldBe((byte)0); + s.LiveRegistrations.ShouldBe(0); + } + + [Fact] + public async Task Waiter_cancellation_never_reaches_the_channel_token() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: true); + using var w1 = new CancellationTokenSource(); + using var w2 = new CancellationTokenSource(); + using var w3 = new CancellationTokenSource(); + + var a = await HoldFirstAsync(s, ch, Req(8, 1)); // bare single, no waiter token + var single = s.SendAsync(Req(8, 2), w1.Token); + ch.Gate!.SetResult(); + await Task.WhenAll(a, single).WaitAsync(Timeout, Ct); // bare single with a waiter token + + var held = await HoldFirstAsync(s, ch, Req(8, 3)); + var m1 = s.SendAsync(Req(8, 4), w2.Token); + var m2 = s.SendAsync(Req(8, 5), w3.Token); + ch.Gate!.SetResult(); + await Task.WhenAll(held, m1, m2).WaitAsync(Timeout, Ct); // MSP with waiter tokens + + ch.Sent.Count.ShouldBe(4); + ch.Sent[3][0].ShouldBe((byte)0x0A); + var waiterTokens = new[] { w1.Token, w2.Token, w3.Token }; + ch.Tokens.ShouldAllBe(t => !waiterTokens.Contains(t)); + ch.Tokens.Distinct().Count().ShouldBe(1); // every call used the scheduler's own stop token + ch.Tokens[0].CanBeCanceled.ShouldBeTrue(); + } + + [Fact] + public async Task Cancellation_registrations_are_released() + { + var ch = new FakeCipChannel(); + await using var s = new RequestScheduler(ch, allowPacking: true); + using var longLived = new CancellationTokenSource(); + + // The hook really counts: queued waiters hold live registrations. + var a = await HoldFirstAsync(s, ch, Req(8)); + var queued = Enumerable.Range(0, 5).Select(_ => s.SendAsync(Req(8), longLived.Token)).ToArray(); + s.LiveRegistrations.ShouldBe(5); + ch.Gate!.SetResult(); + await Task.WhenAll(queued.Append(a)).WaitAsync(Timeout, Ct); + s.LiveRegistrations.ShouldBe(0); + ch.Gate = null; + + var tasks = new Task[10_000]; + for (var i = 0; i < tasks.Length; i++) tasks[i] = s.SendAsync(Req(8), longLived.Token); + await Task.WhenAll(tasks).WaitAsync(TimeSpan.FromSeconds(60), Ct); + + s.LiveRegistrations.ShouldBe(0); + // Cancelling the long-lived token afterwards touches nothing: every registration is gone. + await longLived.CancelAsync(); + s.CancellationCallbacksRun.ShouldBe(0); + tasks.ShouldAllBe(t => t.IsCompletedSuccessfully); + } + + [Fact] + public async Task Dispose_cancels_queued_and_in_flight_waiters() + { + 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); + + await s.DisposeAsync(); + + await Should.ThrowAsync(() => a.WaitAsync(Timeout, Ct)); + await Should.ThrowAsync(() => b.WaitAsync(Timeout, Ct)); + 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 new file mode 100644 index 00000000..e2b07a3f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipChannel.cs @@ -0,0 +1,119 @@ +using System.Buffers.Binary; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; + +/// +/// Scriptable : settable , records every request +/// and the token it was sent with, and answers via . The default responder +/// answers any service with an empty success reply and splits a Multiple Service Packet into +/// per-service replies via . Reused by Tasks 12 and 13. +/// +internal sealed class FakeCipChannel : ICipChannel +{ + private readonly SemaphoreSlim _entered = new(0); + + public FakeCipChannel(int maxRequestBytes = 504) + { + MaxRequestBytes = maxRequestBytes; + Respond = DefaultRespond; + } + + public int MaxRequestBytes { get; set; } + + /// Every request, in send order. + public List Sent { get; } = []; + + /// The token each request was sent with, parallel to . + public List Tokens { get; } = []; + + /// Answers one request with raw CIP reply bytes. + public Func> Respond { get; set; } + + /// Answers one embedded (non-MSP) service; used by the default . + public Func RespondPerService { get; set; } = OkReply; + + /// When set, every send blocks on it (after being recorded) until the test completes it. + public TaskCompletionSource? Gate { get; set; } + + public int SentCount { get { lock (Sent) return Sent.Count; } } + + public async Task> SendAsync(ReadOnlyMemory cipRequest, CancellationToken cancellationToken) + { + var req = cipRequest.ToArray(); + lock (Sent) + { + Sent.Add(req); + Tokens.Add(cancellationToken); + } + _entered.Release(); + if (Gate is { } g) await g.Task.WaitAsync(cancellationToken); + return await Respond(req); + } + + /// Completes once the channel has received requests in total. + public async Task WaitForSendsAsync(int count) + { + while (SentCount < count) + { + if (!await _entered.WaitAsync(TimeSpan.FromSeconds(10))) + throw new TimeoutException($"Expected {count} sends, saw {SentCount}."); + } + } + + /// A success reply for with optional data. + public static byte[] OkReply(byte[] request) => Reply(request[0], 0x00, []); + + public static byte[] Reply(byte requestService, byte generalStatus, byte[] data, params ushort[] extended) + { + var r = new byte[4 + 2 * extended.Length + data.Length]; + r[0] = (byte)(requestService | 0x80); + r[2] = generalStatus; + r[3] = (byte)extended.Length; + for (var i = 0; i < extended.Length; i++) BinaryPrimitives.WriteUInt16LittleEndian(r.AsSpan(4 + 2 * i), extended[i]); + data.CopyTo(r, 4 + 2 * extended.Length); + return r; + } + + /// Splits an MSP request into its embedded requests. + public static List SplitMsp(byte[] msp) + { + const int countAt = 6; // 0A 02 20 02 24 01 + var data = msp.AsSpan(countAt); + int n = BinaryPrimitives.ReadUInt16LittleEndian(data); + var offsets = new int[n]; + for (var i = 0; i < n; i++) offsets[i] = BinaryPrimitives.ReadUInt16LittleEndian(data[(2 + 2 * i)..]); + var result = new List(n); + for (var i = 0; i < n; i++) + { + var end = i + 1 < n ? offsets[i + 1] : data.Length; + result.Add(data[offsets[i]..end].ToArray()); + } + return result; + } + + /// An MSP reply (service 0x8A) carrying the given embedded replies. + public static byte[] MspReply(IReadOnlyList replies, byte generalStatus = 0x00) + { + var body = new List(); + var header = new byte[2 + 2 * replies.Count]; + BinaryPrimitives.WriteUInt16LittleEndian(header, (ushort)replies.Count); + var offset = header.Length; + for (var i = 0; i < replies.Count; i++) + { + BinaryPrimitives.WriteUInt16LittleEndian(header.AsSpan(2 + 2 * i), (ushort)offset); + offset += replies[i].Length; + } + body.AddRange(header); + foreach (var r in replies) body.AddRange(r); + return Reply(0x0A, generalStatus, [.. body]); + } + + /// The default responder: per-service replies, split and re-packed for an MSP. + public Task DefaultRespond(byte[] request) + { + if (request[0] != 0x0A) return Task.FromResult(RespondPerService(request)); + var replies = SplitMsp(request).Select(RespondPerService).ToList(); + var anyFailed = replies.Any(r => r[2] != 0); + return Task.FromResult(MspReply(replies, anyFailed ? (byte)0x1E : (byte)0x00)); + } +} 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 new file mode 100644 index 00000000..ef2a1672 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Logix/LogixTagClientTests.cs @@ -0,0 +1,226 @@ +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.Logix; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Logix; + +public sealed class LogixTagClientTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + private static readonly byte[] Path = CipPath.EncodeSymbolic("TestArray"); + + /// Offset field of a Read/Write Tag Fragmented request (after service, size, path, type?, elements). + private static uint ReadOffset(byte[] req) => BinaryPrimitives.ReadUInt32LittleEndian(req.AsSpan(2 + Path.Length + 2)); + + private static byte[] Bytes(int count, int seed) => Enumerable.Range(0, count).Select(i => (byte)(seed + i)).ToArray(); + + private static byte[] TypedData(ushort type, byte[] data, ushort? handle = null) + { + var header = new List { (byte)type, (byte)(type >> 8) }; + if (handle is { } h) header.AddRange([(byte)h, (byte)(h >> 8)]); + return [.. header, .. data]; + } + + private static (FakeCipChannel Channel, LogixTagClient Client, RequestScheduler Scheduler) Create(int max = 504) + { + var ch = new FakeCipChannel(max); + var s = new RequestScheduler(ch, allowPacking: true); + return (ch, new LogixTagClient(s, ch), s); + } + + [Fact] + public async Task A_read_that_fits_is_one_request_at_offset_zero() + { + var (ch, client, s) = Create(); + await using var _ = s; + ch.RespondPerService = r => FakeCipChannel.Reply(r[0], 0x00, TypedData(LogixTypeCode.DInt, [0x87, 0xD6, 0x12, 0x00])); + + var (type, handle, data) = await client.ReadAsync(Path, 1, Ct); + + type.ShouldBe(LogixTypeCode.DInt); + handle.ShouldBeNull(); + data.ShouldBe(new byte[] { 0x87, 0xD6, 0x12, 0x00 }); + ch.Sent.Count.ShouldBe(1); + ch.Sent[0].ShouldBe(LogixServices.ReadTagFragmented(Path, 1, 0)); // libplctag's wire shape (golden captures) + } + + [Fact] + public async Task Fragmented_read_keeps_the_first_partial_and_continues() + { + var (ch, client, s) = Create(); + await using var _ = s; + ch.RespondPerService = r => ReadOffset(r) switch + { + 0 => FakeCipChannel.Reply(r[0], 0x06, TypedData(LogixTypeCode.DInt, Bytes(100, 0))), + 100 => FakeCipChannel.Reply(r[0], 0x06, TypedData(LogixTypeCode.DInt, Bytes(100, 100))), + 200 => FakeCipChannel.Reply(r[0], 0x00, TypedData(LogixTypeCode.DInt, Bytes(40, 200))), + var o => throw new InvalidOperationException($"unexpected offset {o}"), + }; + + var (type, _, data) = await client.ReadAsync(Path, 60, Ct); + + type.ShouldBe(LogixTypeCode.DInt); + data.Length.ShouldBe(240); + data.ShouldBe(Bytes(240, 0)); // first chunk kept once, type headers stripped, in order + ch.Sent.Select(ReadOffset).ShouldBe(new uint[] { 0, 100, 200 }); + ch.Sent.ShouldAllBe(r => r[0] == 0x52); + } + + [Fact] + public async Task A_fragmented_struct_read_keeps_the_handle_from_the_first_reply() + { + 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: 0x0FCE)); + + var (type, handle, data) = await client.ReadAsync(Path, 1, Ct); + + type.ShouldBe(LogixTypeCode.Struct); + handle.ShouldBe((ushort)0x0FCE); + data.ShouldBe(Bytes(12, 0)); + } + + [Fact] + public async Task A_failed_read_throws_with_the_reply_status() + { + var (ch, client, s) = Create(); + await using var _ = s; + ch.RespondPerService = r => FakeCipChannel.Reply(r[0], 0x04, [], 0x0000); + + var ex = await Should.ThrowAsync(() => client.ReadAsync(Path, 1, Ct)); + ex.Status.ShouldBe(AbStatusCodes.FromCip(0x04, [0x0000])); + } + + [Fact] + public async Task A_fragment_failing_mid_read_throws_with_its_status() + { + var (ch, client, s) = Create(); + await using var _ = s; + ch.RespondPerService = r => ReadOffset(r) == 0 + ? FakeCipChannel.Reply(r[0], 0x06, TypedData(LogixTypeCode.DInt, Bytes(100, 0))) + : FakeCipChannel.Reply(r[0], 0x05, []); + + var ex = await Should.ThrowAsync(() => client.ReadAsync(Path, 60, Ct)); + ex.Status.ShouldBe(AbStatusCodes.FromCip(0x05, [])); + } + + [Fact] + public async Task A_partial_reply_with_no_data_does_not_loop_forever() + { + var (ch, client, s) = Create(); + await using var _ = s; + ch.RespondPerService = r => FakeCipChannel.Reply(r[0], 0x06, TypedData(LogixTypeCode.DInt, [])); + + var ex = await Should.ThrowAsync(() => client.ReadAsync(Path, 60, Ct)); + ex.Status.ShouldBe(AbStatus.BadReply); + ch.Sent.Count.ShouldBe(1); + } + + [Fact] + public async Task A_write_that_fits_is_one_write_tag() + { + var (ch, client, s) = Create(); + await using var _ = s; + var data = Bytes(16, 1); + + await client.WriteAsync(Path, LogixTypeCode.DInt, null, 4, data, Ct); + + ch.Sent.Count.ShouldBe(1); + ch.Sent[0].ShouldBe(LogixServices.WriteTag(Path, LogixTypeCode.DInt, null, 4, data)); + } + + [Fact] + public async Task A_write_exactly_at_the_limit_is_still_one_write_tag() + { + var data = Bytes(16, 1); + var size = LogixServices.WriteTag(Path, LogixTypeCode.DInt, null, 4, data).Length; + var (ch, client, s) = Create(max: size); + await using var _ = s; + + await client.WriteAsync(Path, LogixTypeCode.DInt, null, 4, data, Ct); + + ch.Sent.Count.ShouldBe(1); + ch.Sent[0][0].ShouldBe((byte)0x4D); + } + + [Theory] + [InlineData(LogixTypeCode.DInt, 52)] // room 54 -> 52 (13 DINTs) + [InlineData(LogixTypeCode.LInt, 48)] // room 54 -> 48 (6 LINTs) + [InlineData(LogixTypeCode.Int, 54)] // room 54 -> 54 (27 INTs) + public async Task Fragmented_write_chunks_align_to_element_width(ushort type, int expectedChunk) + { + const int room = 54; + var (ch, client, s) = Create(max: room + 12 + Path.Length); + await using var _ = s; + var width = LogixTypeCode.ByteWidth(type); + var data = Bytes(width * 40, 7); // 40 elements, more than fits in one frame + var elements = (ushort)40; + + await client.WriteAsync(Path, type, null, elements, data, Ct); + + ch.Sent.ShouldAllBe(r => r[0] == 0x53); + ch.Sent.ShouldAllBe(r => r.Length <= ch.MaxRequestBytes); + var reassembled = new List(); + foreach (var r in ch.Sent) + { + var body = r.AsSpan(2 + Path.Length); + BinaryPrimitives.ReadUInt16LittleEndian(body).ShouldBe(type); + BinaryPrimitives.ReadUInt16LittleEndian(body[2..]).ShouldBe(elements); // total element count + var offset = BinaryPrimitives.ReadUInt32LittleEndian(body[4..]); + offset.ShouldBe((uint)reassembled.Count); + var chunk = body[8..]; + (offset % (uint)width).ShouldBe(0u); + reassembled.AddRange(chunk.ToArray()); + } + reassembled.ShouldBe(data); + ch.Sent.Take(ch.Sent.Count - 1).ShouldAllBe(r => r.Length - (2 + Path.Length + 8) == expectedChunk); + } + + [Fact] + public async Task A_fragmented_struct_write_carries_the_handle_in_every_chunk() + { + const int room = 40; + var (ch, client, s) = Create(max: room + 12 + Path.Length); + await using var _ = s; + var data = Bytes(100, 3); // 10 elements of a 10-byte struct + + await client.WriteAsync(Path, LogixTypeCode.Struct, 0x1234, 10, data, Ct); + + ch.Sent.Count.ShouldBe(3); // 40 + 40 + 20 + foreach (var r in ch.Sent) + { + var body = r.AsSpan(2 + Path.Length); + BinaryPrimitives.ReadUInt16LittleEndian(body).ShouldBe(LogixTypeCode.Struct); + BinaryPrimitives.ReadUInt16LittleEndian(body[2..]).ShouldBe((ushort)0x1234); + r.Length.ShouldBeLessThanOrEqualTo(ch.MaxRequestBytes); + } + } + + [Fact] + public async Task A_failed_write_throws_with_the_reply_status() + { + var (ch, client, s) = Create(); + await using var _ = s; + ch.RespondPerService = r => FakeCipChannel.Reply(r[0], 0x0F, []); + + var ex = await Should.ThrowAsync( + () => client.WriteAsync(Path, LogixTypeCode.DInt, null, 1, Bytes(4, 0), Ct)); + ex.Status.ShouldBe(AbStatusCodes.FromCip(0x0F, [])); + } + + [Fact] + public async Task A_failing_fragment_stops_the_write() + { + var (ch, client, s) = Create(max: 54 + 12 + Path.Length); + await using var _ = s; + ch.RespondPerService = r => FakeCipChannel.Reply(r[0], 0x08, []); + + await Should.ThrowAsync( + () => client.WriteAsync(Path, LogixTypeCode.DInt, null, 40, Bytes(160, 0), Ct)); + ch.Sent.Count.ShouldBe(1); + } +}