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

fix(opcuaserver): chunked, per-item-isolated WriteValues with typed partial-failure outcome

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:32:44 -04:00
parent 8ef02506f7
commit 6bf1fb3754
7 changed files with 468 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+91 -12
View File
@@ -503,8 +503,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// </summary>
/// <param name="nodeId">The node identifier of the variable.</param>
/// <param name="value">The new value to write.</param>
/// <param name="quality">The OPC UA quality status code.</param>
/// <param name="statusCode">The full OPC UA status code, sub-code included (#519 — passed to the SDK verbatim).</param>
/// <param name="sourceTimestampUtc">The timestamp of the value in UTC.</param>
/// <param name="realm">Which of the two v3 namespaces the <paramref name="nodeId"/> lives in; the
/// namespace index is resolved from this rather than parsed out of the id string.</param>
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
{
ArgumentException.ThrowIfNullOrEmpty(nodeId);
@@ -516,28 +518,105 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
WriteValueLocked(nodeId, ns, key, value, statusCode, sourceTimestampUtc);
}
/// <summary>Batched <see cref="WriteValue"/>: one <c>Lock</c> acquisition for the whole batch instead of
/// one per value — the publish path's per-value lock take was the steady-state throughput ceiling.
/// Per-item semantics are identical to <see cref="WriteValue"/> (create-on-first-write, #519 verbatim
/// status codes, ClearChangeMasks per node).</summary>
/// <summary>The most items <see cref="WriteValues"/> applies under a SINGLE <c>Lock</c> acquisition before
/// releasing and re-taking it.
/// <para>WHY a cap at all: the lock take is what batching amortises, and the amortisation curve is
/// brutally front-loaded — going from one take per value to one per 512 already removes &gt;99.8% of the
/// takes, so an unbounded hold buys essentially nothing more. What it DOES buy is a stall: the manager-wide
/// <c>Lock</c> is the same lock every SDK service thread (Read / Browse / subscription publish /
/// ConditionRefresh) contends for, and the first poll cycle force-raises EVERY tag (the initial-data push),
/// so batch #1 is a whole device × its UNS fan-out — thousands of <c>CreateVariable</c> calls, each mutating
/// the SDK address space, under one acquisition. 512 is therefore "effectively unbounded for the win,
/// bounded for the stall": chunking hands the SDK a scheduling point between chunks.</para></summary>
private const int MaxWritesPerLockAcquisition = 512;
/// <summary>Batched <see cref="WriteValue"/>: one <c>Lock</c> acquisition per chunk of
/// <see cref="MaxWritesPerLockAcquisition"/> items instead of one per value — the publish path's per-value
/// lock take was the steady-state throughput ceiling. Per-item semantics are identical to
/// <see cref="WriteValue"/> (create-on-first-write, #519 verbatim status codes, ClearChangeMasks per node).
/// <para><b>Items are isolated.</b> A bad item (empty NodeId, or a throw out of the SDK) does not abandon
/// the rest of the batch: it is recorded and the loop continues, and the aggregate surfaces afterwards as
/// a <see cref="ValueWriteBatchException"/>. Dropping the tail would be silent and durable — the poll
/// engine updates its last-value cache BEFORE raising, so a dropped item is not re-raised until that tag's
/// value changes again, leaving the node stale indefinitely.</para></summary>
/// <param name="values">The value writes to apply, in order.</param>
/// <exception cref="ValueWriteBatchException">When one or more items failed. Every other item HAS been
/// applied; the exception carries the first failing NodeId plus the failed/succeeded split so the caller
/// can log and meter honestly.</exception>
public void WriteValues(IReadOnlyList<ValueWrite> values)
{
ArgumentNullException.ThrowIfNull(values);
if (values.Count == 0) return;
// Guard BEFORE the empty-batch early-return, so a pre-start call fails the same legible way every
// other mutator does regardless of what it happens to carry.
EnsureAddressSpaceCreated();
if (values.Count == 0) return;
lock (Lock)
// Pre-pass, OUTSIDE the lock: validate each item and resolve its namespace index + map key into a
// scratch array. Both jobs belong here — validating outside keeps a malformed item from ever reaching
// the SDK, and resolving outside moves the MapKey string allocation (a NodeId + ToString per item)
// off the critical section entirely.
var resolved = new ResolvedWrite[values.Count];
string? firstFailedNodeId = null;
Exception? firstFailure = null;
var failed = 0;
for (var i = 0; i < values.Count; i++)
{
for (var i = 0; i < values.Count; i++)
var nodeId = values[i].NodeId;
if (string.IsNullOrEmpty(nodeId))
{
var v = values[i];
WriteValueLocked(v.NodeId, NamespaceIndexForRealm(v.Realm), MapKey(v.Realm, v.NodeId),
v.Value, v.StatusCode, v.SourceTimestampUtc);
failed++;
firstFailedNodeId ??= nodeId ?? string.Empty;
firstFailure ??= new ArgumentException(
$"WriteValues item {i} has a null or empty NodeId.", nameof(values));
continue;
}
resolved[i] = new ResolvedWrite(
nodeId, NamespaceIndexForRealm(values[i].Realm), MapKey(values[i].Realm, nodeId), IsValid: true);
}
var succeeded = 0;
for (var start = 0; start < values.Count; start += MaxWritesPerLockAcquisition)
{
var end = Math.Min(start + MaxWritesPerLockAcquisition, values.Count);
lock (Lock)
{
for (var i = start; i < end; i++)
{
if (!resolved[i].IsValid) continue;
var v = values[i];
try
{
WriteValueLocked(resolved[i].NodeId, resolved[i].NamespaceIndex, resolved[i].Key,
v.Value, v.StatusCode, v.SourceTimestampUtc);
succeeded++;
}
catch (Exception ex)
{
failed++;
firstFailedNodeId ??= v.NodeId;
firstFailure ??= ex;
}
}
}
}
if (failed > 0)
{
throw new ValueWriteBatchException(
firstFailedNodeId ?? string.Empty, failed, succeeded, firstFailure!);
}
}
/// <summary>One <see cref="WriteValues"/> item after the outside-the-lock pre-pass: its namespace index and
/// map key already resolved, or <paramref name="IsValid"/> false when the item failed validation.</summary>
/// <param name="NodeId">The item's bare <c>s=</c> identifier.</param>
/// <param name="NamespaceIndex">The namespace index resolved from the item's realm.</param>
/// <param name="Key">The realm-qualified map key.</param>
/// <param name="IsValid">False when the item failed the pre-pass and must be skipped.</param>
private readonly record struct ResolvedWrite(string NodeId, ushort NamespaceIndex, string Key, bool IsValid);
/// <summary>The body of a single value write. <b>Caller MUST hold <c>Lock</c></b> — this is factored out
/// solely so <see cref="WriteValues"/> can apply a whole batch under one acquisition; it is byte-identical
/// to what <see cref="WriteValue"/> has always done inside its own <c>lock</c>.</summary>
@@ -2647,7 +2726,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <summary>Guard the address-space mutators against a too-early call. <c>_root</c>
/// is only assigned in <see cref="CreateAddressSpace"/>, which the SDK invokes during
/// <c>StandardServer</c> start; every public mutator (<see cref="WriteValue"/>,
/// <c>StandardServer</c> start; every public mutator (<see cref="WriteValue"/>, <see cref="WriteValues"/>,
/// <see cref="WriteAlarmCondition"/>, <see cref="EnsureFolder"/>, <see cref="EnsureVariable"/>,
/// <see cref="MaterialiseAlarmCondition"/>) and <see cref="ResolveParentFolder"/> assume it has run.
/// If one is invoked before the server has started (a sink wired or a publish replayed before
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
+15 -1
View File
@@ -339,7 +339,12 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// <summary>Mirrors <see cref="HandleAttributeUpdate"/> for a whole burst: ONE
/// <see cref="IOpcUaAddressSpaceSink.WriteValues"/> call, but the write tally and the
/// <c>otopcua.opcua.sink.write</c> meter still advance PER ITEM — batching is a transport change, so
/// dashboards must not drop by the batch factor.</summary>
/// dashboards must not drop by the batch factor.
/// <para>A <see cref="ValueWriteBatchException"/> is a PARTIAL failure: the sink applied every item it
/// could and is telling us which one it couldn't. So it is caught ahead of the general catch to (a) name
/// the offending node in the log — a bare "the batch threw" cannot be acted on — and (b) still count the
/// items that DID land, keeping the per-item counting invariant honest instead of silently under-reporting
/// a whole batch because one item was bad.</para></summary>
private void HandleAttributeBatchUpdate(AttributeValueBatchUpdate msg)
{
try
@@ -348,6 +353,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
Interlocked.Add(ref _writes, msg.Items.Count);
OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair<string, object?>("kind", "value"));
}
catch (ValueWriteBatchException ex)
{
_log.Warning(ex,
"OpcUaPublish: sink.WriteValues partially failed for a {Count}-item batch — {Failed} failed " +
"(first: {Node}), {Succeeded} applied",
msg.Items.Count, ex.FailedCount, ex.FirstFailedNodeId, ex.SucceededCount);
Interlocked.Add(ref _writes, ex.SucceededCount);
OtOpcUaTelemetry.OpcUaSinkWrite.Add(ex.SucceededCount, new KeyValuePair<string, object?>("kind", "value"));
}
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: sink.WriteValues threw for a {Count}-item batch", msg.Items.Count);