From 6e2febbf3d6f2bc60aec638666d91ba15377be61 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:30:57 -0400 Subject: [PATCH] fix(runtime): run driver writes off the mailbox under an in-flight cap (A3) 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 --- .../Drivers/DriverInstanceActor.cs | 116 ++++++++++- ...DriverInstanceActorConcurrentWriteTests.cs | 186 ++++++++++++++++++ 2 files changed, 293 insertions(+), 9 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index f666fd28..3eeaa913 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -194,6 +194,22 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// public static readonly TimeSpan DefaultShutdownTimeout = TimeSpan.FromSeconds(5); + /// + /// Default cap on the writes this actor runs against its driver at once. Writes run OFF the mailbox + /// (a ReceiveAsync 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. + /// + public const int DefaultMaxWritesInFlight = 4; + + /// Self-sent by a write task once it has replied, releasing its in-flight slot. + private sealed class WriteCompleted + { + public static readonly WriteCompleted Instance = new(); + private WriteCompleted() { } + } + private readonly IDriver _driver; /// Phase 6.1 resilience seam: every guarded driver-capability call this actor makes is @@ -217,6 +233,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Bound on the driver shutdown / dispose; see . private readonly TimeSpan _shutdownTimeout; + /// Cap on concurrent writes; see . + private readonly int _maxWritesInFlight; + + /// Writes currently inside the driver. Actor-thread only (incremented on launch, decremented on + /// the self-sent ). + private int _writesInFlight; + + /// Writes waiting for an in-flight slot, in arrival order. Failed fast on every exit from + /// Connected, so a queued write never reaches a driver that is reconnecting. + private readonly Queue<(WriteAttribute Write, IActorRef ReplyTo)> _pendingWrites = new(); + /// /// Set by so the it triggers skips the dispose. A /// supervision restart recreates the actor object around the SAME (the Props closure @@ -347,6 +374,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Optional bound on the driver's shutdown / dispose on stop and on the /// pre-reconnect teardown; defaults to . Exists so a test can prove a /// hung shutdown is abandoned without waiting 5 s. + /// Optional cap on concurrent writes against the driver; defaults to + /// . Writes beyond it queue in arrival order. /// A instance configured to create the wrapped . 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)); /// /// 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 /// when null, and is disabled when zero or negative. /// Bound on the driver's shutdown / dispose; defaults to /// when null. + /// Cap on concurrent writes; defaults to + /// when null or non-positive. 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(_ => 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(_ => 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(_ => @@ -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(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(msg => StartOrQueueWrite(msg, Sender)); + Receive(_ => OnWriteCompleted()); ReceiveAsync(HandleAcknowledgeAsync); ReceiveAsync(HandleSubscribeAsync); ReceiveAsync(_ => UnsubscribeAsync()); @@ -805,6 +846,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // timeout on an inbound write to a transiently-down driver). Synchronous Receive. Receive(_ => Sender.Tell(new WriteAttributeResult(false, "driver not connected"))); + Receive(_ => 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(_ => @@ -956,17 +998,44 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers _initCts = null; } - private async Task HandleWriteAsync(WriteAttribute msg) + /// + /// Launches a write against the driver off the mailbox, or queues it when + /// 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 + /// () replies to the captured sender from the thread pool and then + /// self-sends to free its slot. + /// + 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); + } + } + + /// Frees a write slot and, while Connected, starts the next queued write. + private void OnWriteCompleted() + { + if (_writesInFlight > 0) _writesInFlight--; + if (_pendingWrites.Count > 0 && _driver is IWritable) + { + var (next, replyTo) = _pendingWrites.Dequeue(); + StartOrQueueWrite(next, replyTo); + } + } + + /// Replies to every queued write with ; called on each exit from + /// Connected. In-flight writes are left to finish — they already hold the device and reply with its + /// real outcome. + private void FailQueuedWrites(string reason) + { + while (_pendingWrites.Count > 0) + { + var (_, replyTo) = _pendingWrites.Dequeue(); + replyTo.Tell(new WriteAttributeResult(false, reason)); + } } /// @@ -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(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs new file mode 100644 index 00000000..f44b21f3 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs @@ -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; + +/// +/// Review 02 P2: WriteAttribute was a ReceiveAsync, 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 ReceiveAsync. +/// +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(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( + new DriverInstanceActor.WriteAttribute("tag-1", 1), askBudget); + var second = actor.Ask( + 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( + new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), PresenceBudget); + + var pending = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 7), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + + driver.FireDataChange("tag-a", 3.5); + var published = parent.ExpectMsg(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( + 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( + new DriverInstanceActor.WriteAttribute("tag-0", 0), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + var queued = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-1", 1), PresenceBudget); + + actor.Tell(new DriverInstanceActor.ForceReconnect()); + parent.ExpectMsg(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(); + } + + /// A writable + subscribable driver whose writes either take or hold until released. + private sealed class SlowWritableDriver : IDriver, IWritable, ISubscribable + { + private readonly StubHandle _handle = new(); + private readonly ConcurrentQueue _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 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> WriteAsync(IReadOnlyList 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? OnDataChange; + + public Task SubscribeAsync( + IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) => + Task.FromResult(_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))); + } +}