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

perf(runtime): per-driver index for connectivity fans; comms-loss is one batch

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-16 04:27:20 -04:00
parent 9cf400f1e8
commit b6275f439c
3 changed files with 232 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+96 -24
View File
@@ -50,9 +50,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
/// maintains the <c>FullName→NodeId</c> live-value routing map, and forwards results to the
/// OPC UA publish actor via <c>ForwardBatchToMux</c> — one whole driver production burst per message,
/// and the ONLY value ingress from driver children (<see cref="DriverInstanceActor"/> wraps even a
/// single data change in an <c>AttributeValueBatchPublished</c>). The per-value
/// <c>AttributeValuePublished</c> survives only as the message <see cref="FanCommsLossToDependencyMux"/>
/// fabricates straight to the dependency mux; the host itself no longer receives it.
/// single data change in an <c>AttributeValueBatchPublished</c>). <see cref="FanCommsLossToDependencyMux"/>
/// fabricates a batch of the same shape straight to the dependency mux, so the host neither receives nor
/// sends the per-value <c>AttributeValuePublished</c> any more.
/// </summary>
public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedStash
{
@@ -284,6 +284,34 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// value-subscription set.</summary>
private readonly Dictionary<(string DriverInstanceId, string RawPath), HashSet<NodeRealmRef>> _alarmNodeIdByDriverRef = new();
/// <summary>
/// Secondary index over <see cref="_nodeIdByDriverRef"/>, grouping its entries by
/// <c>DriverInstanceId</c>: <c>DriverInstanceId → [(RawPath, the SAME node set)]</c>. Rebuilt whole in
/// the same <see cref="ApplyDesiredSubscriptions"/> pass that rebuilds the primary — never maintained
/// incrementally, so the two cannot drift.
/// <para>
/// It exists because the connectivity handlers are the only readers that need <i>one driver's</i>
/// entries, and the primary is keyed by the composite <c>(DriverInstanceId, RawPath)</c> — so
/// answering "which refs does this driver own?" meant scanning every entry, i.e. the whole node's
/// address space, for a single driver's disconnect. On a 50k-tag node a driver flap walked 50k
/// keys to find its own few thousand.
/// </para>
/// <para>
/// The node set is held BY REFERENCE, not copied: the entry is captured when the primary's set is
/// created, so any node later added to that set (a UNS reference registered in the same pass) is
/// visible here without a second bookkeeping step.
/// </para>
/// </summary>
private readonly Dictionary<string, List<(string RawPath, HashSet<NodeRealmRef> Nodes)>> _valueRefsByDriver =
new(StringComparer.Ordinal);
/// <summary>The alarm-side twin of <see cref="_valueRefsByDriver"/>: a secondary index over
/// <see cref="_alarmNodeIdByDriverRef"/> grouping its entries by <c>DriverInstanceId</c>, so the
/// connectivity handler annotates one driver's conditions without scanning every condition on the node.
/// Rebuilt whole in the same pass, and likewise holding each condition-node set by reference.</summary>
private readonly Dictionary<string, List<(string RawPath, HashSet<NodeRealmRef> Nodes)>> _alarmRefsByDriver =
new(StringComparer.Ordinal);
/// <summary>
/// Inverse of <see cref="_alarmNodeIdByDriverRef"/>: <c>folder-scoped condition NodeId →
/// (DriverInstanceId, FullName = alarm ConditionId/AlarmFullReference)</c>. Built in the SAME apply
@@ -1231,16 +1259,20 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
var statusCode = msg.Connected ? OpcUaStatus.Good : OpcUaStatus.BadNoCommunication;
var ts = DateTime.UtcNow;
var annotated = 0;
// Fan out to every condition this driver owns. _alarmNodeIdByDriverRef is keyed by
// (DriverInstanceId, RawPath); one driver ref can back several condition NodeIds (identical machines).
foreach (var ((driverId, _), nodeIds) in _alarmNodeIdByDriverRef)
// Fan out to every condition this driver owns, resolved through the per-driver index rather than by
// scanning every condition on the node: _alarmNodeIdByDriverRef is keyed by the composite
// (DriverInstanceId, RawPath), so a filtered walk of it costs the whole address space per flap. One
// driver ref can still back several condition NodeIds (identical machines) — hence the inner loop.
if (_alarmRefsByDriver.TryGetValue(msg.DriverInstanceId, out var alarmEntries))
{
if (driverId != msg.DriverInstanceId) continue;
foreach (var n in nodeIds)
foreach (var (_, nodeIds) in alarmEntries)
{
_opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AlarmQualityUpdate(
n.NodeId, statusCode, ts, n.Realm));
annotated++;
foreach (var n in nodeIds)
{
_opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AlarmQualityUpdate(
n.NodeId, statusCode, ts, n.Realm));
annotated++;
}
}
}
@@ -1258,7 +1290,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <para>
/// <b>The ref set already existed.</b> #481 asked for a driver → all-tag-refs map to be built;
/// <see cref="_nodeIdByDriverRef"/> is exactly that, keyed <c>(DriverInstanceId, RawPath)</c> and
/// rebuilt from the composition's <c>RawTags</c> ∪ <c>UnsReferenceVariables</c> on every apply.
/// rebuilt from the composition's <c>RawTags</c> ∪ <c>UnsReferenceVariables</c> on every apply —
/// read here through the per-driver <see cref="_valueRefsByDriver"/> index, since that composite
/// key cannot answer "this driver's refs" without walking the node's whole address space.
/// Alarm-bearing tags live in the separate <see cref="_alarmNodeIdByDriverRef"/> and are
/// deliberately absent here — a native condition is not a mux value, and its comms quality is
/// annotated by the caller through the Part 9 path instead.
@@ -1286,29 +1320,41 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// Ungated by redundancy role, for the same reason as the native-condition annotation above: a
/// Secondary must keep its scripted-alarm evaluation coherent for failover. The mux fan-out
/// reaches only refs something is actually subscribed to, so a driver whose tags feed no script
/// and no virtual tag costs one dictionary scan and no delivery.
/// and no virtual tag costs one dictionary lookup, one envelope, and no delivery.
/// </para>
/// <para>
/// <b>ONE batch, not one message per ref.</b> The mux's batch receive filters per item against the
/// same registered-interest map its per-value receive consulted, so what each subscriber is handed
/// is unchanged — but a driver owning 5k tags now costs one Tell on its disconnect edge instead of
/// 5k, on the actor that also carries the node's steady value ingress.
/// </para>
/// </remarks>
/// <param name="driverInstanceId">The driver child that just reported comms loss.</param>
private void FanCommsLossToDependencyMux(string driverInstanceId)
{
if (_dependencyMux is null) return;
// Indexed, not a filtered scan of the composite-keyed primary: this used to walk every value ref on
// the node to find one driver's.
if (!_valueRefsByDriver.TryGetValue(driverInstanceId, out var entries) || entries.Count == 0) return;
var ts = DateTime.UtcNow;
var pushed = 0;
foreach (var (driverId, rawPath) in _nodeIdByDriverRef.Keys)
var pushed = entries.Count;
// ONE envelope for the whole driver, not one Tell per ref: a driver owning 5k tags cost 5k messages
// on a single connectivity edge. The mux's batch path filters the items per registered interest
// exactly as it does a steady production burst, so the delivery is unchanged — only the framing is.
var items = new DriverInstanceActor.AttributeValueItem[pushed];
for (var i = 0; i < pushed; i++)
{
if (driverId != driverInstanceId) continue;
_dependencyMux.Tell(new DriverInstanceActor.AttributeValuePublished(
driverInstanceId, rawPath, Value: null, OpcUaStatus.BadNoCommunication, ts));
pushed++;
items[i] = new DriverInstanceActor.AttributeValueItem(
entries[i].RawPath, Value: null, OpcUaStatus.BadNoCommunication, ts);
}
if (pushed > 0)
_log.Debug(
"DriverHost {Node}: driver {Driver} disconnected — pushed BadNoCommunication for {Count} mux ref(s)",
_localNode, driverInstanceId, pushed);
_dependencyMux.Tell(new DriverInstanceActor.AttributeValueBatchPublished(driverInstanceId, items));
// Unconditional: the empty case returned above, so reaching here means at least one ref was pushed.
_log.Debug(
"DriverHost {Node}: driver {Driver} disconnected — pushed BadNoCommunication for {Count} mux ref(s) in one batch",
_localNode, driverInstanceId, pushed);
}
private void ForwardNativeAlarm(DriverInstanceActor.AttributeAlarmPublished msg)
@@ -2455,6 +2501,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
ApplyDesiredSubscriptions(composition);
}
/// <summary>Gets (creating on first use) the per-driver bucket of a secondary index.</summary>
/// <param name="index">Either <see cref="_valueRefsByDriver"/> or <see cref="_alarmRefsByDriver"/>.</param>
/// <param name="driverInstanceId">The driver whose bucket is wanted.</param>
/// <returns>The mutable bucket for that driver.</returns>
private static List<(string RawPath, HashSet<NodeRealmRef> Nodes)> IndexFor(
Dictionary<string, List<(string RawPath, HashSet<NodeRealmRef> Nodes)>> index, string driverInstanceId)
{
if (!index.TryGetValue(driverInstanceId, out var bucket))
index[driverInstanceId] = bucket = new List<(string, HashSet<NodeRealmRef>)>();
return bucket;
}
/// <summary>
/// Hand each running driver child its desired value + alarm subscription set for
/// <paramref name="composition"/>, rebuild the live-value/write routing maps, and forward the
@@ -2509,6 +2567,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_driverRefByAlarmNodeId.Clear();
_alarmMetaByNodeId.Clear();
_nativeAlarmProjector.Clear();
// The per-driver secondary indexes are rebuilt WHOLE alongside their primaries below — same pass,
// same clear — so no incremental maintenance exists to get out of step with them.
_valueRefsByDriver.Clear();
_alarmRefsByDriver.Clear();
foreach (var t in composition.RawTags)
{
var key = (t.DriverInstanceId, t.NodeId); // (DriverInstanceId, RawPath) — RawPath IS the wire-ref
@@ -2517,7 +2579,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// Native alarm → a single Part 9 condition at the RawPath (Raw realm). WP4 adds the referencing
// equipment notifier NodeIds; WP3 wires the single raw condition.
if (!_alarmNodeIdByDriverRef.TryGetValue(key, out var aset))
{
_alarmNodeIdByDriverRef[key] = aset = new HashSet<NodeRealmRef>();
// Indexed on CREATION only, so a repeated key contributes exactly one entry — and the set
// is captured by reference, so later additions to it are seen through the index.
IndexFor(_alarmRefsByDriver, t.DriverInstanceId).Add((t.NodeId, aset));
}
aset.Add(new NodeRealmRef(t.NodeId, AddressSpaceRealm.Raw));
// Inverse 1:1 map for the inbound native-condition ack path (keyed by BARE condition NodeId).
_driverRefByAlarmNodeId[t.NodeId] = key;
@@ -2532,7 +2600,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// Value tag: register the RAW NodeId (Raw realm) AND every referencing UNS NodeId (Uns realm)
// against the SAME driver ref — the single value source fans to all of them.
if (!_nodeIdByDriverRef.TryGetValue(key, out var set))
{
_nodeIdByDriverRef[key] = set = new HashSet<NodeRealmRef>();
IndexFor(_valueRefsByDriver, t.DriverInstanceId).Add((t.NodeId, set));
}
set.Add(new NodeRealmRef(t.NodeId, AddressSpaceRealm.Raw));
_driverRefByNodeId[(AddressSpaceRealm.Raw, t.NodeId)] = key;
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorCommsLossTests.cs
+130
View File
@@ -0,0 +1,130 @@
using Akka.Actor;
using Akka.TestKit;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using static ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness.V3ArtifactSeeder;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// The comms-loss fan (<c>DriverHostActor.FanCommsLossToDependencyMux</c>) end-to-end through a REAL
/// <see cref="DependencyMuxActor"/>, rather than the mux probe the other host suites use.
/// <para>
/// Two properties are pinned here that a probe cannot show. First, the fan is ONE
/// <see cref="DriverInstanceActor.AttributeValueBatchPublished"/> for the whole driver rather than one
/// Tell per ref — a driver owning 5k tags used to cost 5k messages on a single connectivity edge — and
/// the mux's batch path still delivers every item to an interested subscriber, so batching did not
/// quietly drop the delivery it exists to make.
/// </para>
/// <para>
/// Second, the driver scoping: the fan resolves its refs through a per-driver index built alongside
/// <c>_nodeIdByDriverRef</c> instead of scanning every entry in it, and an index is exactly the thing
/// that can key a lookup wrong. The two-driver case is therefore the falsification control — the
/// untouched driver's ref must be absent from the batch, which a scan-and-filter implementation and a
/// correctly-keyed index both satisfy but a mis-keyed one does not.
/// </para>
/// </summary>
public sealed class DriverHostActorCommsLossTests : RuntimeActorTestBase
{
private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-test");
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('c', 64));
/// <summary>Spawns a host whose dependency mux is a REAL <see cref="DependencyMuxActor"/>, applies the
/// seeded deployment, and registers <paramref name="subscriber"/>'s interest in <paramref name="refs"/>.
/// </summary>
/// <param name="db">The factory whose deployment was seeded by <see cref="SeedV3Deployment"/>.</param>
/// <param name="deploymentId">The deployment to dispatch.</param>
/// <param name="subscriber">The probe standing in for a VirtualTagActor / scripted-alarm subscriber.</param>
/// <param name="refs">The RawPaths that subscriber registers interest in.</param>
/// <returns>The applied host actor.</returns>
private IActorRef SpawnHostWithRealMux(
IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId deploymentId, TestProbe subscriber, params string[] refs)
{
var coordinator = CreateTestProbe();
var mux = Sys.ActorOf(DependencyMuxActor.Props());
mux.Tell(new DependencyMuxActor.RegisterInterest(refs, subscriber.Ref));
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" },
dependencyMux: mux,
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
virtualTagHostOverride: CreateTestProbe().Ref));
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
coordinator.ExpectMsg<ApplyAck>(PresenceBudget).Outcome.ShouldBe(ApplyAckOutcome.Applied);
return actor;
}
/// <summary>A driver's comms loss reaches the mux as ONE batch carrying a Bad item per ref it owns, and
/// the mux delivers that whole slice to an interested subscriber in ONE
/// <see cref="VirtualTagActor.DependencyValueBatch"/>.</summary>
[Fact]
public void Comms_loss_reaches_the_mux_as_one_batch_of_bad_items()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
var subscriber = CreateTestProbe();
var actor = SpawnHostWithRealMux(db, deploymentId, subscriber,
"Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp");
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false));
var batch = subscriber.ExpectMsg<VirtualTagActor.DependencyValueBatch>(PresenceBudget);
batch.Items.Select(i => i.TagId).ShouldBe(
new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" }, ignoreOrder: true);
foreach (var item in batch.Items)
{
item.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
item.Value.ShouldBeNull();
}
// ONE batch, not one per ref: the whole point of the change. A per-ref fan would land a second
// single-item DependencyValueBatch here.
subscriber.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
/// <summary>Falsification control for the per-driver index: with TWO drivers configured, one driver's
/// comms loss carries ONLY its own refs — the other driver's tag is untouched.</summary>
[Fact]
public void Comms_loss_batch_carries_only_the_dead_drivers_refs()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
(Driver: "drv-2", DriverName: "Modbus2", Device: "dev2", Tag: "flow", DataType: "Double", Writable: false),
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
var subscriber = CreateTestProbe();
var actor = SpawnHostWithRealMux(db, deploymentId, subscriber,
"Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp", "Plant/Modbus2/dev2/flow");
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false));
var batch = subscriber.ExpectMsg<VirtualTagActor.DependencyValueBatch>(PresenceBudget);
batch.Items.Select(i => i.TagId).ShouldBe(
new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" }, ignoreOrder: true);
subscriber.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorLiveValueTests.cs
+6 -7
View File
@@ -160,7 +160,8 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
// went on evaluating it. These pin the synthetic Bad that closes that hole.
/// <summary>Comms loss pushes a Bad for every value ref the driver owns — not one, and not the
/// alarm-bearing ones (those take the Part 9 quality path instead).</summary>
/// alarm-bearing ones (those take the Part 9 quality path instead) — as ONE batch envelope rather than
/// one Tell per ref.</summary>
[Fact]
public void Driver_comms_loss_pushes_BadNoCommunication_into_the_mux_for_every_ref_it_owns()
{
@@ -177,11 +178,10 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false));
var pushes = new[]
{
mux.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(PresenceBudget),
mux.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(PresenceBudget),
};
// ONE envelope for the whole driver: a driver owning 5k tags cost 5k Tells before this was batched.
var fan = mux.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
fan.DriverInstanceId.ShouldBe("drv-1");
var pushes = fan.Items;
pushes.Select(p => p.FullReference).ShouldBe(
new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" }, ignoreOrder: true);
@@ -194,7 +194,6 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
// Null rather than the last-known value: the honest statement is "no current value". The
// engine's AreInputsReady gate holds the predicate on any Bad, so no script sees the null.
p.Value.ShouldBeNull();
p.DriverInstanceId.ShouldBe("drv-1");
}
mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300));