diff --git a/docs/WorkerFrameProtocol.md b/docs/WorkerFrameProtocol.md index 36fe955..3bab2a3 100644 --- a/docs/WorkerFrameProtocol.md +++ b/docs/WorkerFrameProtocol.md @@ -88,7 +88,9 @@ priority order. A caller enqueues its frame into the control or event queue under a lock, then contends for a single write lock; whichever caller wins drains every frame queued at that moment, control frames first and each class in FIFO order, so a command reply, fault, heartbeat, or shutdown -acknowledgement is never delayed behind a backlog of queued events. Priority +acknowledgement is never delayed behind a backlog of queued events — neither +in the bytes written nor in the flush that marks them delivered (see the +class-boundary flush under flush coalescing below). Priority only reorders *which frame writes next* — it does not affect the sequence value a frame receives (see below), so a caller cannot infer priority class from the wire sequence. @@ -117,13 +119,36 @@ Two failure shapes are distinguished during a drain pass: and every frame still queued, then stops draining entirely so no caller waits forever on a stream that will not recover. -Flushes are coalesced across a drained batch: each frame in the batch is -written to the stream without an individual flush, then one `FlushAsync` -runs after the whole batch, and only then does every successfully-written -frame's completion resolve — so a caller's `WriteAsync` still does not -complete until its bytes are both written *and* flushed, but a batch that -happened to contain several queued frames pays one flush instead of one per -frame. Note the ordering this implies at the peer: the frames reach the pipe +Flushes are coalesced across a *run of same-class frames* inside a drain +pass: each frame in the run is written to the stream without an individual +flush, then one `FlushAsync` runs — at the end of the pass, and additionally +at every control-to-event boundary — and only then does every +successfully-written frame of that run resolve its completion. A caller's +`WriteAsync` therefore still does not complete until its bytes are both +written *and* flushed; what changed is *when* that moment arrives +for a control frame that a pass writes ahead of queued events. It used to be +the end of the pass, so a heartbeat, command reply, fault, or shutdown +acknowledgement was written first but only counted as delivered after up to a +full event batch had been written and flushed behind it. The boundary flush +closes the control run out before the events are written, so the priority +class governs the frame's delivery point and not just its byte order. The +cost stays bounded: a pure-event pass — the event hot path — still pays +exactly one flush however many frames drain together, a run of control +frames still pays one for the whole run (never one per heartbeat, the +syscall-per-frame cost the coalescing removed), and only a pass that actually +mixes both classes pays a second. + +One consequence of the boundary flush is worth stating: a control frame whose +run has already been flushed and completed is out of the drain's +written-but-unflushed set, so a *later* failure in the same pass — a broken +write, or a failed end-of-pass flush — no longer reaches back and fails it. +That is the honest outcome: its bytes were flushed, so it was delivered. A +failure of the boundary flush itself is treated exactly like a failed +end-of-pass flush, and additionally fails the event frame the drain had +already claimed off its queue (nothing else would ever complete it) along +with every frame still queued. + +Note the ordering all of this implies at the peer: the frames reach the pipe before the flush that follows them, so the gateway can read a whole batch while the writer has not yet flushed it. Anything observing the flush itself (a test counting flushes, for instance) must wait for the flush, not infer it @@ -134,12 +159,22 @@ drains them together, so a burst of N events costs one flush rather than N — the coalescing the batch machinery was built for now engages on the event hot path, not only when independent producers happen to queue behind a blocked write. Intra-batch order is preserved (FIFO enqueue under one lock), and a -concurrently queued control frame is still drained ahead of the batch. A -per-frame rejection inside a batch (for example one oversized event) surfaces -from the batch's awaited completions as that frame's +concurrently queued control frame is still drained — and now flushed and +completed — ahead of the batch's remaining events, which is why a batch a +control frame cuts into pays one extra flush while an uninterrupted batch +still pays exactly one. A per-frame rejection inside a batch (for example one +oversized event) surfaces from the batch's awaited completions as that frame's `WorkerFrameProtocolException`; the remaining completions are still observed so none faults unobserved. +The completion is the frame's delivery point, not necessarily the instant its +caller returns. A caller that loses the race for the write lock only observes +its own completion after the winning drainer releases the lock, so its return +remains bounded by that drain pass even though its control frame was flushed +and completed at the class boundary inside it. The boundary flush is what +makes the delivery point honest; unparking a lock-race loser from the winner's +pass would be a separate change to the enqueue-then-contend shape. + Cancellation of a `WriteAsync`/`WriteBatchAsync` call that is still waiting for the write lock when its token fires tombstones the queued frame: the cancelled caller marks its frame under `_gate`, and the draining lock-holder's 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 bdeea32..a58fed6 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs @@ -377,6 +377,14 @@ public sealed class WorkerFrameProtocolTests /// 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. /// Every frame still reaches the wire intact. + /// + /// The burst is all-event on purpose. The control-frame completion decoupling made the drain flush + /// at each control-to-event boundary, so a pass that mixes classes legitimately pays one flush per + /// class run; the property + /// worth pinning is that a run of same-class frames — the event hot path — still costs exactly one + /// flush no matter how many frames drain together. The mixed shape has its own count assertion in + /// . + /// /// /// A task that represents the asynchronous operation. [Fact] @@ -387,7 +395,7 @@ public sealed class WorkerFrameProtocolTests 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); + Task firstWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); await AwaitWithTimeoutAsync(stream.FirstWriteStarted); Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); @@ -410,6 +418,179 @@ public sealed class WorkerFrameProtocolTests } } + /// + /// Control-frame completion decoupling. A control frame's delivery point must not be charged for + /// the event backlog behind it. The priority scheduler already wrote control bytes first, + /// but a frame counts as + /// delivered only once flushed, and the pass deferred its single flush — and every completion — + /// until after the events. The drain now flushes at the control-to-event boundary: with two control + /// frames written and the first event write blocked inside the stream, the flush that closes out the + /// control run has already run, so the heartbeat or reply is on the pipe rather than waiting behind + /// the batch. Exactly two flushes for the pass — one per class run, not one per control frame. + /// + /// The assertion is on the flush, not on the queued callers' returned tasks, because those tasks are + /// still gated by the write lock they lost to the drainer (see the latency contract on + /// WorkerFrameWriter.WriteAsync): the completion resolves at the boundary flush, but a + /// lock-race loser observes it only once the drainer releases the lock. + /// + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainPass_MixedClasses_FlushesControlRunBeforeWritingEvents() + { + WorkerFrameProtocolOptions options = CreateOptions(); + // Frame 1 (control) gates the pass open; frame 3 is the pass's first event write, which blocks + // so the boundary flush can be observed with the event batch still unwritten. + using GatedWriteStream stream = new(secondGateWriteIndex: 3); + WorkerFrameWriter writer = new(stream, options); + + Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + Task secondControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control); + Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + Task eventWrite2 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + await Task.Delay(50); + + stream.ReleaseFirstWrite(); + + // The drain writes both control frames and is now blocked on the first event write. + await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted); + + // The control run was flushed before the event batch was written — not after it. + Assert.Equal(1, stream.FlushCount); + Assert.False(eventWrite1.IsCompleted); + Assert.False(eventWrite2.IsCompleted); + + stream.ReleaseSecondGateWrite(); + await AwaitWithTimeoutAsync( + Task.WhenAll(firstControl, secondControl, eventWrite1, eventWrite2)); + + // One flush per class run: the control run, then the event run at the end of the pass. + Assert.Equal(2, stream.FlushCount); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope frame1 = await reader.ReadAsync(); + WorkerEnvelope frame2 = await reader.ReadAsync(); + WorkerEnvelope frame3 = await reader.ReadAsync(); + WorkerEnvelope frame4 = await reader.ReadAsync(); + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame3.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame4.BodyCase); + Assert.Equal(stream.Length, stream.Position); + } + + /// + /// Control-frame completion decoupling, the inverse guard. An event frame's completion boundary is + /// still the end-of-pass flush: a pure-event pass takes no boundary flush, so with two events + /// already written and the third + /// blocked mid-write, nothing has been flushed and no event can have been reported delivered. Only + /// a class transition may move a flush earlier — a plain event backlog may not. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainPass_PureEventRun_DoesNotFlushBeforeThePassEnds() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using GatedWriteStream stream = new(secondGateWriteIndex: 3); + WorkerFrameWriter writer = new(stream, options); + + Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + Task eventWrite2 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + Task eventWrite3 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + await Task.Delay(50); + + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted); + + // Two event frames written, none flushed: no event frame's delivery point has been reached. + Assert.Equal(0, stream.FlushCount); + Assert.False(eventWrite1.IsCompleted); + Assert.False(eventWrite2.IsCompleted); + + stream.ReleaseSecondGateWrite(); + await AwaitWithTimeoutAsync(Task.WhenAll(eventWrite1, eventWrite2, eventWrite3)); + Assert.Equal(1, stream.FlushCount); + } + + /// + /// Control-frame completion decoupling. The boundary flush is charged per class run, not per + /// control frame: a pass carrying nothing but control frames still pays exactly one flush. Flushing + /// after every control frame + /// would reinstate the syscall-per-heartbeat cost WRK-12 removed. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainPass_PureControlRun_FlushesOnce() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using GatedWriteStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + Task secondControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + Task thirdControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control); + await Task.Delay(50); + + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(Task.WhenAll(firstControl, secondControl, thirdControl)); + + Assert.Equal(1, stream.FlushCount); + } + + /// + /// Control-frame completion decoupling, the new failure window. The boundary flush is a new place + /// the pipe can break with frames written but not yet delivered, so it must fail exactly like the + /// end-of-pass flush: every written control + /// frame fails, and so do the event frame the drain had already claimed off its queue (nothing else + /// would ever complete it) and every frame still queued, so no caller waits forever on a stream that + /// will not recover. The event bytes never reach the wire — the drain stops at the fault. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainPass_WhenBoundaryFlushFails_FailsWrittenClaimedAndQueuedFrames() + { + const string faultMessage = "boundary flush failed"; + WorkerFrameProtocolOptions options = CreateOptions(); + using FlushFaultingGatedStream stream = new(faultMessage); + WorkerFrameWriter writer = new(stream, options); + + Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + Task secondControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control); + Task claimedEvent = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + Task queuedEvent = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + await Task.Delay(50); + + // The drain writes both control frames, claims the first event, and faults on the boundary flush. + stream.ReleaseFirstWrite(); + + // AwaitWithTimeoutAsync turns a frame nobody ever completes into a TimeoutException — a failed + // assertion rather than a hung test run. + foreach (Task write in new[] { firstControl, secondControl, claimedEvent, queuedEvent }) + { + IOException failure = await Assert.ThrowsAsync( + async () => await AwaitWithTimeoutAsync(write)); + Assert.Equal(faultMessage, failure.Message); + } + + // Only the control frames reached the wire; the claimed event was never written. + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope frame1 = await reader.ReadAsync(); + WorkerEnvelope frame2 = await reader.ReadAsync(); + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase); + Assert.Equal(stream.Length, stream.Position); + } + /// /// Verifies a per-frame rejection does not burn a sequence number (WRK-23). The sequence is a /// diagnostic counter, so a gap breaks nothing functionally — but an operator correlating a pipe @@ -877,24 +1058,108 @@ public sealed class WorkerFrameProtocolTests } // A MemoryStream whose first WriteAsync blocks until released, so a test can queue additional frames - // behind an in-progress write and observe the writer's priority ordering. + // behind an in-progress write and observe the writer's priority ordering. A second, optional gate on + // a chosen write index lets a test stop a drain pass mid-flight — at a class boundary, say — and + // sample what the writer has already flushed while the rest of the pass is still unwritten. private sealed class GatedWriteStream : MemoryStream { private readonly SemaphoreSlim _release = new SemaphoreSlim(0); + private readonly SemaphoreSlim _secondGateRelease = new SemaphoreSlim(0); private readonly TaskCompletionSource _firstWriteStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _secondGateWriteStarted = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly int _secondGateWriteIndex; private int _writeCount; private int _flushCount; + /// Initializes a new instance of the GatedWriteStream class. + /// + /// One-based index of a later write to block as well, or 0 (the default) to gate only the first + /// write. Write indexes start at 1, so 0 never matches. + /// + public GatedWriteStream(int secondGateWriteIndex = 0) + { + _secondGateWriteIndex = secondGateWriteIndex; + } + /// Gets a task that completes once the first call has started blocking. public Task FirstWriteStarted => _firstWriteStarted.Task; + /// Gets a task that completes once the second gated call has started blocking. + public Task SecondGateWriteStarted => _secondGateWriteStarted.Task; + /// Gets the number of calls observed so far. public int FlushCount => Volatile.Read(ref _flushCount); /// Releases the first blocked write so it can complete. public void ReleaseFirstWrite() => _release.Release(); + /// Releases the second gated write so it can complete. + public void ReleaseSecondGateWrite() => _secondGateRelease.Release(); + + /// + public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + int writeIndex = Interlocked.Increment(ref _writeCount); + if (writeIndex == 1) + { + _firstWriteStarted.TrySetResult(true); + await _release.WaitAsync(cancellationToken); + } + else if (writeIndex == _secondGateWriteIndex) + { + _secondGateWriteStarted.TrySetResult(true); + await _secondGateRelease.WaitAsync(cancellationToken); + } + + 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) + { + _release.Dispose(); + _secondGateRelease.Dispose(); + } + + base.Dispose(disposing); + } + } + + // A MemoryStream whose first write blocks until released and whose every FlushAsync throws, so a test + // can fault the class-boundary flush with control frames already written and an event frame already + // claimed off its queue. + private sealed class FlushFaultingGatedStream : MemoryStream + { + private readonly SemaphoreSlim _release = new SemaphoreSlim(0); + private readonly TaskCompletionSource _firstWriteStarted = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly string _faultMessage; + private int _writeCount; + + /// Initializes a new instance of the FlushFaultingGatedStream class. + /// Message carried by the every flush throws. + public FlushFaultingGatedStream(string faultMessage) + { + _faultMessage = faultMessage; + } + + /// Gets a task that completes once the first call has started blocking. + public Task FirstWriteStarted => _firstWriteStarted.Task; + + /// Releases the first blocked write so it can complete. + public void ReleaseFirstWrite() => _release.Release(); + /// public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) { @@ -910,8 +1175,7 @@ public sealed class WorkerFrameProtocolTests /// public override Task FlushAsync(CancellationToken cancellationToken) { - Interlocked.Increment(ref _flushCount); - return base.FlushAsync(cancellationToken); + return Task.FromException(new IOException(_faultMessage)); } /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs index cfb20e7..61db4c7 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs @@ -14,9 +14,11 @@ namespace ZB.MOM.WW.MxGateway.Worker.Ipc; /// Writes worker frames to a stream with length-prefixed protobuf serialization. Callers enqueue a /// frame at a and then contend for a single write lock; whoever /// holds the lock drains every queued frame, control frames first, so a reply, fault, or heartbeat is -/// never delayed behind an event backlog. The envelope Sequence is stamped by the -/// draining lock-holder at the moment of writing, so the on-wire order and the stamped sequence always -/// agree even under concurrent callers and priority reordering. +/// never delayed behind an event backlog — neither in the bytes it writes nor in the flush that +/// delivers them, because the drain flushes at every control-to-event boundary rather than only at the +/// end of the pass. The envelope Sequence is stamped by the draining lock-holder at the moment +/// of writing, so the on-wire order and the stamped sequence always agree even under concurrent callers +/// and priority reordering. /// public sealed class WorkerFrameWriter { @@ -24,15 +26,25 @@ public sealed class WorkerFrameWriter { /// Initializes a new instance of the PendingFrame class. /// Worker envelope awaiting write. - public PendingFrame(WorkerEnvelope envelope) + /// Priority class the frame was queued at. + public PendingFrame(WorkerEnvelope envelope, WorkerFrameWritePriority priority) { Envelope = envelope; + IsControl = priority != WorkerFrameWritePriority.Event; Completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); } /// Gets the worker envelope awaiting write. public WorkerEnvelope Envelope { get; } + /// + /// Gets a value indicating whether this frame was queued as control-plane traffic. Recorded at + /// construction from the same expression that picks the queue, so the class the drain sees can + /// never disagree with the queue the frame sits in. The drain uses it to flush and complete + /// written control frames at the moment it turns to events (see ). + /// + public bool IsControl { get; } + /// Gets the completion source signaled once the frame has been written or has failed. public TaskCompletionSource Completion { get; } @@ -95,6 +107,15 @@ public sealed class WorkerFrameWriter /// the canceller behind the very write it is abandoning would defeat the point of cancellation. /// The abandoned frame's completion gets a fault-observing continuation so a write failure after /// the caller unwinds never raises an unobserved-task exception (NEXT-04). + /// + /// Latency contract: a control frame's bytes are written, flushed, and its completion + /// resolved before the events a drain pass writes after it — the delivery point of a heartbeat, + /// reply, fault, or shutdown ack is never charged for the event backlog behind it. The returned + /// task can still be later than that instant for a caller that lost the write-lock race: it only + /// observes its completion after the winning drainer releases the lock, so its own return remains + /// bounded by that pass. That parking is deliberate — the alternative is to race the lock wait + /// against the completion, which buys nothing for the frame's delivery. + /// /// public async Task WriteAsync( WorkerEnvelope envelope, @@ -106,7 +127,7 @@ public sealed class WorkerFrameWriter throw new ArgumentNullException(nameof(envelope)); } - PendingFrame frame = new PendingFrame(envelope); + PendingFrame frame = new PendingFrame(envelope, priority); lock (_gate) { if (priority == WorkerFrameWritePriority.Event) @@ -153,8 +174,12 @@ public sealed class WorkerFrameWriter /// rather than one per frame (WRK-25, realizing the WRK-12 coalescing on the path it was built /// for). Intra-batch order is preserved because the enqueue is atomic under _gate and each /// class queue is FIFO; the control-before-event guarantee still holds because any concurrently - /// queued control frame is drained ahead of this batch by . Every frame's - /// "written and flushed before completion" contract is unchanged. + /// queued control frame is drained ahead of this batch by , and — since + /// the control-frame completion decoupling — is also flushed and completed before this batch's + /// remaining events are written, so a batch in flight does not delay a control frame's delivery. + /// Every frame's "written and flushed before completion" contract is unchanged. An event batch that + /// a control frame cuts into therefore pays one extra flush; an uninterrupted batch still pays + /// exactly one. /// /// Envelopes to write, in order. /// Scheduling priority for the whole batch. @@ -189,7 +214,7 @@ public sealed class WorkerFrameWriter { WorkerEnvelope envelope = envelopes[index] ?? throw new ArgumentException("Batch envelopes must not contain null.", nameof(envelopes)); - frames[index] = new PendingFrame(envelope); + frames[index] = new PendingFrame(envelope, priority); } lock (_gate) @@ -296,14 +321,24 @@ public sealed class WorkerFrameWriter // 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. + // Flushes are coalesced within a priority class rather than blindly across the whole pass (WRK-12 / + // IPC-15, narrowed by the control-frame completion decoupling): each frame is written to the stream + // but not flushed individually, and one FlushAsync runs at the end of the pass — plus one at each + // control-to-event boundary, which flushes and completes the control frames written so far before + // the event backlog behind them is written, instead of after it. Without that boundary flush the + // priority scheduler only got control *bytes* out early: their delivery point, and every waiting + // caller's completion, still sat behind up to a full event batch. + // + // A caller's Completion therefore still signals only after its bytes have been written AND flushed — + // the contract is unchanged, the moment it is reached simply stops being pinned to the end of the + // pass. Cost is bounded: a pure-event pass (the event hot path) still pays exactly one flush, a burst + // of control frames still pays one for the whole burst, and only a pass that actually mixes both + // classes pays a second — never one flush per control frame, which is the syscall-per-heartbeat cost + // WRK-12 removed. private async Task DrainQueuedFramesAsync() { List written = new List(); + bool writtenHoldsControl = false; while (true) { PendingFrame? frame = DequeueNext(); @@ -312,10 +347,40 @@ public sealed class WorkerFrameWriter break; } + if (writtenHoldsControl && !frame.IsControl) + { + // Class transition: the frames written so far include at least one control frame whose + // caller is waiting on delivery. Flush and complete them here rather than parking them + // behind the events this pass is about to write. Charged once per transition, not once + // per control frame. Event frames already in the list ride along — they too are written + // and now flushed, so completing them early is the same contract, earlier. + try + { + await _stream.FlushAsync(CancellationToken.None).ConfigureAwait(false); + } + catch (Exception exception) + { + // Same shape as the end-of-pass flush failure: the bytes reached the stream but the + // flush that guarantees delivery failed, so the pipe is broken. Fail the frame just + // claimed (it is out of its queue and nothing else will ever complete it), every + // written-but-unflushed frame, and everything still queued, then stop draining. + frame.Completion.TrySetException(exception); + FailFrames(written, exception); + FailAllQueued(exception); + return; + } + + // Completed frames leave the list, so a later failure in this pass cannot fail them. + CompleteFrames(written); + written.Clear(); + writtenHoldsControl = false; + } + try { await WriteFrameAsync(frame.Envelope).ConfigureAwait(false); written.Add(frame); + writtenHoldsControl |= frame.IsControl; } catch (WorkerFrameProtocolException exception) when (IsPerFrameRejection(exception)) { @@ -348,13 +413,19 @@ public sealed class WorkerFrameWriter 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. + // broken. Fail every frame still in the batch (the queue was already drained) so no caller + // treats an unflushed write as delivered. Frames a boundary flush already completed are not + // in the list — their bytes were flushed, so this later failure does not reach back to them. FailFrames(written, exception); return; } - foreach (PendingFrame frame in written) + CompleteFrames(written); + } + + private static void CompleteFrames(List frames) + { + foreach (PendingFrame frame in frames) { frame.Completion.TrySetResult(true); }