perf(adminui): pooled DbContext factory; role-grants reads per-call, not per-circuit
This commit is contained in:
@@ -240,6 +240,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
|
||||
_maxWriteAttempts = maxWriteAttempts > 0 ? maxWriteAttempts : 1;
|
||||
|
||||
ReceiveAsync<VirtualTagActor.DependencyValueChanged>(OnValueChangedAsync);
|
||||
ReceiveAsync<VirtualTagActor.DependencyValueBatch>(OnValueBatchAsync);
|
||||
Receive<UpdateHistorizedRefs>(OnUpdateHistorizedRefs);
|
||||
Receive<DrainTick>(_ => OnDrainTick());
|
||||
Receive<DrainResult>(OnDrainResult);
|
||||
@@ -275,6 +276,30 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
|
||||
base.PostStop();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The mux's batched delivery: one message carrying this recorder's whole matched slice of a driver
|
||||
/// burst. Each item is appended by the existing per-value handler, in order — the saving here is the
|
||||
/// mailbox hop, not the append (a genuinely batched append is a later change).
|
||||
/// <para>
|
||||
/// ⚠️ The await below deliberately does NOT use <c>ConfigureAwait(false)</c>. <see
|
||||
/// cref="OnValueChangedAsync"/> reads <c>Self</c> before its own first await, so it must start
|
||||
/// each item ON the actor thread; a context-dropping await here would leave item 2 onward with
|
||||
/// no ActorContext and throw <c>NotSupportedException</c> — the Gitea #491 failure, which
|
||||
/// supervision turns into a restart that silently discards every historized ref. A plain await
|
||||
/// resumes on Akka's ActorTaskScheduler context, so the context is restored before the next
|
||||
/// iteration.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
/// <param name="msg">The batch of matched dependency values.</param>
|
||||
private async Task OnValueBatchAsync(VirtualTagActor.DependencyValueBatch msg)
|
||||
{
|
||||
IReadOnlyList<VirtualTagActor.DependencyValueChanged> items = msg.Items;
|
||||
for (int i = 0; i < items.Count; i++)
|
||||
{
|
||||
await OnValueChangedAsync(items[i]);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task OnValueChangedAsync(VirtualTagActor.DependencyValueChanged msg)
|
||||
{
|
||||
// The mux fans values keyed by the driver ref (msg.TagId == the mux ref). Only historize refs we
|
||||
|
||||
@@ -201,6 +201,10 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
Receive<ApplyScriptedAlarms>(OnApply);
|
||||
Receive<AlarmsLoaded>(OnAlarmsLoaded);
|
||||
Receive<VirtualTagActor.DependencyValueChanged>(OnDependencyChanged);
|
||||
// The mux's batched delivery: one message carrying this host's whole matched slice of a driver
|
||||
// burst. The engine upstream is fed per value, so push each item in order — the saving is the
|
||||
// mailbox hop, not the push.
|
||||
Receive<VirtualTagActor.DependencyValueBatch>(OnDependencyBatch);
|
||||
Receive<EngineEmission>(OnEngineEmission);
|
||||
// Cluster redundancy snapshots (published on the `redundancy-state` topic, subscribed in PreStart)
|
||||
// cache this node's role so OnEngineEmission can gate the cluster-wide alerts publish to the Primary.
|
||||
@@ -395,6 +399,17 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
||||
new DataValueSnapshot(msg.Value, msg.StatusCode, msg.TimestampUtc, msg.TimestampUtc));
|
||||
}
|
||||
|
||||
/// <summary>One batched mux delivery — every item already matched this host's registered interest.
|
||||
/// Loops the per-value handler, preserving in-burst order.</summary>
|
||||
private void OnDependencyBatch(VirtualTagActor.DependencyValueBatch msg)
|
||||
{
|
||||
var items = msg.Items;
|
||||
for (var i = 0; i < items.Count; i++)
|
||||
{
|
||||
OnDependencyChanged(items[i]);
|
||||
}
|
||||
}
|
||||
|
||||
private void OnEngineEmission(EngineEmission msg)
|
||||
{
|
||||
var e = msg.Event;
|
||||
|
||||
@@ -45,6 +45,7 @@ public sealed class DependencyConsumerMuxAdapter : ReceiveActor
|
||||
_mux = mux ?? throw new ArgumentNullException(nameof(mux));
|
||||
|
||||
Receive<VirtualTagActor.DependencyValueChanged>(OnDependencyValueChanged);
|
||||
Receive<VirtualTagActor.DependencyValueBatch>(OnDependencyValueBatch);
|
||||
Receive<ReRegister>(_ => RegisterInterest());
|
||||
}
|
||||
|
||||
@@ -68,4 +69,16 @@ public sealed class DependencyConsumerMuxAdapter : ReceiveActor
|
||||
// Quality is stripped at the mux boundary (value + timestamp only), so forward Good (0).
|
||||
_consumer.OnDependencyValue(msg.TagId, msg.Value, statusCode: 0u, msg.TimestampUtc);
|
||||
}
|
||||
|
||||
/// <summary>The mux's batched delivery: one message carrying this adapter's whole matched slice of a
|
||||
/// driver burst. The consumer contract is per value, so dispatch each item in order — the saving is the
|
||||
/// mailbox hop, not the consumer call.</summary>
|
||||
private void OnDependencyValueBatch(VirtualTagActor.DependencyValueBatch msg)
|
||||
{
|
||||
var items = msg.Items;
|
||||
for (var i = 0; i < items.Count; i++)
|
||||
{
|
||||
OnDependencyValueChanged(items[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,9 +8,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
/// Per-node fan-out router from <see cref="DriverInstanceActor.AttributeValuePublished"/>
|
||||
/// to interested <see cref="VirtualTagActor"/> instances. VirtualTagActor sends
|
||||
/// <see cref="RegisterInterest"/> on start-up listing the tag refs it depends on; the mux
|
||||
/// keeps a map of <c>tagRef → Set<IActorRef></c> and on every AttributeValuePublished
|
||||
/// forwards a <see cref="VirtualTagActor.DependencyValueChanged"/> to each interested
|
||||
/// subscriber.
|
||||
/// keeps a map of <c>tagRef → Set<IActorRef></c> and on every ingress forwards each
|
||||
/// interested subscriber ONE <see cref="VirtualTagActor.DependencyValueBatch"/> carrying that
|
||||
/// subscriber's matched slice of the burst.
|
||||
///
|
||||
/// DriverHostActor forwards driver-child value publishes to this mux (one mux per driver-role
|
||||
/// node). The ONLY driver ingress is the batched
|
||||
@@ -18,8 +18,11 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
/// production burst, which the mux filters per item. The per-value
|
||||
/// <see cref="DriverInstanceActor.AttributeValuePublished"/> receive stays for the comms-loss fan
|
||||
/// ALONE: <c>DriverHostActor.FanCommsLossToDependencyMux</c> fabricates that message and Tells this
|
||||
/// mux directly, never through the host's own receive. Both
|
||||
/// produce the same <see cref="VirtualTagActor.DependencyValueChanged"/> per interested subscriber.
|
||||
/// mux directly, never through the host's own receive.
|
||||
///
|
||||
/// <b>The mux emits ONLY batches</b> (the comms-loss fan as a single-item one), so a subscriber's
|
||||
/// batch receive is its complete hot path. Every subscriber keeps a per-value
|
||||
/// <see cref="VirtualTagActor.DependencyValueChanged"/> receive as well, for direct senders and tests.
|
||||
/// The mux is deliberately not a DPS subscriber — virtual-tag evaluation is local to each node and
|
||||
/// would over-emit if it spanned the cluster.
|
||||
/// </summary>
|
||||
@@ -110,14 +113,20 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
var dep = new VirtualTagActor.DependencyValueChanged(msg.FullReference, msg.Value, msg.TimestampUtc, msg.StatusCode);
|
||||
foreach (var sub in subscribers)
|
||||
{
|
||||
sub.Tell(dep);
|
||||
// A single-item batch, so the mux emits exactly ONE message shape and every subscriber needs
|
||||
// only the batch receive on its hot path. One list per subscriber: batch ownership transfers
|
||||
// to the receiver, so a shared list must never cross mailboxes. This is the comms-loss fan
|
||||
// (one ref, on a driver disconnect), not the steady value path — the extra allocation is noise.
|
||||
sub.Tell(new VirtualTagActor.DependencyValueBatch(new[] { dep }));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Filters one batched driver production burst per item — the batched analogue of
|
||||
/// <see cref="OnAttributeValuePublished"/>. Items nobody registered for are dropped (the common case:
|
||||
/// only a fraction of an address space feeds virtual-tag expressions), so a burst of N values costs one
|
||||
/// message plus one DependencyValueChanged per genuinely interested (item, subscriber) pair.
|
||||
/// <summary>Filters one batched driver production burst per item and delivers each subscriber's matched
|
||||
/// slice as ONE <see cref="VirtualTagActor.DependencyValueBatch"/>. Items nobody registered for are
|
||||
/// dropped (the common case: only a fraction of an address space feeds virtual-tag expressions), so a
|
||||
/// burst of N values costs one inbound message plus one outbound message per genuinely interested
|
||||
/// SUBSCRIBER — not per (item, subscriber) pair as it did when this fanned out per value. That is what
|
||||
/// lets <see cref="VirtualTagActor"/> evaluate its script once per burst instead of once per input.
|
||||
/// <para>The registered set is NOT just virtual-tag inputs: <c>ContinuousHistorizationRecorder</c>
|
||||
/// registers interest in every historized ref, so on a historizing node <see cref="_byRef"/> can be
|
||||
/// large and a sizeable fraction of each burst resolves.</para></summary>
|
||||
@@ -126,6 +135,10 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
// Nobody has registered anything — skip the whole burst rather than probing an empty map per item.
|
||||
if (_byRef.Count == 0) return;
|
||||
|
||||
// Allocated lazily on the first matched item: a burst in which nothing is registered for — still
|
||||
// the common case on a node with a handful of virtual tags — allocates nothing at all.
|
||||
Dictionary<IActorRef, List<VirtualTagActor.DependencyValueChanged>>? perSubscriber = null;
|
||||
|
||||
// Indexed for-loop, not foreach: this is the hot value ingress and an IReadOnlyList<T> enumerator
|
||||
// would allocate once per burst.
|
||||
var count = msg.Items.Count;
|
||||
@@ -134,7 +147,25 @@ public sealed class DependencyMuxActor : ReceiveActor
|
||||
var item = msg.Items[i];
|
||||
if (!_byRef.TryGetValue(item.FullReference, out var subscribers) || subscribers.Count == 0) continue;
|
||||
var dep = new VirtualTagActor.DependencyValueChanged(item.FullReference, item.Value, item.TimestampUtc, item.StatusCode);
|
||||
foreach (var sub in subscribers) sub.Tell(dep);
|
||||
perSubscriber ??= new Dictionary<IActorRef, List<VirtualTagActor.DependencyValueChanged>>();
|
||||
foreach (var sub in subscribers)
|
||||
{
|
||||
if (!perSubscriber.TryGetValue(sub, out var list))
|
||||
{
|
||||
list = new List<VirtualTagActor.DependencyValueChanged>();
|
||||
perSubscriber[sub] = list;
|
||||
}
|
||||
list.Add(dep);
|
||||
}
|
||||
}
|
||||
|
||||
if (perSubscriber is null) return;
|
||||
|
||||
// One message per subscriber, carrying that subscriber's own list. The lists are built here and
|
||||
// never touched again, so handing ownership to the receiver is safe.
|
||||
foreach (var (sub, list) in perSubscriber)
|
||||
{
|
||||
sub.Tell(new VirtualTagActor.DependencyValueBatch(list));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,6 +37,13 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
public sealed record DependencyValueChanged(
|
||||
string TagId, object? Value, DateTime TimestampUtc, uint StatusCode = OpcUaStatus.Good);
|
||||
|
||||
/// <summary>
|
||||
/// One subscriber's slice of one upstream burst: every item in this batch matched
|
||||
/// this subscriber's registered interest set. Never empty. List ownership transfers
|
||||
/// to the receiver (the mux allocates a fresh list per subscriber per burst).
|
||||
/// </summary>
|
||||
public sealed record DependencyValueBatch(IReadOnlyList<DependencyValueChanged> Items);
|
||||
|
||||
/// <summary>Re-assert the last emitted value/quality to the parent bridge, bypassing the value dedup.
|
||||
/// Sent by <see cref="VirtualTagHostActor"/> on every apply (deploy) to a surviving child: a deploy
|
||||
/// re-materialises the child's published UNS node to <c>BadWaitingForInitialData</c>, but an
|
||||
@@ -132,6 +139,7 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
_telemetryHub = telemetryHub;
|
||||
|
||||
Receive<DependencyValueChanged>(OnDependencyChanged);
|
||||
Receive<DependencyValueBatch>(OnDependencyBatch);
|
||||
Receive<ReassertValue>(_ => OnReassertValue());
|
||||
}
|
||||
|
||||
@@ -179,7 +187,41 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
private void OnDependencyChanged(DependencyValueChanged msg)
|
||||
{
|
||||
_dependencies[msg.TagId] = msg.Value;
|
||||
Evaluate(msg.TimestampUtc);
|
||||
}
|
||||
|
||||
/// <summary>Applies EVERY item in the batch to the dependency map first, then evaluates the script
|
||||
/// ONCE. This is the whole point of the batched mux delivery: a burst that moves K of this tag's inputs
|
||||
/// used to arrive as K messages and cost K script evaluations, K-1 of them against a half-updated input
|
||||
/// set whose intermediate results nobody wants (and which the value dedup could not suppress, since each
|
||||
/// intermediate value usually differs). Deliberately NOT a loop over
|
||||
/// <see cref="OnDependencyChanged"/>.
|
||||
/// <para>The single evaluation is stamped with the LAST item's timestamp — the newest sample in the
|
||||
/// burst, and the same stamp the final evaluation of the old per-item fan-out carried.</para></summary>
|
||||
private void OnDependencyBatch(DependencyValueBatch msg)
|
||||
{
|
||||
var items = msg.Items;
|
||||
var count = items.Count;
|
||||
// The mux never sends an empty batch; guard anyway rather than evaluate on no new input.
|
||||
if (count == 0) return;
|
||||
|
||||
// Indexed for-loop, not foreach: this is the hot value ingress and an IReadOnlyList<T> enumerator
|
||||
// would allocate once per burst.
|
||||
for (var i = 0; i < count; i++)
|
||||
{
|
||||
var item = items[i];
|
||||
_dependencies[item.TagId] = item.Value;
|
||||
}
|
||||
|
||||
Evaluate(items[count - 1].TimestampUtc);
|
||||
}
|
||||
|
||||
/// <summary>Recomputes the expression against the current dependency map and emits/dedups the result.
|
||||
/// Shared by the single-value and batched dependency handlers — the only difference between them is how
|
||||
/// many inputs were applied before this runs.</summary>
|
||||
/// <param name="timestampUtc">The triggering dependency change's timestamp, stamped on the result.</param>
|
||||
private void Evaluate(DateTime timestampUtc)
|
||||
{
|
||||
VirtualTagEvalResult result;
|
||||
try
|
||||
{
|
||||
@@ -188,13 +230,13 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.Warning(ex, "VirtualTag {Id}: evaluator threw", _virtualTagId);
|
||||
OnEvaluationFailed($"evaluator threw: {ex.Message}", "Error", msg.TimestampUtc);
|
||||
OnEvaluationFailed($"evaluator threw: {ex.Message}", "Error", timestampUtc);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!result.Success)
|
||||
{
|
||||
OnEvaluationFailed(result.Reason ?? "evaluator failure", "Warning", msg.TimestampUtc);
|
||||
OnEvaluationFailed(result.Reason ?? "evaluator failure", "Warning", timestampUtc);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -219,7 +261,7 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
_lastValue = result.Value;
|
||||
_lastPublishedBad = false;
|
||||
OtOpcUaTelemetry.VirtualTagEval.Add(1, new KeyValuePair<string, object?>("outcome", "ok"));
|
||||
var evalResult = new EvaluationResult(_virtualTagId, result.Value, msg.TimestampUtc, CorrelationId.NewId());
|
||||
var evalResult = new EvaluationResult(_virtualTagId, result.Value, timestampUtc, CorrelationId.NewId());
|
||||
Context.Parent.Tell(evalResult);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user