perf(abcip): native read window >= 64; cold waiters hold no window slot
P8: on the native transport the bounded-parallel read window is also the cap on services per Multiple Service Packet (the device's shared scheduler packs whatever is queued, within the connection's request/reply byte budget), so the default 8 cost a 1000-tag controller ~125 round trips per poll instead of ~10. The effective window (AbCipDriver.ReadWindow) is now max(64, option) when the native tag factory serves reads; libplctag and injected factories keep the option exactly. Option doc + AdminUI help text updated (Razor: Wave 4 live check). P7 addendum: a work item now takes its window slot only after its runtime exists. A cold create queues on its device's single-flight connect gate, and holding a slot there let one dead device's backoff-expiry attempt occupy the whole window for a full Timeout, stalling every other device's reads. Refs #737 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
@@ -117,6 +117,12 @@ public sealed class AbCipDriverOptions
|
||||
/// device round trips (the whole point: a strictly sequential batch pays one full round
|
||||
/// trip per tag); lowering it to <c>1</c> restores strictly sequential reads. Values below
|
||||
/// <c>1</c> are clamped to <c>1</c> at the use site.
|
||||
/// <para>
|
||||
/// On the <see cref="AbCipTransport.Native"/> transport the effective window is at least 64: there the
|
||||
/// window caps how many services one Multiple Service Packet coalesces (the device's shared scheduler packs
|
||||
/// what is queued, bounded by the connection's request/reply byte budget), so a small value only multiplies
|
||||
/// round trips. A value above 64 is honoured; a value below it is raised.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public int MaxConcurrentReads { get; init; } = 8;
|
||||
|
||||
|
||||
@@ -895,7 +895,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
|
||||
// round trips overlap, which a strictly sequential await chain made impossible. Results are
|
||||
// index-addressed, so completion order is irrelevant. Per-item health verdicts are COLLECTED
|
||||
// and written once after the batch: an inline write to the shared _health field would race.
|
||||
var maxConcurrent = Math.Max(1, _options.MaxConcurrentReads);
|
||||
var maxConcurrent = ReadWindow;
|
||||
var outcomes = new ReadOutcome[itemCount];
|
||||
using var window = new SemaphoreSlim(maxConcurrent, maxConcurrent);
|
||||
var tasks = new Task[itemCount];
|
||||
@@ -910,32 +910,44 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
|
||||
ApplyBatchHealth(outcomes, now);
|
||||
return results;
|
||||
|
||||
// The window slot is taken INSIDE each item, after its runtime exists (P7): a cold create queues on its
|
||||
// device's single-flight connect gate, and holding a slot while queued there let one dead device's
|
||||
// backoff-expiry attempt occupy the whole window for a full Timeout, stalling every other device's reads.
|
||||
async Task RunItemAsync(int slot)
|
||||
{
|
||||
// Acquire OUTSIDE the try: a wait that throws never acquired, so it must not release.
|
||||
await window.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
if (slot < groups.Count)
|
||||
{
|
||||
if (slot < groups.Count)
|
||||
{
|
||||
outcomes[slot] = await ReadGroupAsync(groups[slot], results, now, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
var fb = fallbacks[slot - groups.Count];
|
||||
outcomes[slot] = await ReadSingleAsync(
|
||||
fb, fullReferences[fb.OriginalIndex], results, now, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
outcomes[slot] = await ReadGroupAsync(groups[slot], results, now, window, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
else
|
||||
{
|
||||
window.Release();
|
||||
var fb = fallbacks[slot - groups.Count];
|
||||
outcomes[slot] = await ReadSingleAsync(
|
||||
fb, fullReferences[fb.OriginalIndex], results, now, window, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Floor on the read window when the native transport serves this driver (P8). Its runtimes share one
|
||||
/// request scheduler per device that packs whatever is queued when the pump wakes into one Multiple Service
|
||||
/// Packet — bounded by the connection's request AND reply byte budget, so a wide window cannot over-pack. The
|
||||
/// window is therefore the cap on services per round trip: at 8, a 1000-tag controller paid ~125 round trips
|
||||
/// per poll instead of ~10.
|
||||
/// </summary>
|
||||
internal const int NativeMinReadWindow = 64;
|
||||
|
||||
/// <summary>
|
||||
/// The effective bounded-parallel read window: <see cref="AbCipDriverOptions.MaxConcurrentReads"/> (clamped
|
||||
/// to ≥ 1), raised to at least <see cref="NativeMinReadWindow"/> when the native tag factory serves the reads.
|
||||
/// libplctag overlaps per-handle round trips only, so it (and any injected factory) keeps the option as is.
|
||||
/// </summary>
|
||||
internal int ReadWindow => _tagFactory is NativeAbCipTagFactory
|
||||
? Math.Max(NativeMinReadWindow, _options.MaxConcurrentReads)
|
||||
: Math.Max(1, _options.MaxConcurrentReads);
|
||||
|
||||
/// <summary>
|
||||
/// One read work item's health verdict. Returned by <see cref="ReadSingleAsync"/> /
|
||||
/// <see cref="ReadGroupAsync"/> instead of written to the shared health field inline, because
|
||||
@@ -1015,7 +1027,8 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
|
||||
}
|
||||
|
||||
private async Task<ReadOutcome> ReadSingleAsync(
|
||||
AbCipUdtReadFallback fb, string reference, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
|
||||
AbCipUdtReadFallback fb, string reference, DataValueSnapshot[] results, DateTime now,
|
||||
SemaphoreSlim window, CancellationToken ct)
|
||||
{
|
||||
if (!_resolver.TryResolve(reference, out var def))
|
||||
{
|
||||
@@ -1041,82 +1054,91 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
|
||||
{
|
||||
var runtime = await EnsureTagRuntimeAsync(device, def, ct).ConfigureAwait(false);
|
||||
|
||||
// Window slot only now that the runtime exists (see ReadAsync); a wait that throws never acquired.
|
||||
await window.WaitAsync(ct).ConfigureAwait(false);
|
||||
// Serialise Read → GetStatus → Decode against the shared runtime — the cached libplctag
|
||||
// Tag handle is not concurrency-safe and GetStatus returns the LAST op's status.
|
||||
var opLock = device.GetRuntimeLock(def.Name);
|
||||
await opLock.WaitAsync(ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
await runtime.ReadAsync(ct).ConfigureAwait(false);
|
||||
await opLock.WaitAsync(ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
await runtime.ReadAsync(ct).ConfigureAwait(false);
|
||||
|
||||
var status = runtime.GetStatus();
|
||||
if (status == (int)AbStatus.TypeMismatch)
|
||||
{
|
||||
// Final review I4: the controller ANSWERED with a type that disagrees with the authored
|
||||
// one — a deterministic config error, not a comms fault. Keep the handle (evicting would
|
||||
// re-create it — and, for a sole lease holder, re-Forward-Open — on every poll), leave
|
||||
// health alone, warn once per tag.
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadTypeMismatch, null, now);
|
||||
WarnTypeMismatchOnce(def.Name,
|
||||
"AbCip read of {Tag} on device {Device}: the controller's type does not match the authored " +
|
||||
"{DataType}; reporting BadTypeMismatch until the tag's dataType is corrected (logged once per tag)",
|
||||
reference, def.DeviceHostAddress, def.DataType);
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
if (AbCipRefusals.IsDeviceRefusal(status))
|
||||
{
|
||||
// The controller answered with a deterministic refusal: per-tag Bad, keep the handle.
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.MapReadStatus((AbStatus)status), null, now);
|
||||
WarnRefusalOnce(def.Name, "read", reference, def.DeviceHostAddress, ((AbStatus)status).ToString());
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
if (status != 0)
|
||||
{
|
||||
// Evict the stale handle so the next call re-creates it.
|
||||
// A non-zero status can mean the controller dropped the connection or the tag
|
||||
// handle became permanently invalid (e.g. after a PLC download). Evicting
|
||||
// mirrors the probe loop's recreate-on-failure behaviour.
|
||||
EvictRuntime(device, def.Name);
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(null,
|
||||
AbCipStatusMapper.MapStatus(status), null, now);
|
||||
_logger.LogWarning(
|
||||
"AbCip read returned non-zero transport status {Status} for tag {Tag} on device {Device}; " +
|
||||
"evicting cached runtime so next call re-creates it",
|
||||
status, reference, def.DeviceHostAddress);
|
||||
return ReadOutcome.Degraded($"transport status {status} reading {reference}");
|
||||
}
|
||||
var status = runtime.GetStatus();
|
||||
if (status == (int)AbStatus.TypeMismatch)
|
||||
{
|
||||
// Final review I4: the controller ANSWERED with a type that disagrees with the authored
|
||||
// one — a deterministic config error, not a comms fault. Keep the handle (evicting would
|
||||
// re-create it — and, for a sole lease holder, re-Forward-Open — on every poll), leave
|
||||
// health alone, warn once per tag.
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadTypeMismatch, null, now);
|
||||
WarnTypeMismatchOnce(def.Name,
|
||||
"AbCip read of {Tag} on device {Device}: the controller's type does not match the authored " +
|
||||
"{DataType}; reporting BadTypeMismatch until the tag's dataType is corrected (logged once per tag)",
|
||||
reference, def.DeviceHostAddress, def.DataType);
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
if (AbCipRefusals.IsDeviceRefusal(status))
|
||||
{
|
||||
// The controller answered with a deterministic refusal: per-tag Bad, keep the handle.
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.MapReadStatus((AbStatus)status), null, now);
|
||||
WarnRefusalOnce(def.Name, "read", reference, def.DeviceHostAddress, ((AbStatus)status).ToString());
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
if (status != 0)
|
||||
{
|
||||
// Evict the stale handle so the next call re-creates it.
|
||||
// A non-zero status can mean the controller dropped the connection or the tag
|
||||
// handle became permanently invalid (e.g. after a PLC download). Evicting
|
||||
// mirrors the probe loop's recreate-on-failure behaviour.
|
||||
EvictRuntime(device, def.Name);
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(null,
|
||||
AbCipStatusMapper.MapStatus(status), null, now);
|
||||
_logger.LogWarning(
|
||||
"AbCip read returned non-zero transport status {Status} for tag {Tag} on device {Device}; " +
|
||||
"evicting cached runtime so next call re-creates it",
|
||||
status, reference, def.DeviceHostAddress);
|
||||
return ReadOutcome.Degraded($"transport status {status} reading {reference}");
|
||||
}
|
||||
|
||||
var tagPath = AbCipTagPath.TryParse(def.TagPath);
|
||||
var bitIndex = tagPath?.BitIndex;
|
||||
var tagPath = AbCipTagPath.TryParse(def.TagPath);
|
||||
var bitIndex = tagPath?.BitIndex;
|
||||
|
||||
// libplctag never checks the CIP type the controller returned; it decodes the bytes as
|
||||
// the authored type. A scalar whose buffer is not that type's width is mis-typed config
|
||||
// (seen live: a REAL authored as INT read Good -17346), so refuse it rather than publish
|
||||
// garbage as Good. Same-width mismatches (DINT vs REAL) cannot be seen this way.
|
||||
if (!IsArrayTag(def) && bitIndex is null
|
||||
&& ExpectedScalarWidth(def.DataType) is { } width
|
||||
&& runtime.GetSize() is var size && size >= 0 && size != width)
|
||||
{
|
||||
// Same class as the native TypeMismatch above (final review I4): deterministic, device
|
||||
// answered — no Degraded, warn once per tag.
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadTypeMismatch, null, now);
|
||||
WarnTypeMismatchOnce(def.Name,
|
||||
"AbCip read of {Tag} on device {Device} returned {Size} bytes but its authored type {DataType} is {Width} bytes wide; check the tag's dataType (logged once per tag)",
|
||||
reference, def.DeviceHostAddress, size, def.DataType, width);
|
||||
return ReadOutcome.Refused;
|
||||
// libplctag never checks the CIP type the controller returned; it decodes the bytes as
|
||||
// the authored type. A scalar whose buffer is not that type's width is mis-typed config
|
||||
// (seen live: a REAL authored as INT read Good -17346), so refuse it rather than publish
|
||||
// garbage as Good. Same-width mismatches (DINT vs REAL) cannot be seen this way.
|
||||
if (!IsArrayTag(def) && bitIndex is null
|
||||
&& ExpectedScalarWidth(def.DataType) is { } width
|
||||
&& runtime.GetSize() is var size && size >= 0 && size != width)
|
||||
{
|
||||
// Same class as the native TypeMismatch above (final review I4): deterministic, device
|
||||
// answered — no Degraded, warn once per tag.
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadTypeMismatch, null, now);
|
||||
WarnTypeMismatchOnce(def.Name,
|
||||
"AbCip read of {Tag} on device {Device} returned {Size} bytes but its authored type {DataType} is {Width} bytes wide; check the tag's dataType (logged once per tag)",
|
||||
reference, def.DeviceHostAddress, size, def.DataType, width);
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
// Review I-1 — an array tag (the EXPLICIT IsArray flag) decodes the whole buffer into an
|
||||
// element-typed CLR array (int[]/float[]/bool[]/string[]…), INCLUDING a 1-element array
|
||||
// (ElementCount 1). Scalar tags keep the single-value path.
|
||||
var value = IsArrayTag(def)
|
||||
? runtime.DecodeArray(def.DataType, Math.Max(1, def.ElementCount))
|
||||
: runtime.DecodeValue(def.DataType, bitIndex);
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now);
|
||||
return ReadOutcome.Success;
|
||||
}
|
||||
finally
|
||||
{
|
||||
opLock.Release();
|
||||
}
|
||||
// Review I-1 — an array tag (the EXPLICIT IsArray flag) decodes the whole buffer into an
|
||||
// element-typed CLR array (int[]/float[]/bool[]/string[]…), INCLUDING a 1-element array
|
||||
// (ElementCount 1). Scalar tags keep the single-value path.
|
||||
var value = IsArrayTag(def)
|
||||
? runtime.DecodeArray(def.DataType, Math.Max(1, def.ElementCount))
|
||||
: runtime.DecodeValue(def.DataType, bitIndex);
|
||||
results[fb.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now);
|
||||
return ReadOutcome.Success;
|
||||
}
|
||||
finally
|
||||
{
|
||||
opLock.Release();
|
||||
window.Release();
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
@@ -1185,7 +1207,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
|
||||
/// per-tag fallback list are unaffected.
|
||||
/// </summary>
|
||||
private async Task<ReadOutcome> ReadGroupAsync(
|
||||
AbCipUdtReadGroup group, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
|
||||
AbCipUdtReadGroup group, DataValueSnapshot[] results, DateTime now, SemaphoreSlim window, CancellationToken ct)
|
||||
{
|
||||
var parent = group.ParentDefinition;
|
||||
|
||||
@@ -1199,53 +1221,62 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
|
||||
{
|
||||
var runtime = await EnsureTagRuntimeAsync(device, parent, ct).ConfigureAwait(false);
|
||||
|
||||
// Window slot only now that the runtime exists (see ReadAsync); a wait that throws never acquired.
|
||||
await window.WaitAsync(ct).ConfigureAwait(false);
|
||||
// Serialise the whole-UDT Read → GetStatus → per-member DecodeValueAt against the shared
|
||||
// parent runtime — the members decode from that one handle's buffer.
|
||||
var opLock = device.GetRuntimeLock(parent.Name);
|
||||
await opLock.WaitAsync(ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
await runtime.ReadAsync(ct).ConfigureAwait(false);
|
||||
await opLock.WaitAsync(ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
await runtime.ReadAsync(ct).ConfigureAwait(false);
|
||||
|
||||
var status = runtime.GetStatus();
|
||||
if (status == (int)AbStatus.TypeMismatch)
|
||||
{
|
||||
// Final review I4: deterministic — keep the handle, no health signal, warn once.
|
||||
StampGroupStatus(group, results, now, AbCipStatusMapper.BadTypeMismatch);
|
||||
WarnTypeMismatchOnce(parent.Name,
|
||||
"AbCip whole-UDT read of {Parent} on device {Device}: the controller's type does not match " +
|
||||
"the authored type; {MemberCount} members report BadTypeMismatch (logged once per tag)",
|
||||
group.ParentName, parent.DeviceHostAddress, group.Members.Count);
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
if (AbCipRefusals.IsDeviceRefusal(status))
|
||||
{
|
||||
StampGroupStatus(group, results, now, AbCipStatusMapper.MapReadStatus((AbStatus)status));
|
||||
WarnRefusalOnce(parent.Name, "whole-UDT read", group.ParentName, parent.DeviceHostAddress, ((AbStatus)status).ToString());
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
if (status != 0)
|
||||
{
|
||||
EvictRuntime(device, parent.Name);
|
||||
var mapped = AbCipStatusMapper.MapStatus(status);
|
||||
StampGroupStatus(group, results, now, mapped);
|
||||
_logger.LogWarning(
|
||||
"AbCip whole-UDT read returned non-zero transport status {Status} for parent {Parent} " +
|
||||
"on device {Device}; {MemberCount} member values stamped with mapped status",
|
||||
status, group.ParentName, parent.DeviceHostAddress, group.Members.Count);
|
||||
return ReadOutcome.Degraded($"transport status {status} reading UDT {group.ParentName}");
|
||||
}
|
||||
var status = runtime.GetStatus();
|
||||
if (status == (int)AbStatus.TypeMismatch)
|
||||
{
|
||||
// Final review I4: deterministic — keep the handle, no health signal, warn once.
|
||||
StampGroupStatus(group, results, now, AbCipStatusMapper.BadTypeMismatch);
|
||||
WarnTypeMismatchOnce(parent.Name,
|
||||
"AbCip whole-UDT read of {Parent} on device {Device}: the controller's type does not match " +
|
||||
"the authored type; {MemberCount} members report BadTypeMismatch (logged once per tag)",
|
||||
group.ParentName, parent.DeviceHostAddress, group.Members.Count);
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
if (AbCipRefusals.IsDeviceRefusal(status))
|
||||
{
|
||||
StampGroupStatus(group, results, now, AbCipStatusMapper.MapReadStatus((AbStatus)status));
|
||||
WarnRefusalOnce(parent.Name, "whole-UDT read", group.ParentName, parent.DeviceHostAddress, ((AbStatus)status).ToString());
|
||||
return ReadOutcome.Refused;
|
||||
}
|
||||
if (status != 0)
|
||||
{
|
||||
EvictRuntime(device, parent.Name);
|
||||
var mapped = AbCipStatusMapper.MapStatus(status);
|
||||
StampGroupStatus(group, results, now, mapped);
|
||||
_logger.LogWarning(
|
||||
"AbCip whole-UDT read returned non-zero transport status {Status} for parent {Parent} " +
|
||||
"on device {Device}; {MemberCount} member values stamped with mapped status",
|
||||
status, group.ParentName, parent.DeviceHostAddress, group.Members.Count);
|
||||
return ReadOutcome.Degraded($"transport status {status} reading UDT {group.ParentName}");
|
||||
}
|
||||
|
||||
foreach (var member in group.Members)
|
||||
{
|
||||
var value = runtime.DecodeValueAt(member.Definition.DataType, member.Offset, bitIndex: null);
|
||||
results[member.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now);
|
||||
foreach (var member in group.Members)
|
||||
{
|
||||
var value = runtime.DecodeValueAt(member.Definition.DataType, member.Offset, bitIndex: null);
|
||||
results[member.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now);
|
||||
}
|
||||
return ReadOutcome.Success;
|
||||
}
|
||||
finally
|
||||
{
|
||||
opLock.Release();
|
||||
}
|
||||
return ReadOutcome.Success;
|
||||
}
|
||||
finally
|
||||
{
|
||||
opLock.Release();
|
||||
window.Release();
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
|
||||
+1
-1
@@ -21,7 +21,7 @@
|
||||
<div class="col-md-3">
|
||||
<label class="form-label">Max concurrent reads</label>
|
||||
<InputNumber @bind-Value="_form.MaxConcurrentReads" @bind-Value:after="EmitAsync" class="form-control form-control-sm" />
|
||||
<div class="form-text">Read work items in flight at once. Default 8; 1 = sequential.</div>
|
||||
<div class="form-text">Read work items in flight at once. Default 8; 1 = sequential (libplctag). The native transport always uses at least 64 — it packs concurrent reads into one request.</div>
|
||||
</div>
|
||||
<div class="col-md-3">
|
||||
<div class="form-check form-switch mt-4">
|
||||
|
||||
Reference in New Issue
Block a user