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

fix(ab-transport): harden Task 10 scheduler and tag client (dispose faults, pump guard, read cap, write validation, 8-byte wide-struct alignment)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-07 16:22:40 -04:00
parent 1c7fd5d5de
commit 5211975067
5 changed files with 184 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/RequestScheduler.cs
+50 -9
View File
@@ -117,7 +117,7 @@ public sealed class RequestScheduler : IAsyncDisposable
/// <summary>
/// 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 <see cref="AbTransportException"/> when the channel
/// or the MSP envelope fails; cancels when <paramref name="ct"/> fires.
/// or the MSP envelope fails, or the scheduler is disposed; cancels only when <paramref name="ct"/> fires.
/// </summary>
public Task<CipReply> SendAsync(byte[] request, CancellationToken ct)
{
@@ -154,9 +154,10 @@ public sealed class RequestScheduler : IAsyncDisposable
return p.Tcs.Task;
}
private List<Pending> TakeBatch()
/// <summary>Moves the next batch from the queue into <paramref name="batch"/>. Filling a caller-owned
/// list means a throw part-way through still leaves every taken waiter reachable for faulting.</summary>
private void TakeBatch(List<Pending> batch)
{
var batch = new List<Pending>();
lock (_lock)
{
var sizes = new List<int>();
@@ -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<Pending>();
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<Pending> batch, Exception ex)
{
var error = ex as AbTransportException
?? new AbTransportException(AbStatus.BadConnection, $"Request scheduler failed: {ex.Message}", ex);
List<Pending> 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<Pending> 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
}
}
/// <summary>
/// The error a waiter sees when the scheduler (i.e. its connection) goes away under it. Deliberately
/// NOT an <see cref="OperationCanceledException"/>: 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.
/// </summary>
private static AbTransportException DisposedError() =>
new(AbStatus.BadConnection, "Scheduler disposed.");
/// <summary>Stops the pump. In-flight and queued waiters fault with
/// <see cref="AbTransportException"/> (<see cref="AbStatus.BadConnection"/>).</summary>
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();
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs
+27 -4
View File
@@ -19,6 +19,14 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann
/// size, type, struct handle, element count, offset.</summary>
private const int FragmentedWriteOverhead = 12;
/// <summary>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.</summary>
internal const int MaxReadBytes = 2 * 1024 * 1024;
/// <summary>Chunk alignment when one element is wider than a frame (Logix lays structures out on
/// at most 8-byte boundaries).</summary>
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
/// <summary>
/// Write <paramref name="data"/> to the tag at <paramref name="path"/>. One Write Tag when the
/// request fits <see cref="ICipChannel.MaxRequestBytes"/>; otherwise Write Tag Fragmented chunks of
/// <c>MaxRequestBytes - 12 - path.Length</c> bytes aligned down to the element width.
/// <c>MaxRequestBytes - 12 - path.Length</c> bytes aligned down to the element width (or to 8 bytes
/// when a single element is wider than a frame).
/// </summary>
/// <remarks>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.</remarks>
/// <exception cref="ArgumentException">For an atomic type, <paramref name="data"/> is not exactly
/// <paramref name="elements"/> × the type's width.</exception>
public async Task WriteAsync(byte[] path, ushort type, ushort? handle, ushort elements,
ReadOnlyMemory<byte> 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,
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/RequestSchedulerTests.cs
+41 -5
View File
@@ -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<AbTransportException>(() => b.WaitAsync(Timeout, Ct)); // already taken
var exC = await Should.ThrowAsync<AbTransportException>(() => c.WaitAsync(Timeout, Ct)); // still queued
var exD = await Should.ThrowAsync<AbTransportException>(() => 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<OperationCanceledException>(() => a.WaitAsync(Timeout, Ct));
await Should.ThrowAsync<OperationCanceledException>(() => b.WaitAsync(Timeout, Ct));
// Not OperationCanceledException: upstream would swallow that as "unsubscribed".
var inFlight = await Should.ThrowAsync<AbTransportException>(() => a.WaitAsync(Timeout, Ct));
inFlight.Status.ShouldBe(AbStatus.BadConnection);
var queued = await Should.ThrowAsync<AbTransportException>(() => 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<ObjectDisposedException>(() => s.SendAsync(Req(8), CancellationToken.None));
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipChannel.cs
+10 -1
View File
@@ -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;
}
/// <summary>When set, reading <see cref="MaxRequestBytes"/> throws it (simulates a broken channel).</summary>
public Exception? MaxRequestBytesThrows { get; set; }
/// <summary>Every request, in send order.</summary>
public List<byte[]> Sent { get; } = [];
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Logix/LogixTagClientTests.cs
+56
View File
@@ -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<byte>();
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<AbTransportException>(() => 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<ArgumentException>(
() => 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()
{