diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index ed397b0a..351cb662 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -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); + /// Batched : one Lock 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 (create-on-first-write, #519 verbatim + /// status codes, ClearChangeMasks per node). + /// The value writes to apply, in order. + public void WriteValues(IReadOnlyList 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); + } } } + /// The body of a single value write. Caller MUST hold Lock — this is factored out + /// solely so can apply a whole batch under one acquisition; it is byte-identical + /// to what has always done inside its own lock. + /// The bare s= identifier of the variable. + /// The namespace index resolved from the write's realm. + /// The realm-qualified map key for . + /// The new value to write. + /// The full OPC UA status code, sub-code included. + /// The timestamp of the value in UTC. + 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); + } + /// /// Apply a full Part 9 alarm-condition write. When a real has /// been materialised for (via ), diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs index b2df9615..dcb135f5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs @@ -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); + /// + /// Overrides the interface's default implementation on purpose: that default loops + /// , which would re-take the node manager's SDK-wide Lock once per value. + /// Forwarding to the manager's own batched entry point takes it ONCE for the whole batch. + public void WriteValues(IReadOnlyList values) + => _nodeManager.WriteValues(values); + /// public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm) => _nodeManager.WriteAlarmCondition(alarmNodeId, state, sourceTimestampUtc, realm); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs new file mode 100644 index 00000000..ace33503 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs @@ -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; + +/// +/// Value-pipeline batching (Task 6) — applies a whole +/// batch under ONE acquisition of the manager-wide SDK Lock instead of one per value (the +/// per-value take was the steady-state throughput ceiling on the publish path). +/// These tests pin the observable 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 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 +/// Bad. +/// Boots a real through — the +/// same harness uses — so the assertions read the real SDK node +/// state an OPC UA client would receive. +/// +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}"); + + /// 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. + [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 + { + 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(); + } + + /// A batch item naming a NodeId that was never declared lazily creates the variable, exactly + /// as the single-value create-on-first-write path does — + /// and it does so per realm, so a Raw and a UNS item sharing an s= id create two nodes. + [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 + { + 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(); + } + + /// An empty batch neither throws nor touches the address space. + [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.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; + } + + /// Cleans up the PKI root directory. + public void Dispose() + { + if (Directory.Exists(_pkiRoot)) + { + try { Directory.Delete(_pkiRoot, recursive: true); } + catch { /* best-effort cleanup */ } + } + } +}