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

feat(opcuaserver): node manager batched WriteValues under one lock

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:13:46 -04:00
parent adebad711f
commit 8d7350ea73
3 changed files with 219 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+50 -15
View File
@@ -513,25 +513,60 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
var ns = NamespaceIndexForRealm(realm);
var key = MapKey(realm, nodeId);
lock (Lock)
{
// CreateVariable mutates the SDK address space (_root.AddChild + AddPredefinedNode),
// so it MUST run under Lock — the SDK's subscription/ConditionRefresh threads take it too.
if (!_variables.TryGetValue(key, out var variable))
{
variable = CreateVariable(nodeId, ns);
_variables[key] = variable;
}
WriteValueLocked(nodeId, ns, key, value, statusCode, sourceTimestampUtc);
}
variable.Value = value;
// #519 — the caller's status code reaches the SDK verbatim, sub-code included. This used to
// re-expand a 3-state enum, so every Bad became a bare StatusCodes.Bad (0x80000000) and a
// driver's BadNoCommunication was indistinguishable from BadTypeMismatch at the client.
variable.StatusCode = new StatusCode(statusCode);
variable.Timestamp = sourceTimestampUtc;
variable.ClearChangeMasks(SystemContext, includeChildren: false);
/// <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>
/// <param name="values">The value writes to apply, in order.</param>
public void WriteValues(IReadOnlyList<ValueWrite> values)
{
ArgumentNullException.ThrowIfNull(values);
if (values.Count == 0) return;
EnsureAddressSpaceCreated();
lock (Lock)
{
for (var i = 0; i < values.Count; i++)
{
var v = values[i];
WriteValueLocked(v.NodeId, NamespaceIndexForRealm(v.Realm), MapKey(v.Realm, v.NodeId),
v.Value, v.StatusCode, v.SourceTimestampUtc);
}
}
}
/// <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>
/// <param name="nodeId">The bare <c>s=</c> identifier of the variable.</param>
/// <param name="ns">The namespace index resolved from the write's realm.</param>
/// <param name="key">The realm-qualified map key for <paramref name="nodeId"/>.</param>
/// <param name="value">The new value to write.</param>
/// <param name="statusCode">The full OPC UA status code, sub-code included.</param>
/// <param name="sourceTimestampUtc">The timestamp of the value in UTC.</param>
private void WriteValueLocked(string nodeId, ushort ns, string key, object? value, uint statusCode, DateTime sourceTimestampUtc)
{
// CreateVariable mutates the SDK address space (_root.AddChild + AddPredefinedNode),
// so it MUST run under Lock — the SDK's subscription/ConditionRefresh threads take it too.
if (!_variables.TryGetValue(key, out var variable))
{
variable = CreateVariable(nodeId, ns);
_variables[key] = variable;
}
variable.Value = value;
// #519 — the caller's status code reaches the SDK verbatim, sub-code included. This used to
// re-expand a 3-state enum, so every Bad became a bare StatusCodes.Bad (0x80000000) and a
// driver's BadNoCommunication was indistinguishable from BadTypeMismatch at the client.
variable.StatusCode = new StatusCode(statusCode);
variable.Timestamp = sourceTimestampUtc;
// Stays INSIDE the lock, per node: this is how the SDK notifies monitored items of the change.
variable.ClearChangeMasks(SystemContext, includeChildren: false);
}
/// <summary>
/// Apply a full Part 9 alarm-condition write. When a real <see cref="AlarmConditionState"/> has
/// been materialised for <paramref name="alarmNodeId"/> (via <see cref="MaterialiseAlarmCondition"/>),
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs
+7
View File
@@ -24,6 +24,13 @@ public sealed class SdkAddressSpaceSink : IOpcUaAddressSpaceSink, ISurgicalAddre
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
=> _nodeManager.WriteValue(nodeId, value, statusCode, sourceTimestampUtc, realm);
/// <inheritdoc />
/// <remarks>Overrides the interface's default implementation on purpose: that default loops
/// <see cref="WriteValue"/>, which would re-take the node manager's SDK-wide <c>Lock</c> once per value.
/// Forwarding to the manager's own batched entry point takes it ONCE for the whole batch.</remarks>
public void WriteValues(IReadOnlyList<ValueWrite> values)
=> _nodeManager.WriteValues(values);
/// <inheritdoc />
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
=> _nodeManager.WriteAlarmCondition(alarmNodeId, state, sourceTimestampUtc, realm);
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs
+162
View File
@@ -0,0 +1,162 @@
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>
/// Value-pipeline batching (Task 6) — <see cref="OtOpcUaNodeManager.WriteValues"/> applies a whole
/// batch under ONE acquisition of the manager-wide SDK <c>Lock</c> instead of one per value (the
/// per-value take was the steady-state throughput ceiling on the publish path).
/// <para>These tests pin the <b>observable</b> contract rather than the locking: every item lands with
/// its own Value / StatusCode / Timestamp, an item for an undeclared NodeId is lazily created exactly
/// as <see cref="OtOpcUaNodeManager.WriteValue"/> does, and an empty batch does nothing at all.
/// A non-Good status is included so the #519 verbatim-sub-code passthrough is proven on the batched
/// path too — a batch that re-projected status through a 3-state enum would collapse it to a bare
/// <c>Bad</c>.</para>
/// <para>Boots a real <see cref="OtOpcUaSdkServer"/> through <see cref="OpcUaApplicationHost"/> — the
/// same harness <see cref="NodeManagerArrayTests"/> uses — so the assertions read the real SDK node
/// state an OPC UA client would receive.</para>
/// </summary>
public sealed class NodeManagerBatchWriteTests : IDisposable
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private readonly string _pkiRoot = Path.Combine(
Path.GetTempPath(),
$"otopcua-batchwrite-{Guid.NewGuid():N}");
/// <summary>Every item of a multi-item batch lands on its own node with its own value, status code
/// (including a non-Good sub-code) and source timestamp — no item is dropped, and none overwrites
/// another's state.</summary>
[Fact]
public async Task WriteValues_lands_every_value()
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
foreach (var name in new[] { "a", "b", "c" })
{
nm.EnsureVariable($"eq-1/{name}", parentFolderNodeId: null, displayName: name, dataType: "Double",
writable: false, realm: AddressSpaceRealm.Uns);
}
var t1 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
var t2 = new DateTime(2026, 8, 15, 10, 0, 1, DateTimeKind.Utc);
var t3 = new DateTime(2026, 8, 15, 10, 0, 2, DateTimeKind.Utc);
nm.WriteValues(new List<ValueWrite>
{
new("eq-1/a", 1.5d, OpcUaStatus.Good, t1, AddressSpaceRealm.Uns),
new("eq-1/b", null, OpcUaStatus.BadNoCommunication, t2, AddressSpaceRealm.Uns),
new("eq-1/c", 3.5d, 0x80740000u /* BadTypeMismatch */, t3, AddressSpaceRealm.Uns),
});
var a = nm.TryGetVariable("eq-1/a");
a.ShouldNotBeNull();
a!.Value.ShouldBe(1.5d);
a.StatusCode.ShouldBe((StatusCode)StatusCodes.Good);
a.Timestamp.ShouldBe(t1);
var b = nm.TryGetVariable("eq-1/b");
b.ShouldNotBeNull();
b!.Value.ShouldBeNull();
b.StatusCode.ShouldBe((StatusCode)StatusCodes.BadNoCommunication);
b.StatusCode.ShouldNotBe((StatusCode)StatusCodes.Bad,
"the batched path must hand the status sub-code to the SDK verbatim (#519)");
b.Timestamp.ShouldBe(t2);
var c = nm.TryGetVariable("eq-1/c");
c.ShouldNotBeNull();
c!.Value.ShouldBe(3.5d);
c.StatusCode.ShouldBe((StatusCode)StatusCodes.BadTypeMismatch);
c.Timestamp.ShouldBe(t3);
await host.DisposeAsync();
}
/// <summary>A batch item naming a NodeId that was never declared lazily creates the variable, exactly
/// as the single-value <see cref="OtOpcUaNodeManager.WriteValue"/> create-on-first-write path does —
/// and it does so per realm, so a Raw and a UNS item sharing an <c>s=</c> id create two nodes.</summary>
[Fact]
public async Task WriteValues_creates_missing_variables()
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
nm.TryGetVariable("eq-1/lazy").ShouldBeNull();
nm.TryGetVariable("Plant/dev1/lazy", AddressSpaceRealm.Raw).ShouldBeNull();
var ts = new DateTime(2026, 8, 15, 11, 0, 0, DateTimeKind.Utc);
nm.WriteValues(new List<ValueWrite>
{
new("eq-1/lazy", 42, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
new("Plant/dev1/lazy", 7, OpcUaStatus.Good, ts, AddressSpaceRealm.Raw),
});
var uns = nm.TryGetVariable("eq-1/lazy");
uns.ShouldNotBeNull();
uns!.Value.ShouldBe(42);
uns.StatusCode.ShouldBe((StatusCode)StatusCodes.Good);
uns.Timestamp.ShouldBe(ts);
var raw = nm.TryGetVariable("Plant/dev1/lazy", AddressSpaceRealm.Raw);
raw.ShouldNotBeNull();
raw!.Value.ShouldBe(7);
await host.DisposeAsync();
}
/// <summary>An empty batch neither throws nor touches the address space.</summary>
[Fact]
public async Task Empty_batch_is_a_no_op()
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
var before = nm.VariableCount;
Should.NotThrow(() => nm.WriteValues([]));
nm.VariableCount.ShouldBe(before);
await host.DisposeAsync();
}
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
{
var host = new OpcUaApplicationHost(
new OpcUaApplicationHostOptions
{
ApplicationName = "OtOpcUa.BatchWriteTest",
ApplicationUri = $"urn:OtOpcUa.BatchWriteTest:{Guid.NewGuid():N}",
OpcUaPort = AllocateFreePort(),
PublicHostname = "localhost",
PkiStoreRoot = _pkiRoot,
},
NullLogger<OpcUaApplicationHost>.Instance);
var server = new OtOpcUaSdkServer();
await host.StartAsync(server, Ct);
return (host, server);
}
private static int AllocateFreePort()
{
using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
listener.Start();
var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
listener.Stop();
return port;
}
/// <summary>Cleans up the PKI root directory.</summary>
public void Dispose()
{
if (Directory.Exists(_pkiRoot))
{
try { Directory.Delete(_pkiRoot, recursive: true); }
catch { /* best-effort cleanup */ }
}
}
}