perf(worker): value cache borrows the write-once event's instances — three clones per OnDataChange removed
MxAccessValueCache.Set deep-copied the Value (recursive for an MxArray), the SourceTimestamp, and the Statuses RepeatedField (container plus every MxStatusProxy) on every OnDataChange. The aliasing audit found all three removable: the sink enqueues the event first — which stamps WorkerSequence/WorkerTimestamp inside the queue lock — and only then runs the postPublish hook that reaches Set, so the event is write-once by then and the queue's ownership invariant forbids later mutation. The producer never reuses instances (fresh MxEvent per mapper call, fresh MxValue per convert), and the alias already existed on the read side: MxAccessSession.SucceededRead puts the cache's own Value/SourceTimestamp/status references on every BulkReadResult, which the worker only serializes onto the IPC pipe. Set and CachedValue now carry the ownership contract: the cache holds borrowed references into an enqueued, write-once MxEvent; consumers may read and serialize, never mutate. Mutation would corrupt the still-queued event AND invalidate QueuedEvent.Size — the enqueue-time memoized serialized size the byte-budgeted Drain charges — so a grown message could overshoot the negotiated frame max and fault the session with MessageTooLarge. MxAccessEventQueue's class remark, which claimed the cache keeps an independent snapshot, is corrected to point at the borrow. MxAccessWriteCompletionCache.Record keeps its parallel statuses.Clone() deliberately, with a cross-reference explaining why: it takes a bare RepeatedField whose provenance its signature cannot constrain, and it is on the command-rate write path, not the streaming hot path. Tests: Set_StoresIndependentSnapshot_UnaffectedByLaterEventMutation codified the invariant being reversed, so it is replaced by Set_BorrowsTheEventsOwnInstances_ByOwnershipContract (Assert.Same on Value, SourceTimestamp, and the status row). Adds the missing cached-read-path test to MxAccessCommandExecutorTests — nothing in the worker exercised was_cached == true end to end — asserting the cache hit, reference identity out to the BulkReadResult, and that no COM call is made for the read. Not built or tested here: these are net48/x86 worker files that cannot compile on the macOS tree. Verification is deferred to the windev gate.
This commit is contained in:
Joseph Doherty
@@ -669,6 +669,107 @@ public sealed class MxAccessCommandExecutorTests
|
||||
Assert.Contains("RemoveItem:90:900", fakeComObject.OperationNames);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies ReadBulk's cached fast path end to end — the
|
||||
/// <c>was_cached = true</c> half of the command, which nothing else in
|
||||
/// the worker suite exercises. With the tag already added AND advised
|
||||
/// and a value in the per-session cache, the executor answers from the
|
||||
/// cache: the result is successful and flagged cached, and no COM call
|
||||
/// is made for the read at all, so the subscription the caller did not
|
||||
/// create is left exactly as it was.
|
||||
///
|
||||
/// It also pins the borrow contract through the whole read path: the
|
||||
/// <c>Value</c>, <c>SourceTimestamp</c>, and status row on the
|
||||
/// <c>BulkReadResult</c> are the cached event's own instances, not
|
||||
/// copies (see <see cref="MxAccessValueCache.Set"/>). The worker only
|
||||
/// serializes them onto the IPC pipe, so the alias never escapes the
|
||||
/// process.
|
||||
///
|
||||
/// Driven through <see cref="MxAccessCommandExecutor"/> directly rather
|
||||
/// than <see cref="MxAccessStaSession"/> because seeding the cache needs
|
||||
/// a handle on it: <see cref="MxAccessSession.Create"/> only shares the
|
||||
/// sink's cache for the production <c>MxAccessBaseEventSink</c>, which
|
||||
/// casts the COM object to <c>LMXProxyServerClass</c> and so cannot take
|
||||
/// a fake, while <c>CreateForTesting</c> accepts the cache directly. The
|
||||
/// cached path neither waits nor pumps, so it needs no STA.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Execute_ReadBulk_WhenTagIsAdvisedAndCached_ServesTheCachedInstancesWithoutTouchingTheSubscription()
|
||||
{
|
||||
FakeMxAccessComObject fakeComObject = new(
|
||||
registerHandle: 92,
|
||||
addItemHandle: 920);
|
||||
MxAccessValueCache valueCache = new();
|
||||
using MxAccessSession session = MxAccessSession.CreateForTesting(
|
||||
mxAccessServer: fakeComObject,
|
||||
eventSink: new NoopEventSink(),
|
||||
valueCache: valueCache);
|
||||
MxAccessCommandExecutor executor = new(
|
||||
session,
|
||||
new ZB.MOM.WW.MxGateway.Worker.Conversion.VariantConverter());
|
||||
|
||||
// Registry half of the fast path: the tag must resolve to a live item
|
||||
// handle on this server AND carry an advice, or TryGetCachedReadFor
|
||||
// falls through to the AddItem/Advise snapshot lifecycle.
|
||||
MxCommandReply registerReply = executor.Execute(
|
||||
CreateRegisterCommand("register-before-cached-read", "client-a"));
|
||||
MxCommandReply addItemReply = executor.Execute(
|
||||
CreateAddItemCommand("add-before-cached-read", 92, "Galaxy.Tag.Value"));
|
||||
MxCommandReply adviseReply = executor.Execute(
|
||||
CreateAdviseCommand("advise-before-cached-read", 92, 920));
|
||||
Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code);
|
||||
Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code);
|
||||
Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code);
|
||||
|
||||
// Cache half: stand in for the event sink's post-publish hook, which
|
||||
// records the event it just enqueued.
|
||||
MxEvent cachedEvent = new()
|
||||
{
|
||||
Family = MxEventFamily.OnDataChange,
|
||||
ServerHandle = 92,
|
||||
ItemHandle = 920,
|
||||
Quality = 192,
|
||||
SourceTimestamp = Timestamp.FromDateTime(new(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc)),
|
||||
Value = new MxValue
|
||||
{
|
||||
DataType = MxDataType.Integer,
|
||||
VariantType = "VT_I4",
|
||||
Int32Value = 7788,
|
||||
},
|
||||
OnDataChange = new OnDataChangeEvent(),
|
||||
};
|
||||
cachedEvent.Statuses.Add(new MxStatusProxy { Category = MxStatusCategory.Ok });
|
||||
valueCache.Set(92, 920, cachedEvent);
|
||||
|
||||
MxCommandReply reply = executor.Execute(CreateReadBulkCommand(
|
||||
"read-bulk-cached",
|
||||
serverHandle: 92,
|
||||
tagAddresses: new[] { "Galaxy.Tag.Value" },
|
||||
timeoutMs: 80));
|
||||
|
||||
Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
|
||||
Assert.Equal(MxCommandKind.ReadBulk, reply.Kind);
|
||||
BulkReadResult result = Assert.Single(reply.ReadBulk.Results);
|
||||
Assert.True(result.WasSuccessful);
|
||||
Assert.True(result.WasCached);
|
||||
Assert.Equal("Galaxy.Tag.Value", result.TagAddress);
|
||||
Assert.Equal(920, result.ItemHandle);
|
||||
Assert.Equal(192, result.Quality);
|
||||
Assert.Equal(7788, result.Value.Int32Value);
|
||||
|
||||
// Borrowed, not copied — all the way from the event handed to
|
||||
// MxAccessValueCache.Set out to the reply the worker serializes.
|
||||
Assert.Same(cachedEvent.Value, result.Value);
|
||||
Assert.Same(cachedEvent.SourceTimestamp, result.SourceTimestamp);
|
||||
Assert.Same(cachedEvent.Statuses[0], Assert.Single(result.Statuses));
|
||||
|
||||
// No second AddItem, and above all no UnAdvise/RemoveItem: only the
|
||||
// three setup calls ever reached MXAccess.
|
||||
Assert.Equal(
|
||||
new[] { "Register:client-a", "AddItem:92:Galaxy.Tag.Value", "Advise:92:920" },
|
||||
fakeComObject.OperationNames);
|
||||
}
|
||||
|
||||
/// <summary>Verifies that ReadBulk with no payload returns an invalid request error.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
|
||||
@@ -48,14 +48,28 @@ public sealed class MxAccessValueCacheTests
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// Pins the ownership contract: <c>Set</c> borrows the event's own
|
||||
/// protobuf sub-messages instead of deep-copying them, so <c>TryGet</c>
|
||||
/// hands back the very <c>Value</c>, <c>SourceTimestamp</c>, and
|
||||
/// <c>MxStatusProxy</c> instances the caller passed in.
|
||||
///
|
||||
/// This is safe only because the event is write-once by the time
|
||||
/// <c>Set</c> runs: the sink enqueues it (which stamps the worker
|
||||
/// sequence and timestamp) and only then post-publishes it here, and
|
||||
/// <see cref="MxAccessEventQueue"/>'s ownership invariant forbids
|
||||
/// mutating an enqueued event. This test therefore asserts reference
|
||||
/// identity and deliberately does NOT mutate the event afterwards —
|
||||
/// doing so is exactly what the contract forbids, and it would also
|
||||
/// invalidate the serialized size the queue memoized at enqueue.
|
||||
///
|
||||
/// It replaced a test asserting the opposite (an independent deep-copied
|
||||
/// snapshot). The three clones that test pinned — the MxValue, the
|
||||
/// Timestamp, and the RepeatedField plus every status row in it — ran on
|
||||
/// every OnDataChange and bought nothing: the read path already aliased
|
||||
/// the cache's instances into every BulkReadResult.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Set_StoresIndependentSnapshot_UnaffectedByLaterEventMutation()
|
||||
public void Set_BorrowsTheEventsOwnInstances_ByOwnershipContract()
|
||||
{
|
||||
MxAccessValueCache cache = new();
|
||||
Timestamp sourceTimestamp = Timestamp.FromDateTime(new(2026, 5, 19, 9, 0, 0, DateTimeKind.Utc));
|
||||
@@ -63,19 +77,12 @@ public sealed class MxAccessValueCacheTests
|
||||
|
||||
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.Same(mxEvent.Value, cached.Value);
|
||||
Assert.Same(mxEvent.SourceTimestamp, cached.SourceTimestamp);
|
||||
Assert.Same(mxEvent.Statuses, cached.Statuses);
|
||||
Assert.Same(mxEvent.Statuses[0], Assert.Single(cached.Statuses));
|
||||
Assert.Equal(192, cached.Quality);
|
||||
Assert.Equal(sourceTimestamp, cached.SourceTimestamp);
|
||||
Assert.Single(cached.Statuses);
|
||||
Assert.Equal(MxStatusCategory.Ok, cached.Statuses[0].Category);
|
||||
}
|
||||
|
||||
/// <summary>Verifies that TryGet returns false for unknown handles.</summary>
|
||||
|
||||
@@ -20,8 +20,11 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
|
||||
/// <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"/>).
|
||||
/// Enqueue via the mapper and satisfy this. "Does not mutate" is the load-
|
||||
/// bearing half for the post-publish value cache, which deliberately borrows
|
||||
/// the enqueued event's own value/timestamp/status instances rather than
|
||||
/// copying them (see <see cref="MxAccessValueCache.Set"/>) — it reads and
|
||||
/// serializes them, and this invariant is what keeps that safe.
|
||||
/// <para>
|
||||
/// The byte-budgeted <see cref="Drain(uint, int)"/> relies on the same invariant: each
|
||||
/// event's serialized size is measured once at enqueue and stored beside it, which is
|
||||
|
||||
@@ -57,7 +57,12 @@ public sealed class MxAccessValueCache
|
||||
/// <summary>Records a fresh OnDataChange payload for the given handle pair.</summary>
|
||||
/// <param name="serverHandle">MXAccess server handle.</param>
|
||||
/// <param name="itemHandle">MXAccess item handle.</param>
|
||||
/// <param name="mxEvent">The protobuf MxEvent created by the event mapper.</param>
|
||||
/// <param name="mxEvent">
|
||||
/// The protobuf MxEvent created by the event mapper, already handed to
|
||||
/// the outbound queue. The cache borrows this event's
|
||||
/// <c>Value</c>/<c>SourceTimestamp</c>/<c>Statuses</c> instances rather
|
||||
/// than copying them — see the ownership contract in the method body.
|
||||
/// </param>
|
||||
public void Set(
|
||||
int serverHandle,
|
||||
int itemHandle,
|
||||
@@ -68,20 +73,39 @@ 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();
|
||||
|
||||
// Ownership contract: the cache BORROWS, it does not copy. CachedValue
|
||||
// retains the event's own Value, SourceTimestamp, and Statuses
|
||||
// instances.
|
||||
//
|
||||
// Sound because the event is write-once by the time this runs.
|
||||
// MxAccessBaseEventSink.EnqueueEvent calls eventQueue.Enqueue first —
|
||||
// which stamps WorkerSequence/WorkerTimestamp inside the queue lock —
|
||||
// and only then runs the postPublish hook that lands here; the queue's
|
||||
// ownership invariant (see MxAccessEventQueue's class remarks) forbids
|
||||
// mutating an event after it is enqueued. The producer side never
|
||||
// reuses instances either: the mapper builds a fresh MxEvent, and the
|
||||
// VariantConverter a fresh MxValue, per COM callback. The three deep
|
||||
// copies this replaced (the MxValue — recursive for an MxArray — the
|
||||
// Timestamp, and the RepeatedField container plus every MxStatusProxy
|
||||
// in it) therefore bought nothing but garbage on the worker's hottest
|
||||
// path.
|
||||
//
|
||||
// Consumers of TryGet / TryWaitForUpdate may read and serialize what
|
||||
// they get back; they must never mutate it. ReadBulk already depends on
|
||||
// that: MxAccessSession.SucceededRead puts these very instances on the
|
||||
// BulkReadResult it returns, which the worker only serializes onto the
|
||||
// IPC pipe — worker↔gateway is a process boundary, so no gateway-side
|
||||
// consumer can alias them. Mutating one would corrupt the event still
|
||||
// queued for the outbound stream AND invalidate QueuedEvent.Size, the
|
||||
// serialized size memoized at enqueue that the byte-budgeted Drain
|
||||
// charges against its budget: a message grown after enqueue could
|
||||
// overshoot the negotiated frame max and fault the session with
|
||||
// MessageTooLarge (WorkerPipeSession.FaultOnOversizedEventAsync).
|
||||
//
|
||||
// Value is always set for OnDataChange; SourceTimestamp can be null when
|
||||
// the source timestamp could not be parsed. CachedValue's parameters are
|
||||
// annotated non-null but have always accepted a runtime null for both,
|
||||
// and the read side null-checks accordingly.
|
||||
long key = CreateItemKey(serverHandle, itemHandle);
|
||||
lock (syncRoot)
|
||||
{
|
||||
@@ -91,10 +115,10 @@ public sealed class MxAccessValueCache
|
||||
|
||||
entries[key] = new CachedValue(
|
||||
nextVersion,
|
||||
cachedValue,
|
||||
mxEvent.Value,
|
||||
mxEvent.Quality,
|
||||
cachedTimestamp,
|
||||
mxEvent.Statuses.Clone());
|
||||
mxEvent.SourceTimestamp,
|
||||
mxEvent.Statuses);
|
||||
}
|
||||
|
||||
// Signaled outside the lock: the waiter re-takes syncRoot (through
|
||||
@@ -228,23 +252,35 @@ public sealed class MxAccessValueCache
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Snapshot of the most recent OnDataChange payload for a handle pair.
|
||||
/// The most recent OnDataChange payload for a handle pair.
|
||||
/// <see cref="Version"/> increments by one on every <see cref="Set"/>
|
||||
/// call so the bulk read executor can detect "a new value arrived
|
||||
/// since I started waiting".
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Plain readonly struct (not a record) so this compiles under the
|
||||
/// worker's net48 target, which lacks <c>IsExternalInit</c>.
|
||||
/// <para>
|
||||
/// Borrowed references, not copies. <see cref="Value"/>,
|
||||
/// <see cref="SourceTimestamp"/>, and <see cref="Statuses"/> are the
|
||||
/// very instances hanging off the MxEvent that was enqueued for the
|
||||
/// outbound stream — <see cref="Set"/> carries the full ownership
|
||||
/// contract. Read them and serialize them; never mutate them and
|
||||
/// never hand them to something that will. That event is write-once
|
||||
/// from the moment it is enqueued, and its serialized size is
|
||||
/// memoized at that point for the byte-budgeted drain.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Plain readonly struct (not a record) so this compiles under the
|
||||
/// worker's net48 target, which lacks <c>IsExternalInit</c>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public readonly struct CachedValue
|
||||
{
|
||||
/// <summary>Initializes a new cached value snapshot.</summary>
|
||||
/// <param name="version">Version counter incremented on each update.</param>
|
||||
/// <param name="value">The MXAccess value.</param>
|
||||
/// <param name="value">The MXAccess value, borrowed from the enqueued event.</param>
|
||||
/// <param name="quality">The MXAccess quality code.</param>
|
||||
/// <param name="sourceTimestamp">The source timestamp of the value.</param>
|
||||
/// <param name="statuses">The MXAccess status codes.</param>
|
||||
/// <param name="sourceTimestamp">The source timestamp of the value, borrowed from the enqueued event.</param>
|
||||
/// <param name="statuses">The MXAccess status codes, borrowed from the enqueued event.</param>
|
||||
public CachedValue(
|
||||
ulong version,
|
||||
MxValue value,
|
||||
@@ -262,16 +298,16 @@ public sealed class MxAccessValueCache
|
||||
/// <summary>Monotonic per-handle version counter.</summary>
|
||||
public ulong Version { get; }
|
||||
|
||||
/// <summary>The cached MxValue payload.</summary>
|
||||
/// <summary>The OnDataChange event's own MxValue payload. Read-only to consumers.</summary>
|
||||
public MxValue Value { get; }
|
||||
|
||||
/// <summary>Quality code from the OnDataChange event.</summary>
|
||||
public int Quality { get; }
|
||||
|
||||
/// <summary>Source timestamp from the OnDataChange event.</summary>
|
||||
/// <summary>The OnDataChange event's own source timestamp. Read-only to consumers.</summary>
|
||||
public Timestamp SourceTimestamp { get; }
|
||||
|
||||
/// <summary>MxStatusProxy entries from the OnDataChange event.</summary>
|
||||
/// <summary>The OnDataChange event's own MxStatusProxy collection. Read-only to consumers.</summary>
|
||||
public RepeatedField<MxStatusProxy> Statuses { get; }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -73,6 +73,19 @@ public sealed class MxAccessWriteCompletionCache
|
||||
ulong version = entries.TryGetValue(key, out CompletionEntry existing)
|
||||
? existing.Version + 1
|
||||
: 1UL;
|
||||
|
||||
// Still a defensive copy, deliberately — this is NOT an oversight
|
||||
// left behind by the borrow that MxAccessValueCache.Set adopted (see
|
||||
// the ownership contract there). The value cache is handed the whole
|
||||
// enqueued MxEvent, so the queue's write-once ownership invariant
|
||||
// covers everything it retains. This method is handed a bare
|
||||
// RepeatedField whose provenance its signature cannot constrain:
|
||||
// the production sink does pass an enqueued event's Statuses, but
|
||||
// callers that build and keep their own rows are equally valid
|
||||
// against this API, and a borrowed alias would then let a later
|
||||
// mutation rewrite an already-recorded completion. The write path is
|
||||
// command-rate, not the per-OnDataChange streaming hot path, so the
|
||||
// clone costs nothing worth reclaiming.
|
||||
entries[key] = new CompletionEntry(version, statuses.Clone());
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user