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

fix(runtime): run driver writes off the mailbox under an in-flight cap (A3)

Browse Source 
WriteAttribute was a ReceiveAsync, suspending the child's mailbox for the
whole device round-trip (up to 5 s): writes serialised, the second of two
operator writes to a slow device missed the host's 8 s Ask and was
reverted although the device took it, and every data change queued behind
the write. Writes are now a synchronous Receive that launches the device
call on the thread pool (sender captured, reply from the task, slot freed
by a self-sent WriteCompleted) under a per-driver cap (default 4; excess
writes queue in order). Queued writes fail fast on every exit from
Connected. Subscribe/unsubscribe stay ReceiveAsync.

Review ref: 02/P2.

Refs #731

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 06:30:57 -04:00
parent bfd5b7b015
commit 6e2febbf3d
2 changed files with 293 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
+107 -9
View File
@@ -194,6 +194,22 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// </summary>
public static readonly TimeSpan DefaultShutdownTimeout = TimeSpan.FromSeconds(5);
/// <summary>
/// Default cap on the writes this actor runs against its driver at once. Writes run OFF the mailbox
/// (a <c>ReceiveAsync</c> used to suspend it for the whole device round-trip, so writes serialised and
/// every data change queued behind them — the second of two operator writes to a slow device missed the
/// host's 8 s Ask and was reverted although the device took it). The cap keeps a burst from flooding a
/// device with concurrent commands; writes beyond it queue in order and start as slots free.
/// </summary>
public const int DefaultMaxWritesInFlight = 4;
/// <summary>Self-sent by a write task once it has replied, releasing its in-flight slot.</summary>
private sealed class WriteCompleted
{
public static readonly WriteCompleted Instance = new();
private WriteCompleted() { }
}
private readonly IDriver _driver;
/// <summary>Phase 6.1 resilience seam: every guarded driver-capability call this actor makes is
@@ -217,6 +233,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <summary>Bound on the driver shutdown / dispose; see <see cref="DefaultShutdownTimeout"/>.</summary>
private readonly TimeSpan _shutdownTimeout;
/// <summary>Cap on concurrent writes; see <see cref="DefaultMaxWritesInFlight"/>.</summary>
private readonly int _maxWritesInFlight;
/// <summary>Writes currently inside the driver. Actor-thread only (incremented on launch, decremented on
/// the self-sent <see cref="WriteCompleted"/>).</summary>
private int _writesInFlight;
/// <summary>Writes waiting for an in-flight slot, in arrival order. Failed fast on every exit from
/// <c>Connected</c>, so a queued write never reaches a driver that is reconnecting.</summary>
private readonly Queue<(WriteAttribute Write, IActorRef ReplyTo)> _pendingWrites = new();
/// <summary>
/// Set by <see cref="PreRestart"/> so the <see cref="PostStop"/> it triggers skips the dispose. A
/// supervision restart recreates the actor object around the SAME <see cref="IDriver"/> (the Props closure
@@ -347,6 +374,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <param name="shutdownTimeout">Optional bound on the driver's shutdown / dispose on stop and on the
/// pre-reconnect teardown; defaults to <see cref="DefaultShutdownTimeout"/>. Exists so a test can prove a
/// hung shutdown is abandoned without waiting 5 s.</param>
/// <param name="maxWritesInFlight">Optional cap on concurrent writes against the driver; defaults to
/// <see cref="DefaultMaxWritesInFlight"/>. Writes beyond it queue in arrival order.</param>
/// <returns>A <see cref="Akka.Actor.Props"/> instance configured to create the wrapped <see cref="DriverInstanceActor"/>.</returns>
public static Props Props(
IDriver driver,
@@ -357,7 +386,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
IDriverCapabilityInvoker? invoker = null,
TimeSpan? healthPollInterval = null,
TimeSpan? commsLossTimeout = null,
TimeSpan? shutdownTimeout = null) =>
TimeSpan? shutdownTimeout = null,
int? maxWritesInFlight = null) =>
Akka.Actor.Props.Create(() => new DriverInstanceActor(
driver,
reconnectInterval ?? DefaultReconnectInterval,
@@ -367,7 +397,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
invoker,
healthPollInterval,
commsLossTimeout,
shutdownTimeout));
shutdownTimeout,
maxWritesInFlight));
/// <summary>
/// Returns true when the driver should boot in DEV-STUB mode based on host platform and
@@ -406,6 +437,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <see cref="DefaultCommsLossTimeout"/> when null, and is disabled when zero or negative.</param>
/// <param name="shutdownTimeout">Bound on the driver's shutdown / dispose; defaults to
/// <see cref="DefaultShutdownTimeout"/> when null.</param>
/// <param name="maxWritesInFlight">Cap on concurrent writes; defaults to
/// <see cref="DefaultMaxWritesInFlight"/> when null or non-positive.</param>
public DriverInstanceActor(
IDriver driver,
TimeSpan reconnectInterval,
@@ -415,8 +448,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
IDriverCapabilityInvoker? invoker = null,
TimeSpan? healthPollInterval = null,
TimeSpan? commsLossTimeout = null,
TimeSpan? shutdownTimeout = null)
TimeSpan? shutdownTimeout = null,
int? maxWritesInFlight = null)
{
_maxWritesInFlight = maxWritesInFlight is > 0 ? maxWritesInFlight.Value : DefaultMaxWritesInFlight;
_driver = driver;
_invoker = invoker ?? NullDriverCapabilityInvoker.Instance;
_driverInstanceId = driver.DriverInstanceId;
@@ -506,6 +541,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// "write timeout". Synchronous Receive: Sender.Tell on the actor thread is safe.
Receive<WriteAttribute>(_ =>
Sender.Tell(new WriteAttributeResult(false, "driver not connected")));
// A write launched while Connected can finish after the transition out of it; its slot still frees.
Receive<WriteCompleted>(_ => OnWriteCompleted());
// An ack arriving while still connecting can't reach the upstream alarm system; drop it (the ack is
// fire-and-forget — no reply to surface — and the OPC UA condition state already committed locally).
Receive<RouteAlarmAck>(_ =>
@@ -574,10 +611,14 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
{
_log.Info("DriverInstance {Id}: ForceReconnect requested by admin; re-entering Reconnecting", _driverInstanceId);
DetachSubscription();
FailQueuedWrites("driver not connected");
Become(Reconnecting);
PublishHealthSnapshot();
});
ReceiveAsync<WriteAttribute>(HandleWriteAsync);
// Synchronous Receive: the device call runs off the mailbox (see StartOrQueueWrite), so data changes
// and further writes are never head-of-line blocked behind a slow device.
Receive<WriteAttribute>(msg => StartOrQueueWrite(msg, Sender));
Receive<WriteCompleted>(_ => OnWriteCompleted());
ReceiveAsync<RouteAlarmAck>(HandleAcknowledgeAsync);
ReceiveAsync<Subscribe>(HandleSubscribeAsync);
ReceiveAsync<Unsubscribe>(_ => UnsubscribeAsync());
@@ -805,6 +846,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// timeout on an inbound write to a transiently-down driver). Synchronous Receive.
Receive<WriteAttribute>(_ =>
Sender.Tell(new WriteAttributeResult(false, "driver not connected")));
Receive<WriteCompleted>(_ => OnWriteCompleted());
// An ack arriving while reconnecting can't reach the upstream alarm system; drop it (fire-and-forget,
// no reply — the OPC UA condition state already committed locally on the Part 9 ack).
Receive<RouteAlarmAck>(_ =>
@@ -956,17 +998,44 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
_initCts = null;
}
private async Task HandleWriteAsync(WriteAttribute msg)
/// <summary>
/// Launches a write against the driver off the mailbox, or queues it when
/// <see cref="_maxWritesInFlight"/> writes are already inside the driver. Runs on the actor thread; the
/// only actor state it touches is the in-flight count and the queue. The device round-trip itself
/// (<see cref="RunWriteAsync"/>) replies to the captured sender from the thread pool and then
/// self-sends <see cref="WriteCompleted"/> to free its slot.
/// </summary>
private void StartOrQueueWrite(WriteAttribute msg, IActorRef replyTo)
{
var replyTo = Sender;
if (_driver is not IWritable writable)
{
replyTo.Tell(new WriteAttributeResult(false, "Driver does not implement IWritable"));
return;
}
if (_writesInFlight >= _maxWritesInFlight)
{
_pendingWrites.Enqueue((msg, replyTo));
return;
}
_writesInFlight++;
// Everything the task needs is captured HERE, on the actor thread — ResolveHost reads the driver,
// and Self must not be touched off-context.
var self = Self;
var hostKey = ResolveHost(msg.TagId);
var invoker = _invoker;
_ = RunWriteAsync(writable, invoker, hostKey, msg, replyTo, self);
}
private static async Task RunWriteAsync(
IWritable writable, IDriverCapabilityInvoker invoker, string hostKey,
WriteAttribute msg, IActorRef replyTo, IActorRef self)
{
// Off the actor context from here on: the first await yields, and nothing below reads actor state.
await Task.Yield();
var request = new[] { new WriteRequest(msg.TagId, msg.Value) };
// Bound the write so a hung backend can't pin this actor forever — retry stays
// Bound the write so a hung backend can't hold a slot forever — retry stays
// off by default, but a stalled call still needs an answer.
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
try
@@ -978,8 +1047,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// therefore authoritative regardless of any configured Write retry. Per-tag opt-in to retries via
// WriteIdempotentAttribute is the unshipped forward path (see WriteIdempotentAttribute). The outer
// cts is retained as a hard backstop above the (typically tighter) tier timeout.
var results = await _invoker.ExecuteWriteAsync(
ResolveHost(msg.TagId),
var results = await invoker.ExecuteWriteAsync(
hostKey,
isIdempotent: false,
async ct => await writable.WriteAsync(request, ct),
cts.Token);
@@ -999,6 +1068,33 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
{
replyTo.Tell(new WriteAttributeResult(false, ex.Message));
}
finally
{
self.Tell(WriteCompleted.Instance);
}
}
/// <summary>Frees a write slot and, while Connected, starts the next queued write.</summary>
private void OnWriteCompleted()
{
if (_writesInFlight > 0) _writesInFlight--;
if (_pendingWrites.Count > 0 && _driver is IWritable)
{
var (next, replyTo) = _pendingWrites.Dequeue();
StartOrQueueWrite(next, replyTo);
}
}
/// <summary>Replies to every queued write with <paramref name="reason"/>; called on each exit from
/// <c>Connected</c>. In-flight writes are left to finish — they already hold the device and reply with its
/// real outcome.</summary>
private void FailQueuedWrites(string reason)
{
while (_pendingWrites.Count > 0)
{
var (_, replyTo) = _pendingWrites.Dequeue();
replyTo.Tell(new WriteAttributeResult(false, reason));
}
}
/// <summary>
@@ -1246,6 +1342,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
{
_log.Warning("DriverInstance {Id}: disconnect observed ({Reason}); reconnecting", _driverInstanceId, reason);
DetachSubscription();
FailQueuedWrites("driver not connected");
RecordFault();
Become(Reconnecting);
PublishHealthSnapshot();
@@ -1643,6 +1740,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
}
DetachSubscription();
FailQueuedWrites("driver stopped");
// MUST happen: the IDriver instance can outlive this actor (the host respawns a child around the
// same driver object), so a missing unsubscribe accumulates a handler per respawn holding a dead Self.
DetachRediscoverySource();
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs
+186
View File
@@ -0,0 +1,186 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// Review 02 P2: <c>WriteAttribute</c> was a <c>ReceiveAsync</c>, which suspends the mailbox for the whole
/// device round-trip (up to the 5 s bound). Writes therefore serialised — the SECOND of two operator writes to
/// a ~3 s device missed the host's 8 s Ask and was reverted although the device took it — and every data
/// change queued behind the write. Writes now run off the mailbox under a per-driver in-flight cap; the
/// subscribe/unsubscribe pair that must serialise against the subscription handle stays <c>ReceiveAsync</c>.
/// </summary>
public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBase
{
private static readonly TimeSpan WriteLatency = TimeSpan.FromMilliseconds(1200);
private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect(IDriver driver, int? maxWritesInFlight = null)
{
var parent = CreateTestProbe();
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
driver, commsLossTimeout: TimeSpan.Zero, maxWritesInFlight: maxWritesInFlight));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(m => m.Connected, PresenceBudget);
return (actor, parent);
}
[Fact]
public async Task Two_concurrent_slow_writes_both_complete_within_one_device_round_trip()
{
var driver = new SlowWritableDriver { Latency = WriteLatency };
var (actor, _) = Connect(driver);
var sw = Stopwatch.StartNew();
var askBudget = TimeSpan.FromSeconds(8); // the host's Ask
var first = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
new DriverInstanceActor.WriteAttribute("tag-1", 1), askBudget);
var second = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
new DriverInstanceActor.WriteAttribute("tag-2", 2), askBudget);
var results = await Task.WhenAll(first, second);
sw.Stop();
results.ShouldAllBe(r => r.Success);
driver.Writes.Select(w => w.FullReference).ShouldBe(new[] { "tag-1", "tag-2" }, ignoreOrder: true);
// Serialised writes take two latencies; concurrent ones take one. The margin covers a loaded machine.
sw.Elapsed.ShouldBeLessThan(WriteLatency + WriteLatency / 2,
"the second write waited for the first — writes are still serialised on the mailbox");
driver.MaxConcurrentWrites.ShouldBe(2, "positive control: both writes were inside the driver at once");
}
[Fact]
public async Task Data_changes_are_delivered_while_a_write_is_pending()
{
var driver = new SlowWritableDriver { HoldWrites = true };
var (actor, parent) = Connect(driver);
await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), PresenceBudget);
var pending = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
new DriverInstanceActor.WriteAttribute("tag-a", 7), PresenceBudget);
AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget);
driver.FireDataChange("tag-a", 3.5);
var published = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
published.Items.Single().Value.ShouldBe(3.5);
pending.IsCompleted.ShouldBeFalse("the write is still held — the data change did not wait behind it");
// Positive control: releasing the device completes the write with its real outcome.
driver.ReleaseWrites();
(await pending).Success.ShouldBeTrue();
}
[Fact]
public async Task Writes_beyond_the_in_flight_cap_queue_and_drain_in_order()
{
var driver = new SlowWritableDriver { HoldWrites = true };
var (actor, _) = Connect(driver, maxWritesInFlight: 2);
var asks = Enumerable.Range(0, 4)
.Select(i => actor.Ask<DriverInstanceActor.WriteAttributeResult>(
new DriverInstanceActor.WriteAttribute($"tag-{i}", i), PresenceBudget))
.ToArray();
AwaitCondition(() => driver.WritesEntered == 2, PresenceBudget);
// Settle: a third write entering the driver would show up here.
await Task.Delay(300);
driver.WritesEntered.ShouldBe(2, "the cap let a third write into the driver");
driver.ReleaseWrites();
var results = await Task.WhenAll(asks);
results.ShouldAllBe(r => r.Success);
driver.Writes.Select(w => w.FullReference).ShouldBe(new[] { "tag-0", "tag-1", "tag-2", "tag-3" });
}
[Fact]
public async Task Queued_writes_fail_fast_when_the_driver_leaves_Connected()
{
var driver = new SlowWritableDriver { HoldWrites = true };
var (actor, parent) = Connect(driver, maxWritesInFlight: 1);
var inFlight = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
new DriverInstanceActor.WriteAttribute("tag-0", 0), PresenceBudget);
AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget);
var queued = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
new DriverInstanceActor.WriteAttribute("tag-1", 1), PresenceBudget);
actor.Tell(new DriverInstanceActor.ForceReconnect());
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(m => !m.Connected, PresenceBudget);
var queuedResult = await queued;
queuedResult.Success.ShouldBeFalse();
queuedResult.Reason.ShouldBe("driver not connected");
driver.WritesEntered.ShouldBe(1, "a queued write must never reach a driver that is reconnecting");
// The in-flight one still answers with the device's real outcome once it is released.
driver.ReleaseWrites();
(await inFlight).Success.ShouldBeTrue();
}
/// <summary>A writable + subscribable driver whose writes either take <see cref="Latency"/> or hold until released.</summary>
private sealed class SlowWritableDriver : IDriver, IWritable, ISubscribable
{
private readonly StubHandle _handle = new();
private readonly ConcurrentQueue<WriteRequest> _writes = new();
private TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
private int _entered;
private int _inside;
private int _maxInside;
public TimeSpan Latency { get; init; }
public bool HoldWrites { get; init; }
public IReadOnlyList<WriteRequest> Writes => _writes.ToArray();
public int WritesEntered => Volatile.Read(ref _entered);
public int MaxConcurrentWrites => Volatile.Read(ref _maxInside);
public string DriverInstanceId => "slow-writer-1";
public string DriverType => "SlowWriter";
public void ReleaseWrites() => _release.TrySetResult();
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);
public async Task<IReadOnlyList<WriteResult>> WriteAsync(IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
{
Interlocked.Increment(ref _entered);
var inside = Interlocked.Increment(ref _inside);
int seen;
while (inside > (seen = Volatile.Read(ref _maxInside)))
Interlocked.CompareExchange(ref _maxInside, inside, seen);
try
{
if (HoldWrites) await _release.Task.WaitAsync(cancellationToken);
else if (Latency > TimeSpan.Zero) await Task.Delay(Latency, cancellationToken);
foreach (var w in writes) _writes.Enqueue(w);
return writes.Select(_ => new WriteResult(OpcUaStatus.Good)).ToList();
}
finally
{
Interlocked.Decrement(ref _inside);
}
}
public event EventHandler<DataChangeEventArgs>? OnDataChange;
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) =>
Task.FromResult<ISubscriptionHandle>(_handle);
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask;
public void FireDataChange(string fullRef, object? value) =>
OnDataChange?.Invoke(this, new DataChangeEventArgs(
_handle, fullRef, new DataValueSnapshot(value, OpcUaStatus.Good, DateTime.UtcNow, DateTime.UtcNow)));
}
}