Skip to content

fix: Body Pump #52

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
st0o0 merged 21 commits into from
Jun 24, 2026
Merged

fix: Body Pump #52

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
762848d
feat(body): update IBodyDrainTarget interface and remove DrainContinue
st0o0 Jun 24, 2026
5de10a0
feat(body): unseal BodyDrainSlot, add FlowControlledDrainSlot
st0o0 Jun 24, 2026
1fa000a
fix(test): update starvation guard tests after removal
st0o0 Jun 24, 2026
a493eef
feat(h2): add Reserve/Refund to FlowController for window reservation
st0o0 Jun 24, 2026
c447452
feat(body): implement BodyPumpBase with credit system and adaptive bu…
st0o0 Jun 24, 2026
4f85d62
fix(body): fix Cancel stale-id leak and document IsStreamEligible con…
st0o0 Jun 24, 2026
ea7bac4
refactor(body): rewrite SerialBodyPump on BodyPumpBase
st0o0 Jun 24, 2026
0a5da93
refactor(body): rewrite MultiplexedBodyPump on BodyPumpBase
st0o0 Jun 24, 2026
d4c5781
refactor(body): rewrite FlowControlledBodyPump on BodyPumpBase
st0o0 Jun 24, 2026
ec83b71
feat(body): wire OnOutboundFlushed → AddCredit across all protocols
st0o0 Jun 24, 2026
f814f6b
test(body): expand credit-driven body pump test coverage
st0o0 Jun 24, 2026
ae7e1a4
chore(body): remove dead code from previous pump iterations
st0o0 Jun 24, 2026
5400f22
fix(body): fix pump stall on async PipeReader streams
st0o0 Jun 24, 2026
358abf8
fix(h2): fix FlowControlledBodyPump OnWindowUpdate deadlock + window …
st0o0 Jun 24, 2026
8f7c366
fix(body): reclaim credit after synchronous read completion
st0o0 Jun 24, 2026
6de49b9
fix(body): add inline AddCredit to all EmitDataFrames implementations
st0o0 Jun 24, 2026
aa0afe4
fix(body): prevent double slot cleanup on drain complete + fix test
st0o0 Jun 24, 2026
7638365
test(body): add regression tests for 6 body pump bugfixes
st0o0 Jun 24, 2026
1e9567f
fix(body): restore client-side backpressure and remove sync credit re…
st0o0 Jun 24, 2026
04c428d
test(body): add edge case tests for pump cancellation, boundaries, an…
st0o0 Jun 24, 2026
b33386d
refactor(body): simplify pump architecture
st0o0 Jun 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 13 additions & 101 deletions src/TurboHTTP.Tests/Protocol/Body/BodyDrainSlotSpec.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,10 @@ public void Initialize_should_set_identity_fields()
var cts = new CancellationTokenSource();
using var linked = CancellationTokenSource.CreateLinkedTokenSource(cts.Token);

slot.Initialize(42, stream, contentLength: 3, CancellationToken.None, linked);
slot.Initialize(42, stream, CancellationToken.None, linked);

Assert.Equal(42, slot.StreamId);
Assert.Same(stream, slot.BodyStream);
Assert.Equal(3L, slot.ContentLength);
Assert.Same(linked, slot.LinkedCts);
}

Expand All @@ -32,66 +31,16 @@ public void BeginRead_should_set_IsReadInFlight()
}

[Fact(Timeout = 5000)]
public void CompleteSyncRead_should_clear_IsReadInFlight()
public void CompleteRead_should_clear_IsReadInFlight()
{
var slot = new BodyDrainSlot<int>();
slot.BeginRead();

slot.CompleteSyncRead();
slot.CompleteRead();

Assert.False(slot.IsReadInFlight);
}

[Fact(Timeout = 5000)]
public void CompleteAsyncRead_should_clear_IsReadInFlight_and_reset_sync_counter()
{
var slot = new BodyDrainSlot<int>();
slot.BeginRead();
slot.IncrementSyncReads(100);
slot.IncrementSyncReads(100);

slot.CompleteAsyncRead();

Assert.False(slot.IsReadInFlight);
Assert.Equal(0, slot.ConsecutiveSyncReads);
}

[Fact(Timeout = 5000)]
public void IncrementSyncReads_should_return_false_below_threshold()
{
var slot = new BodyDrainSlot<int>();

var result = slot.IncrementSyncReads(3);

Assert.False(result);
Assert.Equal(1, slot.ConsecutiveSyncReads);
}

[Fact(Timeout = 5000)]
public void IncrementSyncReads_should_return_true_at_threshold()
{
var slot = new BodyDrainSlot<int>();

slot.IncrementSyncReads(3);
slot.IncrementSyncReads(3);
var result = slot.IncrementSyncReads(3);

Assert.True(result);
Assert.Equal(3, slot.ConsecutiveSyncReads);
}

