merge: Task 10 — RequestScheduler (MSP coalescing) + LogixTagClient fragment loops
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,301 @@
|
||||
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, or the scheduler is disposed; cancels only 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;
|
||||
}
|
||||
|
||||
/// <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)
|
||||
{
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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 = 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
|
||||
{
|
||||
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)
|
||||
{
|
||||
// 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)
|
||||
{
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <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)
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
}
|
||||
_stop.Cancel();
|
||||
try { await _pump.ConfigureAwait(false); } catch { /* stopping */ }
|
||||
|
||||
List<Pending> remaining;
|
||||
lock (_lock)
|
||||
{
|
||||
remaining = [.. _queue];
|
||||
_queue.Clear();
|
||||
}
|
||||
var error = DisposedError();
|
||||
foreach (var p in remaining) p.Fail(error, this);
|
||||
_signal.Dispose();
|
||||
_stop.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
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;
|
||||
|
||||
/// <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));
|
||||
|
||||
/// <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;
|
||||
if (data.Count > MaxReadBytes)
|
||||
throw new AbTransportException(AbStatus.TooLarge,
|
||||
$"Fragmented read exceeded {MaxReadBytes} bytes; the controller kept answering Partial.");
|
||||
}
|
||||
|
||||
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 (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;
|
||||
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. 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,
|
||||
$"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}.");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user