perf(worker): control-frame completions resolve at the class-transition flush, not after the event batch
The two-class writer already got control bytes out ahead of a queued event backlog, but a frame counts as delivered only once flushed, and the drain deferred its single FlushAsync — and every TrySetResult — to the end of the pass. A heartbeat, command reply, fault, or shutdown ack was therefore written first and completed last, behind up to a full 128-frame event batch. The drain now records each frame's priority class on PendingFrame and flushes at every control-to-event boundary, completing and clearing the written set there. Cost stays bounded: a pure-event pass still pays exactly one flush, a run of control frames still pays one for the run, and only a pass that mixes both classes pays a second — never one flush per control frame, the syscall-per-heartbeat cost WRK-12 removed. A boundary flush that itself fails is a new failure window and is handled like the end-of-pass flush failure, additionally failing the event frame the drain had already claimed off its queue and every frame still queued. Frames a boundary flush completed leave the written set, so a later failure in the same pass can no longer reach back and fail an already-delivered control frame. The awaited task of a caller that lost the write-lock race is still bounded by the winning drainer's pass — that enqueue-then-contend parking is unchanged and now documented on WriteAsync and in docs/WorkerFrameProtocol.md.
This commit is contained in:
Joseph Doherty
@@ -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.
|
||||
/// <para>
|
||||
/// 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
|
||||
/// <see cref="DrainPass_MixedClasses_FlushesControlRunBeforeWritingEvents"/>.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <em>bytes</em> 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.
|
||||
/// <para>
|
||||
/// 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
|
||||
/// <c>WorkerFrameWriter.WriteAsync</c>): the completion resolves at the boundary flush, but a
|
||||
/// lock-race loser observes it only once the drainer releases the lock.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<IOException>(
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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<bool> _firstWriteStarted =
|
||||
new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private readonly TaskCompletionSource<bool> _secondGateWriteStarted =
|
||||
new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private readonly int _secondGateWriteIndex;
|
||||
private int _writeCount;
|
||||
private int _flushCount;
|
||||
|
||||
/// <summary>Initializes a new instance of the GatedWriteStream class.</summary>
|
||||
/// <param name="secondGateWriteIndex">
|
||||
/// 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.
|
||||
/// </param>
|
||||
public GatedWriteStream(int secondGateWriteIndex = 0)
|
||||
{
|
||||
_secondGateWriteIndex = secondGateWriteIndex;
|
||||
}
|
||||
|
||||
/// <summary>Gets a task that completes once the first <see cref="WriteAsync"/> call has started blocking.</summary>
|
||||
public Task FirstWriteStarted => _firstWriteStarted.Task;
|
||||
|
||||
/// <summary>Gets a task that completes once the second gated <see cref="WriteAsync"/> call has started blocking.</summary>
|
||||
public Task SecondGateWriteStarted => _secondGateWriteStarted.Task;
|
||||
|
||||
/// <summary>Gets the number of <see cref="FlushAsync"/> calls observed so far.</summary>
|
||||
public int FlushCount => Volatile.Read(ref _flushCount);
|
||||
|
||||
/// <summary>Releases the first blocked write so it can complete.</summary>
|
||||
public void ReleaseFirstWrite() => _release.Release();
|
||||
|
||||
/// <summary>Releases the second gated write so it can complete.</summary>
|
||||
public void ReleaseSecondGateWrite() => _secondGateRelease.Release();
|
||||
|
||||
/// <inheritdoc />
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override Task FlushAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
Interlocked.Increment(ref _flushCount);
|
||||
return base.FlushAsync(cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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<bool> _firstWriteStarted =
|
||||
new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private readonly string _faultMessage;
|
||||
private int _writeCount;
|
||||
|
||||
/// <summary>Initializes a new instance of the FlushFaultingGatedStream class.</summary>
|
||||
/// <param name="faultMessage">Message carried by the <see cref="IOException"/> every flush throws.</param>
|
||||
public FlushFaultingGatedStream(string faultMessage)
|
||||
{
|
||||
_faultMessage = faultMessage;
|
||||
}
|
||||
|
||||
/// <summary>Gets a task that completes once the first <see cref="WriteAsync"/> call has started blocking.</summary>
|
||||
public Task FirstWriteStarted => _firstWriteStarted.Task;
|
||||
|
||||
/// <summary>Releases the first blocked write so it can complete.</summary>
|
||||
public void ReleaseFirstWrite() => _release.Release();
|
||||
|
||||
/// <inheritdoc />
|
||||
public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
|
||||
{
|
||||
@@ -910,8 +1175,7 @@ public sealed class WorkerFrameProtocolTests
|
||||
/// <inheritdoc />
|
||||
public override Task FlushAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
Interlocked.Increment(ref _flushCount);
|
||||
return base.FlushAsync(cancellationToken);
|
||||
return Task.FromException(new IOException(_faultMessage));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
Reference in New Issue
Block a user