perf(worker): event hot-path allocation + flush cuts (WRK-06/11/12, IPC-15)

WRK-06: MxStatusProxyConverter caches the four resolved FieldInfo per status
type in a static ConcurrentDictionary (the GetField metadata scan ran 4x per
status per event on the STA path). GetValue+Convert.ToInt32 still run per event
(late-bound RCW). Exceptions byte-identical: missing-field message unchanged
(ResolveField, not cached on throw via GetOrAdd); null-value message unchanged.

WRK-11: MxAccessEventQueue.Enqueue takes ownership of the passed MxEvent -
stamps WorkerSequence/WorkerTimestamp on it in place and enqueues it, no
Clone(). Audited all 3 callers (base/alarm event sinks, provider-mode handler):
each builds a fresh event per Enqueue, none reuse it. MxAccessValueCache.Set now
deep-copies its retained Value/SourceTimestamp/Statuses so the cache snapshot
never aliases the queue-owned (later serialized) event. Net: alarm/other events
clone nothing (was full clone); data-change clones payload-only.

WRK-12: WorkerFrameWriter coalesces the flush across a drained batch - each
frame is written but not flushed individually; one FlushAsync after the batch,
then all written frames complete. Preserves the written+flushed completion
contract; a burst of N events costs 1 flush, not N. On write failure the whole
in-flight batch + queue fail so no caller hangs.

IPC-15 (doc): the multi-event WorkerEnvelope body remains unimplemented (wire
still carries one event per worker_event frame); gateway.md Performance section
now distinguishes the shipped flush-coalescing from that deferred proto change.

net48-safe (no init/records; readonly struct cache entry). Worker builds x86
only - verification on windev. Tests added: converter cache-reuse, queue
ownership-transfer, value-cache snapshot independence, writer batch-flush-once.

Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW
This commit is contained in:
Joseph Doherty
2026-07-09 15:52:49 -04:00
parent 73ce824f6c
commit f61c816acf
9 changed files with 303 additions and 28 deletions
@@ -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;
/// <summary>Converts MXAccess MXSTATUS_PROXY COM objects to protobuf MxStatusProxy messages.</summary>
public sealed class MxStatusProxyConverter
{
/// <summary>
/// Per-type cache of the four resolved <see cref="FieldInfo"/> objects a
/// status conversion needs. The status type is stable (the interop
/// <c>MXSTATUS_PROXY</c> struct in production; a fixed test double in
/// tests), so the expensive <see cref="Type.GetField(string, BindingFlags)"/>
/// metadata scan is resolved once per type and reused. Keyed by
/// <see cref="Type"/> so a plain-CLR test double and the real interop
/// struct each get their own entry, keeping the converter interop-agnostic.
/// </summary>
private static readonly ConcurrentDictionary<Type, StatusFields> FieldCache = new();
/// <summary>Converts a single status object to a protobuf message, reflecting all fields and diagnostics.</summary>
/// <param name="status">COM status object to convert.</param>
/// <returns>The converted protobuf status message.</returns>
@@ -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);
}
/// <summary>
/// Resolves (and caches) the four <see cref="FieldInfo"/> 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
/// <see cref="MxStatusConversionException"/> the per-field lookup used to
/// throw — and, because <see cref="ConcurrentDictionary{TKey,TValue}.GetOrAdd(TKey, Func{TKey, TValue})"/>
/// does not store a value when the factory throws, a bad type keeps
/// failing identically on every call rather than being cached.
/// </summary>
/// <param name="statusType">Runtime type of the status object being converted.</param>
/// <returns>The resolved field set for <paramref name="statusType"/>.</returns>
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,
};
}
/// <summary>
/// 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 <c>IsExternalInit</c>.
/// </summary>
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; }
}
}
@@ -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<PendingFrame> written = new List<PendingFrame>();
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<PendingFrame> 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<byte>(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(
@@ -8,6 +8,17 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
/// <summary>
/// Thread-safe queue for MxAccess events with capacity overflow and fault tracking.
/// </summary>
/// <remarks>
/// Ownership invariant: <see cref="Enqueue"/> takes ownership of the
/// <see cref="MxEvent"/> 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
/// <see cref="MxEvent"/> 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 <see cref="MxAccessValueCache.Set"/>).
/// </remarks>
public sealed class MxAccessEventQueue
{
/// <summary>
@@ -110,8 +121,11 @@ public sealed class MxAccessEventQueue
/// <summary>
/// Enqueues an MxAccess event, assigning a sequence number and timestamp.
/// Takes ownership of <paramref name="mxEvent"/>: 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.
/// </summary>
/// <param name="mxEvent">MXAccess event to enqueue.</param>
/// <param name="mxEvent">Freshly built MXAccess event to enqueue; ownership transfers to the queue.</param>
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);
}
@@ -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());
}
}