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

fix(abcip): Task 13 review r1 - bounded listing pager, exclusive large reads, template bound, safe lease release

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-07 17:07:05 -04:00
parent 78e60a5a75
commit b2b9aa04ed
5 changed files with 182 additions and 15 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
+12 -4
View File
@@ -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;
/// <summary>True when this request must never share a Multiple Service Packet with another.</summary>
public bool Exclusive { get; } = exclusive;
public TaskCompletionSource<CipReply> Tcs { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
public LinkedListNode<Pending>? Node { get; set; }
@@ -119,12 +122,16 @@ public sealed class RequestScheduler : IAsyncDisposable
/// status is not an exception here. Faults with <see cref="AbTransportException"/> when the channel
/// 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)
/// <param name="request">The CIP request.</param>
/// <param name="ct">Cancels this waiter only.</param>
/// <param name="exclusive">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).</param>
public Task<CipReply> SendAsync(byte[] request, CancellationToken ct, bool exclusive = false)
{
ArgumentNullException.ThrowIfNull(request);
if (ct.IsCancellationRequested) return Task.FromCanceled<CipReply>(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;
}
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTagEnumerator.cs
+35 -7
View File
@@ -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<AbCipTagCreateParams, (RequestScheduler Scheduler, Func<ValueTask> Release)> _acquire;
private readonly ILogger _logger;
/// <summary>Production constructor: leases from <paramref name="pool"/>.</summary>
/// <param name="pool">The connection pool (the factory passes <see cref="DeviceSessionPool.Shared"/>).</param>
@@ -28,9 +31,14 @@ internal sealed class NativeAbCipTagEnumerator : IAbCipTagEnumerator
/// <summary>Test seam: supplies the scheduler and its release directly.</summary>
/// <param name="acquire">Acquires the scheduler for a device, with the callback that releases it.</param>
/// <param name="logger">Receives release failures; null discards them.</param>
internal NativeAbCipTagEnumerator(
Func<AbCipTagCreateParams, (RequestScheduler Scheduler, Func<ValueTask> Release)> acquire) =>
Func<AbCipTagCreateParams, (RequestScheduler Scheduler, Func<ValueTask> Release)> acquire,
ILogger? logger = null)
{
_acquire = acquire;
_logger = logger ?? NullLogger.Instance;
}
/// <inheritdoc />
public async IAsyncEnumerable<AbCipDiscoveredTag> 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:";
/// <summary>Page cap for one listing; a real controller needs a small fraction of this.</summary>
internal const int MaxPages = 10_000;
/// <summary>Byte cap for one listing (64 MiB).</summary>
internal const long MaxListingBytes = 64L * 1024 * 1024;
/// <summary>Reads <c>@tags</c> or <c>Program:X.@tags</c> in full.</summary>
/// <param name="scheduler">The device's shared scheduler.</param>
/// <param name="listingName">The listing pseudo-tag name.</param>
@@ -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
/// <summary>Upper bound on waiting for the lease release (and its Forward Close).</summary>
private static readonly TimeSpan DisposeWait = TimeSpan.FromSeconds(2);
/// <summary>Releases, waiting at most 2 s; never throws.</summary>
public static void ReleaseBounded(Func<ValueTask> release)
/// <summary>Releases, waiting at most 2 s; never throws, whatever the release delegate does.</summary>
public static void ReleaseBounded(Func<ValueTask> 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.");
}
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/NativeAbCipTemplateReader.cs
+15 -4
View File
@@ -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
/// <summary>Bytes of the template definition header not counted in the Read Template length.</summary>
private const int DefinitionHeaderBytes = 23;
/// <summary>Upper bound on a template's Read Template length; Logix templates are far smaller.</summary>
internal const long MaxTemplateBytes = 64 * 1024;
private readonly Func<AbCipTagCreateParams, (RequestScheduler Scheduler, ICipChannel Channel, Func<ValueTask> Release)> _acquire;
private readonly ILogger _logger;
/// <summary>Production constructor: leases from <paramref name="pool"/>.</summary>
/// <param name="pool">The connection pool (the factory passes <see cref="DeviceSessionPool.Shared"/>).</param>
@@ -32,9 +38,14 @@ internal sealed class NativeAbCipTemplateReader : IAbCipTemplateReader
/// <summary>Test seam: supplies the scheduler, channel and release directly.</summary>
/// <param name="acquire">Acquires the connection parts for a device.</param>
/// <param name="logger">Receives release failures; null discards them.</param>
internal NativeAbCipTemplateReader(
Func<AbCipTagCreateParams, (RequestScheduler Scheduler, ICipChannel Channel, Func<ValueTask> Release)> acquire) =>
Func<AbCipTagCreateParams, (RequestScheduler Scheduler, ICipChannel Channel, Func<ValueTask> Release)> acquire,
ILogger? logger = null)
{
_acquire = acquire;
_logger = logger ?? NullLogger.Instance;
}
/// <inheritdoc />
public async Task<byte[]> 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);
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTagEnumeratorTests.cs
+74
View File
@@ -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<AbTransportException>(
() => 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<AbTransportException>(
() => 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<OperationCanceledException>(() => 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);
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipTemplateReaderTests.cs
+46
View File
@@ -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<AbTransportException>(
() => reader.ReadAsync(Params, 0x0FCE, TestContext.Current.CancellationToken));
ex.Status.ShouldBe(AbStatus.BadReply);
ch.Sent.ShouldNotContain(r => r[0] == 0x4C);
}
}