perf(runtime): Direct-mode apply threads the in-hand artifact (1 SQL read per apply, not 3)
This commit is contained in:
Joseph Doherty
@@ -1752,10 +1752,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
|
||||
// Trigger the OPC UA address-space rebuild so the local SDK reflects the new
|
||||
// composition. The publish actor handles the load-compose-diff-apply pipeline; we
|
||||
// just forward the same correlation id so the audit trail joins up.
|
||||
_opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId));
|
||||
//
|
||||
// The already-read bytes ride along. ReconcileDrivers just read this exact blob out of central
|
||||
// SQL and handed it back (it is non-null here — the null case returned above), so a bytes-less
|
||||
// RebuildAddressSpace would make the publish actor re-read the same immutable row, and the
|
||||
// SubscribeBulk pass would read it a THIRD time: three round-trips per apply for one blob that
|
||||
// cannot have changed between them. Threading it is also the shape the FetchAndCache apply has
|
||||
// always had (HandleFetchedForApply), for the harder reason that it has no ConfigDb to re-read.
|
||||
// Consequence: OpcUaPublishActor.LoadArtifact / LoadLatestArtifact are no longer on any apply
|
||||
// path — they remain only for a legacy/dev caller that Tells RebuildAddressSpace without bytes.
|
||||
_opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId, appliedBlob));
|
||||
// SubscribeBulk pass: hand each driver its desired tag references so live values flow into
|
||||
// the just-rebuilt address space instead of staying BadWaitingForInitialData.
|
||||
PushDesiredSubscriptions(deploymentId);
|
||||
PushDesiredSubscriptionsFromArtifact(deploymentId, appliedBlob);
|
||||
CacheAppliedArtifact(deploymentId, revision, appliedBlob);
|
||||
// Reaching here means the artifact came from central, so the node is no longer serving a
|
||||
// cache-sourced configuration. Note this only clears on a real apply: a dispatch of the
|
||||
@@ -1987,10 +1996,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
|
||||
/// the OPC UA sink and resume after reconnects. Drivers with no configured tags get an empty set
|
||||
/// (which clears any stale subscription from a previous deployment).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The <b>bootstrap-restore</b> entry point only. Neither apply path reaches it any more: both
|
||||
/// <see cref="ApplyAndAck"/> and <see cref="HandleFetchedForApply"/> already hold the artifact bytes
|
||||
/// and call <see cref="PushDesiredSubscriptionsFromArtifact"/> directly, so this ConfigDb read exists
|
||||
/// for <see cref="RestoreApplied"/>, which restores from a deployment id recovered out of
|
||||
/// NodeDeploymentState and may hold no blob of its own.
|
||||
/// </remarks>
|
||||
private void PushDesiredSubscriptions(DeploymentId deploymentId)
|
||||
{
|
||||
// Phase 4 belt-and-suspenders: Direct-only read (the FetchAndCache path calls
|
||||
// PushDesiredSubscriptionsFromArtifact with bytes in hand). No ConfigDb ⇒ nothing to load.
|
||||
// Phase 4 belt-and-suspenders: a ConfigDb read, so it can only be the restore path on a node that
|
||||
// has one (every apply path calls PushDesiredSubscriptionsFromArtifact with bytes in hand).
|
||||
// No ConfigDb ⇒ nothing to load.
|
||||
if (_dbFactory is null)
|
||||
{
|
||||
_log.Warning("DriverHost {Node}: no ConfigDb to load artifact for SubscribeBulk ({Id}); skipping", _localNode, deploymentId);
|
||||
|
||||
@@ -8,6 +8,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
|
||||
@@ -199,6 +200,47 @@ public sealed class DriverHostActorTests : RuntimeActorTestBase
|
||||
plan.PredicateSource.ShouldContain("ctx.GetTag");
|
||||
}
|
||||
|
||||
/// <summary>Verifies that a Direct-mode apply hands the artifact bytes it has already read to the
|
||||
/// address-space rebuild instead of leaving the publish actor to re-read the same row from central SQL.</summary>
|
||||
/// <remarks>
|
||||
/// The Direct apply reads <c>Deployment.ArtifactBlob</c> once (<c>ReconcileDrivers</c> returns it as
|
||||
/// <c>appliedBlob</c>) and then used to Tell <c>RebuildAddressSpace</c> with NO bytes — so
|
||||
/// <c>OpcUaPublishActor.HandleRebuild</c> re-read the row, and <c>PushDesiredSubscriptions</c> read it
|
||||
/// a third time: three central-SQL reads of one immutable blob per apply. Carrying the in-hand bytes
|
||||
/// collapses that to one. The FetchAndCache path has always been shaped this way, because it has no
|
||||
/// ConfigDb to re-read from — this asserts Direct mode now matches it. Byte-equality (rather than
|
||||
/// merely non-null) is the assertion that matters: a rebuild fed the WRONG artifact would materialise
|
||||
/// a configuration nobody deployed.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public void Direct_mode_apply_hands_the_already_read_artifact_to_the_rebuild()
|
||||
{
|
||||
var db = NewInMemoryDbFactory();
|
||||
var deploymentId = V3ArtifactSeeder.SeedV3Deployment(db, RevA,
|
||||
rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) },
|
||||
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
|
||||
|
||||
byte[] seeded;
|
||||
using (var seedCtx = db.CreateDbContext())
|
||||
seeded = seedCtx.Deployments.Single(d => d.DeploymentId == deploymentId.Value).ArtifactBlob!;
|
||||
seeded.Length.ShouldBeGreaterThan(0);
|
||||
|
||||
var coordinator = CreateTestProbe();
|
||||
var publish = CreateTestProbe();
|
||||
var actor = Sys.ActorOf(DriverHostActor.Props(
|
||||
db, TestNode, coordinator.Ref,
|
||||
localRoles: new HashSet<string> { "driver" },
|
||||
opcUaPublishActor: publish.Ref));
|
||||
|
||||
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
|
||||
coordinator.ExpectMsg<ApplyAck>(PresenceBudget).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
||||
|
||||
var rebuild = publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(PresenceBudget);
|
||||
rebuild.DeploymentId.ShouldBe(deploymentId);
|
||||
rebuild.Artifact.ShouldNotBeNull();
|
||||
rebuild.Artifact.ShouldBe(seeded);
|
||||
}
|
||||
|
||||
private static DeploymentId SeedDeploymentWithScriptedAlarm(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> db, RevisionHash rev)
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user