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

feat(ab-transport): MSP-coalescing request scheduler and fragmented Logix read/write

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-07 16:15:05 -04:00
parent 110fb861e9
commit 1c7fd5d5de
5 changed files with 1075 additions and 0 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
+260
View File
@@ -0,0 +1,260 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
/// <summary>
/// Per-connection CIP request scheduler. Keeps exactly one request in flight on its
/// <see cref="ICipChannel"/>; 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 <see cref="ICipChannel.MaxRequestBytes"/>
/// into one Multiple Service Packet. Each waiter receives its own embedded <see cref="CipReply"/>.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
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<CipReply> Tcs { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
public LinkedListNode<Pending>? Node { get; set; }
/// <summary>Stores the waiter's registration; disposes it at once if the waiter already completed.</summary>
public void SetRegistration(CancellationTokenRegistration registration, RequestScheduler owner)
{
lock (this)
{
if (!_completed)
{
_registration = registration;
_hasRegistration = true;
return;
}
}
registration.Dispose();
Interlocked.Decrement(ref owner._liveRegistrations);
}
/// <summary>Marks the waiter complete exactly once and releases its registration (finding 10).
/// Returns false if it was already completed.</summary>
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<Pending> _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);
}
/// <summary>Test hook: waiter cancellation registrations that have not yet been disposed.</summary>
internal int LiveRegistrations => Volatile.Read(ref _liveRegistrations);
/// <summary>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.</summary>
internal int CancellationCallbacksRun => Volatile.Read(ref _cancellationCallbacksRun);
/// <summary>Test hook: requests still queued (not yet taken into a batch).</summary>
internal int QueuedCount { get { lock (_lock) return _queue.Count; } }
/// <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.
/// </summary>
public Task<CipReply> SendAsync(byte[] request, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(request);
if (ct.IsCancellationRequested) return Task.FromCanceled<CipReply>(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<Pending> TakeBatch()
{
var batch = new List<Pending>();
lock (_lock)
{
var sizes = new List<int>();
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<Pending> 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<byte[]>(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<Pending> remaining;
lock (_lock)
{
remaining = [.. _queue];
_queue.Clear();
}
foreach (var p in remaining) p.Cancel(_stop.Token, this);
_signal.Dispose();
_stop.Dispose();
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs
+113
View File
@@ -0,0 +1,113 @@
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Logix;
/// <summary>
/// Logix tag read/write over a connection's <see cref="RequestScheduler"/>, including the fragment
/// loops for values that do not fit in one frame.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel channel)
{
private const byte StatusSuccess = 0x00;
private const byte StatusPartial = 0x06;
/// <summary>Fixed bytes of a Write Tag Fragmented request besides path and data: service, path
/// size, type, struct handle, element count, offset.</summary>
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));
/// <summary>
/// Read <paramref name="elements"/> elements of the tag at the encoded <paramref name="path"/>.
/// A Partial (0x06) reply keeps its data and the read continues at <c>offset = bytes received so
/// far</c> 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.
/// </summary>
/// <remarks>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.</remarks>
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<byte>(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());
}
/// <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.
/// </summary>
public async Task WriteAsync(byte[] path, ushort type, ushort? handle, ushort elements,
ReadOnlyMemory<byte> 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}.");
}
}