diff --git a/gateway.md b/gateway.md index 0a6c8d0..9f1e452 100644 --- a/gateway.md +++ b/gateway.md @@ -884,7 +884,12 @@ Baseline choices: Optimizations after parity: - batch commands where MXAccess semantics allow, -- batch events from worker to gateway while preserving order, +- batch events from worker to gateway while preserving order. Implemented so + far: the worker frame writer coalesces the *flush* across a drained batch of + event frames, so a burst costs one flush syscall rather than one per event + (WRK-12). Still not implemented: a multi-event `WorkerEnvelope` body that + packs several events into a single frame — the wire still carries one event + per `worker_event` frame (IPC-15), so this remains an additive-proto change, - optional data-change coalescing by item handle, - memory-mapped payload slabs for very large arrays, - shared schema for typed values to avoid raw COM marshaling at the gateway, diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Conversion/MxStatusProxyConverterTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Conversion/MxStatusProxyConverterTests.cs index 4f732c3..4224573 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Conversion/MxStatusProxyConverterTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Conversion/MxStatusProxyConverterTests.cs @@ -106,6 +106,35 @@ public sealed class MxStatusProxyConverterTests Assert.Contains("success", exception.Message); } + /// + /// Verifies that repeated conversions of the same status type produce + /// identical results. WRK-06 caches the resolved FieldInfo objects per + /// type after the first conversion; the cached path must yield the same + /// message the uncached first call produced. + /// + [Fact] + public void Convert_RepeatedForSameType_ProducesIdenticalResults() + { + FakeMxStatusProxy status = new() + { + success = 0, + category = 3, + detectedBy = 1, + detail = 21, + }; + + // First call populates the per-type FieldInfo cache; the second reuses it. + MxStatusProxy first = _converter.Convert(status); + MxStatusProxy second = _converter.Convert(status); + + Assert.Equal(first, second); + Assert.Equal(MxStatusCategory.CommunicationError, second.Category); + Assert.Equal(MxStatusSource.RespondingLmx, second.DetectedBy); + Assert.Equal(0, second.Success); + Assert.Equal(21, second.Detail); + Assert.Equal("Invalid reference", second.DiagnosticText); + } + public struct FakeMxStatusProxy { public short success; diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs index 8921762..c80b447 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs @@ -373,6 +373,43 @@ public sealed class WorkerFrameProtocolTests Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame3.BodyCase); } + /// + /// Verifies the writer coalesces the flush across a batch of frames drained together: four frames + /// queued behind an in-progress write drain in a single pass and share one FlushAsync, not four + /// (WRK-12 / IPC-15). Every frame still reaches the wire intact. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_WhenBatchDrainedTogether_FlushesOnce() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using GatedWriteStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + // A blocked first write occupies the writer and holds the lock while more frames queue behind it. + Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + Task eventWrite2 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + Task eventWrite3 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + await Task.Delay(50); + + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(Task.WhenAll(firstWrite, eventWrite1, eventWrite2, eventWrite3)); + + // Four frames written in one drain pass => exactly one flush. + Assert.Equal(1, stream.FlushCount); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + for (int index = 0; index < 4; index++) + { + WorkerEnvelope frame = await reader.ReadAsync(); + Assert.NotEqual(WorkerEnvelope.BodyOneofCase.None, frame.BodyCase); + } + } + /// Verifies a zero negotiated frame maximum keeps the constructor default (IPC-02). [Fact] public void AdoptNegotiatedMaxMessageBytes_WithZero_KeepsDefault() @@ -451,9 +488,12 @@ public sealed class WorkerFrameProtocolTests private readonly TaskCompletionSource _firstWriteStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); private int _writeCount; + private int _flushCount; public Task FirstWriteStarted => _firstWriteStarted.Task; + public int FlushCount => Volatile.Read(ref _flushCount); + public void ReleaseFirstWrite() => _release.Release(); public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) @@ -467,6 +507,12 @@ public sealed class WorkerFrameProtocolTests await base.WriteAsync(buffer, offset, count, cancellationToken); } + public override Task FlushAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _flushCount); + return base.FlushAsync(cancellationToken); + } + protected override void Dispose(bool disposing) { if (disposing) diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs index 27d684b..84f8e1d 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs @@ -29,6 +29,26 @@ public sealed class MxAccessEventQueueTests Assert.False(queue.TryDequeue(out _)); } + /// + /// Verifies that Enqueue takes ownership of the passed event instead of + /// cloning it: the dequeued instance is the very reference passed in, and + /// the worker sequence/timestamp are stamped on that same instance + /// (WRK-11). + /// + [Fact] + public void Enqueue_TakesOwnershipOfPassedEventInstance() + { + MxAccessEventQueue queue = new(capacity: 4); + MxEvent original = CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10); + + queue.Enqueue(original); + + Assert.True(queue.TryDequeue(out WorkerEvent? dequeued)); + Assert.Same(original, dequeued?.Event); + Assert.Equal(1UL, original.WorkerSequence); + Assert.NotNull(original.WorkerTimestamp); + } + /// Verifies that Drain removes at most the requested number of events. [Fact] public void Drain_RemovesAtMostRequestedEvents() diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs index 7e5cb4f..fded4a2 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs @@ -46,6 +46,37 @@ public sealed class MxAccessValueCacheTests Assert.Equal(999, other.Value.Int32Value); } + /// + /// Verifies that Set stores an independent deep-copied snapshot: mutating + /// the source event's protobuf sub-messages after caching does not alter + /// the cached value. WRK-11 stopped the event sink cloning before enqueue, + /// so the same MxEvent instance now flows to the outbound queue; the cache + /// must own its own copy so the two never share mutable state. + /// + [Fact] + public void Set_StoresIndependentSnapshot_UnaffectedByLaterEventMutation() + { + MxAccessValueCache cache = new(); + Timestamp sourceTimestamp = Timestamp.FromDateTime(new(2026, 5, 19, 9, 0, 0, DateTimeKind.Utc)); + MxEvent mxEvent = BuildEvent(serverHandle: 7, itemHandle: 21, intValue: 100, quality: 192, sourceTimestamp); + + cache.Set(7, 21, mxEvent); + + // Mutate the event in place after it was cached — as if it kept flowing + // through the (unrelated) outbound path. None of this must reach the cache. + mxEvent.Value.Int32Value = 999; + mxEvent.Quality = 0; + mxEvent.SourceTimestamp = Timestamp.FromDateTime(new(2030, 1, 1, 0, 0, 0, DateTimeKind.Utc)); + mxEvent.Statuses[0].Category = MxStatusCategory.SecurityError; + + Assert.True(cache.TryGet(7, 21, out MxAccessValueCache.CachedValue cached)); + Assert.Equal(100, cached.Value.Int32Value); + Assert.Equal(192, cached.Quality); + Assert.Equal(sourceTimestamp, cached.SourceTimestamp); + Assert.Single(cached.Statuses); + Assert.Equal(MxStatusCategory.Ok, cached.Statuses[0].Category); + } + /// Verifies that TryGet returns false for unknown handles. [Fact] public void TryGet_WithUnknownHandle_ReturnsFalse() diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Conversion/MxStatusProxyConverter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Conversion/MxStatusProxyConverter.cs index 8deff4d..036cd9c 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Conversion/MxStatusProxyConverter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Conversion/MxStatusProxyConverter.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Globalization; using System.Reflection; @@ -9,6 +10,17 @@ namespace ZB.MOM.WW.MxGateway.Worker.Conversion; /// Converts MXAccess MXSTATUS_PROXY COM objects to protobuf MxStatusProxy messages. public sealed class MxStatusProxyConverter { + /// + /// Per-type cache of the four resolved objects a + /// status conversion needs. The status type is stable (the interop + /// MXSTATUS_PROXY struct in production; a fixed test double in + /// tests), so the expensive + /// metadata scan is resolved once per type and reused. Keyed by + /// so a plain-CLR test double and the real interop + /// struct each get their own entry, keeping the converter interop-agnostic. + /// + private static readonly ConcurrentDictionary FieldCache = new(); + /// Converts a single status object to a protobuf message, reflecting all fields and diagnostics. /// COM status object to convert. /// The converted protobuf status message. @@ -20,10 +32,11 @@ public sealed class MxStatusProxyConverter } Type statusType = status.GetType(); - int success = ReadInt32Field(status, statusType, "success"); - int rawCategory = ReadInt32Field(status, statusType, "category"); - int rawDetectedBy = ReadInt32Field(status, statusType, "detectedBy"); - int detail = ReadInt32Field(status, statusType, "detail"); + StatusFields fields = GetFields(statusType); + int success = ReadInt32Field(status, statusType, fields.Success); + int rawCategory = ReadInt32Field(status, statusType, fields.Category); + int rawDetectedBy = ReadInt32Field(status, statusType, fields.DetectedBy); + int detail = ReadInt32Field(status, statusType, fields.Detail); return new MxStatusProxy { @@ -82,6 +95,43 @@ public sealed class MxStatusProxyConverter private static int ReadInt32Field( object value, + Type valueType, + FieldInfo field) + { + object? fieldValue = field.GetValue(value); + if (fieldValue is null) + { + throw new MxStatusConversionException( + $"Status object field '{field.Name}' on type '{valueType.FullName}' is null."); + } + + return System.Convert.ToInt32(fieldValue, CultureInfo.InvariantCulture); + } + + /// + /// Resolves (and caches) the four objects for the + /// given status type. The first resolution for a type performs the + /// reflection scan; every subsequent conversion of that type reuses the + /// cached entry. A type missing a required field throws the same + /// the per-field lookup used to + /// throw — and, because + /// does not store a value when the factory throws, a bad type keeps + /// failing identically on every call rather than being cached. + /// + /// Runtime type of the status object being converted. + /// The resolved field set for . + private static StatusFields GetFields(Type statusType) + { + return FieldCache.GetOrAdd( + statusType, + type => new StatusFields( + ResolveField(type, "success"), + ResolveField(type, "category"), + ResolveField(type, "detectedBy"), + ResolveField(type, "detail"))); + } + + private static FieldInfo ResolveField( Type valueType, string fieldName) { @@ -92,14 +142,7 @@ public sealed class MxStatusProxyConverter $"Status object type '{valueType.FullName}' does not expose required field '{fieldName}'."); } - object? fieldValue = field.GetValue(value); - if (fieldValue is null) - { - throw new MxStatusConversionException( - $"Status object field '{fieldName}' on type '{valueType.FullName}' is null."); - } - - return System.Convert.ToInt32(fieldValue, CultureInfo.InvariantCulture); + return field; } private static MxStatusCategory MapCategory(int rawCategory) @@ -134,4 +177,32 @@ public sealed class MxStatusProxyConverter _ => MxStatusSource.Unknown, }; } + + /// + /// The four resolved status fields cached per type. Plain readonly struct + /// (not a record) so it compiles under the worker's net48 target, which + /// lacks IsExternalInit. + /// + private readonly struct StatusFields + { + public StatusFields( + FieldInfo success, + FieldInfo category, + FieldInfo detectedBy, + FieldInfo detail) + { + Success = success; + Category = category; + DetectedBy = detectedBy; + Detail = detail; + } + + public FieldInfo Success { get; } + + public FieldInfo Category { get; } + + public FieldInfo DetectedBy { get; } + + public FieldInfo Detail { get; } + } } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs index 5ae2e58..28734ad 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs @@ -112,36 +112,77 @@ public sealed class WorkerFrameWriter // Runs only under _writeLock. Drains control frames before event frames, stamping and writing each. // The stream write itself is not cancellable: a frame is written atomically or fails, never left // half-written on the pipe because a caller gave up waiting. + // + // Flushes are coalesced across the whole drained batch (WRK-12 / IPC-15): each frame is written to + // the stream but not flushed individually; a single FlushAsync runs after the batch, then every + // successfully-written frame is completed. A caller's Completion therefore still signals only after + // its bytes have been written AND flushed, so the "written and flushed" contract is unchanged — but + // a burst of N events now costs one flush syscall instead of N. private async Task DrainQueuedFramesAsync() { + List written = new List(); while (true) { PendingFrame? frame = DequeueNext(); if (frame is null) { - return; + break; } try { await WriteFrameAsync(frame.Envelope).ConfigureAwait(false); - frame.Completion.TrySetResult(true); + written.Add(frame); } catch (WorkerFrameProtocolException exception) when (IsPerFrameRejection(exception)) { // Validation, empty-payload, and oversized-frame errors are specific to this frame and - // do not damage the stream; fail only this frame and keep draining the rest. + // do not damage the stream; fail only this frame and keep draining the rest. Nothing was + // written for it, so it needs no flush. frame.Completion.TrySetException(exception); } catch (Exception exception) { - // A stream write/flush failure means the pipe is broken; fail this frame and every frame - // still queued so no caller awaits forever, then stop draining. + // A stream write failure means the pipe is broken; fail this frame, every frame already + // written this batch but not yet flushed, and every frame still queued so no caller + // awaits forever, then stop draining. frame.Completion.TrySetException(exception); + FailFrames(written, exception); FailAllQueued(exception); return; } } + + if (written.Count == 0) + { + return; + } + + try + { + await _stream.FlushAsync(CancellationToken.None).ConfigureAwait(false); + } + catch (Exception exception) + { + // The batch reached the stream but the flush that guarantees delivery failed: the pipe is + // broken. Fail every frame in the batch (the queue was already drained) so no caller treats + // an unflushed write as delivered. + FailFrames(written, exception); + return; + } + + foreach (PendingFrame frame in written) + { + frame.Completion.TrySetResult(true); + } + } + + private static void FailFrames(List frames, Exception exception) + { + foreach (PendingFrame frame in frames) + { + frame.Completion.TrySetException(exception); + } } private static bool IsPerFrameRejection(WorkerFrameProtocolException exception) @@ -211,14 +252,14 @@ public sealed class WorkerFrameWriter // Serialize once into a single buffer that carries the 4-byte length prefix followed by the // payload, then issue one stream write. This avoids a second serialization pass, a separate - // prefix array, and a separate prefix write. + // prefix array, and a separate prefix write. The flush is deferred to the end of the drained + // batch (see DrainQueuedFramesAsync) so a burst of frames shares one flush. int frameLength = sizeof(uint) + payloadLength; byte[] frame = new byte[frameLength]; WriteUInt32LittleEndian(frame, (uint)payloadLength); envelope.WriteTo(new Span(frame, sizeof(uint), payloadLength)); await _stream.WriteAsync(frame, 0, frameLength, CancellationToken.None).ConfigureAwait(false); - await _stream.FlushAsync(CancellationToken.None).ConfigureAwait(false); } private static void WriteUInt32LittleEndian( diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs index 51fe92f..fe09401 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs @@ -8,6 +8,17 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; /// /// Thread-safe queue for MxAccess events with capacity overflow and fault tracking. /// +/// +/// Ownership invariant: takes ownership of the +/// passed to it — it stamps the worker sequence and +/// timestamp on that same instance and enqueues it directly, without +/// cloning (WRK-11). Every caller must therefore pass a freshly built +/// that it does not retain, reuse, or mutate after the +/// call returns. All production callers (MxAccessBaseEventSink, +/// MxAccessAlarmEventSink, AlarmCommandHandler) build a new event per +/// Enqueue via the mapper and satisfy this; the value cache stores its own +/// independent snapshot (see ). +/// public sealed class MxAccessEventQueue { /// @@ -110,8 +121,11 @@ public sealed class MxAccessEventQueue /// /// Enqueues an MxAccess event, assigning a sequence number and timestamp. + /// Takes ownership of : the sequence and timestamp + /// are stamped on that same instance and it is enqueued without cloning, so + /// the caller must pass a freshly built event it does not reuse afterwards. /// - /// MXAccess event to enqueue. + /// Freshly built MXAccess event to enqueue; ownership transfers to the queue. public void Enqueue(MxEvent mxEvent) { if (mxEvent is null) @@ -132,13 +146,17 @@ public sealed class MxAccessEventQueue throw new MxAccessEventQueueOverflowException(capacity); } - MxEvent queuedEvent = mxEvent.Clone(); - queuedEvent.WorkerSequence = ++lastEventSequence; - queuedEvent.WorkerTimestamp = Timestamp.FromDateTime(DateTime.UtcNow); + // WRK-11: stamp the sequence/timestamp on the caller's own event and + // enqueue that same instance under the lock instead of cloning. See + // the ownership invariant on the class summary — the caller hands the + // event over exclusively, so a defensive Clone() here is pure + // overhead on the hottest path. + mxEvent.WorkerSequence = ++lastEventSequence; + mxEvent.WorkerTimestamp = Timestamp.FromDateTime(DateTime.UtcNow); WorkerEvent workerEvent = new() { - Event = queuedEvent, + Event = mxEvent, }; events.Enqueue(workerEvent); } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs index a4ba24d..f82429b 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs @@ -40,6 +40,20 @@ public sealed class MxAccessValueCache throw new ArgumentNullException(nameof(mxEvent)); } + // WRK-11: the event sink no longer clones before enqueue, so the passed + // mxEvent is the very instance handed to the outbound queue. Deep-copy + // the value/timestamp/statuses payload we retain here so the cache's + // snapshot stays independent of the enqueued (and later serialized) + // event — the two must never share mutable protobuf sub-messages. + // Value is always set for OnDataChange; SourceTimestamp may be unset when + // the source timestamp could not be parsed, so both are cloned only when + // present. The null-forgiving result matches CachedValue's non-null- + // annotated parameters, which already accepted a runtime-null value or + // timestamp before WRK-11 (the ternary keeps the compiler's null-state + // from poisoning to maybe-null, which a plain null check would do). + MxValue cachedValue = mxEvent.Value is null ? null! : mxEvent.Value.Clone(); + Timestamp cachedTimestamp = mxEvent.SourceTimestamp is null ? null! : mxEvent.SourceTimestamp.Clone(); + long key = CreateItemKey(serverHandle, itemHandle); lock (syncRoot) { @@ -49,10 +63,10 @@ public sealed class MxAccessValueCache entries[key] = new CachedValue( nextVersion, - mxEvent.Value, + cachedValue, mxEvent.Quality, - mxEvent.SourceTimestamp, - mxEvent.Statuses); + cachedTimestamp, + mxEvent.Statuses.Clone()); } }