feat(runtime): DriverHostActor batched value fan-out
Replaces the temporary per-item unwrap bridge with a real batched handler: one AttributeValueBatchPublished burst now costs ONE hop to the dependency mux and ONE OpcUaPublishActor.AttributeValueBatchUpdate, instead of one message per value. Routing per item is unchanged — the v3 Batch 4 single-source fan-out (raw NodeId + every referencing UNS NodeId, identical value/status/timestamp), with an unmapped ref Debug-logged and dropped without disturbing the rest of the burst. A burst in which nothing resolves sends no publish message at all rather than an empty batch. Indexed for-loop over Items is deliberate: no interface-enumerator allocation per burst on the hottest routing seam. ForwardToMux and its registrations stay for the comms-loss fan and per-tag drivers.
This commit is contained in:
@@ -48,7 +48,8 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
/// Children (DriverInstance/VirtualTag/etc.) are fully wired: <c>DispatchDeployment</c> drives
|
||||
/// <see cref="PushDesiredSubscriptions"/> which runs the spawn-plan via <see cref="IDriverFactory"/>,
|
||||
/// maintains the <c>FullName→NodeId</c> live-value routing map, and forwards results to the
|
||||
/// OPC UA publish actor via <c>ForwardToMux</c>.
|
||||
/// OPC UA publish actor via <c>ForwardToMux</c> (per value) and <c>ForwardBatchToMux</c> (one whole
|
||||
/// driver production burst per message — the primary path for a batch-capable driver).
|
||||
/// </summary>
|
||||
public sealed class DriverHostActor : ReceiveActor, IWithTimers
|
||||
{
|
||||
@@ -976,19 +977,47 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Temporary bridge: unwrap a batched driver publish back into the per-value
|
||||
/// <see cref="ForwardToMux"/> path, so the batching ingress seam lands without changing any
|
||||
/// downstream behaviour.
|
||||
/// Routes one batched driver production burst — the batched analogue of
|
||||
/// <see cref="ForwardToMux"/>, and the primary value ingress for a batch-capable driver.
|
||||
/// <para>
|
||||
/// The routing is identical per item: each <c>FullReference</c> (the tag's RawPath) resolves
|
||||
/// through <see cref="_nodeIdByDriverRef"/> to the v3 Batch 4 single-source fan-out — the RAW
|
||||
/// NodeId AND every referencing UNS NodeId, each carrying IDENTICAL value/status/timestamp — and
|
||||
/// an unmapped ref is Debug-logged and dropped without disturbing the rest of the burst. What
|
||||
/// changes is the delivery: the whole burst costs ONE hop to the dependency mux and ONE
|
||||
/// <see cref="ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AttributeValueBatchUpdate"/> to
|
||||
/// the publish actor, instead of one message per value.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
// TODO(Task 8 plan 2026-08-15): replace with batched fan-out — one mux hop and one OPC UA publish
|
||||
// hop per burst instead of one per value, which is the whole point of the batching work.
|
||||
private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished batch)
|
||||
private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished msg)
|
||||
{
|
||||
foreach (var i in batch.Items)
|
||||
// One envelope per burst; the mux filters per item (it registered a batch receive in Task 9).
|
||||
_dependencyMux?.Tell(msg);
|
||||
|
||||
if (_opcUaPublishActor is null) return;
|
||||
|
||||
// Indexed for-loop, not foreach: IReadOnlyList<AttributeValueItem> enumerates through an interface
|
||||
// enumerator that would allocate once per burst on the hottest routing seam in the server.
|
||||
List<ValueWrite>? writes = null;
|
||||
for (var idx = 0; idx < msg.Items.Count; idx++)
|
||||
{
|
||||
ForwardToMux(new DriverInstanceActor.AttributeValuePublished(
|
||||
batch.DriverInstanceId, i.FullReference, i.Value, i.StatusCode, i.TimestampUtc));
|
||||
var item = msg.Items[idx];
|
||||
if (_nodeIdByDriverRef.TryGetValue((msg.DriverInstanceId, item.FullReference), out var nodeIds))
|
||||
{
|
||||
writes ??= new List<ValueWrite>(msg.Items.Count * 2);
|
||||
foreach (var n in nodeIds)
|
||||
writes.Add(new ValueWrite(n.NodeId, item.Value, item.StatusCode, item.TimestampUtc, n.Realm));
|
||||
}
|
||||
else if (_log.IsDebugEnabled)
|
||||
{
|
||||
_log.Debug("DriverHost {Node}: no bound NodeId for ({Driver},{Ref}) — value dropped",
|
||||
_localNode, msg.DriverInstanceId, item.FullReference);
|
||||
}
|
||||
}
|
||||
|
||||
// Nothing resolved ⇒ no message at all, rather than an empty batch for the sink to unpack.
|
||||
if (writes is not null)
|
||||
_opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AttributeValueBatchUpdate(writes));
|
||||
}
|
||||
|
||||
private void ForwardToMux(DriverInstanceActor.AttributeValuePublished msg)
|
||||
|
||||
Reference in New Issue
Block a user