perf(worker): unpark awaited control-frame writers from the winning drain pass
WriteAsync enqueued its frame and then contended unconditionally for the write lock, so a caller that lost the race stayed in WaitAsync until the winning drainer released — even though that winner writes, flushes, and completes the loser's control frame at the control-to-event class boundary, part-way through its pass. The boundary flush made the delivery point honest; the awaited task was still charged for the whole event backlog it had just been flushed ahead of. WriteAsync now awaits its own frame's completion racing the lock acquisition. Completion first: the caller returns at its frame's delivery point and the outstanding acquisition is detached, not dropped — a continuation drains whatever is queued and releases, so the lock is never acquired and silently held and a frame enqueued between the previous drainer's last dequeue and its release is still written. Lock first: drain as before. Cancellation keeps the WRK-22 tombstone semantics exactly, and a wait cancelled after the caller has already detached releases nothing (SemaphoreSlim hands no count to a wait it cancels), so no count leaks and no queued frame is stranded. A token that fires after the frame's completion won the race changes nothing — the frame was delivered. WriteBatchAsync deliberately keeps the plain wait-then-drain shape: its last completion resolves at the end-of-pass flush anyway. Three tests: the latency win (a control caller returning while the winning WriteBatchAsync event burst is demonstrably still blocked mid-pass), a mixed-priority concurrency soak pinning exactly-once writes and a single drainer, and the cancel-after-detach corner (a wrongly released count would surface as the drainer's own Release throwing SemaphoreFullException). edited on macOS, windev verification pending (plan Task 11). Verified here by compiling and running WorkerFrameWriter plus the writer suite against net10.0 in a scratch harness: 31/31 pass, and the two behaviour-pinning tests fail against the pre-change parked implementation.
This commit is contained in:
Joseph Doherty
@@ -428,10 +428,10 @@ public sealed class WorkerFrameProtocolTests
|
||||
/// 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.
|
||||
/// The assertion here is on the flush and on the <em>event</em> callers, whose delivery point is
|
||||
/// still the end-of-pass flush. That the control frame's own caller returns at the boundary — the
|
||||
/// lock-parking this pass used to impose on it — is the separate subject of
|
||||
/// <see cref="WriteAsync_ControlFrameLosingLockRaceToEventBatch_ReturnsAtItsOwnCompletion"/>.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
@@ -912,6 +912,201 @@ public sealed class WorkerFrameProtocolTests
|
||||
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, f4.BodyCase);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Frame-writer lock-parking, closed. The class-boundary flush made a control frame's <em>delivery
|
||||
/// point</em> honest, but the caller that lost the write-lock race still could not observe it: it
|
||||
/// sat in <c>WaitAsync</c> until the winning drainer released the lock, so its awaited task was
|
||||
/// charged for the whole event backlog the boundary flush had just jumped it ahead of. The caller
|
||||
/// now races its own frame's completion against the lock acquisition, so it returns at the boundary.
|
||||
/// <para>
|
||||
/// The winner here is a <c>WriteBatchAsync</c> event burst — the production hot path, the event
|
||||
/// drain loop's own call — gated so the pass is stopped inside the batch, after the boundary flush
|
||||
/// that delivered the control frame and long before the pass ends. The control caller returning
|
||||
/// while the drainer is demonstrably still blocked mid-batch is the whole property; under the parked
|
||||
/// shape this await could not return until <c>ReleaseSecondGateWrite</c>, so a regression shows up
|
||||
/// as <c>AwaitWithTimeoutAsync</c>'s <see cref="TimeoutException"/> rather than as a hang.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task WriteAsync_ControlFrameLosingLockRaceToEventBatch_ReturnsAtItsOwnCompletion()
|
||||
{
|
||||
WorkerFrameProtocolOptions options = CreateOptions();
|
||||
// Write 1 (the batch's first event) gates the pass open; write 3 is the first event written
|
||||
// after the control frame's boundary flush, so blocking it stops the drain with the control
|
||||
// frame delivered and the rest of the batch still unwritten.
|
||||
using GatedWriteStream stream = new(secondGateWriteIndex: 3);
|
||||
WorkerFrameWriter writer = new(stream, options);
|
||||
|
||||
WorkerEnvelope[] batch = new[]
|
||||
{
|
||||
CreateEventEnvelope(workerSequence: 1),
|
||||
CreateEventEnvelope(workerSequence: 2),
|
||||
CreateEventEnvelope(workerSequence: 3),
|
||||
CreateEventEnvelope(workerSequence: 4),
|
||||
};
|
||||
|
||||
Task batchWrite = writer.WriteBatchAsync(batch, WorkerFrameWritePriority.Event);
|
||||
await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
|
||||
|
||||
// The control frame loses the lock race to the batch and is drained by it.
|
||||
Task controlWrite = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control);
|
||||
await Task.Delay(50);
|
||||
|
||||
stream.ReleaseFirstWrite();
|
||||
await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted);
|
||||
|
||||
// The boundary flush ran and the drain is now blocked inside the batch behind it.
|
||||
Assert.Equal(1, stream.FlushCount);
|
||||
Assert.False(batchWrite.IsCompleted);
|
||||
|
||||
// The latency win: the loser returns here, with the winner's pass still in flight.
|
||||
await AwaitWithTimeoutAsync(controlWrite);
|
||||
Assert.False(batchWrite.IsCompleted);
|
||||
|
||||
stream.ReleaseSecondGateWrite();
|
||||
await AwaitWithTimeoutAsync(batchWrite);
|
||||
|
||||
// One flush per class run, unchanged: the control run's boundary flush, then the batch's.
|
||||
Assert.Equal(2, stream.FlushCount);
|
||||
|
||||
// Detaching the abandoned lock wait strands nothing: every frame is on the wire exactly once,
|
||||
// in drain order, with contiguous write-time sequences.
|
||||
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();
|
||||
WorkerEnvelope frame5 = await reader.ReadAsync();
|
||||
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, 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(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame5.BodyCase);
|
||||
Assert.Equal(
|
||||
new ulong[] { 1, 2, 3, 4, 5 },
|
||||
new[] { frame1.Sequence, frame2.Sequence, frame3.Sequence, frame4.Sequence, frame5.Sequence });
|
||||
Assert.Equal(stream.Length, stream.Position);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Frame-writer lock-parking, the "nothing is stranded" half. A caller that returns on its
|
||||
/// completion abandons a live write-lock acquisition; that acquisition still carries drain
|
||||
/// responsibility, because a frame can be enqueued after the winning drainer's last dequeue and
|
||||
/// before its release. Under heavy mixed-priority concurrency — every caller racing its completion
|
||||
/// against the lock, so detached acquisitions pile up — every frame must still be written exactly
|
||||
/// once, and the write lock must still admit exactly one drainer: a double-drain would interleave
|
||||
/// two passes over the same stream, and a double-release would either do that or throw
|
||||
/// <see cref="SemaphoreFullException"/> out of a later drain. Contiguous 1..N sequences with no
|
||||
/// duplicates and no trailing bytes is the observable form of both.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task WriteAsync_UnderMixedPriorityConcurrency_WritesEveryQueuedFrameExactlyOnce()
|
||||
{
|
||||
const int perClass = 40;
|
||||
WorkerFrameProtocolOptions options = CreateOptions();
|
||||
using MemoryStream stream = new();
|
||||
WorkerFrameWriter writer = new(stream, options);
|
||||
|
||||
Task[] writes = new Task[perClass * 2];
|
||||
for (int index = 0; index < perClass; index++)
|
||||
{
|
||||
writes[index * 2] = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
|
||||
writes[(index * 2) + 1] = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
|
||||
}
|
||||
|
||||
await AwaitWithTimeoutAsync(Task.WhenAll(writes));
|
||||
|
||||
// A detached acquisition drains whatever it finds and releases; this write goes through the
|
||||
// same lock afterwards, so it can only succeed if the lock was left in a usable state.
|
||||
await AwaitWithTimeoutAsync(
|
||||
writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control));
|
||||
|
||||
const int total = (perClass * 2) + 1;
|
||||
int controlCount = 0;
|
||||
int eventCount = 0;
|
||||
stream.Position = 0;
|
||||
WorkerFrameReader reader = new(stream, options);
|
||||
for (int index = 0; index < total; index++)
|
||||
{
|
||||
WorkerEnvelope frame = await reader.ReadAsync();
|
||||
Assert.Equal((ulong)(index + 1), frame.Sequence);
|
||||
if (frame.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerEvent)
|
||||
{
|
||||
eventCount++;
|
||||
}
|
||||
else
|
||||
{
|
||||
controlCount++;
|
||||
}
|
||||
}
|
||||
|
||||
Assert.Equal(perClass + 1, controlCount);
|
||||
Assert.Equal(perClass, eventCount);
|
||||
// No frame was written twice and none was left queued.
|
||||
Assert.Equal(stream.Length, stream.Position);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Frame-writer lock-parking, the cancellation corner. A caller that returned on its completion
|
||||
/// leaves a live lock acquisition behind; if its token then fires, that acquisition is cancelled
|
||||
/// after the caller is long gone. <see cref="SemaphoreSlim"/> hands no count to a wait it cancels,
|
||||
/// so the detached continuation must release nothing on that path — releasing there would push the
|
||||
/// count past the maximum and make the drainer's own <c>Release</c> throw
|
||||
/// <see cref="SemaphoreFullException"/>, which is exactly what awaiting the drainer's write here
|
||||
/// detects. The late cancellation must also not retroactively cancel the call that already
|
||||
/// returned, nor tombstone a frame that is already on the wire.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task WriteAsync_TokenCancelledAfterCompletionFirstReturn_LeavesTheWriteLockIntact()
|
||||
{
|
||||
WorkerFrameProtocolOptions options = CreateOptions();
|
||||
using GatedWriteStream stream = new(secondGateWriteIndex: 3);
|
||||
WorkerFrameWriter writer = new(stream, options);
|
||||
|
||||
// The drainer holds the lock, blocked writing its own control frame.
|
||||
Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
|
||||
await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
Task lateControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control, cts.Token);
|
||||
Task eventWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
|
||||
await Task.Delay(50);
|
||||
|
||||
// The drain writes both control frames, flushes them at the class boundary, then blocks on the
|
||||
// event write — so the cancellable caller returns on its completion with its acquisition live.
|
||||
stream.ReleaseFirstWrite();
|
||||
await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted);
|
||||
await AwaitWithTimeoutAsync(lateControl);
|
||||
|
||||
// Cancel the acquisition nobody is waiting on any more.
|
||||
cts.Cancel();
|
||||
|
||||
stream.ReleaseSecondGateWrite();
|
||||
|
||||
// A count released on the cancelled path would surface here, as the drainer's release throwing.
|
||||
await AwaitWithTimeoutAsync(Task.WhenAll(firstControl, eventWrite));
|
||||
|
||||
// The lock is still usable, and the cancelled token did not recall the delivered frame.
|
||||
await AwaitWithTimeoutAsync(
|
||||
writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control));
|
||||
|
||||
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.GatewayHello, frame4.BodyCase);
|
||||
Assert.Equal(stream.Length, stream.Position);
|
||||
}
|
||||
|
||||
private static WorkerEnvelope CreateGatewayHelloEnvelope(ulong sequence = 1)
|
||||
{
|
||||
return new WorkerEnvelope
|
||||
|
||||
Reference in New Issue
Block a user