From 8d5340a0b45bec47cf98083df38dc9b7382caaa4 Mon Sep 17 00:00:00 2001 From: Zakaria El Orche Date: Thu, 13 Aug 2026 19:00:19 +0000 Subject: [PATCH] feat(core): add HTTP/2 flow-controlled bodies --- flash/docs/http2/DECISIONS.md | 30 +++ flash/docs/http2/FLOW-CONTROL.md | 48 ++++ flash/docs/http2/IMPLEMENTATION-PLAN.md | 41 ++-- flash/docs/http2/STREAMS.md | 9 +- .../http2/message/Http2BodyBenchmark.java | 105 ++++++++ .../relism/flash/http2/Http2Connection.java | 115 +++++++-- .../flash/http2/Http2ConnectionScratch.java | 11 + .../dev/relism/flash/http2/Http2Limits.java | 11 +- .../flash/http2/Http2StreamDispatcher.java | 170 +++++++++++-- .../flash/http2/message/DataBufferPool.java | 69 ++++++ .../flash/http2/message/Http2RequestBody.java | 232 ++++++++++++++++++ .../http2/message/Http2ResponseWriter.java | 188 ++++++++++++-- .../http2/stream/Http2FlowController.java | 108 ++++++++ .../flash/http2/stream/Http2Stream.java | 188 +++++++++++++- .../flash/http2/stream/Http2StreamTable.java | 21 +- .../flash/http2/Http2BackpressureTest.java | 37 +++ .../http2/Http2ConnectionIntegrationTest.java | 143 +++++++++++ .../http2/message/Http2LargeResponseTest.java | 54 ++++ .../http2/message/Http2RequestBodyTest.java | 86 +++++++ .../message/Http2ResponseWriterTest.java | 33 +++ .../http2/stream/Http2FlowControlTest.java | 81 ++++++ 21 files changed, 1679 insertions(+), 101 deletions(-) create mode 100644 flash/docs/http2/FLOW-CONTROL.md create mode 100644 flash/src/jmh/java/dev/relism/flash/http2/message/Http2BodyBenchmark.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/message/DataBufferPool.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/stream/Http2FlowController.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/Http2BackpressureTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/message/Http2LargeResponseTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/message/Http2RequestBodyTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/stream/Http2FlowControlTest.java diff --git a/flash/docs/http2/DECISIONS.md b/flash/docs/http2/DECISIONS.md index f965bcf..c073a40 100644 --- a/flash/docs/http2/DECISIONS.md +++ b/flash/docs/http2/DECISIONS.md @@ -951,3 +951,33 @@ the lifecycle benchmark remains at the allocation noise floor. advertised limit or reader size from measurements; do not add a sleep-based dispatch delay. --- + +## DEC-28 — Align the receive window with a coalescing bounded DATA pool + +**Context.** The demultiplexer cannot block waiting for an application handler, but delaying +WINDOW_UPDATE only provides backpressure after the peer has spent the window it already owns. A +pool smaller than that outstanding credit can be exhausted legitimately. Allocating one buffer per +DATA frame is also unsafe because many tiny or heavily padded frames can consume little payload +storage while exhausting an object-per-frame pool. + +**Decision.** Advertise 1 MiB at both the connection and stream receive levels and back the +connection with exactly 64 reusable 16 KiB buffers (also 1 MiB). Adjacent DATA payloads coalesce +into available tail space; padding contributes to flow credit but not storage. WINDOW_UPDATE is +sent at half-window consumption, never merely on receipt. Bodies at or below 64 KiB with a known +length use one reusable contiguous stream buffer and dispatch at END_STREAM; all other bodies +dispatch immediately onto the same protocol-neutral `RequestBody` over a blocking pooled source. + +Response DATA uses the same serialized writer with a progress cursor. A stream object itself is +the executor task for initial handling and resumptions, so no per-resume closure is created. +WINDOW_UPDATE only schedules work; it never reads an application `InputStream` on the demux thread. + +**Consequence.** Outstanding peer credit and worst-case pooled payload storage match exactly, +small frames do not multiply objects, slow consumers withhold credit naturally, and streaming +responses resume without recursive writer completion or demux blocking. The 100 MiB bidirectional +integration test remains bounded, while JMH measures the streaming request and response paths at +the allocation noise floor. + +**Revisit when.** Production memory/throughput measurements justify a different window. Change the +window and pool byte capacity together; never raise credit independently of bounded storage. + +--- diff --git a/flash/docs/http2/FLOW-CONTROL.md b/flash/docs/http2/FLOW-CONTROL.md new file mode 100644 index 0000000..a2ff2b7 --- /dev/null +++ b/flash/docs/http2/FLOW-CONTROL.md @@ -0,0 +1,48 @@ +# HTTP/2 bodies and flow control + +HTTP/2 applies flow control independently to the connection and to every stream. Flash advertises +a 1 MiB receive window at both levels and sends WINDOW_UPDATE only after the application has +consumed at least half a window. A DATA frame decrements both windows by its complete payload +length, including the pad-length byte and padding; only its unpadded data reaches the handler. + +## Request bodies + +Known bodies up to 64 KiB remain in one reusable contiguous stream buffer. Their handler is +dispatched at END_STREAM, and `RequestBody.bytes()` performs the only allocation: the byte array +returned to application code. For a 1,024-byte body JMH reports exactly 1,040 B/op, the array plus +its object header, with no framework allocation around it. + +Larger or unknown-length bodies dispatch after request headers. DATA is copied out of the frame +reader into a connection-owned pool of 64 reusable 16 KiB buffers. Small adjacent frames coalesce +inside a buffer, so the pool is bounded by bytes rather than frame count. The existing +`RequestBody.stream()` blocks only the handler's virtual thread when data is absent. Buffers return +to the pool as reads consume them, and that consumption reopens both receive windows. If a handler +does not read its body, the normal post-handler drain performs the same bounded consumption. + +The connection window and pool both cover exactly 1 MiB, so the peer can never hold more credit +than the server can store before backpressure takes effect. Per-stream accepted body bytes remain +bounded by `MAX_REQUEST_BODY_SIZE`. Declared content length is parsed without a String and checked +against the unpadded DATA total at END_STREAM. + +## Responses + +Fixed byte arrays, known-length streams and unknown-length streams all use one resumable +`Http2ResponseWriter`. It emits DATA frames no larger than the peer's frame limit, the available +connection window, the available stream window and the reusable 16 KiB relay buffer. A +WINDOW_UPDATE schedules the stream on the shared virtual-thread executor; application streams are +never read by the demultiplexer. + +`Response.chunked(InputStream)` means unknown-length streaming at the application API. HTTP/2 has +no chunked transfer coding, so Flash emits ordinary DATA followed by END_STREAM and never sends a +`transfer-encoding` field. `Response.stream(InputStream, length)` emits `content-length` and fails +the stream if the source ends before that length. + +## Verification + +- A real Java HTTP/2 client uploads and downloads 100 MiB over TLS; both directions are validated + byte-for-byte without materializing the test payload. +- A synthetic 100 MiB response proves serialized scratch storage stays below 64 KiB. +- h2spec sections 5, 6.1, 6.9 and 8: 50 passed, one h2spec-skipped case, zero failures. +- Clean Maven build with JMH sources: 633 tests, no failures. +- JMH request streaming: 159.408 ns/op, 0.001 B/op, no GC. +- JMH response streaming frame: 219.090 ns/op, 0.002 B/op, no GC. diff --git a/flash/docs/http2/IMPLEMENTATION-PLAN.md b/flash/docs/http2/IMPLEMENTATION-PLAN.md index e6dd387..653f174 100644 --- a/flash/docs/http2/IMPLEMENTATION-PLAN.md +++ b/flash/docs/http2/IMPLEMENTATION-PLAN.md @@ -71,8 +71,8 @@ Status values: `not started` / `in progress` / `blocked` / `done`. | 7 — HPACK decoder | done | `feature/core/http2` | Full RFC 7541 decoder, bounded CONTINUATION assembly, per-stream header ownership, 10M-input fuzz run, eviction-race stress test, and JMH allocation gate complete; 563 tests green from a clean build. | | 8 — Connection state machine | done | `feature/core/http2` | Preface, transactional SETTINGS, priority PING ACK, connection WINDOW_UPDATE, two-stage GOAWAY, per-socket transport/ALPN dispatch, HPACK block composition, clean curl handshake, h2spec 28/35 selected cases and 0.008 B/op JMH gate complete. Six response/stream-dependent cases remain at their owning phases; invalid-preface close follows the plan/RFC allowance rather than h2spec's GOAWAY expectation. | | 9 — HPACK encoder + h2 response path | done | `feature/core/http2` | Stateless static-table HPACK encoder; precompiled status/content-type/date fields; reusable response writer with header filtering, bounds, CONTINUATION splitting and fixed DATA happy path; HTTP/1/2 serializer parity test. EX-46 fixed the one-digit Date day-of-month bug. JMH: 174.309 ns/op, 0.001 B/op (noise floor), no GC. 603/603 tests green from a clean `-Pjmh` build. | -| 10 — Stream state machine + dispatch | done | `feature/core/http2` | Explicit stream transition table, bounded primitive stream table and pool, pseudo-header/message validation, protocol-neutral `Request` assembly, virtual-thread dispatch and exception path, cancellation-safe release, raw h2c + Java HTTP/2 integration. h2spec sections 5/8: 37/39; the two content-length/DATA accounting cases are owned by Phase 11. JMH pooled lifecycle: 458.499 ns/op, 0.003 B/op, no GC. 618/618 tests green from a clean `-Pjmh` build. | -| 11 — DATA, flow control, bodies | not started | — | — | +| 10 — Stream state machine + dispatch | done | `feature/core/http2` | Explicit stream transition table, bounded primitive stream table and pool, pseudo-header/message validation, protocol-neutral `Request` assembly, virtual-thread dispatch and exception path, cancellation-safe release, raw h2c + Java HTTP/2 integration. Phase 11 closed the two deferred content-length/DATA cases; h2spec sections 5/8 are now 39/39. JMH pooled lifecycle: 458.499 ns/op, 0.003 B/op, no GC. 618/618 tests green at phase closure. | +| 11 — DATA, flow control, bodies | done | `feature/core/http2` | Two-level receive/send flow control, consumption-driven WINDOW_UPDATE hysteresis, bounded/coalescing DATA pool, inline and blocking streaming request bodies through the existing `RequestBody`, resumable fixed/known/unknown response streams, content-length and empty-DATA validation. Real TLS HTTP/2 transfer: 100 MiB upload + 100 MiB download verified byte-for-byte. h2spec combined sections 5, 6.1, 6.9 and 8: 50 passed, 1 tool-skipped, 0 failed. JMH: inline materialization exactly one 1,040-byte array; request streaming 0.001 B/op; response streaming 0.002 B/op; full pooled lifecycle 0.003 B/op. 633/633 tests green from a clean `-Pjmh` build. | | 12 — Trailers, half-close, gRPC | not started | — | — | | 13 — Security hardening & abuse resistance | not started | — | — | | 14 — h2c prior knowledge + proxy support | not started | — | — | @@ -2558,8 +2558,7 @@ rules, the dispatch model, and the resource-release contract. existing `HttpServerTest` suite against an h2 client. - [x] `FastPathRouterImpl` unchanged. - [x] 0 B/op for the pooled protocol-side h2 GET lifecycle (0.003 B/op JMH noise floor). -- [~] `h2spec` sections 5 and 8: 37/39 green. Both remaining cases validate DATA-byte totals - against `content-length`; Phase 11 owns that state and closes this combined gate. +- [x] `h2spec` sections 5 and 8: 39/39 green after DATA-byte accounting landed. --- @@ -2571,13 +2570,13 @@ backpressure. ### Files Created: -- `h2/stream/Http2FlowController.java` — connection and stream windows, both directions. -- `h2/message/Http2RequestBody.java` — DATA frames → the `RequestBody` contract. -- `h2/message/DataBufferPool.java` — the fixed-size buffer free list. +- `http2/stream/Http2FlowController.java` — connection and stream windows, both directions. +- `http2/message/Http2RequestBody.java` — DATA frames → the `RequestBody` contract. +- `http2/message/DataBufferPool.java` — the fixed-size buffer free list. Modified: -- `h2/Http2Connection.java` — DATA dispatch. -- `h2/message/Http2ResponseWriter.java` — multi-frame and streaming bodies. +- `http2/Http2Connection.java` — DATA dispatch. +- `http2/message/Http2ResponseWriter.java` — multi-frame and streaming bodies. - `models/RequestBody.java` — accept an h2 backing (the Phase 6 refactor made this possible). ### Tasks @@ -2635,16 +2634,16 @@ Modified: - Streaming path: 0 B/op at steady state; all buffers come from `DataBufferPool`. ### Safety checks -- [ ] Connection-level **and** stream-level WINDOW_UPDATE both sent -- [ ] Window overflow (> 2^31-1) rejected -- [ ] Window underflow (peer exceeds its window) rejected with the correct scope -- [ ] Padding counted toward flow control -- [ ] Flow control accounted for RST streams until settled -- [ ] `content-length` verified against actual DATA -- [ ] Empty DATA frame flood bounded -- [ ] `DataBufferPool` bounded; exhaustion applies backpressure rather than allocating without +- [x] Connection-level **and** stream-level WINDOW_UPDATE both sent +- [x] Window overflow (> 2^31-1) rejected +- [x] Window underflow (peer exceeds its window) rejected with the correct scope +- [x] Padding counted toward flow control +- [x] Flow control accounted for RST streams until settled +- [x] `content-length` verified against actual DATA +- [x] Empty DATA frame flood bounded +- [x] `DataBufferPool` bounded; exhaustion applies backpressure rather than allocating without limit -- [ ] Body size bounded by `Http2Limits.MAX_REQUEST_BODY_SIZE` when no handler consumes it +- [x] Body size bounded by `Http2Limits.MAX_REQUEST_BODY_SIZE` when no handler consumes it ### Tests - `Http2FlowControlTest` — the classic scenarios: a 10 MB upload with a 64 KB window; a @@ -2663,9 +2662,9 @@ Modified: and the dispatch-on-END_STREAM optimization with its rationale. ### DoD -- [ ] 100 MB upload and 100 MB download both correct, both bounded memory. -- [ ] `h2spec` DATA and WINDOW_UPDATE sections green. -- [ ] Small-body path allocates exactly one `byte[]` (the user's body). +- [x] 100 MB upload and 100 MB download both correct, both bounded memory. +- [x] `h2spec` DATA and WINDOW_UPDATE sections green (13 passed, one tool-skipped, zero failed). +- [x] Small-body path allocates exactly one `byte[]` (1,040 B/op for a 1,024-byte body). --- diff --git a/flash/docs/http2/STREAMS.md b/flash/docs/http2/STREAMS.md index fd42729..031e4a6 100644 --- a/flash/docs/http2/STREAMS.md +++ b/flash/docs/http2/STREAMS.md @@ -40,10 +40,9 @@ while dispatch is pending. ## Verification -- Clean Maven build with JMH sources: 618 tests, no failures. -- h2spec sections 5 and 8: 37/39. The two remaining cases require request DATA byte accounting and - are completed with body flow control. +- Phase 11 clean Maven build with JMH sources: 633 tests, no failures. +- h2spec sections 5 and 8: 39/39 after request DATA byte accounting landed. - Java `HttpClient` negotiates HTTP/2 over TLS and runs an existing parameterized route unchanged. - curl prior-knowledge h2c receives a valid `200` response and body. -- JMH pooled lifecycle (HPACK decode, request assembly, response write and release): - 458.499 ns/op, 0.003 B/op, no GC. +- Current JMH pooled lifecycle (HPACK decode, request assembly, response write and release): + 483.571 ns/op, 0.003 B/op, no GC. diff --git a/flash/src/jmh/java/dev/relism/flash/http2/message/Http2BodyBenchmark.java b/flash/src/jmh/java/dev/relism/flash/http2/message/Http2BodyBenchmark.java new file mode 100644 index 0000000..c265ea7 --- /dev/null +++ b/flash/src/jmh/java/dev/relism/flash/http2/message/Http2BodyBenchmark.java @@ -0,0 +1,105 @@ +package dev.relism.flash.http2.message; + +import dev.relism.flash.http.ContentType; +import dev.relism.flash.models.RequestBody; +import dev.relism.flash.models.Response; +import java.io.IOException; +import java.io.InputStream; +import java.util.concurrent.TimeUnit; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; + +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 3) +@Measurement(iterations = 5) +@Fork(2) +@State(Scope.Thread) +public class Http2BodyBenchmark { + private static final Http2RequestBody.ConsumptionListener NOOP = bytes -> {}; + + private final byte[] payload = new byte[1024]; + private final byte[] target = new byte[1024]; + private DataBufferPool pool; + private Http2RequestBody source; + private RequestBody body; + private Response response; + private Http2ResponseWriter responseWriter; + private ResettableInputStream responseSource; + + @Setup(Level.Trial) + public void setup() throws IOException { + pool = new DataBufferPool(16_384, 1); + source = new Http2RequestBody(pool); + body = new RequestBody(); + response = new Response(200, ContentType.BINARY); + responseWriter = new Http2ResponseWriter(); + responseSource = new ResettableInputStream(payload); + source.begin(-1, false, NOOP); + source.offer(1, payload, 0, payload.length, payload.length); + source.finish(1); + source.read(target); + } + + @Benchmark + public byte[] inlineBytes() { + source.begin(payload.length, true, NOOP); + source.offer(1, payload, 0, payload.length, payload.length); + source.finish(1); + body.reset(source, payload.length, null, 0, 0); + return body.bytes(); + } + + @Benchmark + public int streamingRead() throws IOException { + source.begin(-1, false, NOOP); + source.offer(1, payload, 0, payload.length, payload.length); + source.finish(1); + return source.read(target, 0, target.length); + } + + @Benchmark + public int streamingResponseFrame() throws IOException { + responseSource.rewind(); + response.reset(200, ContentType.BINARY).stream(responseSource, payload.length); + responseWriter.startFlowControlled( + response, 1, false, false, true, false, false, 16_384, 32_768, 16_384); + return responseWriter.length(); + } + + private static final class ResettableInputStream extends InputStream { + private final byte[] source; + private int position; + + ResettableInputStream(byte[] source) { + this.source = source; + } + + void rewind() { + position = 0; + } + + @Override + public int read() { + return position == source.length ? -1 : source[position++] & 0xff; + } + + @Override + public int read(byte[] target, int offset, int length) { + if (position == source.length) return -1; + int count = Math.min(length, source.length - position); + System.arraycopy(source, position, target, offset, count); + position += count; + return count; + } + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java b/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java index a33eeb6..68bc4bd 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java @@ -1,5 +1,6 @@ package dev.relism.flash.http2; +import dev.relism.flash.bytes.Pairs; import dev.relism.flash.http2.Http2ConnectionScratch.ControlIntent; import dev.relism.flash.http2.Http2ConnectionScratch.ControlKind; import dev.relism.flash.http2.frame.FrameFlags; @@ -8,7 +9,10 @@ import dev.relism.flash.http2.frame.FrameType; import dev.relism.flash.http2.frame.FrameValidator; import dev.relism.flash.http2.frame.Http2FrameReader; import dev.relism.flash.http2.frame.Http2FrameWriter; +import dev.relism.flash.http2.frame.Padding; import dev.relism.flash.http2.hpack.HeaderSink; +import dev.relism.flash.http2.message.DataBufferPool; +import dev.relism.flash.http2.stream.Http2FlowController; import dev.relism.flash.http2.stream.Http2Stream; import dev.relism.flash.http2.stream.Http2StreamState; import dev.relism.flash.http2.stream.Http2StreamTable; @@ -37,10 +41,13 @@ public final class Http2Connection implements ConnectionProtocol { private final Http2Settings.StreamWindowUpdater streamWindows; private final long settingsAckTimeoutMs; private final Http2HeaderBlockDecoder headerBlocks = new Http2HeaderBlockDecoder(); - private final Http2StreamTable streams = new Http2StreamTable(Http2Limits.MAX_CONCURRENT_STREAMS); + private final DataBufferPool dataBuffers = + new DataBufferPool(Http2Limits.MAX_FRAME_SIZE_LOCAL, Http2Limits.DATA_BUFFER_POOL_SIZE); + private final Http2StreamTable streams = + new Http2StreamTable(Http2Limits.MAX_CONCURRENT_STREAMS, dataBuffers); private static final HeaderSink DISCARD_HEADERS = (name, value, never) -> {}; - private long connectionSendWindow = 65_535; + private Http2FlowController flowController; private int outstandingLocalSettings; private long oldestSettingsSentNanos; private int lastProcessedStreamId; @@ -68,7 +75,8 @@ public final class Http2Connection implements ConnectionProtocol { this.streamWindows = delta -> { try { - streams.adjustAllSendWindows(delta); + if (flowController == null) streams.adjustAllSendWindows(delta); + else flowController.applyInitialWindowDelta(streams, delta); } catch (IllegalStateException overflow) { throw Http2Exception.FLOW_CONTROL_ERROR; } @@ -80,12 +88,16 @@ public final class Http2Connection implements ConnectionProtocol { @Override public void run(ConnectionContext ctx) throws IOException { Http2FrameWriter writer = new Http2FrameWriter(ctx.rawOut()::write); + flowController = + new Http2FlowController( + (streamId, increment) -> sendWindowUpdate(writer, streamId, increment)); streamDispatcher = new Http2StreamDispatcher( ctx, writer, peerSettings, streams, + flowController, (streamId, error) -> sendRstStream(writer, streamId, error)); try { run(ctx.in(), writer, ctx.stopped()); @@ -96,6 +108,11 @@ public final class Http2Connection implements ConnectionProtocol { void run(BufferedByteSource input, Http2FrameWriter writer, BooleanSupplier stopped) throws IOException { + if (flowController == null) { + flowController = + new Http2FlowController( + (streamId, increment) -> sendWindowUpdate(writer, streamId, increment)); + } Http2FrameReader reader = new Http2FrameReader(input); runPrepared(input, reader, writer, stopped); } @@ -206,6 +223,10 @@ public final class Http2Connection implements ConnectionProtocol { pendingHeaderStream = streams.acquire(streamId); refusingHeaderStream = pendingHeaderStream == null; + if (pendingHeaderStream != null) { + flowController.initializeStreamSendWindow( + pendingHeaderStream, peerSettings.initialWindowSize()); + } HeaderSink sink = refusingHeaderStream ? DISCARD_HEADERS : pendingHeaderStream.headerBlock(); if (headerBlocks.accept(frame, sink)) completeHeaders(writer, frame.streamId()); } @@ -224,6 +245,8 @@ public final class Http2Connection implements ConnectionProtocol { } else { Http2Stream stream = pendingHeaderStream; if (streamDispatcher != null) stream.validateHeaders(); + boolean dispatch = + stream.prepareRequestBody(flowController, headerBlocks.endStream()); stream.transition( headerBlocks.endStream() ? Http2StreamState.Event.RECV_HEADERS_ES @@ -233,7 +256,7 @@ public final class Http2Connection implements ConnectionProtocol { streams.remove(streamId); streams.release(stream); if (!gracefulStarted) startGracefulShutdown(writer); - } else if (headerBlocks.endStream()) { + } else if (dispatch) { enqueueDispatch(stream); } } @@ -250,17 +273,47 @@ public final class Http2Connection implements ConnectionProtocol { } private void receiveData(FrameHeader frame) { - Http2Stream stream = streamForFrame(frame.streamId()); - stream.transition( - FrameFlags.isEndStream(frame.flags()) - ? Http2StreamState.Event.RECV_DATA_ES - : Http2StreamState.Event.RECV_DATA); - if (frame.length() != 0) { + flowController.receiveConnectionBytes(frame.length()); + Http2Stream stream = streams.get(frame.streamId()); + if (stream == null) { + discardConnectionBytes(frame.length()); + if (frame.streamId() > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR; throw new Http2StreamException( - frame.streamId(), Http2ErrorCode.INTERNAL_ERROR, "request DATA support is not active"); + frame.streamId(), Http2ErrorCode.STREAM_CLOSED, "stream is closed"); } - if (FrameFlags.isEndStream(frame.flags()) && streamDispatcher != null) { - enqueueDispatch(stream); + boolean bodyAccepted = false; + try { + stream.transition( + FrameFlags.isEndStream(frame.flags()) + ? Http2StreamState.Event.RECV_DATA_ES + : Http2StreamState.Event.RECV_DATA); + flowController.receiveStreamBytes(stream, frame.length()); + long unpadded = + Padding.unpad( + frame.buffer(), + frame.payloadOffset(), + frame.length(), + FrameFlags.isPadded(frame.flags())); + int dataOffset = Pairs.hi(unpadded); + int dataLength = Pairs.lo(unpadded); + if (frame.length() == 0) { + if (stream.incrementEmptyDataFrames() + > Http2Limits.MAX_EMPTY_DATA_FRAMES_PER_STREAM) { + throw new Http2StreamException( + frame.streamId(), Http2ErrorCode.ENHANCE_YOUR_CALM, "empty DATA frame limit exceeded"); + } + } else { + stream.resetEmptyDataFrames(); + } + stream.receiveData(frame.buffer(), dataOffset, dataLength, frame.length()); + bodyAccepted = true; + if (FrameFlags.isEndStream(frame.flags())) { + stream.finishRequestBody(); + if (streamDispatcher != null && !stream.dispatched()) enqueueDispatch(stream); + } + } catch (RuntimeException failure) { + if (!bodyAccepted) discardConnectionBytes(frame.length()); + throw failure; } } @@ -276,6 +329,7 @@ public final class Http2Connection implements ConnectionProtocol { streams.remove(stream.id()); if (releaseDeferred) { stream.cancel(); + if (stream.responseStarted() && !stream.responseInFlight()) streams.release(stream); } else { streams.release(stream); } @@ -286,6 +340,7 @@ public final class Http2Connection implements ConnectionProtocol { throw new Http2StreamException( stream.id(), Http2ErrorCode.REFUSED_STREAM, "dispatch queue is full"); } + stream.markDispatched(); dispatchQueue[dispatchCount++] = stream; } @@ -299,13 +354,6 @@ public final class Http2Connection implements ConnectionProtocol { } } - private Http2Stream streamForFrame(int streamId) { - Http2Stream stream = streams.get(streamId); - if (stream != null) return stream; - if (streamId > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR; - throw new Http2StreamException(streamId, Http2ErrorCode.STREAM_CLOSED, "stream is closed"); - } - private void receiveSettings(FrameHeader frame, Http2FrameWriter writer) throws IOException { boolean ack = FrameFlags.isAck(frame.flags()); if (ack) { @@ -346,16 +394,16 @@ public final class Http2Connection implements ConnectionProtocol { return; } try { - stream.adjustSendWindow(increment); + flowController.increaseStreamSendWindow(stream, increment); } catch (IllegalStateException overflow) { throw new Http2StreamException( frame.streamId(), Http2ErrorCode.FLOW_CONTROL_ERROR, "stream window overflow"); } + if (streamDispatcher != null) streamDispatcher.streamWindowUpdated(stream); return; } - long next = connectionSendWindow + increment; - if (next > Integer.MAX_VALUE) throw Http2Exception.FLOW_CONTROL_ERROR; - connectionSendWindow = next; + flowController.increaseConnectionSendWindow(increment); + if (streamDispatcher != null) streamDispatcher.connectionWindowUpdated(); } private void receiveGoAway(FrameHeader frame) { @@ -386,6 +434,21 @@ public final class Http2Connection implements ConnectionProtocol { writer.writePriority(rst); } + private void sendWindowUpdate(Http2FrameWriter writer, int streamId, int increment) + throws IOException { + ControlIntent update = scratch.acquire(ControlKind.SETTINGS_OR_OTHER); + update.windowUpdate(streamId, increment); + writer.writePriority(update); + } + + private void discardConnectionBytes(int bytes) { + try { + flowController.discarded(bytes); + } catch (IOException failure) { + throw new IllegalStateException("failed to restore connection flow-control window", failure); + } + } + private void closeStreamAfterError(int streamId) { Http2Stream stream = streams.remove(streamId); if (stream == null) return; @@ -446,7 +509,7 @@ public final class Http2Connection implements ConnectionProtocol { } public long connectionSendWindow() { - return connectionSendWindow; + return flowController == null ? 65_535 : flowController.connectionSendWindow(); } public int peerLastStreamId() { @@ -459,7 +522,7 @@ public final class Http2Connection implements ConnectionProtocol { void reset() { peerSettings.reset(); - connectionSendWindow = 65_535; + flowController = null; outstandingLocalSettings = 0; oldestSettingsSentNanos = 0; lastProcessedStreamId = 0; diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2ConnectionScratch.java b/flash/src/main/java/dev/relism/flash/http2/Http2ConnectionScratch.java index ee5b175..ff6e049 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2ConnectionScratch.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2ConnectionScratch.java @@ -119,6 +119,17 @@ final class Http2ConnectionScratch { length = 17 + debugLength; } + void windowUpdate(int streamId, int increment) { + buffer[0] = 0; + buffer[1] = 0; + buffer[2] = 4; + buffer[3] = (byte) FrameType.WINDOW_UPDATE.code(); + buffer[4] = 0; + writeUInt31(buffer, 5, streamId); + writeUInt31(buffer, 9, increment); + length = 13; + } + private static void writeUInt31(byte[] target, int off, int value) { writeUInt32(target, off, value & 0x7FFF_FFFF); } diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java b/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java index 2b31fea..70c90fc 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java @@ -111,6 +111,15 @@ public final class Http2Limits { */ public static final int MAX_EMPTY_DATA_FRAMES_PER_STREAM = 1_000; + /** Largest request body retained contiguously before dispatching its handler. */ + public static final int INLINE_BODY_THRESHOLD = 64 * 1024; + + /** Hard limit for request body bytes accepted on one stream. */ + public static final int MAX_REQUEST_BODY_SIZE = 100 * 1024 * 1024; + + /** Number of frame-sized buffers available to streaming request bodies on one connection. */ + public static final int DATA_BUFFER_POOL_SIZE = 64; + /** * The value of {@code SETTINGS_INITIAL_WINDOW_SIZE} Flash advertises for every new stream: * deliberately large (1 MiB, versus the RFC default of 65535) so that a normal-sized @@ -125,7 +134,7 @@ public final class Http2Limits { * windows, and sizing for that worst case would commit 100 MiB of receive window to every * connection regardless of load. */ - public static final int CONNECTION_WINDOW_SIZE_LOCAL = 16 * 1_048_576; + public static final int CONNECTION_WINDOW_SIZE_LOCAL = 1_048_576; /** * The HPACK dynamic table size Flash's decoder honours, in bytes of RFC 7541 §4.1 accounting. RFC diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java b/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java index 0a82c8c..f54babd 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java @@ -3,6 +3,7 @@ package dev.relism.flash.http2; import dev.relism.flash.http.HttpMethod; import dev.relism.flash.http2.frame.Http2FrameWriter; import dev.relism.flash.http2.message.Http2ResponseWriter; +import dev.relism.flash.http2.stream.Http2FlowController; import dev.relism.flash.http2.stream.Http2Stream; import dev.relism.flash.http2.stream.Http2StreamState; import dev.relism.flash.http2.stream.Http2StreamTable; @@ -16,7 +17,7 @@ import lombok.extern.slf4j.Slf4j; /** Dispatches completed request streams without blocking the connection demultiplexer. */ @Slf4j -final class Http2StreamDispatcher { +final class Http2StreamDispatcher implements Http2Stream.ResponseSink { @FunctionalInterface interface FailureSink { void fail(int streamId, Http2ErrorCode errorCode) throws IOException; @@ -26,7 +27,10 @@ final class Http2StreamDispatcher { private final Http2FrameWriter frameWriter; private final Http2Settings peerSettings; private final Http2StreamTable streams; + private final Http2FlowController flowController; private final FailureSink failures; + private final Http2Stream[] resumeScratch = + new Http2Stream[Http2Limits.MAX_CONCURRENT_STREAMS]; private volatile boolean firstResponse = true; Http2StreamDispatcher( @@ -34,28 +38,65 @@ final class Http2StreamDispatcher { Http2FrameWriter frameWriter, Http2Settings peerSettings, Http2StreamTable streams, + Http2FlowController flowController, FailureSink failures) { this.context = context; this.frameWriter = frameWriter; this.peerSettings = peerSettings; this.streams = streams; + this.flowController = flowController; this.failures = failures; } + void streamWindowUpdated(Http2Stream stream) { + scheduleResume(stream); + } + + void connectionWindowUpdated() { + int count = streams.copyValues(resumeScratch); + for (int i = 0; i < count; i++) { + Http2Stream stream = resumeScratch[i]; + resumeScratch[i] = null; + scheduleResume(stream); + } + } + + private void scheduleResume(Http2Stream stream) { + if (!stream.responseStarted() || stream.cancelled()) return; + if (!stream.beginResponseBatch()) return; + stream.markResumeTask(); + try { + context.executor().execute(stream); + } catch (RejectedExecutionException rejected) { + stream.endResponseBatch(); + failAndRelease(stream, Http2ErrorCode.REFUSED_STREAM, rejected); + } + } + void dispatch(Http2Stream stream) { if (stream.cancelled()) { streams.release(stream); return; } stream.markDispatched(); + stream.responseSink(this); try { - context.executor().execute(() -> handle(stream)); + context.executor().execute(stream); } catch (RejectedExecutionException rejected) { failAndRelease(stream, Http2ErrorCode.REFUSED_STREAM, rejected); } } + @Override + public void handleRequest(Http2Stream stream) { + handle(stream); + } + private void handle(Http2Stream stream) { + if (stream.cancelled()) { + streams.release(stream); + return; + } try { Request request = stream.assembleRequest(context.remoteAddress(), context.sslSocket()); Response pooled = stream.resetResponse(); @@ -74,6 +115,7 @@ final class Http2StreamDispatcher { else if (result != null) response.setBody(result); } + request.drain(); Http2ResponseWriter responseWriter = stream.responseWriter(); if (stream.cancelled()) { request.recycle(); @@ -81,44 +123,120 @@ final class Http2StreamDispatcher { streams.release(stream); return; } - boolean prepared; + boolean headRequest = request.method() == HttpMethod.HEAD; + int reserved; + int used; synchronized (this) { boolean tableUpdate = firstResponse; - prepared = - responseWriter.prepare( - response, - stream.id(), - request.method() == HttpMethod.HEAD, - context.configuration().isSendDate(), - true, - context.configuration().isH2HuffmanDynamicValues(), - tableUpdate, - peerSettings.maxFrameSize(), - peerSettings.maxHeaderListSize(), - (int) Math.min(stream.sendWindow(), Integer.MAX_VALUE)); - if (prepared) { - firstResponse = false; - stream.transition(Http2StreamState.Event.SEND_HEADERS_ES); - request.recycle(); - if (response == pooled) pooled.recycle(); - streams.remove(stream.id()); - frameWriter.write(responseWriter); + reserved = flowController.reserveSend(stream, peerSettings.maxFrameSize()); + used = 0; + try { + used = + responseWriter.startFlowControlled( + response, + stream.id(), + headRequest, + context.configuration().isSendDate(), + true, + context.configuration().isH2HuffmanDynamicValues(), + tableUpdate, + peerSettings.maxFrameSize(), + peerSettings.maxHeaderListSize(), + reserved); + } finally { + flowController.refundSend(stream, reserved - used); } + firstResponse = false; } - if (!prepared) { - request.recycle(); - if (response == pooled) pooled.recycle(); - failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, null); + request.recycle(); + if (response == pooled) pooled.recycle(); + stream.markResponseStarted(); + applyBatchTransition(stream, responseWriter); + if (!stream.beginResponseBatch()) { + throw new IllegalStateException("response batch already in flight"); } + frameWriter.write(responseWriter); } catch (Exception failure) { failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, failure); } } + private void tryResumeResponse(Http2Stream stream) { + if (stream.cancelled()) { + stream.endResponseBatch(); + streams.release(stream); + return; + } + Http2ResponseWriter responseWriter = stream.responseWriter(); + if (responseWriter.finished()) { + stream.endResponseBatch(); + streams.remove(stream.id()); + streams.release(stream); + return; + } + int reserved = flowController.reserveSend(stream, peerSettings.maxFrameSize()); + if (reserved == 0) { + stream.endResponseBatch(); + return; + } + try { + int used = 0; + try { + used = responseWriter.resume(peerSettings.maxFrameSize(), reserved); + } finally { + flowController.refundSend(stream, reserved - used); + } + applyBatchTransition(stream, responseWriter); + frameWriter.write(responseWriter); + } catch (Exception failure) { + stream.endResponseBatch(); + failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, failure); + } + } + + @Override + public void resumeResponse(Http2Stream stream) { + tryResumeResponse(stream); + } + + private static void applyBatchTransition( + Http2Stream stream, Http2ResponseWriter responseWriter) { + if (responseWriter.headersInBatch()) { + if (responseWriter.finished() && responseWriter.dataBytesInBatch() == 0) { + stream.transition(Http2StreamState.Event.SEND_HEADERS_ES); + return; + } + stream.transition(Http2StreamState.Event.SEND_HEADERS); + } + if (responseWriter.dataBytesInBatch() != 0 || responseWriter.endStreamInBatch()) { + stream.transition( + responseWriter.endStreamInBatch() + ? Http2StreamState.Event.SEND_DATA_ES + : Http2StreamState.Event.SEND_DATA); + } + } + + @Override + public void responseBatchCompleted(Http2Stream stream) { + stream.endResponseBatch(); + if (stream.id() == 0) return; + if (stream.cancelled() || stream.responseWriter().finished()) { + streams.remove(stream.id()); + streams.release(stream); + } else { + scheduleResume(stream); + } + } + private void failAndRelease(Http2Stream stream, Http2ErrorCode error, Exception cause) { if (stream.id() == 0) return; if (cause != null) log.error("HTTP/2 stream {} failed", stream.id(), cause); streams.remove(stream.id()); + try { + stream.cancel(); + } catch (RuntimeException cancellationFailure) { + log.debug("Failed to cancel HTTP/2 stream {} cleanly", stream.id(), cancellationFailure); + } try { failures.fail(stream.id(), error); } catch (IOException writeFailure) { diff --git a/flash/src/main/java/dev/relism/flash/http2/message/DataBufferPool.java b/flash/src/main/java/dev/relism/flash/http2/message/DataBufferPool.java new file mode 100644 index 0000000..48409cc --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/message/DataBufferPool.java @@ -0,0 +1,69 @@ +package dev.relism.flash.http2.message; + +/** Bounded connection-owned free list of frame-sized request-body buffers. */ +public final class DataBufferPool { + static final class DataBuffer { + final byte[] bytes; + DataBuffer next; + int position; + int length; + int flowControlledBytes; + + DataBuffer(int size) { + bytes = new byte[size]; + } + + void reset() { + next = null; + position = 0; + length = 0; + flowControlledBytes = 0; + } + } + + private final int bufferSize; + private final int maxBuffers; + private DataBuffer free; + private int created; + private int available; + + public DataBufferPool(int bufferSize, int maxBuffers) { + if (bufferSize < 1 || maxBuffers < 1) { + throw new IllegalArgumentException("bufferSize and maxBuffers must be positive"); + } + this.bufferSize = bufferSize; + this.maxBuffers = maxBuffers; + } + + synchronized DataBuffer acquire() { + DataBuffer buffer = free; + if (buffer != null) { + free = buffer.next; + available--; + buffer.reset(); + return buffer; + } + if (created == maxBuffers) return null; + created++; + return new DataBuffer(bufferSize); + } + + synchronized void release(DataBuffer buffer) { + buffer.reset(); + buffer.next = free; + free = buffer; + available++; + } + + public synchronized int createdCount() { + return created; + } + + public synchronized int availableCount() { + return available; + } + + public int capacity() { + return maxBuffers; + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java new file mode 100644 index 0000000..9d92d2a --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java @@ -0,0 +1,232 @@ +package dev.relism.flash.http2.message; + +import dev.relism.flash.http2.Http2ErrorCode; +import dev.relism.flash.http2.Http2Limits; +import dev.relism.flash.http2.Http2StreamException; +import dev.relism.flash.http2.message.DataBufferPool.DataBuffer; +import java.io.IOException; +import java.io.InputStream; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; + +/** Reusable request-body source fed by the connection demultiplexer. */ +public final class Http2RequestBody extends InputStream { + @FunctionalInterface + public interface ConsumptionListener { + void consumed(int flowControlledBytes) throws IOException; + } + + private final DataBufferPool pool; + private final ReentrantLock lock = new ReentrantLock(); + private final Condition dataAvailable = lock.newCondition(); + private final byte[] oneByte = new byte[1]; + private byte[] inline; + private DataBuffer head; + private DataBuffer tail; + private ConsumptionListener listener; + private long declaredLength; + private long received; + private int inlinePosition; + private int inlineFlowControlledBytes; + private boolean inlineMode; + private boolean finished; + + public Http2RequestBody(DataBufferPool pool) { + this.pool = pool; + } + + public void begin(long declaredLength, boolean inlineMode, ConsumptionListener listener) { + releaseQueued(); + this.declaredLength = declaredLength; + this.inlineMode = inlineMode; + this.listener = listener; + received = 0; + inlinePosition = 0; + inlineFlowControlledBytes = 0; + finished = false; + if (inlineMode && inline == null) inline = new byte[Http2Limits.INLINE_BODY_THRESHOLD]; + } + + public void offer( + int streamId, byte[] source, int offset, int length, int flowControlledBytes) { + long next = received + length; + if (next > Http2Limits.MAX_REQUEST_BODY_SIZE) { + throw new Http2StreamException( + streamId, Http2ErrorCode.PROTOCOL_ERROR, "request body exceeds configured limit"); + } + if (declaredLength >= 0 && next > declaredLength) { + throw new Http2StreamException( + streamId, Http2ErrorCode.PROTOCOL_ERROR, "request body exceeds content-length"); + } + if (length == 0) { + notifyConsumed(flowControlledBytes); + return; + } + if (inlineMode) { + if (next > inline.length) { + throw new Http2StreamException( + streamId, Http2ErrorCode.PROTOCOL_ERROR, "inline request body exceeded its bound"); + } + System.arraycopy(source, offset, inline, (int) received, length); + received = next; + inlineFlowControlledBytes += flowControlledBytes; + return; + } + + lock.lock(); + try { + int remaining = length; + int sourcePosition = offset; + while (remaining > 0) { + if (tail == null || tail.length == tail.bytes.length) { + DataBuffer buffer = pool.acquire(); + if (buffer == null) { + throw new Http2StreamException( + streamId, + Http2ErrorCode.ENHANCE_YOUR_CALM, + "request body buffer pool exhausted"); + } + if (tail == null) head = buffer; + else tail.next = buffer; + tail = buffer; + } + int copied = Math.min(remaining, tail.bytes.length - tail.length); + System.arraycopy(source, sourcePosition, tail.bytes, tail.length, copied); + tail.length += copied; + sourcePosition += copied; + remaining -= copied; + } + tail.flowControlledBytes += flowControlledBytes; + received = next; + dataAvailable.signal(); + } finally { + lock.unlock(); + } + } + + public void finish(int streamId) { + if (declaredLength >= 0 && received != declaredLength) { + throw new Http2StreamException( + streamId, + Http2ErrorCode.PROTOCOL_ERROR, + "content-length does not match received DATA bytes"); + } + lock.lock(); + try { + finished = true; + dataAvailable.signalAll(); + } finally { + lock.unlock(); + } + } + + public int cancel() { + int discarded; + lock.lock(); + try { + finished = true; + discarded = inlineFlowControlledBytes + releaseQueuedLocked(); + inlineFlowControlledBytes = 0; + dataAvailable.signalAll(); + } finally { + lock.unlock(); + } + return discarded; + } + + public long declaredLength() { + return declaredLength; + } + + @Override + public int read() throws IOException { + int count = read(oneByte, 0, 1); + return count < 0 ? -1 : oneByte[0] & 0xff; + } + + @Override + public int read(byte[] target, int offset, int length) throws IOException { + if (length == 0) return 0; + if (inlineMode) return readInline(target, offset, length); + + DataBuffer consumed = null; + int copied; + int flowControlled = 0; + lock.lock(); + try { + while (head == null && !finished) { + try { + dataAvailable.await(); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IOException("interrupted while waiting for request DATA", interrupted); + } + } + if (head == null) return -1; + DataBuffer buffer = head; + copied = Math.min(length, buffer.length - buffer.position); + System.arraycopy(buffer.bytes, buffer.position, target, offset, copied); + buffer.position += copied; + if (buffer.position == buffer.length) { + head = buffer.next; + if (head == null) tail = null; + flowControlled = buffer.flowControlledBytes; + consumed = buffer; + } + } finally { + lock.unlock(); + } + if (consumed != null) { + pool.release(consumed); + notifyConsumed(flowControlled); + } + return copied; + } + + private int readInline(byte[] target, int offset, int length) throws IOException { + if (!finished) { + throw new IOException("inline request body is not complete"); + } + if (inlinePosition == received) return -1; + int copied = (int) Math.min(length, received - inlinePosition); + System.arraycopy(inline, inlinePosition, target, offset, copied); + inlinePosition += copied; + if (inlinePosition == received && inlineFlowControlledBytes != 0) { + int flowControlled = inlineFlowControlledBytes; + inlineFlowControlledBytes = 0; + notifyConsumed(flowControlled); + } + return copied; + } + + private void notifyConsumed(int bytes) { + if (bytes == 0 || listener == null) return; + try { + listener.consumed(bytes); + } catch (IOException failure) { + cancel(); + throw new IllegalStateException("failed to update request flow-control window", failure); + } + } + + private void releaseQueued() { + lock.lock(); + try { + releaseQueuedLocked(); + } finally { + lock.unlock(); + } + } + + private int releaseQueuedLocked() { + int flowControlled = 0; + while (head != null) { + DataBuffer released = head; + head = released.next; + flowControlled += released.flowControlledBytes; + pool.release(released); + } + tail = null; + return flowControlled; + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java index 0fa8c70..4c7b256 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java @@ -5,6 +5,7 @@ import dev.relism.flash.http.ContentType; import dev.relism.flash.http.DateHeader; import dev.relism.flash.http.HttpStatus; import dev.relism.flash.http2.Http2ErrorCode; +import dev.relism.flash.http2.Http2Limits; import dev.relism.flash.http2.Http2StreamException; import dev.relism.flash.http2.frame.FrameFlags; import dev.relism.flash.http2.frame.FrameType; @@ -13,6 +14,8 @@ import dev.relism.flash.http2.frame.WriteIntent; import dev.relism.flash.http2.hpack.HpackEncoder; import dev.relism.flash.models.Response; import dev.relism.flash.models.ResponseSerializer; +import java.io.IOException; +import java.io.InputStream; /** * Reusable per-stream HTTP/2 response serializer. It prepares a complete small response outside the @@ -30,13 +33,23 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize private final ByteWriter headerBlock; private final ByteWriter output; private final FrameWriteBuffer frames; - private final byte[] decimalScratch = new byte[10]; + private final byte[] decimalScratch = new byte[20]; + private final byte[] relay = new byte[Http2Limits.MAX_FRAME_SIZE_LOCAL]; private WriteIntent next; private boolean huffmanDynamicValues; private int streamId; private long headerListSize; private long maxHeaderListSize; private Completion completion; + private byte[] fixedBody; + private InputStream streamBody; + private long bodyRemaining; + private int fixedPosition; + private boolean unknownLength; + private boolean finished; + private boolean headersInBatch; + private boolean endStreamInBatch; + private int dataBytesInBatch; public Http2ResponseWriter() { this(1024, 2048); @@ -117,6 +130,157 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize return true; } + /** Starts a response whose DATA may span multiple flow-control windows. */ + public int startFlowControlled( + Response response, + int streamId, + boolean headRequest, + boolean sendDate, + boolean sendContentLength, + boolean huffmanDynamicValues, + boolean emitTableSizeUpdate, + int maxFrameSize, + long maxHeaderListSize, + int availableFlowWindow) + throws IOException { + if (streamId <= 0) throw new IllegalArgumentException("streamId must be positive"); + if (maxFrameSize <= 0 || availableFlowWindow < 0) { + throw new IllegalArgumentException("frame size must be positive and flow window non-negative"); + } + headerBlock.reset(); + output.reset(); + this.streamId = streamId; + this.huffmanDynamicValues = huffmanDynamicValues; + this.maxHeaderListSize = maxHeaderListSize; + headerListSize = 0; + next = null; + headersInBatch = true; + endStreamInBatch = false; + dataBytesInBatch = 0; + fixedPosition = 0; + fixedBody = response.isStreaming() ? null : response.getBody(); + streamBody = response.isStreaming() ? response.getStream() : null; + unknownLength = response.isStreaming() && response.isChunked(); + if (response.isStreaming() && !unknownLength && response.getStreamLength() < 0) { + throw new IllegalArgumentException("known response stream length must not be negative"); + } + bodyRemaining = + response.isStreaming() + ? (unknownLength ? -1 : response.getStreamLength()) + : (fixedBody == null ? 0 : fixedBody.length); + long representationLength = bodyRemaining; + boolean representationUnknownLength = unknownLength; + + int statusCode = response.getStatusCode(); + boolean bodyForbidden = + statusCode == 204 || statusCode == 304 || (statusCode >= 100 && statusCode < 200); + if (headRequest || bodyForbidden) { + fixedBody = null; + streamBody = null; + unknownLength = false; + bodyRemaining = 0; + } + + if (emitTableSizeUpdate) HpackEncoder.writeDynamicTableSizeUpdateZero(headerBlock); + writeStatus(statusCode); + writeContentType(response.getContentType()); + if (sendDate) { + addHeaderListSize(4, 29); + headerBlock.writeBytes(DateHeader.hpackBytes()); + } + if (sendContentLength && !bodyForbidden && !representationUnknownLength) { + addHeaderListSize(CONTENT_LENGTH_NAME_LENGTH, decimalLength(representationLength)); + writeDecimalLiteral(28, representationLength); + } + ResponseSerializer.forEachCustomField(response, this); + + boolean hasBody = unknownLength || bodyRemaining > 0; + writeHeaderFrames(maxFrameSize, !hasBody); + finished = !hasBody; + if (hasBody && availableFlowWindow > 0) { + appendData(maxFrameSize, availableFlowWindow); + } + return dataBytesInBatch; + } + + /** Serializes the next DATA batch after a WINDOW_UPDATE or previous write completion. */ + public int resume(int maxFrameSize, int availableFlowWindow) throws IOException { + if (finished || availableFlowWindow <= 0) return 0; + output.reset(); + next = null; + headersInBatch = false; + endStreamInBatch = false; + dataBytesInBatch = 0; + appendData(maxFrameSize, availableFlowWindow); + return dataBytesInBatch; + } + + private void appendData(int maxFrameSize, int availableFlowWindow) throws IOException { + int target = Math.min(relay.length, Math.min(maxFrameSize, availableFlowWindow)); + int count; + boolean end; + if (fixedBody != null) { + count = (int) Math.min(target, bodyRemaining); + frames.beginFrame( + FrameType.DATA, count == bodyRemaining ? FrameFlags.END_STREAM : 0, streamId); + output.writeBytes(fixedBody, fixedPosition, count); + frames.endFrame(); + fixedPosition += count; + bodyRemaining -= count; + end = bodyRemaining == 0; + } else { + int limit = unknownLength ? target : (int) Math.min(target, bodyRemaining); + count = 0; + boolean eof = false; + while (count < limit) { + int read = streamBody.read(relay, count, limit - count); + if (read < 0) { + eof = true; + break; + } + if (read == 0) { + int one = streamBody.read(); + if (one < 0) { + eof = true; + break; + } + relay[count++] = (byte) one; + } else { + count += read; + } + } + if (!unknownLength) { + bodyRemaining -= count; + if (eof && bodyRemaining != 0) { + throw new IOException("streaming response ended before its declared length"); + } + } + end = unknownLength ? eof : bodyRemaining == 0; + frames.beginFrame(FrameType.DATA, end ? FrameFlags.END_STREAM : 0, streamId); + output.writeBytes(relay, 0, count); + frames.endFrame(); + } + dataBytesInBatch = count; + endStreamInBatch = end; + finished = end; + } + + public boolean finished() { + return finished; + } + + public boolean headersInBatch() { + return headersInBatch; + } + + public boolean endStreamInBatch() { + return endStreamInBatch; + } + + public int dataBytesInBatch() { + return dataBytesInBatch; + } + @Override public void accept( byte[] name, int nameOff, int nameLen, byte[] value, int valueOff, int valueLen) { @@ -151,10 +315,10 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize } } - private void writeDecimalLiteral(int nameIndex, int value) { + private void writeDecimalLiteral(int nameIndex, long value) { int length = decimalLength(value); int offset = decimalScratch.length - length; - int current = value; + long current = value; for (int i = decimalScratch.length - 1; i >= offset; i--) { decimalScratch[i] = (byte) ('0' + current % 10); current /= 10; @@ -209,17 +373,13 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize return true; } - private static int decimalLength(int value) { - if (value < 10) return 1; - if (value < 100) return 2; - if (value < 1000) return 3; - if (value < 10000) return 4; - if (value < 100000) return 5; - if (value < 1000000) return 6; - if (value < 10000000) return 7; - if (value < 100000000) return 8; - if (value < 1000000000) return 9; - return 10; + private static int decimalLength(long value) { + int length = 1; + while (value >= 10) { + value /= 10; + length++; + } + return length; } @Override diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2FlowController.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2FlowController.java new file mode 100644 index 0000000..3836c24 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2FlowController.java @@ -0,0 +1,108 @@ +package dev.relism.flash.http2.stream; + +import dev.relism.flash.http2.Http2ErrorCode; +import dev.relism.flash.http2.Http2Exception; +import dev.relism.flash.http2.Http2Limits; +import dev.relism.flash.http2.Http2StreamException; +import java.io.IOException; + +/** Connection-level half of HTTP/2's two-level flow-control accounting. */ +public final class Http2FlowController { + @FunctionalInterface + public interface WindowUpdateSink { + void update(int streamId, int increment) throws IOException; + } + + private final WindowUpdateSink updates; + private int receiveWindow = Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL; + private int consumedSinceUpdate; + private long sendWindow = 65_535; + + public Http2FlowController(WindowUpdateSink updates) { + this.updates = updates; + } + + public synchronized void receiveConnectionBytes(int bytes) { + if (bytes < 0) throw new IllegalArgumentException("bytes must not be negative"); + if (bytes > receiveWindow) throw Http2Exception.FLOW_CONTROL_ERROR; + receiveWindow -= bytes; + } + + public void consumed(Http2Stream stream, int bytes) throws IOException { + int connectionIncrement = 0; + synchronized (this) { + consumedSinceUpdate += bytes; + if (consumedSinceUpdate >= Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL / 2) { + connectionIncrement = consumedSinceUpdate; + receiveWindow += connectionIncrement; + consumedSinceUpdate = 0; + } + } + int streamIncrement = stream.consumedReceiveBytes(bytes); + if (streamIncrement != 0) updates.update(stream.id(), streamIncrement); + if (connectionIncrement != 0) updates.update(0, connectionIncrement); + } + + public void discarded(int bytes) throws IOException { + int increment = 0; + synchronized (this) { + consumedSinceUpdate += bytes; + if (consumedSinceUpdate >= Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL / 2) { + increment = consumedSinceUpdate; + receiveWindow += increment; + consumedSinceUpdate = 0; + } + } + if (increment != 0) updates.update(0, increment); + } + + public synchronized int reserveSend(Http2Stream stream, int requested) { + int streamWindow = stream.sendWindow(); + if (requested <= 0 || sendWindow <= 0 || streamWindow <= 0) return 0; + int granted = + (int) + Math.min(requested, Math.min(sendWindow, Math.min(streamWindow, Integer.MAX_VALUE))); + sendWindow -= granted; + stream.adjustSendWindow(-granted); + return granted; + } + + public synchronized void refundSend(Http2Stream stream, int bytes) { + if (bytes == 0) return; + sendWindow += bytes; + stream.adjustSendWindow(bytes); + } + + public synchronized void increaseConnectionSendWindow(int increment) { + long next = sendWindow + increment; + if (next > Integer.MAX_VALUE) throw Http2Exception.FLOW_CONTROL_ERROR; + sendWindow = next; + } + + public synchronized void increaseStreamSendWindow(Http2Stream stream, int increment) { + stream.adjustSendWindow(increment); + } + + public synchronized void initializeStreamSendWindow(Http2Stream stream, int initialWindow) { + stream.adjustSendWindow(initialWindow - 65_535); + } + + public synchronized void applyInitialWindowDelta(Http2StreamTable streams, int delta) { + streams.adjustAllSendWindows(delta); + } + + public void receiveStreamBytes(Http2Stream stream, int bytes) { + if (!stream.receiveBytes(bytes)) { + throw new Http2StreamException( + stream.id(), Http2ErrorCode.FLOW_CONTROL_ERROR, "stream receive window exceeded"); + } + } + + public synchronized int connectionReceiveWindow() { + return receiveWindow; + } + + public synchronized long connectionSendWindow() { + return sendWindow; + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java index 94b6645..c73b9e7 100644 --- a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java @@ -4,9 +4,12 @@ import dev.relism.flash.bytes.PooledSlice; import dev.relism.flash.http.ContentType; import dev.relism.flash.http.HttpMethod; import dev.relism.flash.http2.Http2ErrorCode; +import dev.relism.flash.http2.Http2Limits; import dev.relism.flash.http2.Http2StreamException; import dev.relism.flash.http2.hpack.HpackHeaderBlock; +import dev.relism.flash.http2.message.DataBufferPool; import dev.relism.flash.http2.message.Http2HeaderMap; +import dev.relism.flash.http2.message.Http2RequestBody; import dev.relism.flash.http2.message.Http2ResponseWriter; import dev.relism.flash.http2.message.PseudoHeaders; import dev.relism.flash.models.Request; @@ -14,11 +17,22 @@ import dev.relism.flash.models.RequestBody; import dev.relism.flash.models.RequestLine; import dev.relism.flash.models.Response; import dev.relism.flash.routing.AbstractRouter; +import dev.relism.fpr.core.ByteView; +import java.io.IOException; import java.net.InetSocketAddress; import javax.net.ssl.SSLSocket; /** Per-stream request, response, decoded-header and write state. */ -public final class Http2Stream implements Http2ResponseWriter.Completion { +public final class Http2Stream + implements Http2ResponseWriter.Completion, Http2RequestBody.ConsumptionListener, Runnable { + public interface ResponseSink { + void handleRequest(Http2Stream stream); + + void responseBatchCompleted(Http2Stream stream); + + void resumeResponse(Http2Stream stream); + } + private static final byte[] HTTP_2 = {'H', 'T', 'T', 'P', '/', '2'}; private final HpackHeaderBlock headerBlock = new HpackHeaderBlock(); @@ -26,24 +40,37 @@ public final class Http2Stream implements Http2ResponseWriter.Completion { private final Http2HeaderMap headers = new Http2HeaderMap(); private final RequestLine requestLine = new RequestLine(); private final RequestBody requestBody = new RequestBody(); + private final Http2RequestBody http2Body; private final Request request = new Request(); private final Response response = new Response(200, ContentType.TEXT_PLAIN); private final Http2ResponseWriter responseWriter = new Http2ResponseWriter(); private final PooledSlice path = new PooledSlice(); private final PooledSlice query = new PooledSlice(); private final PooledSlice protocol = new PooledSlice(); + private final PooledSlice scanName = new PooledSlice(); + private final PooledSlice scanValue = new PooledSlice(); private int id; private Http2StreamState state = Http2StreamState.IDLE; private int sendWindow = 65_535; + private int receiveWindow = Http2Limits.INITIAL_WINDOW_SIZE_LOCAL; + private int consumedReceiveBytes; + private int emptyDataFrames; private Http2StreamTable owner; private Object routeScratch; private volatile boolean dispatched; private volatile boolean cancelled; private boolean headersValidated; + private Http2FlowController flowController; + private ResponseSink responseSink; + private volatile boolean responseInFlight; + private volatile boolean responseStarted; + private boolean releaseClaimed; + private volatile boolean resumeTask; Http2Stream poolNext; - Http2Stream() { + Http2Stream(DataBufferPool dataBuffers) { + http2Body = new Http2RequestBody(dataBuffers); responseWriter.completion(this); protocol.reset(HTTP_2, 0, HTTP_2.length); } @@ -53,9 +80,17 @@ public final class Http2Stream implements Http2ResponseWriter.Completion { this.owner = owner; state = Http2StreamState.IDLE; sendWindow = 65_535; + receiveWindow = Http2Limits.INITIAL_WINDOW_SIZE_LOCAL; + consumedReceiveBytes = 0; + emptyDataFrames = 0; dispatched = false; cancelled = false; headersValidated = false; + responseInFlight = false; + responseStarted = false; + releaseClaimed = false; + resumeTask = false; + responseSink = null; headerBlock.reset(); } @@ -95,7 +130,7 @@ public final class Http2Stream implements Http2ResponseWriter.Completion { id, Http2ErrorCode.PROTOCOL_ERROR, "unsupported request method"); } requestLine.reset(method, path, question < 0 ? null : query, protocol, headers); - requestBody.reset(null, 0, null, 0, 0); + requestBody.reset(http2Body, http2Body.declaredLength(), null, 0, 0); return Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket); } @@ -105,6 +140,63 @@ public final class Http2Stream implements Http2ResponseWriter.Completion { headersValidated = true; } + public boolean prepareRequestBody(Http2FlowController flowController, boolean endStream) { + this.flowController = flowController; + long contentLength = parseContentLength(); + if (contentLength < 0 && endStream) contentLength = 0; + boolean inline = contentLength >= 0 && contentLength <= Http2Limits.INLINE_BODY_THRESHOLD; + http2Body.begin(contentLength, inline, this); + if (endStream) http2Body.finish(id); + return endStream || !inline; + } + + public void receiveData(byte[] source, int offset, int length, int flowControlledBytes) { + http2Body.offer(id, source, offset, length, flowControlledBytes); + } + + public void finishRequestBody() { + http2Body.finish(id); + } + + private long parseContentLength() { + long parsed = -1; + for (int i = 0; i < headerBlock.count(); i++) { + headerBlock.get(i, scanName, scanValue); + if (!equals(scanName, "content-length")) continue; + long value = parseDecimal(scanValue); + if (parsed >= 0 && parsed != value) { + throw new Http2StreamException( + id, Http2ErrorCode.PROTOCOL_ERROR, "conflicting content-length fields"); + } + parsed = value; + } + return parsed; + } + + private long parseDecimal(ByteView value) { + if (value.length() == 0) { + throw new Http2StreamException(id, Http2ErrorCode.PROTOCOL_ERROR, "empty content-length"); + } + long parsed = 0; + for (int i = 0; i < value.length(); i++) { + int digit = (value.byteAt(i) & 0xff) - '0'; + if (digit < 0 || digit > 9 || parsed > (Http2Limits.MAX_REQUEST_BODY_SIZE - digit) / 10L) { + throw new Http2StreamException( + id, Http2ErrorCode.PROTOCOL_ERROR, "invalid or oversized content-length"); + } + parsed = parsed * 10 + digit; + } + return parsed; + } + + private static boolean equals(ByteView value, String expected) { + if (value.length() != expected.length()) return false; + for (int i = 0; i < value.length(); i++) { + if ((value.byteAt(i) & 0xff) != expected.charAt(i)) return false; + } + return true; + } + public Response resetResponse() { return response.reset(200, ContentType.TEXT_PLAIN); } @@ -129,6 +221,42 @@ public final class Http2Stream implements Http2ResponseWriter.Completion { return responseWriter; } + public void responseSink(ResponseSink responseSink) { + this.responseSink = responseSink; + } + + public void markResponseStarted() { + responseStarted = true; + } + + public boolean responseStarted() { + return responseStarted; + } + + public void markResumeTask() { + resumeTask = true; + } + + public synchronized boolean beginResponseBatch() { + if (responseInFlight) return false; + responseInFlight = true; + return true; + } + + public synchronized void endResponseBatch() { + responseInFlight = false; + } + + public synchronized boolean responseInFlight() { + return responseInFlight; + } + + synchronized boolean claimRelease() { + if (releaseClaimed) return false; + releaseClaimed = true; + return true; + } + public Object routeScratch(AbstractRouter router) { if (routeScratch == null) routeScratch = router.newScratch(); return routeScratch; @@ -144,25 +272,71 @@ public final class Http2Stream implements Http2ResponseWriter.Completion { public void cancel() { cancelled = true; + int discarded = http2Body.cancel(); + if (discarded != 0 && flowController != null) { + try { + flowController.discarded(discarded); + } catch (IOException failure) { + throw new IllegalStateException("failed to restore discarded flow-control bytes", failure); + } + } } public boolean cancelled() { return cancelled; } - public int sendWindow() { + public synchronized int sendWindow() { return sendWindow; } - public void adjustSendWindow(int delta) { + public synchronized void adjustSendWindow(int delta) { long adjusted = (long) sendWindow + delta; if (adjusted > Integer.MAX_VALUE) throw new IllegalStateException("stream window overflow"); sendWindow = (int) adjusted; } + public synchronized boolean receiveBytes(int bytes) { + if (bytes > receiveWindow) return false; + receiveWindow -= bytes; + return true; + } + + public synchronized int consumedReceiveBytes(int bytes) { + consumedReceiveBytes += bytes; + if (consumedReceiveBytes < Http2Limits.INITIAL_WINDOW_SIZE_LOCAL / 2) { + return 0; + } + int increment = consumedReceiveBytes; + receiveWindow += increment; + consumedReceiveBytes = 0; + return increment; + } + + public int incrementEmptyDataFrames() { + return ++emptyDataFrames; + } + + public void resetEmptyDataFrames() { + emptyDataFrames = 0; + } + + @Override + public void consumed(int flowControlledBytes) throws IOException { + flowController.consumed(this, flowControlledBytes); + } + @Override public void responseWriteCompleted() { - Http2StreamTable table = owner; - if (table != null) table.release(this); + ResponseSink sink = responseSink; + if (sink != null) sink.responseBatchCompleted(this); + } + + @Override + public void run() { + ResponseSink sink = responseSink; + if (sink == null) return; + if (resumeTask) sink.resumeResponse(this); + else sink.handleRequest(this); } } diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java index 3588ed7..4aa2d8b 100644 --- a/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java @@ -1,5 +1,7 @@ package dev.relism.flash.http2.stream; +import dev.relism.flash.http2.Http2Limits; +import dev.relism.flash.http2.message.DataBufferPool; import java.util.Arrays; /** Fixed-capacity primitive stream-id table using linear-probed open addressing. */ @@ -17,8 +19,15 @@ public final class Http2StreamTable { private int size; private Http2Stream free; private int created; + private final DataBufferPool dataBuffers; public Http2StreamTable(int maxEntries) { + this( + maxEntries, + new DataBufferPool(Http2Limits.MAX_FRAME_SIZE_LOCAL, Http2Limits.DATA_BUFFER_POOL_SIZE)); + } + + public Http2StreamTable(int maxEntries, DataBufferPool dataBuffers) { if (maxEntries < 1) throw new IllegalArgumentException("maxEntries must be positive"); int capacity = 1; while (capacity < maxEntries * 2) capacity <<= 1; @@ -26,6 +35,7 @@ public final class Http2StreamTable { values = new Http2Stream[capacity]; mask = capacity - 1; this.maxEntries = maxEntries; + this.dataBuffers = dataBuffers; } public synchronized Http2Stream get(int streamId) { @@ -51,7 +61,7 @@ public final class Http2StreamTable { stream.poolNext = null; } else { if (created == maxEntries) return null; - stream = new Http2Stream(); + stream = new Http2Stream(dataBuffers); created++; } stream.reset(streamId, this); @@ -60,6 +70,7 @@ public final class Http2StreamTable { } public synchronized void release(Http2Stream stream) { + if (!stream.claimRelease()) return; stream.clear(); stream.poolNext = free; free = stream; @@ -92,6 +103,14 @@ public final class Http2StreamTable { } } + public synchronized int copyValues(Http2Stream[] target) { + int count = 0; + for (int i = 0; i < keys.length && count < target.length; i++) { + if (keys[i] != 0) target[count++] = values[i]; + } + return count; + } + public synchronized void adjustAllSendWindows(int delta) { for (int i = 0; i < keys.length; i++) { if (keys[i] == 0) continue; diff --git a/flash/src/test/java/dev/relism/flash/http2/Http2BackpressureTest.java b/flash/src/test/java/dev/relism/flash/http2/Http2BackpressureTest.java new file mode 100644 index 0000000..4471098 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/Http2BackpressureTest.java @@ -0,0 +1,37 @@ +package dev.relism.flash.http2; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import dev.relism.flash.http2.message.DataBufferPool; +import dev.relism.flash.http2.message.Http2RequestBody; +import dev.relism.flash.http2.stream.Http2FlowController; +import dev.relism.flash.http2.stream.Http2Stream; +import dev.relism.flash.http2.stream.Http2StreamTable; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class Http2BackpressureTest { + @Test + void windowUpdatesAreWithheldUntilTheHandlerConsumesQueuedData() throws Exception { + AtomicInteger updates = new AtomicInteger(); + Http2FlowController flow = + new Http2FlowController((streamId, increment) -> updates.addAndGet(increment)); + Http2Stream stream = new Http2StreamTable(1).acquire(1); + DataBufferPool pool = new DataBufferPool(Http2Limits.MAX_FRAME_SIZE_LOCAL, 32); + Http2RequestBody body = new Http2RequestBody(pool); + body.begin(-1, false, bytes -> flow.consumed(stream, bytes)); + byte[] frame = new byte[Http2Limits.MAX_FRAME_SIZE_LOCAL]; + + for (int i = 0; i < 32; i++) { + flow.receiveConnectionBytes(frame.length); + flow.receiveStreamBytes(stream, frame.length); + body.offer(1, frame, 0, frame.length, frame.length); + } + assertEquals(0, updates.get(), "receiving alone must not reopen either window"); + + body.finish(1); + assertEquals(32L * frame.length, body.readAllBytes().length); + assertEquals(2 * 32 * frame.length, updates.get()); + assertEquals(32, pool.availableCount()); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java b/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java index 8abd20b..32a006c 100644 --- a/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java @@ -11,6 +11,7 @@ import dev.relism.flash.http2.hpack.HpackDecoder; import dev.relism.flash.http2.hpack.HpackEncoder; import dev.relism.flash.tls.TestKeystores; import dev.relism.flash.tls.TlsConfig; +import java.io.ByteArrayInputStream; import java.io.EOFException; import java.io.InputStream; import java.net.ServerSocket; @@ -81,6 +82,115 @@ class Http2ConnectionIntegrationTest { assertEquals("42:localhost:" + port, response.body()); } + @Test + void javaHttpClientUploadsAndDownloadsFlowControlledBodies(@TempDir Path directory) + throws Exception { + int port = freePort(); + Path keystore = + TestKeystores.build( + directory, + "http2-bodies.p12", + "changeit", + TestKeystores.Entry.of("server", "localhost", "localhost")); + byte[] upload = new byte[2 * 1024 * 1024]; + for (int i = 0; i < upload.length; i++) upload[i] = (byte) (i * 31); + byte[] download = new byte[2 * 1024 * 1024 + 17]; + for (int i = 0; i < download.length; i++) download[i] = (byte) (i * 17); + + app = + FlashApp.create( + FlashConfiguration.builder() + .port(port) + .host("127.0.0.1") + .tls(TlsConfig.keystore(keystore, "changeit")) + .http2Enabled(true) + .build()); + app.post("/echo", (request, response) -> request.body().bytes()); + app.get("/fixed", (request, response) -> response.body(download)); + app.get( + "/stream", + (request, response) -> response.chunked(new ByteArrayInputStream(download))); + app.start(); + + HttpClient client = + HttpClient.newBuilder() + .sslContext(TestKeystores.trustAllClientContext()) + .version(HttpClient.Version.HTTP_2) + .build(); + HttpResponse echoed = + client.send( + HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/echo")) + .POST(HttpRequest.BodyPublishers.ofByteArray(upload)) + .build(), + HttpResponse.BodyHandlers.ofByteArray()); + HttpResponse fixed = + client.send( + HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/fixed")).GET().build(), + HttpResponse.BodyHandlers.ofByteArray()); + HttpResponse streamed = + client.send( + HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/stream")).GET().build(), + HttpResponse.BodyHandlers.ofByteArray()); + + assertArrayEquals(upload, echoed.body()); + assertArrayEquals(download, fixed.body()); + assertArrayEquals(download, streamed.body()); + assertTrue(streamed.headers().firstValue("transfer-encoding").isEmpty()); + } + + @Test + void hundredMegabyteUploadAndDownloadRemainStreaming(@TempDir Path directory) throws Exception { + int port = freePort(); + long length = 100L * 1024 * 1024; + Path keystore = + TestKeystores.build( + directory, + "http2-large-bodies.p12", + "changeit", + TestKeystores.Entry.of("server", "localhost", "localhost")); + app = + FlashApp.create( + FlashConfiguration.builder() + .port(port) + .host("127.0.0.1") + .tls(TlsConfig.keystore(keystore, "changeit")) + .http2Enabled(true) + .build()); + app.post( + "/upload", + (request, response) -> { + long count = verifyPattern(request.body().stream()); + return Long.toString(count); + }); + app.get( + "/download", + (request, response) -> response.stream(new PatternInputStream(length), length)); + app.start(); + + HttpClient client = + HttpClient.newBuilder() + .sslContext(TestKeystores.trustAllClientContext()) + .version(HttpClient.Version.HTTP_2) + .build(); + HttpResponse upload = + client.send( + HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/upload")) + .POST(HttpRequest.BodyPublishers.ofInputStream(() -> new PatternInputStream(length))) + .build(), + HttpResponse.BodyHandlers.ofString()); + HttpResponse download = + client.send( + HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/download")) + .GET() + .build(), + HttpResponse.BodyHandlers.ofInputStream()); + + assertEquals(Long.toString(length), upload.body()); + try (InputStream body = download.body()) { + assertEquals(length, verifyPattern(body)); + } + } + @Test void bodylessGetRunsExistingRouteAndReturnsHeadersAndData() throws Exception { int port = freePort(); @@ -401,6 +511,39 @@ class Http2ConnectionIntegrationTest { throw new AssertionError(); } + private static long verifyPattern(InputStream input) throws Exception { + byte[] buffer = new byte[64 * 1024]; + long position = 0; + int count; + while ((count = input.read(buffer)) >= 0) { + for (int i = 0; i < count; i++) assertEquals((byte) (position++ * 31), buffer[i]); + } + return position; + } + + private static final class PatternInputStream extends InputStream { + private final long length; + private long position; + + PatternInputStream(long length) { + this.length = length; + } + + @Override + public int read() { + if (position == length) return -1; + return (byte) (position++ * 31) & 0xff; + } + + @Override + public int read(byte[] target, int offset, int requested) { + if (position == length) return -1; + int count = (int) Math.min(requested, length - position); + for (int i = 0; i < count; i++) target[offset + i] = (byte) (position++ * 31); + return count; + } + } + private static Http2TestFrames.WireFrame readFrame(InputStream input) throws Exception { byte[] header = input.readNBytes(9); if (header.length != 9) throw new EOFException("truncated frame header"); diff --git a/flash/src/test/java/dev/relism/flash/http2/message/Http2LargeResponseTest.java b/flash/src/test/java/dev/relism/flash/http2/message/Http2LargeResponseTest.java new file mode 100644 index 0000000..b55735c --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/message/Http2LargeResponseTest.java @@ -0,0 +1,54 @@ +package dev.relism.flash.http2.message; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.relism.flash.http.ContentType; +import dev.relism.flash.models.Response; +import java.io.InputStream; +import org.junit.jupiter.api.Test; + +class Http2LargeResponseTest { + @Test + void hundredMegabyteStreamUsesOneBoundedReusableFrameBuffer() throws Exception { + long length = 100L * 1024 * 1024; + Response response = + new Response(200, ContentType.BINARY).stream(new RepeatingInputStream(length), length); + Http2ResponseWriter writer = new Http2ResponseWriter(); + long written = + writer.startFlowControlled( + response, 1, false, false, true, false, false, 16_384, 32_768, 16_384); + int largestBuffer = writer.buffer().length; + while (!writer.finished()) { + written += writer.resume(16_384, 16_384); + largestBuffer = Math.max(largestBuffer, writer.buffer().length); + } + + assertEquals(length, written); + assertTrue(largestBuffer <= 65_536, "serialized storage must not scale with body length"); + } + + private static final class RepeatingInputStream extends InputStream { + private long remaining; + + RepeatingInputStream(long remaining) { + this.remaining = remaining; + } + + @Override + public int read() { + if (remaining == 0) return -1; + remaining--; + return 0x5a; + } + + @Override + public int read(byte[] target, int offset, int length) { + if (remaining == 0) return -1; + int count = (int) Math.min(length, remaining); + java.util.Arrays.fill(target, offset, offset + count, (byte) 0x5a); + remaining -= count; + return count; + } + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/message/Http2RequestBodyTest.java b/flash/src/test/java/dev/relism/flash/http2/message/Http2RequestBodyTest.java new file mode 100644 index 0000000..1b9b57b --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/message/Http2RequestBodyTest.java @@ -0,0 +1,86 @@ +package dev.relism.flash.http2.message; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import dev.relism.flash.http2.Http2ErrorCode; +import dev.relism.flash.http2.Http2Limits; +import dev.relism.flash.http2.Http2StreamException; +import dev.relism.flash.models.RequestBody; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class Http2RequestBodyTest { + @Test + void inlineBodyFeedsTheProtocolNeutralRequestBodyWithOneMaterialization() { + AtomicInteger consumed = new AtomicInteger(); + Http2RequestBody source = new Http2RequestBody(new DataBufferPool(16, 1)); + source.begin(3, true, consumed::addAndGet); + source.offer(1, "abc".getBytes(StandardCharsets.US_ASCII), 0, 3, 5); + source.finish(1); + RequestBody body = new RequestBody(); + body.reset(source, 3, null, 0, 0); + + assertArrayEquals("abc".getBytes(StandardCharsets.US_ASCII), body.bytes()); + assertEquals(5, consumed.get()); + assertEquals(3, body.contentLength()); + } + + @Test + void streamingBodyReusesAndReturnsPooledBuffers() throws Exception { + DataBufferPool pool = new DataBufferPool(8, 2); + AtomicInteger consumed = new AtomicInteger(); + Http2RequestBody source = new Http2RequestBody(pool); + source.begin(-1, false, consumed::addAndGet); + source.offer(1, new byte[] {1, 2, 3, 4, 5, 6, 7, 8}, 0, 8, 8); + source.offer(1, new byte[] {9, 10}, 0, 2, 2); + source.finish(1); + + assertArrayEquals(new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, source.readAllBytes()); + assertEquals(10, consumed.get()); + assertEquals(2, pool.createdCount()); + assertEquals(2, pool.availableCount()); + } + + @Test + void contentLengthMismatchIsAProtocolStreamError() { + Http2RequestBody source = new Http2RequestBody(new DataBufferPool(8, 1)); + source.begin(4, true, bytes -> {}); + source.offer(3, new byte[] {1, 2, 3}, 0, 3, 3); + + Http2StreamException failure = + assertThrows(Http2StreamException.class, () -> source.finish(3)); + assertEquals(Http2ErrorCode.PROTOCOL_ERROR, failure.errorCode()); + } + + @Test + void boundedPoolNeverAllocatesPastItsCapacity() { + DataBufferPool pool = new DataBufferPool(4, 1); + Http2RequestBody source = new Http2RequestBody(pool); + source.begin(-1, false, bytes -> {}); + source.offer(1, new byte[] {1, 2, 3, 4}, 0, 4, 4); + + Http2StreamException failure = + assertThrows( + Http2StreamException.class, + () -> source.offer(1, new byte[] {2}, 0, 1, 1)); + assertEquals(Http2ErrorCode.ENHANCE_YOUR_CALM, failure.errorCode()); + assertEquals(1, pool.createdCount()); + } + + @Test + void unknownLengthBodyCannotExceedTheConfiguredMaximum() { + Http2RequestBody source = new Http2RequestBody(new DataBufferPool(8, 1)); + source.begin(-1, false, bytes -> {}); + + Http2StreamException failure = + assertThrows( + Http2StreamException.class, + () -> + source.offer( + 1, new byte[1], 0, Http2Limits.MAX_REQUEST_BODY_SIZE + 1, 1)); + assertEquals(Http2ErrorCode.PROTOCOL_ERROR, failure.errorCode()); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java b/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java index 6b695ce..633af58 100644 --- a/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java @@ -12,6 +12,7 @@ import dev.relism.flash.http2.frame.FrameType; import dev.relism.flash.http2.hpack.HpackDecoder; import dev.relism.flash.models.Response; import dev.relism.fpr.core.ByteView; +import java.io.ByteArrayInputStream; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; @@ -99,6 +100,38 @@ class Http2ResponseWriterTest { () -> writer.prepare(response, 5, false, false, true, false, false, 16_384, 41, 65_535)); } + @Test + void flowControlledHeadPreservesKnownRepresentationLength() throws Exception { + Response response = + new Response(200, ContentType.BINARY) + .stream(new ByteArrayInputStream(new byte[] {1, 2, 3, 4}), 4); + Http2ResponseWriter writer = new Http2ResponseWriter(); + + writer.startFlowControlled( + response, 1, true, false, true, false, false, 16_384, 4096, 16_384); + Parsed parsed = parse(writer); + + assertEquals(List.of(FrameType.HEADERS), parsed.types); + assertTrue(decode(parsed.headerBlock).contains("content-length=4")); + } + + @Test + void unknownLengthStreamUsesNativeDataWithoutTransferEncoding() throws Exception { + Response response = + new Response(200, ContentType.BINARY) + .chunked(new ByteArrayInputStream(new byte[] {1, 2, 3, 4})); + Http2ResponseWriter writer = new Http2ResponseWriter(); + + writer.startFlowControlled( + response, 1, false, false, true, false, false, 16_384, 4096, 16_384); + Parsed parsed = parse(writer); + List fields = decode(parsed.headerBlock); + + assertFalse(fields.stream().anyMatch(field -> field.startsWith("content-length="))); + assertFalse(fields.stream().anyMatch(field -> field.startsWith("transfer-encoding="))); + assertEquals(List.of(FrameType.HEADERS, FrameType.DATA), parsed.types); + } + private static Parsed parse(Http2ResponseWriter writer) { Parsed parsed = new Parsed(); byte[] wire = writer.buffer(); diff --git a/flash/src/test/java/dev/relism/flash/http2/stream/Http2FlowControlTest.java b/flash/src/test/java/dev/relism/flash/http2/stream/Http2FlowControlTest.java new file mode 100644 index 0000000..2d5fe4b --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/stream/Http2FlowControlTest.java @@ -0,0 +1,81 @@ +package dev.relism.flash.http2.stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import dev.relism.flash.http2.Http2Exception; +import dev.relism.flash.http2.Http2Limits; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class Http2FlowControlTest { + @Test + void receiveWindowsReopenAtHalfWindowAtBothLevels() throws Exception { + AtomicInteger connectionUpdates = new AtomicInteger(); + AtomicInteger streamUpdates = new AtomicInteger(); + Http2FlowController controller = + new Http2FlowController( + (streamId, increment) -> { + if (streamId == 0) connectionUpdates.addAndGet(increment); + else streamUpdates.addAndGet(increment); + }); + Http2Stream stream = new Http2StreamTable(1).acquire(1); + int half = Http2Limits.INITIAL_WINDOW_SIZE_LOCAL / 2; + + controller.receiveConnectionBytes(half); + controller.receiveStreamBytes(stream, half); + controller.consumed(stream, half); + + assertEquals(half, connectionUpdates.get()); + assertEquals(half, streamUpdates.get()); + assertEquals(Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL, controller.connectionReceiveWindow()); + } + + @Test + void connectionAndStreamUnderflowUseTheirCorrectErrorScope() { + Http2FlowController controller = new Http2FlowController((streamId, increment) -> {}); + Http2Stream stream = new Http2StreamTable(1).acquire(1); + + assertSame( + Http2Exception.FLOW_CONTROL_ERROR, + assertThrows( + Http2Exception.class, + () -> + controller.receiveConnectionBytes( + Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL + 1))); + assertEquals( + dev.relism.flash.http2.Http2ErrorCode.FLOW_CONTROL_ERROR, + assertThrows( + dev.relism.flash.http2.Http2StreamException.class, + () -> + controller.receiveStreamBytes( + stream, Http2Limits.INITIAL_WINDOW_SIZE_LOCAL + 1)) + .errorCode()); + } + + @Test + void sendReservationHonoursBothWindowsAndRejectsOverflow() { + Http2FlowController controller = new Http2FlowController((streamId, increment) -> {}); + Http2Stream stream = new Http2StreamTable(1).acquire(1); + + assertEquals(65_535, controller.reserveSend(stream, 100_000)); + assertEquals(0, controller.reserveSend(stream, 1)); + controller.increaseConnectionSendWindow(Integer.MAX_VALUE); + assertSame( + Http2Exception.FLOW_CONTROL_ERROR, + assertThrows(Http2Exception.class, () -> controller.increaseConnectionSendWindow(1))); + } + + @Test + void emptyDataFrameCounterCrossesTheConfiguredLimitDeterministically() { + Http2Stream stream = new Http2StreamTable(1).acquire(1); + for (int i = 1; i <= Http2Limits.MAX_EMPTY_DATA_FRAMES_PER_STREAM; i++) { + assertEquals(i, stream.incrementEmptyDataFrames()); + } + assertEquals( + Http2Limits.MAX_EMPTY_DATA_FRAMES_PER_STREAM + 1, stream.incrementEmptyDataFrames()); + stream.resetEmptyDataFrames(); + assertEquals(1, stream.incrementEmptyDataFrames()); + } +}