[Fact(Timeout = 5000)]
public void ResetSyncReads_should_zero_counter()
{
var slot = new BodyDrainSlot<int>();
slot.IncrementSyncReads(100);
slot.IncrementSyncReads(100);

slot.ResetSyncReads();

Assert.Equal(0, slot.ConsecutiveSyncReads);
}

[Fact(Timeout = 5000)]
public void MarkOrphaned_should_set_IsOrphaned()
{
Expand All @@ -103,49 +52,18 @@ public void MarkOrphaned_should_set_IsOrphaned()
}

[Fact(Timeout = 5000)]
public void MarkDrainComplete_should_set_IsDrainComplete()
{
var slot = new BodyDrainSlot<int>();

slot.MarkDrainComplete();

Assert.True(slot.IsDrainComplete);
}

[Fact(Timeout = 5000)]
public void StoreLimbo_should_set_HasLimbo_and_LimboData()
{
var slot = new BodyDrainSlot<int>();
var data = new byte[] { 10, 20, 30 };

slot.StoreLimbo(data);

Assert.True(slot.HasLimbo);
Assert.Equal(data, slot.LimboData.ToArray());
}

[Fact(Timeout = 5000)]
public void ShrinkLimbo_should_advance_slice()
public void ReservedWindow_should_default_to_zero()
{
var slot = new BodyDrainSlot<int>();
var data = new byte[] { 1, 2, 3, 4, 5 };
slot.StoreLimbo(data);

slot.ShrinkLimbo(2);

Assert.Equal([3, 4, 5], slot.LimboData.ToArray());
Assert.Equal(0, slot.ReservedWindow);
}

[Fact(Timeout = 5000)]
public void ClearLimbo_should_clear_HasLimbo_and_LimboData()
public void ReservedWindow_should_persist_across_reads()
{
var slot = new BodyDrainSlot<int>();
slot.StoreLimbo(new byte[] { 1, 2 });

slot.ClearLimbo();

Assert.False(slot.HasLimbo);
Assert.True(slot.LimboData.IsEmpty);
slot.ReservedWindow = 8 * 1024;
Assert.Equal(8 * 1024, slot.ReservedWindow);
}

[Fact(Timeout = 5000)]
Expand Down Expand Up @@ -194,7 +112,7 @@ public void DisposeResources_should_dispose_buffer_and_LinkedCts()
var stream = new MemoryStream([]);
var cts = new CancellationTokenSource();
using var linked = CancellationTokenSource.CreateLinkedTokenSource(cts.Token);
slot.Initialize(1, stream, null, CancellationToken.None, linked);
slot.Initialize(1, stream, CancellationToken.None, linked);
slot.EnsureBuffer(256);

slot.DisposeResources();
Expand All @@ -210,38 +128,32 @@ public void Reset_should_clear_all_state()
var stream = new MemoryStream([1, 2, 3]);
var cts = new CancellationTokenSource();
using var linked = CancellationTokenSource.CreateLinkedTokenSource(cts.Token);
slot.Initialize(7, stream, 3, CancellationToken.None, linked);
slot.Initialize(7, stream, CancellationToken.None, linked);
slot.EnsureBuffer(256);
slot.BeginRead();
slot.MarkOrphaned();
slot.StoreLimbo(new byte[] { 9 });
slot.MarkDrainComplete();
slot.IncrementSyncReads(100);
slot.ReservedWindow = 16 * 1024;

slot.Reset();

Assert.Equal(0, slot.StreamId);
Assert.Null(slot.BodyStream);
Assert.Null(slot.ContentLength);
#pragma warning disable xUnit1051 // SUT behavior: asserts slot resets RequestCt to default, not test cooperative cancellation
Assert.Equal(default, slot.RequestCt);
#pragma warning restore xUnit1051
Assert.Null(slot.LinkedCts);
Assert.Null(slot.Buffer);
Assert.Equal(0, slot.ReservedWindow);
Assert.False(slot.IsReadInFlight);
Assert.False(slot.IsOrphaned);
Assert.False(slot.HasLimbo);
Assert.True(slot.LimboData.IsEmpty);
Assert.False(slot.IsDrainComplete);
Assert.Equal(0, slot.ConsecutiveSyncReads);
}

[Fact(Timeout = 5000)]
public void Reset_should_work_for_long_streamId()
{
var slot = new BodyDrainSlot<long>();
var stream = new MemoryStream([]);
slot.Initialize(99L, stream, null, CancellationToken.None, null);
slot.Initialize(99L, stream, CancellationToken.None, null);

slot.Reset();

Expand Down
Loading