From c529948745e0193ebcafc03c374c10b012ea9262 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:42:50 -0400 Subject: [PATCH] perf(mqtt): publish a write batch concurrently, serialised per topic; writes feed health (R06:M-8) WriteAsync published one awaited QoS 1 round-trip per request, so 50 writes to a stalled broker cost 50 publish deadlines. Different topics now publish concurrently (at most 8 in flight); same-topic writes stay in request order. A caller cancel still fills every slot. A successful publish advances LastSuccessfulRead. MiniBroker now frames packets properly and records/PUBACKs PUBLISHes. Refs #744 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/Mqtt.md | 5 + .../MqttDriver.cs | 104 ++++++++++++--- .../MiniBroker.cs | 119 +++++++++++++++--- .../MqttDriverSessionLifecycleTests.cs | 73 +++++++++++ 4 files changed, 272 insertions(+), 29 deletions(-) diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index b9e62582..d5c33835 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -491,6 +491,11 @@ with their real type; a birth type the encoder cannot represent (any `*Array`, D matters and is deliberately not configurable: a retained command would be replayed by the broker to the device on *every* future subscribe — including the device's own reconnect — so one stale write would be re-applied forever. +**A batch publishes concurrently, one topic at a time (R06:M-8).** Writes to *different* topics in one +`WriteAsync` go out together, at most 8 in flight, so a stalled broker costs a batch about one publish deadline +rather than one per request; writes to the *same* topic stay sequential in request order, so the last one wins. +A successful publish (PUBACK, or a sent QoS 0 command) advances `LastSuccessfulRead` like an inbound message. + **Plain QoS 1 is at-least-once.** Know the consequences before pointing a Plain `writeTopic` at a non-idempotent device topic: diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs index 26cdeab6..dec7569c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs @@ -104,6 +104,12 @@ public sealed class MqttDriver /// private const string DiscoveryFolderName = "Mqtt"; + /// + /// Ceiling on PUBLISHes one write batch keeps in flight (R06:M-8). Bounded so a large batch cannot + /// flood a broker's inflight window; different topics share it, one topic's writes stay sequential. + /// + internal const int MaxConcurrentPublishes = 8; + /// How often the host-connectivity poll samples . private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1); @@ -209,6 +215,9 @@ public sealed class MqttDriver private DriverHealth _health = new(DriverState.Unknown, null, null); + /// UTC ticks of the last successful write PUBLISH; 0 = none. Answer evidence for health. + private long _lastPublishUtcTicks; + /// Guards the generation bump against the generation-checked health write. private readonly object _healthLock = new(); @@ -446,7 +455,7 @@ public sealed class MqttDriver return stored; } - var lastMessage = connection.LastMessageUtc ?? stored.LastSuccessfulRead; + var lastMessage = Latest(connection.LastMessageUtc ?? stored.LastSuccessfulRead, LastPublishUtc); return connection.State switch { MqttConnectionState.Connected => new DriverHealth(DriverState.Healthy, lastMessage, stored.LastError), @@ -708,9 +717,12 @@ public sealed class MqttDriver { ArgumentNullException.ThrowIfNull(writes); - var results = new WriteResult[writes.Count]; + var results = new WriteResult?[writes.Count]; var connection = _connection; + // Pass 1 — everything that needs no broker: resolve, encode, refuse. What survives is grouped by + // destination topic in REQUEST order. + var byTopic = new Dictionary>(StringComparer.Ordinal); for (var i = 0; i < writes.Count; i++) { var w = writes[i]; @@ -738,22 +750,70 @@ public sealed class MqttDriver continue; } + if (!byTopic.TryGetValue(encoded.Topic, out var queue)) + { + byTopic[encoded.Topic] = queue = []; + } + + queue.Add((i, encoded, w.FullReference)); + } + + // Pass 2 — publish (R06:M-8). Different topics go out concurrently, at most MaxConcurrentPublishes + // in flight, so a stalled broker costs a batch ~one publish deadline instead of one per request. + // Writes to the SAME topic stay strictly sequential in request order: the device applies them in + // arrival order, and the last one must win. + if (byTopic.Count > 0) + { + using var slots = new SemaphoreSlim(MaxConcurrentPublishes, MaxConcurrentPublishes); + await Task.WhenAll(byTopic.Values.Select(queue => PublishQueueAsync(connection!, queue, slots, results, cancellationToken))) + .ConfigureAwait(false); + } + + // Every slot is filled — a caller cancellation leaves the unattempted ones BadCommunicationError + // rather than throwing, so the batch shape holds. + var filled = new WriteResult[writes.Count]; + for (var i = 0; i < filled.Length; i++) + { + filled[i] = results[i] ?? new WriteResult(StatusBadCommunicationError); + } + + return filled; + } + + /// Publishes one topic's writes in request order, each under a shared concurrency slot. + private async Task PublishQueueAsync( + MqttConnection connection, + List<(int Index, MqttEncodedWrite Encoded, string Reference)> queue, + SemaphoreSlim slots, + WriteResult?[] results, + CancellationToken cancellationToken) + { + foreach (var (index, encoded, reference) in queue) + { + try + { + await slots.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return; // caller-cancelled: the remaining slots are filled by the caller + } + try { await connection .PublishAsync(encoded.Topic, encoded.Payload, encoded.Qos, retain: false, cancellationToken) .ConfigureAwait(false); - results[i] = new WriteResult(0u); + results[index] = new WriteResult(0u); + + // A PUBACK (or a sent QoS 0 command) is the broker answering — evidence the session works, + // so it feeds LastSuccessfulRead alongside inbound messages. + Interlocked.Exchange(ref _lastPublishUtcTicks, DateTime.UtcNow.Ticks); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - // Caller-cancelled: fill the remainder rather than throwing, so the batch shape holds. - for (var j = i; j < writes.Count; j++) - { - results[j] = new WriteResult(StatusBadCommunicationError); - } - - return results; + results[index] = new WriteResult(StatusBadCommunicationError); + return; } catch (Exception ex) { @@ -761,13 +821,15 @@ public sealed class MqttDriver ex, "MQTT driver '{DriverId}': write to '{Reference}' could not be published on '{Topic}'.", _driverInstanceId, - w.FullReference, + reference, encoded.Topic); - results[i] = new WriteResult(StatusBadCommunicationError); + results[index] = new WriteResult(StatusBadCommunicationError); + } + finally + { + slots.Release(); } } - - return results; } /// Maps an encode refusal to the status an OPC UA client sees. @@ -1528,6 +1590,20 @@ public sealed class MqttDriver OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, previous, next)); } + /// The last successful write publish, or null. + private DateTime? LastPublishUtc + { + get + { + var ticks = Interlocked.Read(ref _lastPublishUtcTicks); + return ticks == 0 ? null : new DateTime(ticks, DateTimeKind.Utc); + } + } + + /// The later of two optional instants. + private static DateTime? Latest(DateTime? a, DateTime? b) + => a is null ? b : b is null ? a : (a > b ? a : b); + /// Barrier-protected read of the multi-threaded health field. private DriverHealth ReadHealth() => Volatile.Read(ref _health); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs index 77d0b762..5cb47e49 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs @@ -5,7 +5,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; /// /// The smallest broker that can complete an MQTT 3.1.1 handshake: accept TCP, answer CONNECT with -/// a CONNACK, answer PINGREQ with PINGRESP, and nothing else. That is enough for MQTTnet to report +/// a CONNACK, answer PINGREQ with PINGRESP, SUBSCRIBE with a SUBACK (unless told not to), and record +/// PUBLISHes (PUBACKing QoS 1 only when told to). It never forwards a PUBLISH to a subscriber. That is enough for MQTTnet to report /// a real connection — which is what makes a real drop () drive a real /// DisconnectedAsync and a real reconnect, rather than a test seam standing in for one. /// @@ -107,6 +108,96 @@ internal sealed class MiniBroker : IDisposable return [.. ack]; } + /// + /// When true, a QoS 1 PUBLISH is answered with a PUBACK. False (the default) is a broker that takes + /// the PUBLISH and never acknowledges it — the stalled-broker shape a write deadline exists for. + /// + public bool AckPublish { get; set; } + + /// Every PUBLISH received, in arrival order, as (topic, payload). + public System.Collections.Concurrent.ConcurrentQueue<(string Topic, byte[] Payload)> Published { get; } = new(); + + /// Handles one complete packet; returns false when the connection must close. + private async Task HandlePacketAsync(NetworkStream stream, byte[] packet) + { + var packetType = packet[0] & 0xF0; + if (packetType == 0x10) + { + await stream.WriteAsync(new byte[] { 0x20, 0x02, 0x00, ConnAckReturnCode }, _cts.Token); + + // A real broker closes the socket straight after refusing, which is what drives + // MQTTnet's DisconnectedAsync. Mirroring it keeps the fake honest. + return ConnAckReturnCode == ReturnCodeAccepted; + } + + if (packetType == 0xC0) + { + await stream.WriteAsync(PingResp, _cts.Token); + } + else if (packetType == 0x80 && AnswerSubscribe && TryBuildSubAck(packet, packet.Length) is { } subAck) + { + await stream.WriteAsync(subAck, _cts.Token); + } + else if (packetType == 0x30) + { + var qos = (packet[0] >> 1) & 0x03; + var at = HeaderLength(packet); + var topicLength = (packet[at] << 8) | packet[at + 1]; + var topic = System.Text.Encoding.UTF8.GetString(packet, at + 2, topicLength); + at += 2 + topicLength; + byte idHigh = 0, idLow = 0; + if (qos > 0) + { + idHigh = packet[at]; + idLow = packet[at + 1]; + at += 2; + } + + Published.Enqueue((topic, packet[at..])); + if (qos == 1 && AckPublish) + { + await stream.WriteAsync(new byte[] { 0x40, 0x02, idHigh, idLow }, _cts.Token); + } + } + + return true; + } + + /// Length of a packet's fixed header (type byte + variable-length remaining length). + private static int HeaderLength(byte[] packet) + { + var at = 1; + while ((packet[at] & 0x80) != 0) at++; + return at + 1; + } + + /// Removes one complete packet from the front of , if one is there. + private static bool TryTakePacket(List pending, out byte[] packet) + { + packet = []; + if (pending.Count < 2) return false; + + var at = 1; + var remaining = 0; + var multiplier = 1; + byte digit; + do + { + if (at >= pending.Count) return false; + digit = pending[at++]; + remaining += (digit & 0x7F) * multiplier; + multiplier *= 128; + } + while ((digit & 0x80) != 0); + + var total = at + remaining; + if (pending.Count < total) return false; + + packet = pending.GetRange(0, total).ToArray(); + pending.RemoveRange(0, total); + return true; + } + /// Total TCP connections accepted since construction — counts reconnect attempts. public int AcceptedCount => Volatile.Read(ref _acceptedCount); @@ -178,7 +269,8 @@ internal sealed class MiniBroker : IDisposable private async Task ServeAsync(TcpClient client) { - var buffer = new byte[1024]; + var buffer = new byte[4096]; + var pending = new List(); try { var stream = client.GetStream(); @@ -195,25 +287,22 @@ internal sealed class MiniBroker : IDisposable continue; } - var packetType = buffer[0] & 0xF0; - if (packetType == 0x10) + // Frame properly: several packets can share one TCP read (concurrent publishes do), and + // one packet can span two. + pending.AddRange(buffer.AsSpan(0, read).ToArray()); + var close = false; + while (TryTakePacket(pending, out var packet)) { - await stream.WriteAsync(new byte[] { 0x20, 0x02, 0x00, ConnAckReturnCode }, _cts.Token); - - // A real broker closes the socket straight after refusing, which is what drives - // MQTTnet's DisconnectedAsync. Mirroring it keeps the fake honest. - if (ConnAckReturnCode != ReturnCodeAccepted) + if (!await HandlePacketAsync(stream, packet)) { + close = true; break; } } - else if (packetType == 0xC0) + + if (close) { - await stream.WriteAsync(PingResp, _cts.Token); - } - else if (packetType == 0x80 && AnswerSubscribe && TryBuildSubAck(buffer, read) is { } subAck) - { - await stream.WriteAsync(subAck, _cts.Token); + break; } } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs index 76d12cdd..6abfa293 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs @@ -288,6 +288,79 @@ public sealed class MqttDriverSessionLifecycleTests driver.GetHealth().State.ShouldBe(DriverState.Healthy); } + private static RawTagEntry WritableTag(string rawPath, string writeTopic) => new( + rawPath, + $$"""{"topic":"{{writeTopic}}/state","payloadFormat":"Scalar","dataType":"Float64","writeTopic":"{{writeTopic}}"}""", + WriteIdempotent: false); + + /// + /// R06:M-8. A write batch used to publish strictly serially, one awaited QoS 1 round-trip per + /// request — against a stalled broker, N writes × the publish deadline. Different topics now publish + /// concurrently: all four PUBLISHes reach a broker that never PUBACKs well inside ONE 30 s deadline + /// (serial code would have sent exactly one). Cancelling the batch then fills every slot, so the + /// batch shape holds. + /// + [Fact] + public async Task AWriteBatch_PublishesDifferentTopicsConcurrently() + { + using var broker = new MiniBroker(); // never PUBACKs + var options = BrokerOptions(broker.Port) with + { + ConnectTimeoutSeconds = 30, + RawTags = [.. Enumerable.Range(0, 4).Select(i => WritableTag($"Plant/Mqtt/dev1/Sp{i}", $"cmd/sp{i}"))], + }; + await using var driver = new MqttDriver(options, "d", null); + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + + using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + var write = driver.WriteAsync( + [.. Enumerable.Range(0, 4).Select(i => new WriteRequest($"Plant/Mqtt/dev1/Sp{i}", (double)i))], + cts.Token); + + await WaitForAsync(() => broker.Published.Count == 4, "all four PUBLISHes to be in flight at once"); + write.IsCompleted.ShouldBeFalse("no PUBACK has arrived, so nothing can have completed"); + + await cts.CancelAsync(); + var results = await write.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + results.Count.ShouldBe(4); + results.ShouldAllBe(r => r.StatusCode == 0x80050000u); // BadCommunicationError, every slot filled + } + + /// + /// Writes to the SAME topic in one batch stay in request order — last write wins at the device — + /// while the batch as a whole is concurrent. The broker PUBACKs here, so every slot is Good — and + /// those acks advance LastSuccessfulRead (writes feed health). + /// + [Fact] + public async Task AWriteBatch_KeepsSameTopicWritesInRequestOrder() + { + using var broker = new MiniBroker { AckPublish = true }; + var options = BrokerOptions(broker.Port) with + { + RawTags = [WritableTag("Plant/Mqtt/dev1/Sp", "cmd/sp"), WritableTag("Plant/Mqtt/dev1/Other", "cmd/other")], + }; + await using var driver = new MqttDriver(options, "d", null); + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + driver.GetHealth().LastSuccessfulRead.ShouldBeNull("nothing has answered yet"); + + var results = await driver.WriteAsync( + [ + new WriteRequest("Plant/Mqtt/dev1/Sp", 1.0), + new WriteRequest("Plant/Mqtt/dev1/Other", 9.0), + new WriteRequest("Plant/Mqtt/dev1/Sp", 2.0), + new WriteRequest("Plant/Mqtt/dev1/Sp", 3.0), + ], + TestContext.Current.CancellationToken); + + results.ShouldAllBe(r => r.StatusCode == 0u); + broker.Published.Where(p => p.Topic == "cmd/sp") + .Select(p => System.Text.Encoding.UTF8.GetString(p.Payload)) + .ShouldBe(["1", "2", "3"]); + + // A PUBACK is the broker answering, so writes feed the evidence clock the comms-loss watchdog reads. + driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull(); + } + private static async Task WaitForAsync(Func condition, string what) { var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);