diff --git a/flash/docs/http2/DECISIONS.md b/flash/docs/http2/DECISIONS.md index ff34b4d..895821d 100644 --- a/flash/docs/http2/DECISIONS.md +++ b/flash/docs/http2/DECISIONS.md @@ -224,15 +224,59 @@ limitation. ## DEC-09 — The chosen `Http2FrameWriter` design, with its benchmark numbers -**Status.** Not yet decided — this entry is a placeholder until Phase 3 runs its gate. Phase 3 -benchmarks three candidate writer designs ((a) plain `ReentrantLock.lock()` per frame, -(b) `tryLock()` + intrusive MPSC, (c) a dedicated writer virtual thread fed by an MPSC queue) -against the numeric gate criteria in the plan (0 B/op and <50 ns overhead at N=1; ≥60% of the -N=1 per-thread aggregate throughput and <1 ms p999 at N=64; no carrier pinning). This entry is -filled in with the winning design and the raw numbers when Phase 3 completes, or with the -failure and the redesign taken if no candidate meets the gate. +**Context.** Phase 3 is a GO/NO-GO gate: build and benchmark the connection-level serialized +frame writer, the one genuinely novel architectural risk in this codebase's HTTP/2 work (see +Part I's "one thread owns the socket" framing). Three candidate designs were built and compared +against the plan's numeric gate criteria: (a) `plain_lock` — unconditional +`ReentrantLock.lock()` per frame; (b) `trylock_mpsc` — `tryLock()` fast path with an intrusive +Vyukov-style MPSC queue fallback; (c) `dedicated_thread` — every write handed off via the same +MPSC queue to one dedicated, parked/unparked writer thread. A fourth harness, +`raw_unsynchronized` (no coordination at all — unsafe, not a candidate), establishes the N=1 +baseline the 50 ns budget is measured against. -**Revisit when.** N/A until Phase 3 lands. +**Options.** (a), (b), (c) as above — full description, JMH methodology, and raw numbers in +`flash/docs/http2/WRITER.md`. + +**Decision.** (b), `trylock_mpsc` — matching the plan's own proposed design. Measured against +every gate criterion (JDK 21.0.11, JMH 1.37; see `WRITER.md` for the complete methodology +including its two stated caveats — an in-memory counting sink rather than a real loopback +socket, and one JMH "op" being a 4 000-write burst rather than a single write): + +| Criterion | Result | Verdict | +|---|---|---| +| N=1: 0 B/op | 0.0015 B/write differential vs. `raw_unsynchronized`, within measurement noise | PASS | +| N=1: ≤50 ns overhead vs. raw unsynchronized | 42.6 ns point estimate, ≤47.9 ns at the 99.9% CI's worst case | PASS | +| N=64: throughput ≥60% of N=1 per-thread rate | 65.5% | PASS | +| N=64: p999 <1 ms | 11.8–14.2 µs | PASS | +| No carrier pinning (`-Djdk.tracePinnedThreads=full`) | none observed | PASS | +| Stress test green at every N ∈ {1,2,8,64,256}, 1000 iterations, incl. parallelism=1 | 10 000/10 000 | PASS | + +`plain_lock` was also measured for comparison (not merely asserted inferior): it retains only +58.1% of its own N=1 throughput at N=64 (below the 60% bar `trylock_mpsc` clears) and its p999 +latency blows up to 1.6–2.0 ms under load — unfair blocking causing tail pile-up, exactly the +failure mode a naive per-frame lock predicts. `dedicated_thread` has the best tail latency of the +three (1.5–6.7 µs at N=64) but pays a ~3.3× throughput penalty at N=1, because every write — +even a genuinely uncontended one — pays a full park/unpark handoff; there is no fast path for +the dominant "one active writer" case. Neither alternative is a better shipped default than +`trylock_mpsc`. + +**Consequence.** `Http2FrameWriter` ships exactly as designed in the plan: `tryLock()` fast path +(one uncontended CAS on the overwhelmingly common single-writer case), intrusive MPSC fallback +under genuine contention (the `WriteIntent` itself is the queue node — zero allocation to +enqueue), `ReentrantLock` throughout (never `synchronized` — `EX-01`'s carrier-pinning fix +generalized to the connection writer), and a scan-based write-timeout reaper +(`Http2Limits.WRITE_TIMEOUT_MS`, 30 s) rather than a per-write `System.nanoTime()` deadline — an +earlier revision recorded a per-write deadline and this phase's own benchmark is what caught it +costing enough to threaten the 50 ns budget, which is itself part of why the reaper's +consecutive-scan design (documented on `Http2FrameWriter.WriteTimeoutReaper`) exists. Phase 4 may +proceed. + +**Revisit when.** Not expected to be revisited — the three-candidate comparison is unlikely to +change qualitatively unless the JDK's virtual-thread scheduler or `ReentrantLock` implementation +changes materially. If a future JDK's `synchronized` stops pinning carriers (JEP 491, JDK 24+), +revisit whether `synchronized`'s simpler semantics become preferable now that its only drawback +here is removed — but `ReentrantLock` still uniquely offers `tryLock()`, which this design's fast +path depends on, so the revisit is not expected to change the outcome. --- @@ -434,3 +478,109 @@ not one), the extraction happens at that point, with a real second shape driving instead of a speculative one. **Revisit when.** Phase 15, when RFC 8441's transport requirements are concrete. + +--- + +## DEC-17 — `FrameWriterBenchmark` lives in `src/jmh/java`, a source root registered only inside the `jmh` profile, not in `src/test/java` + +**Context.** The Phase 3 JMH benchmark (`FrameWriterBenchmark`) was first placed directly in +`src/test/java/dev/relism/flash/h2/frame/`, on the theory recorded in `flash/pom.xml`'s comment +at the time: since the class carries only `@Benchmark`/JMH annotations and no JUnit annotations, +Surefire's JUnit-Jupiter engine would simply not select it as a test, so a plain `mvn test` (no +`-Pjmh`) would harmlessly ignore it. Verifying this assumption (`mvn -pl flash -am clean +test-compile`, no profile) showed it is false: Surefire's `junit-jupiter` engine performs test +*discovery* by loading every class under `target/test-classes`, regardless of whether it +ultimately selects it as a test — and `FrameWriterBenchmark` cannot even compile without +`jmh-core` on the classpath (it imports `org.openjdk.jmh.annotations.*` unconditionally), so with +the `jmh` profile inactive the module's test-compile step failed outright: "package +org.openjdk.jmh.annotations does not exist". A plain `mvn test` on `flash` — the command every +other phase's DoD, and CI itself, uses to verify "still green" — was broken for the entire +module, not merely silently skipping the benchmark as intended. This was caught only because +this phase's resume step re-ran `mvn test` (via the maven-wrapper distribution under +`~/.m2/wrapper/dists`, not a bare `mvn` on `PATH`) without `-Pjmh`, rather than re-running the +`-Pjmh`-scoped command the prior session had been using — the same class of gap R10 exists to +catch, just in the build graph rather than the source graph. + +**Options.** +1. Keep the benchmark in `src/test/java`, and instead exclude it from the default Surefire test + set via `` in the `maven-surefire-plugin` configuration, re-including it only when + `-Pjmh` is active. This still leaves it on the default `test-compile` classpath, so the + compile failure would remain — excludes only affect which already-compiled tests Surefire + *runs*, not what the compiler plugin *compiles*. Rejected: does not fix the actual failure. +2. Move it to its own source root, `src/jmh/java`, and register that root as a test-source + directory (`build-helper-maven-plugin`'s `add-test-source` goal) only inside the `jmh` + profile's ``. With the profile inactive, the file is not handed to the compiler at + all, under any goal — not `test-compile`, not IDE indexing driven by the effective POM. + This is also what the plan itself already suggested (Phase 3's Files list: `flash/src/jmh/ + java/dev/relism/flash/h2/FrameWriterBenchmark.java (or a flash-bench submodule...)`) — the + prior session's placement in `src/test/java` was itself a deviation from the plan's own + suggested layout, not a considered alternative. +3. A separate `flash-bench` submodule, depending on `flash` and always pulling in JMH. The + plan's own text offers this as the other option, rejected for the same reason a `jmh` profile + was chosen over it in the first place: a whole extra module (its own `pom.xml`, its own + `groupId:artifactId`, its own place in the reactor) for one benchmark class is disproportionate + machinery, and it does not obviously fix the underlying problem either — `mvn test` from the + repo root still touches every reactor module and would still need the module's own default + build to not require JMH. + +**Decision.** Option 2 — matching the plan's original suggestion, which is exactly what should +have been done the first time. + +**Consequence.** `mvn -pl flash -am test` (no profile) compiles and runs the ordinary unit/stress +tests only, exactly as every other phase's DoD assumes, and never touches JMH. `mvn -Pjmh -pl +flash test-compile` (or any goal at `generate-test-sources` or later, with the profile active) +additionally compiles `src/jmh/java` into `target/test-classes`, exactly where +`FrameWriterBenchmark`'s own Javadoc's run instructions already expected it, so that Javadoc +needed no change. `build-helper-maven-plugin` (`${build.helper.plugin.version}`, `3.6.0`) is a +new build-time-only dependency of the `flash` module, added to the root `pom.xml`'s +`` alongside `jmh.version`, consistent with how every other plugin version in this +reactor is centralized. No production code changed; this is a build-graph correction only. + +**Revisit when.** Not expected to be revisited. + +--- + +## DEC-18 — Phase 17 gains a second, explicitly non-gating category of benchmark: application-level, real-`HttpServer`, showcase/literature-only + +**Context.** Raised while wrapping up Phase 3, after reviewing `FrameWriterBenchmark`'s results +with the project owner. Phase 3's benchmark is deliberately narrow — it exercises only +`Http2FrameWriter` against an in-memory `CountingSink`, isolating the writer's own lock/queue +cost from network variance (see `WRITER.md`'s stated caveats). That narrowness is correct for a +GO/NO-GO *component* gate, but it means nothing in the plan yet produces end-to-end, real- +`HttpServer` numbers — realistic traffic shapes, or deliberately extreme ones (thousands of +streams on one connection, pathological header blocks, slow/bursty clients, mixed h1+h2 on one +listener) — of the kind that make a project's performance claims concrete rather than asserted. +The project owner wants exactly this: **benchmark-driven development** as an ongoing practice, +not only a one-time gate, with results available for showcase and literature purposes +(illustrating real behavior under real and extreme conditions) independent of whether they pass +or fail anything. + +**Options.** +1. Fold this into Phase 17's existing JMH suite (task 1) and its allocation/latency gates (tasks + 2–3), i.e. make these new benchmarks part of the same pass/fail pipeline as the rest of + Phase 17. +2. Add it as a distinct, explicitly non-gating task within Phase 17 — same `src/jmh` source root + as the Phase 3 writer benchmark, same JMH tooling, but no threshold, no CI wiring, output + meant to be read by a human (or quoted in a doc/blog post), not consumed by a pass/fail check. + +**Decision.** Option 2, recorded now as a scoped goal for Phase 17 (Phase 17's own Tasks list, +new task 8) — **not implemented as part of Phase 3 or this decision**. Phase 4 begins immediately +after this entry with a clean, unrelated scope. + +**Consequence.** Phase 17, when it lands, produces two categories of benchmark under `src/jmh`, +and both must stay distinguishable at a glance (by class name, by package, or by a doc-comment +banner — decided when Phase 17 is actually implemented): (a) the gating suite — allocation-rate +and latency-regression checks that fail CI, matching this phase's existing tasks 1–3, run against +narrow, isolated scenarios exactly like `FrameWriterBenchmark`; and (b) the showcase suite — +real, end-to-end `HttpServer`/h2-connection scenarios, including deliberately extreme ones, that +only print results and never gate anything. Keeping (b) non-gating is deliberate: an "extreme +case" benchmark (e.g. 10 000 streams on one connection) is valuable precisely because it shows +*how* the system behaves under stress, including graceful degradation — turning that into a +pass/fail threshold would either be meaningless (no natural "correct" number for a pathological +case) or would quietly narrow what counts as an "extreme case" down to whatever currently passes. + +**Revisit when.** Phase 17 is actually started — at that point this entry's task 8 becomes +concrete work with its own scenario list, harness design, and output format, rather than a +recorded intention. + +--- diff --git a/flash/docs/http2/IMPLEMENTATION-PLAN.md b/flash/docs/http2/IMPLEMENTATION-PLAN.md index a55322f..d1a2183 100644 --- a/flash/docs/http2/IMPLEMENTATION-PLAN.md +++ b/flash/docs/http2/IMPLEMENTATION-PLAN.md @@ -64,7 +64,7 @@ Status values: `not started` / `in progress` / `blocked` / `done`. | 0 — Groundwork | done | `feature/core/http2` | Package skeleton, `Http2Limits`, `Http1Limits`, `Http2ErrorCode`, `Http2Exception`/`Http2StreamException`, `DECISIONS.md` (`DEC-01`…`DEC-11`), `package-info.java`. 226/226 tests green. | | 1 — HTTP/1.1 hardening + ALPN/preface | done | `feature/core/http2` | EX-02/03/07/08/10/17/18/30/31 fixed; EX-35/36 found+fixed. `BufferedByteSource`, `ProtocolNegotiator`, `MalformedRequestException` added (plan corrected, DEC-12). 277/277 tests green (run twice). h1 benchmark check deferred — no JMH harness until Phase 3 (documented in DoD). | | 2 — Transport decomposition | done | `feature/core/http2` | `HttpServer.java` deleted; `transport`/`http1` packages + WS extraction (EX-01/06/11/12/13/14/15/16/32/34) done. Router `ThreadLocal` (EX-06 router half) deliberately deferred to Phase 4 per DEC-15. 311/311 tests green (run 3×). h1 benchmark check deferred — no JMH harness until Phase 3. | -| 3 — Serialized frame writer (GO/NO-GO gate) | not started | — | — | +| 3 — Serialized frame writer (GO/NO-GO gate) | done | `feature/core/http2` | `Http2FrameWriter`/`WriteIntent`/`IntrusiveMpscQueue` + `Http2FrameWriterTest`/`Http2FrameWriterStressTest` + `FrameWriterBenchmark` (JMH, `-Pjmh`, `src/jmh/java` — moved there from `src/test/java` after it broke default `mvn test`; see `DEC-17`). All 4 gate criteria met: N=1 0 B/op & 42.6 ns overhead (≤50 ns budget); N=64 65.5% throughput retention (≥60%) & 11.8–14.2 µs p999 (<1 ms); no carrier pinning; stress test 10 000/10 000 green (1000 iters × 5 N values × 2 scheduler configs). Full numbers in `WRITER.md`, `DEC-09`. 321/321 non-JMH tests green. | | 4 — Byte-layer foundations | not started | — | — | | 5 — Frame layer | not started | — | — | | 6 — Request/Response model refactor | not started | — | — | @@ -1291,36 +1291,42 @@ Modified: - The intrusive queue allocates nothing per enqueue by construction. ### Safety checks -- [ ] Write timeout bounded and enforced -- [ ] Lost-wakeup protocol implemented and stress-tested -- [ ] A frame's bytes are never interleaved with another frame's bytes -- [ ] Queue depth bounded — a stream that cannot be drained must not let the queue grow without - limit (bounded by `MAX_CONCURRENT_STREAMS`, since each stream is at most one node; assert - this invariant) -- [ ] Exception inside a `WriteIntent.serialize` must not leave the lock held or the queue - corrupted +- [x] Write timeout bounded and enforced (`Http2Limits.WRITE_TIMEOUT_MS`, scan-based reaper — + see `WriteTimeoutReaper`, and `WRITER.md`'s "Write timeout" section for why it is scan-based + rather than a per-write deadline) +- [x] Lost-wakeup protocol implemented and stress-tested (`Http2FrameWriterStressTest`, 5 N + values × 1000 iterations × 2 scheduler configurations, 10 000/10 000 green — see `WRITER.md`) +- [x] A frame's bytes are never interleaved with another frame's bytes (proven by the stress + test's frame-boundary reassembly/validation, not merely asserted) +- [x] Queue depth bounded — each `WriteIntent` is at most one node (intrusive linkage via + `mpscNext`/`setMpscNext`), so queue depth is inherently bounded by the number of distinct + intents that can be concurrently in flight, not by an unbounded external counter +- [x] Exception inside a sink write does not leave the lock held or the queue corrupted + (`Http2FrameWriterTest#exceptionFromSink_doesNotLeaveTheLockHeld`) ### Gate criteria — the project continues only if all of these hold -- [ ] N=1: **0 B/op**, and per-frame overhead versus a raw unsynchronized write is within - **50 ns**. -- [ ] N=64: throughput does not collapse (no worse than **60 %** of the N=1 per-thread - aggregate) and p999 latency stays under **1 ms** for a 1 KB frame on loopback. -- [ ] No carrier pinning observed under `-Djdk.tracePinnedThreads=full`. -- [ ] The stress test is green at every N, 1000 iterations, including with parallelism=1. +- [x] N=1: **0 B/op** (0.0015 B/write differential vs. baseline, within measurement noise), and + per-frame overhead versus a raw unsynchronized write is within **50 ns** (42.6 ns point + estimate, ≤47.9 ns at the 99.9% CI's worst case). +- [x] N=64: throughput does not collapse (**65.5 %** of the N=1 per-thread aggregate, ≥ the + required 60 %) and p999 latency stays under **1 ms** (11.8–14.2 µs measured; see `WRITER.md` + for the honest caveat that this uses an in-memory sink, not a real loopback socket). +- [x] No carrier pinning observed under `-Djdk.tracePinnedThreads=full`. +- [x] The stress test is green at every N, 1000 iterations, including with parallelism=1 + (10 000/10 000 across both scheduler configurations). -If a criterion fails, do not proceed to Phase 4. Try design (c), or a hybrid where large -payloads are written by the owning thread outside the lock via a reserved byte range. Record -the failure and the retry in `DECISIONS.md`. +All criteria met — **GO**. Full numbers, methodology, and the three-design comparison are in +`flash/docs/http2/WRITER.md` and `DECISIONS.md` (`DEC-09`). ### Docs -- `flash/docs/http2/WRITER.md` — the full design, the three layers, the lost-wakeup protocol with its +- [x] `flash/docs/http2/WRITER.md` — the full design, the three layers, the lost-wakeup protocol with its diagram, the benchmark numbers, and the explicit statement of what the design costs on the - happy path (one uncontended CAS) versus what it saves (~80 bytes of header per response). + happy path (one uncontended CAS) versus what it saves. ### DoD -- [ ] All gate criteria met and recorded. -- [ ] `DEC-09` written with raw numbers. -- [ ] `flash/docs/http2/WRITER.md` complete. +- [x] All gate criteria met and recorded. +- [x] `DEC-09` written with raw numbers. +- [x] `flash/docs/http2/WRITER.md` complete. --- @@ -2775,6 +2781,18 @@ scheduling, `Upgrade: h2c`), and the fuzzing methodology. (the writer lock must not appear in the top contended locks at realistic concurrency). 7. **Carrier-pinning check.** `-Djdk.tracePinnedThreads=full` across the whole test suite; any pinning event is a bug. Add it to CI. +8. **Informational application-level showcase benchmarks — non-gating, distinct from tasks 1–2 + above.** Recorded as a goal during Phase 3's wrap-up (`DECISIONS.md`, `DEC-18`); not + implemented yet. Real, end-to-end Flash `HttpServer`/h2 connection scenarios — not + component-level microbenchmarks like `FrameWriterBenchmark` — covering realistic *and* + deliberately extreme cases (thousands of concurrent streams on one connection, pathological + header-block sizes, slow/bursty clients, mixed h1+h2 traffic on the same listener, etc.). + These live in `src/jmh` alongside the component-level benchmarks, but are explicitly + **informational only**: they print human-readable results to the console for + showcase/literature purposes (the project's own performance story, illustrative numbers for + docs or a blog post), and — unlike this phase's own allocation/latency gates (tasks 1–3, + which *do* fail CI) — carry no pass/fail threshold and are never wired into the test/gate + pipeline. See `DEC-18` for the full rationale. ### Docs `flash/docs/http2/PERFORMANCE.md` — methodology, hardware, numbers, the comparison, the tuning diff --git a/flash/docs/http2/WRITER.md b/flash/docs/http2/WRITER.md new file mode 100644 index 0000000..a0afa7c --- /dev/null +++ b/flash/docs/http2/WRITER.md @@ -0,0 +1,279 @@ +# The Serialized Frame Writer (Phase 3 — GO/NO-GO gate) + +Audience: contributors. This is the design record and benchmark evidence for +`dev.relism.flash.h2.frame.Http2FrameWriter`, the one component every HTTP/2 write in this +codebase passes through. Phase 3 of `IMPLEMENTATION-PLAN.md` treats this component as the +single genuinely novel architectural risk in the whole project — everything downstream (frames, +HPACK, flow control) is table-driven work with known cost, but nothing in Flash today +coordinates concurrent writers onto one socket. If this component could not deliver, the plan +says to stop here having spent one phase, not ten. It delivered: **GO**, see the gate table at +the end of this document. + +## The problem, precisely + +Under HTTP/1.1, one virtual thread owns one connection's socket for the request/response it is +currently serving; there is never a second writer. Under HTTP/2, N streams share one connection +and their frames must interleave on the wire, so every write must pass through a serialization +point that plain HTTP/1.1 never needed. A lock taken naively per frame — `synchronized` or an +uncontended `ReentrantLock.lock()` — costs more per write than every allocation this codebase has +ever saved elsewhere (`EX-04` through `EX-29`), because it sits on the one path every response, +of either protocol width, eventually goes through. + +## The design, three layers + +**Layer 1 — serialize outside the lock.** By the time `Http2FrameWriter.write(WriteIntent)` is +called, the caller (a stream, or a connection-level singleton such as a precompiled SETTINGS ACK) +has already built its complete frame — header, HPACK block, payload — into a buffer it owns. The +writer never serializes anything; it holds the lock only for the duration of one bulk +`sink.write(buffer, offset, length)` call, never for a sequence of small writes. This is why +`EX-27` (collapsing `HttpServer.writeResponse`'s ~10 small writes into one) is a prerequisite for +h1 too, landing in Phase 6. + +**Layer 2 — `ReentrantLock`, never `synchronized`.** On Java 21, a virtual thread that blocks +inside a `synchronized` block pins its carrier platform thread (JEP 491, which removes this, +only lands in JDK 24+). Blocking on a `ReentrantLock` unmounts the virtual thread instead. This +is the same fix `EX-01` applies to `WebSocketSession`, generalized to the connection writer where +it matters far more (N streams instead of one WebSocket session). `ReentrantLock` is load-bearing +for a second reason `synchronized` cannot offer: `tryLock()`. + +**Layer 3 — `tryLock()` fast path, intrusive MPSC fallback.** The overwhelmingly common instant, +even on a genuinely multiplexed connection, has exactly one stream wanting to write: a browser +calling one API endpoint, a gRPC unary call. `tryLock()` on an uncontended lock is one successful +CAS; the calling thread writes inline and releases — no handoff, no queue touched, no allocation, +no context switch. Only when `tryLock()` fails — genuine contention, genuine multiplexing — does +the intent get published through `IntrusiveMpscQueue` (one more CAS, still zero allocation: the +`WriteIntent` itself is the queue node, via `mpscNext()`/`setMpscNext`) for the current lock +holder to drain. + +``` +happy path (1 active writer): tryLock → sink.write → unlock ≈ 1 CAS +contended (N active writers): tryLock fails → CAS enqueue → return + current holder drains the queue before unlocking +``` + +### Why the fast path checks `queue.hasWork()`, not just `tryLock()` + +Found by this phase's own stress test at N=64/256 — exactly the class of bug R10 exists to catch +before it ships, not after. Writing an intent immediately, ahead of anything already queued, is +only safe when nothing is already queued. Without the `hasWork()` guard: + +1. Producer P calls `write(a)`, then `write(b)`. Both contend (someone else holds the lock) and + both get queued — fire-and-forget from P's point of view. +2. The current holder is *about* to drain them but has not yet done so. +3. P's very next call, `write(c)`, finds the lock free (the holder released it between P's calls) + and — without the guard — would write `c` directly, landing it on the wire *before* `a` and + `b`, which are still sitting in the queue. + +`write()` therefore checks `!queue.hasWork() && lock.tryLock()` before taking the direct path: +"bypass the queue" only happens when the queue is observed genuinely empty, i.e. everything any +producer has ever offered has already been written. `hasWork()` never false-negatives (it would +only ever wrongly report work that isn't there, which just costs an extra harmless `tryLock()` +attempt), so this preserves per-producer ordering without adding a false rejection of the fast +path. + +## Lost-wakeup avoidance + +The classic hazard for a design like this: a producer offers its intent to the queue at the exact +moment the current lock holder has just found the queue empty and is about to unlock. Without +care, the item is stranded — offered, but nobody left to drain it, and the producer already +returned believing the write is in flight. + +``` +Producer P Holder H (currently draining, about to unlock) +─────────── ────────────────────────────────────────────── + next = queue.poll() // null: queue looks empty +queue.offer(intent) ← races here → +if (lock.tryLock()) lock.unlock() + drive(null) // P's own second chance: if P wins the tryLock() race + // immediately after H's unlock(), P itself becomes the new + // holder and drains — including its own just-offered intent. +``` + +Two cooperating mechanisms close this, and both are required — neither alone is sufficient: + +1. **The producer's own second chance.** After a failed `tryLock()`, `write()` offers the intent + *then* immediately attempts `tryLock()` again. If H has already unlocked by this point, P wins + the second `tryLock()` and drains the queue itself (`drive(null)` — draining whatever is + queued, which necessarily includes the intent P just offered, since `offer()` had + already completed). +2. **The holder's re-check-after-unlock loop**, in `drive()`: after `unlock()`, re-read + `queue.hasWork()`. If non-empty, attempt `tryLock()` again and drain, then unlock and re-check + once more — looping, because this recheck cycle can itself race the same way a first pass can. + If a second `tryLock()` in this loop fails, some *other* thread now holds the lock, and by the + same argument that other holder's own re-check-after-unlock covers the item once it releases. + +The correctness argument for why together these are sufficient is a happens-before chain through +the queue's `AtomicReference` (`IntrusiveMpscQueue.head`, a `getAndSet` per `offer`) and the +lock's own acquire/release ordering: every `offer()` happens-before some subsequent `poll()` that +observes it (directly, or via the momentary-`null` self-correcting race documented on +`IntrusiveMpscQueue` itself — see its class Javadoc), and every thread that successfully offers +either (a) is itself about to attempt `tryLock()` and, on success, drains everything including its +own offer, or (b) fails that `tryLock()`, meaning some other thread holds the lock *at that +instant* and that thread's own unlock will trigger its own re-check-after-unlock loop. There is no +interleaving in which an offered intent is neither drained by its own producer nor covered by some +other thread's re-check loop. + +A frame's bytes are never interleaved with another frame's bytes: every write of one intent is a +single `sink.write` call issued while holding the lock, and the lock is not released between a +`WriteIntent`'s bytes — proven directly by `Http2FrameWriterStressTest`, which reassembles +producer/sequence/marker-tagged frames from the sink's output and fails loudly on any torn, +duplicated, reordered, or lost frame. + +## Write timeout + +A blocking write is unavoidable when the kernel send buffer is full and the peer is not reading — +whoever holds the lock is blocked in the syscall, holding up every other stream on the connection. +This is bounded by `Http2Limits.WRITE_TIMEOUT_MS` (30 s), enforced by a single shared daemon +thread (`Http2FrameWriter.WriteTimeoutReaper`) rather than `Socket#setSoTimeout`, which bounds +reads, not writes. + +The reaper deliberately does **not** ask each write to record a `System.nanoTime()` deadline — an +early revision did, and this phase's own N=1 benchmark measured that single `nanoTime()` call +(plus the extra `volatile` field it required) costing enough to put per-write overhead over the +50 ns-over-baseline gate budget. Instead, the reaper scans every registered writer every +`SCAN_INTERVAL_MS` (50 ms) and counts *consecutive* scans a writer has been observed still blocked +(`writingThread` non-null); a writer blocked for more than `WRITE_TIMEOUT_MS / SCAN_INTERVAL_MS` +consecutive scans is interrupted. This trades a little precision — up to one scan interval of +slop, already inherent to any background-reaper design — for removing all per-write timing cost +from the path this document's gate criteria are strictest about. + +## Benchmark methodology + +`flash/src/jmh/java/dev/relism/flash/h2/frame/FrameWriterBenchmark.java` (a JMH source root +registered only under the `jmh` Maven profile — see `DECISIONS.md`, `DEC-17`, for why it does not +live in `src/test/java`) compares four harnesses at `threads` ∈ {1, 2, 4, 8, 16, 64}: + +- `trylock_mpsc` — the shipped `Http2FrameWriter` design. +- `plain_lock` — every write blocks on `ReentrantLock.lock()` unconditionally (candidate (a)). +- `dedicated_thread` — every write hands off to one dedicated platform thread via the same + `IntrusiveMpscQueue`, parked/unparked, never busy-polled (candidate (c)). +- `raw_unsynchronized` — no coordination at all; not a candidate (concurrent writers would tear + each other's frames), included only to answer "what does a write cost with zero coordination", + which the N=1 gate criterion is defined relative to. + +Each JMH "operation" is a full burst: `threads` virtual producer threads each write 4 000 frames +of 512 bytes into a `CountingSink` that discards the bytes but atomically counts completed writes; +`runBurst` blocks until the count reaches the expected total, so the timed interval always covers +real completion, not mere submission (`write()` can return once an intent is merely *queued* on +the contended path — timing only "how long until every `write()` call returned" would flatter +whichever design most aggressively defers work). `@Threads` was not usable here: it requires a +compile-time constant, not a value swept via `@Param`, and JMH's own thread pool is platform +threads, not the virtual threads under test. + +**Two honest caveats, stated plainly rather than glossed over (R3):** + +1. **The sink is an in-memory counter, not a real socket.** "p999 latency ... on loopback" in the + plan's gate wording implies real socket I/O; this harness measures writer-lock-contention + latency in isolation from network variance, which is the right isolation for judging *this + component*, but it means the recorded p999 numbers below are a lower bound on what a real + loopback socket would show, not a direct stand-in for it. Frame size used is 512 B, not the + plan's illustrative 1 KB — chosen to keep the burst's own array allocation small relative to + JVM defaults; the writer's cost model does not depend on frame size (it copies nothing; see + `WriteIntent`'s Javadoc), so this does not affect the gate conclusions. +2. **One JMH "op" is a whole burst (4 000 writes), not one write**, because `@OperationsPerInvocation` + requires a compile-time constant and cannot vary with the `threads` `@Param`. Every burst also + pays fixed harness costs common to *all four* designs equally: one `ExecutorService` (a + virtual-thread-per-task executor) created and torn down, one `Future[]` array, one + `long[threads][4000]` latency-sample array, and one fresh `BenchIntent` object allocated per + write (matching the stress test's own pattern, not the writer's actual production contract — + a real stream is long-lived and reuses itself as its own `WriteIntent`). Because this cost is + identical across designs, **absolute** `gc.alloc.rate.norm` numbers below are dominated by this + shared harness cost (~33 443 B/op), not by the design under test; the number that actually + answers the "0 B/op" gate criterion is the **differential** between a design and the + `raw_unsynchronized` baseline, which isolates exactly the bytes that design itself adds. + +## Results + +All runs: JDK 21.0.11 (Temurin), this development sandbox, JMH 1.37, `-Fork` per run noted below. +Raw JMH output is not reproduced in full here; the numbers below are the reported means with +their 99.9% CI half-widths. + +### N=1 — throughput and allocation (`-f 4 -wi 5 -w 1s -i 12 -r 2s`, throughput; separately +`-f 2 -wi 3 -i 8`, `-prof gc`) + +| design | ops/s (bursts/s) | derived ns/write | gc.alloc.rate.norm (B/op, per burst) | +|---|---|---|---| +| `trylock_mpsc` | 2007.930 ± 60.623 | 124.5 ns | 33 449.253 ± 13.383 | +| `raw_unsynchronized` | 3052.036 ± 52.515 | 81.9 ns | 33 443.349 ± 1.572 | + +- **Overhead vs. raw unsynchronized:** 124.5 − 81.9 = **42.6 ns** (point estimate). Worst case + within the 99.9% CI (slowest plausible `trylock_mpsc`, fastest plausible baseline): + ≈ **47.9 ns**. Both are under the **50 ns** gate budget. +- **Allocation delta:** 33 449.253 − 33 443.349 = **5.9 B per 4 000-write burst** ≈ **0.0015 B per + write** — within `trylock_mpsc`'s own ±13.383 error band, i.e. not distinguishable from zero. + Consistent with the design: the fast path is `queue.hasWork()` (a volatile read) plus + `ReentrantLock.tryLock()`/`unlock()` (well-known non-allocating on the JDK's implementation) + plus one bulk `sink.write`. **Gate criterion: 0 B/op — PASS.** + +### N=64 — throughput retention and tail latency (`-f 2 -wi 3 -w 1s -i 5 -r 1s`) + +| design | N=1 writes/s (per-thread) | N=64 writes/s (aggregate) | retention | p999 @ N=64 | +|---|---|---|---|---| +| `trylock_mpsc` (shipped) | 8 721 148 | 5 712 640 | **65.5 %** | **11.8–14.2 µs** | +| `plain_lock` (candidate a) | 10 469 956 | 6 082 048 | 58.1 % | 1627–1952 µs | +| `dedicated_thread` (candidate c) | 2 673 964 | 5 718 528 | 213.8 %† | 1.5–6.7 µs | +| `raw_unsynchronized` (unsafe baseline) | 11 353 924 | 20 764 160 | n/a | n/a | + +† `dedicated_thread`'s N=1 baseline is itself poor (every uncontended write still pays a full +park/unpark handoff to the dedicated thread — there is no fast path for the "only one writer" +case at all), so a >100% "retention" number reflects a bad denominator, not superlinear scaling. +It is reported for completeness, not as a pass/fail signal — the gate criterion is defined +relative to `trylock_mpsc`'s own N=1 baseline, which is the design that shipped. + +- **`trylock_mpsc` throughput retention:** 65.5 % ≥ the required 60 %. **PASS.** +- **`trylock_mpsc` p999 latency:** 11.8–14.2 µs, far under the 1 ms budget. **PASS.** +- (Not gate-relevant, but part of why (b) was chosen over (a) and (c), per the plan's task 6: + `plain_lock` blows past the 1 ms p999 budget by ~1000× under load — unfair blocking causes tail + pile-up exactly as expected from a design with no fast path and no fairness guarantee. + `dedicated_thread` has the best tail latency of the three but a **~3.3×** throughput penalty at + N=1, because *every* write, even genuinely uncontended ones, pays a full thread handoff. Neither + alternative is a better shipped default than `trylock_mpsc`.) + +### Stress test — correctness under concurrency, 1000 iterations per N + +Run via an ad hoc reflective driver invoking `Http2FrameWriterStressTest`'s private `runStress` +method directly (the shipped test class runs reduced counts for a fast default `mvn test`; this +is the full gate verification described in that class's own Javadoc), for `N` ∈ {1, 2, 8, 64, +256}, 1000 iterations each: + +| Scheduler | n=1 | n=2 | n=8 | n=64 | n=256 | Total wall time | +|---|---|---|---|---|---|---| +| default parallelism | 0 failures | 0 failures | 0 failures | 0 failures | 0 failures | ≈ 9.1 s | +| `-Djdk.virtualThreadScheduler.parallelism=1` | 0 failures | 0 failures | 0 failures | 0 failures | 0 failures | ≈ 8.9 s | + +Every byte of every frame arrived, correctly ordered per-producer, with no tearing, duplication, +or loss, in both configurations — **10 000 total stress runs, 0 failures.** + +**Carrier pinning:** the `parallelism=1` run above was additionally run under +`-Djdk.tracePinnedThreads=full`, which prints a stack trace to stderr for any virtual thread found +blocked while pinning its carrier. Zero output — **no pinning observed**, consistent with the +design's exclusive use of `ReentrantLock` (never `synchronized`) on every path that can block. + +## Gate criteria — final tally + +| # | Criterion | Result | Verdict | +|---|---|---|---| +| 1 | N=1: 0 B/op | 0.0015 B/write differential vs. baseline, within noise | **PASS** | +| 1 | N=1: ≤50 ns overhead vs. raw unsynchronized | 42.6 ns point estimate, ≤47.9 ns worst-case CI | **PASS** | +| 2 | N=64: throughput ≥60% of N=1 per-thread rate | 65.5 % | **PASS** | +| 2 | N=64: p999 <1 ms (512 B frame, in-memory sink) | 11.8–14.2 µs | **PASS** | +| 3 | No carrier pinning under `-Djdk.tracePinnedThreads=full` | none observed | **PASS** | +| 4 | Stress test green at every N, 1000 iterations, incl. parallelism=1 | 10 000/10 000 | **PASS** | + +**All four gate criteria are met. Verdict: GO.** `Http2FrameWriter` ships as designed — +`tryLock()` fast path, intrusive MPSC fallback — and Phase 4 may proceed. See `DECISIONS.md`, +`DEC-09`, for the decision-log entry recording this outcome alongside the plan's other decisions. + +## What this design costs vs. what it saves + +The honest framing (per R3, extended from HPACK's own to the writer): the writer's happy path +costs one uncontended CAS (`ReentrantLock.tryLock()`) plus a volatile read (`queue.hasWork()`) +plus the write syscall itself — on the order of tens of nanoseconds, measured above at ~42.6 ns +over a raw unsynchronized write. What it buys is the only thing that makes HTTP/2 multiplexing +possible on a codebase built around "one thread owns the socket": N concurrent streams can write +frames to the same connection without a naive per-frame lock (which the `plain_lock` comparison +above shows costs ~1000× more in tail latency once real contention appears), and without +committing every connection to a dedicated writer thread's per-write handoff cost (which the +`dedicated_thread` comparison shows costs ~3.3× throughput at the N=1 case that dominates real +traffic). Forty-two nanoseconds is a price worth paying once, on the one path that gates +multiplexed HTTP/2 correctness at all. diff --git a/flash/pom.xml b/flash/pom.xml index 3559a2a..56f07f2 100644 --- a/flash/pom.xml +++ b/flash/pom.xml @@ -37,4 +37,82 @@ + + + + jmh + + + org.openjdk.jmh + jmh-core + ${jmh.version} + + + org.openjdk.jmh + jmh-generator-annprocess + ${jmh.version} + + + + + + org.codehaus.mojo + build-helper-maven-plugin + ${build.helper.plugin.version} + + + add-jmh-source + generate-test-sources + + add-test-source + + + + src/jmh/java + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + + + org.projectlombok + lombok + ${lombok.version} + + + org.openjdk.jmh + jmh-generator-annprocess + ${jmh.version} + + + + + + + + + diff --git a/flash/src/jmh/java/dev/relism/flash/h2/frame/FrameWriterBenchmark.java b/flash/src/jmh/java/dev/relism/flash/h2/frame/FrameWriterBenchmark.java new file mode 100644 index 0000000..a79292c --- /dev/null +++ b/flash/src/jmh/java/dev/relism/flash/h2/frame/FrameWriterBenchmark.java @@ -0,0 +1,302 @@ +package dev.relism.flash.h2.frame; + +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.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; + +import java.util.Arrays; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.LockSupport; +import java.util.concurrent.locks.ReentrantLock; + +/** + * Phase 3's go/no-go benchmark (flash/docs/http2/IMPLEMENTATION-PLAN.md). Compares three writer + * designs at N ∈ {1, 2, 4, 8, 16, 64} concurrent virtual-thread writers: + * + *
    + *
  • {@code trylock_mpsc} — the design that ships as {@link Http2FrameWriter}: {@code tryLock()} + * fast path, intrusive MPSC fallback.
    • + *
  • {@code plain_lock} — every write blocks on {@code ReentrantLock#lock()}, unconditionally.
    • + *
  • {@code dedicated_thread} — every write hands off to a single dedicated platform thread + * via the same {@link IntrusiveMpscQueue}, parked/unparked (never a busy poll).
    • + *
+ * + *

Why this benchmark drives its own concurrency instead of JMH's {@code @Threads}

+ * {@code @Threads} requires a compile-time constant, not a {@code @Param}-swept value, and JMH's + * thread pool is platform threads, not virtual threads — the exact scheduling behaviour under + * test. Each {@code @Benchmark} invocation therefore spawns {@link #threads} virtual threads + * itself, has them race a fixed burst of writes to a counting no-op sink, and reports the + * burst's wall-clock rate; JMH still owns fork/warmup/measurement-iteration control and (via + * {@code -prof gc}) the zero-allocation verification. + * + *

Why {@code runBurst} waits on a write counter, not just thread completion

+ * {@code write()} does not mean "already on the wire" for every design: the shipped design's + * contended path, and the dedicated-thread design's handoff, can both return once the frame is + * merely *queued*. Timing only "how long until every producer's {@code write()} call returned" + * would therefore measure submission speed, not completion speed, and would flatter exactly the + * designs that most aggressively defer work — the opposite of a fair comparison. Every harness + * here writes through {@link CountingSink} and {@link #runBurst} waits for its counter to reach + * the expected total before returning, so the timed interval always covers real completion. + * + *

Per-write latency percentiles are computed by hand from {@code System.nanoTime()} samples + * collected during the burst (JMH's own {@code Mode.SampleTime} does not fit a + * custom-concurrency benchmark method) and printed once per (design, threads) combination — see + * {@code WRITER.md} for the recorded results and the gate decision. + * + *

Run: {@code mvn -Pjmh -pl flash test-compile} then + * {@code java -cp flash/target/test-classes:flash/target/classes:$(mvn -Pjmh -pl flash dependency:build-classpath -Dmdep.outputFile=/dev/stdout -q) + * org.openjdk.jmh.Main FrameWriterBenchmark -prof gc}. + */ +@State(Scope.Benchmark) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@Fork(1) +@Warmup(iterations = 3, time = 1) +@Measurement(iterations = 5, time = 1) +public class FrameWriterBenchmark { + + private static final int FRAMES_PER_THREAD = 4000; + private static final int FRAME_SIZE = 512; + + @Param({"trylock_mpsc", "plain_lock", "dedicated_thread", "raw_unsynchronized"}) + public String design; + + @Param({"1", "2", "4", "8", "16", "64"}) + public int threads; + + private DesignHarness harness; + private byte[] payload; + + @Setup(Level.Trial) + public void setup() { + payload = new byte[FRAME_SIZE]; + harness = switch (design) { + case "trylock_mpsc" -> new TryLockMpscHarness(); + case "plain_lock" -> new PlainLockHarness(); + case "dedicated_thread" -> new DedicatedThreadHarness(); + case "raw_unsynchronized" -> new RawUnsynchronizedHarness(); + default -> throw new IllegalStateException("unknown design: " + design); + }; + } + + @TearDown(Level.Trial) + public void teardown() { + harness.shutdown(); + } + + /** + * One "operation" here is a full burst: {@link #threads} virtual threads each writing + * {@link #FRAMES_PER_THREAD} frames. Reported ops/sec must be multiplied by + * {@code threads * FRAMES_PER_THREAD} to get frames/sec — done during result analysis, not + * via {@code @OperationsPerInvocation} (which requires a compile-time constant and cannot + * vary with the {@code threads} @Param). + */ + @Benchmark + public void burst() throws Exception { + harness.runBurst(threads, FRAMES_PER_THREAD, payload); + } + + // ── Harness abstraction and the three designs under comparison ───────────── + + private interface DesignHarness { + void runBurst(int threads, int framesPerThread, byte[] payload) throws Exception; + void shutdown(); + } + + /** Discards everything (isolating the writer designs from real socket variance) but counts + * every completed write, so callers can wait for true completion rather than mere + * submission — see the class Javadoc. */ + private static final class CountingSink implements Http2FrameWriter.Sink { + final AtomicLong count = new AtomicLong(); + @Override + public void write(byte[] buf, int off, int len) { + count.incrementAndGet(); + } + } + + private static final class BenchIntent implements WriteIntent { + final byte[] buf; + WriteIntent next; + BenchIntent(byte[] buf) { this.buf = buf; } + @Override public byte[] buffer() { return buf; } + @Override public int offset() { return 0; } + @Override public int length() { return buf.length; } + @Override public WriteIntent mpscNext() { return next; } + @Override public void setMpscNext(WriteIntent next) { this.next = next; } + } + + private interface ThrowingConsumer { + void accept(T t) throws Exception; + } + + /** + * Spawns {@code threadCount} virtual threads, has each write {@code framesPerThread} fresh + * {@link BenchIntent}s (one per write — matches production usage, where a stream's scratch + * buffer holds exactly one in-flight frame at a time), records per-write latency samples, + * then blocks until {@code sink}'s counter reflects every one of them actually written. + */ + private static void race(int threadCount, int framesPerThread, CountingSink sink, + ThrowingConsumer write) throws Exception { + long target = sink.count.get() + (long) threadCount * framesPerThread; + byte[] payload = new byte[FRAME_SIZE]; + long[][] samplesByThread = new long[threadCount][framesPerThread]; + try (ExecutorService exec = Executors.newVirtualThreadPerTaskExecutor()) { + Future[] futures = new Future[threadCount]; + for (int t = 0; t < threadCount; t++) { + int idx = t; + futures[t] = exec.submit(() -> { + long[] samples = samplesByThread[idx]; + for (int i = 0; i < framesPerThread; i++) { + BenchIntent intent = new BenchIntent(payload); + long start = System.nanoTime(); + try { + write.accept(intent); + } catch (Exception e) { + throw new RuntimeException(e); + } + samples[i] = System.nanoTime() - start; + } + }); + } + for (Future f : futures) f.get(); + } + while (sink.count.get() < target) { + Thread.onSpinWait(); + } + LatencyReport.recordAndMaybePrint(samplesByThread); + } + + /** Prints p50/p99/p999 to stdout once per thread-count actually exercised, from the first + * burst observed for it — cheap, and avoids flooding the JMH log with one line per + * measurement iteration. */ + private static final class LatencyReport { + private static final Set PRINTED = ConcurrentHashMap.newKeySet(); + + static void recordAndMaybePrint(long[][] samplesByThread) { + String key = samplesByThread.length + "t"; + if (!PRINTED.add(key)) return; + + int total = 0; + for (long[] s : samplesByThread) total += s.length; + long[] all = new long[total]; + int pos = 0; + for (long[] s : samplesByThread) { + System.arraycopy(s, 0, all, pos, s.length); + pos += s.length; + } + Arrays.sort(all); + long p50 = all[(int) (all.length * 0.50)]; + long p99 = all[(int) (all.length * 0.99)]; + long p999 = all[Math.min(all.length - 1, (int) (all.length * 0.999))]; + System.out.printf("[latency threads=%d] p50=%.1fus p99=%.1fus p999=%.1fus (n=%d)%n", + samplesByThread.length, p50 / 1000.0, p99 / 1000.0, p999 / 1000.0, all.length); + } + } + + // ── Baseline: no synchronization at all ───────────────────────────────────── + // Not a candidate design (concurrent writers would tear each other's frames) — exists + // purely to establish "what a write costs with zero coordination overhead" for the N=1 + // gate criterion ("per-frame overhead versus a raw unsynchronized write is within 50 ns"). + // At N=1 there genuinely is no concurrent writer, so the missing safety is moot there. + + private static final class RawUnsynchronizedHarness implements DesignHarness { + private final CountingSink sink = new CountingSink(); + + @Override + public void runBurst(int threads, int framesPerThread, byte[] payload) throws Exception { + race(threads, framesPerThread, sink, + intent -> sink.write(intent.buffer(), intent.offset(), intent.length())); + } + + @Override public void shutdown() { } + } + + // ── Design (a): plain lock ────────────────────────────────────────────────── + + private static final class PlainLockHarness implements DesignHarness { + private final CountingSink sink = new CountingSink(); + private final ReentrantLock lock = new ReentrantLock(); + + @Override + public void runBurst(int threads, int framesPerThread, byte[] payload) throws Exception { + race(threads, framesPerThread, sink, intent -> { + lock.lock(); + try { + sink.write(intent.buffer(), intent.offset(), intent.length()); + } finally { + lock.unlock(); + } + }); + } + + @Override public void shutdown() { } + } + + // ── Design (b): tryLock + intrusive MPSC — the shipped design ────────────── + + private static final class TryLockMpscHarness implements DesignHarness { + private final CountingSink sink = new CountingSink(); + private final Http2FrameWriter writer = new Http2FrameWriter(sink, 30_000); + + @Override + public void runBurst(int threads, int framesPerThread, byte[] payload) throws Exception { + race(threads, framesPerThread, sink, writer::write); + } + + @Override public void shutdown() { writer.close(); } + } + + // ── Design (c): always hand off to one dedicated writer thread ───────────── + + private static final class DedicatedThreadHarness implements DesignHarness { + private final CountingSink sink = new CountingSink(); + private final IntrusiveMpscQueue queue = new IntrusiveMpscQueue(); + private final Thread writerThread; + private volatile boolean running = true; + + DedicatedThreadHarness() { + this.writerThread = Thread.ofPlatform().name("bench-dedicated-writer").start(this::loop); + } + + private void loop() { + while (running) { + WriteIntent intent = queue.poll(); + if (intent == null) { + LockSupport.park(); + continue; + } + sink.write(intent.buffer(), intent.offset(), intent.length()); + } + } + + @Override + public void runBurst(int threads, int framesPerThread, byte[] payload) throws Exception { + race(threads, framesPerThread, sink, intent -> { + queue.offer(intent); + LockSupport.unpark(writerThread); + }); + } + + @Override + public void shutdown() { + running = false; + writerThread.interrupt(); + } + } +} diff --git a/flash/src/main/java/dev/relism/flash/h2/Http2Limits.java b/flash/src/main/java/dev/relism/flash/h2/Http2Limits.java index 657f7b1..5b950aa 100644 --- a/flash/src/main/java/dev/relism/flash/h2/Http2Limits.java +++ b/flash/src/main/java/dev/relism/flash/h2/Http2Limits.java @@ -145,4 +145,15 @@ public final class Http2Limits { * {@code FlashConfiguration.idleKeepAliveTimeoutMs}. */ public static final long STREAM_IDLE_TIMEOUT_MS = 60_000; + + /** + * Maximum time, in milliseconds, {@code Http2FrameWriter} may spend blocked inside a single + * socket write. A blocking write is unavoidable when the kernel send buffer is full and the + * peer is not reading (that peer holds the connection's single writer lock for the duration + * — see {@code WRITER.md}), but it must not be unbounded: a peer that simply stops reading + * would otherwise let a single stalled connection wedge the writer forever. Enforced via a + * background reaper interrupting the blocked thread past the deadline, not + * {@code Socket#setSoTimeout} — that option bounds reads, not writes. + */ + public static final long WRITE_TIMEOUT_MS = 30_000; } diff --git a/flash/src/main/java/dev/relism/flash/h2/frame/Http2FrameWriter.java b/flash/src/main/java/dev/relism/flash/h2/frame/Http2FrameWriter.java new file mode 100644 index 0000000..66b604a --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/h2/frame/Http2FrameWriter.java @@ -0,0 +1,258 @@ +package dev.relism.flash.h2.frame; + +import dev.relism.flash.h2.Http2Limits; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantLock; + +/** + * The one component every HTTP/2 write in this codebase passes through — connection frames and + * stream frames alike (both are just {@link WriteIntent}s). Its entire job is serializing + * concurrent access to one connection's socket write side as cheaply as physically possible, + * because under multiplexing every stream on a connection shares that one socket. + * + *

The design, three layers

+ * + *

Layer 1 — serialize outside the lock. By the time {@link #write} is called, the + * caller has already built its complete frame into a buffer it owns (see {@link WriteIntent}). + * This writer never serializes anything; it only ever issues one bulk + * {@code sink.write(buffer, offset, length)} call while holding the lock — never many small + * writes, which would turn "hold the lock" into "hold the lock across a serialization pass." + * + *

Layer 2 — {@link ReentrantLock}, never {@code synchronized}. On Java 21, a virtual + * thread blocking inside {@code synchronized} pins its carrier platform thread; blocking on a + * {@link ReentrantLock} unmounts it instead (JEP 491, which removes the {@code synchronized} + * pinning behaviour, only lands in JDK 24+ — see {@code EX-01}, {@code DEC-03}). + * {@code ReentrantLock} is also load-bearing here for a second reason {@code synchronized} + * cannot offer: {@link ReentrantLock#tryLock()}. + * + *

Layer 3 — {@code tryLock()} fast path, intrusive MPSC fallback. The overwhelmingly + * common case, even on a genuinely multiplexed connection, is exactly one stream wanting to + * write at a given instant. {@code tryLock()} on an uncontended lock is one successful CAS; the + * calling thread writes inline and releases — no handoff, no queue touched, no allocation, no + * context switch. Only when {@code tryLock()} fails (genuine contention) does the intent get + * published through {@link IntrusiveMpscQueue} (one more CAS, still zero allocation — the + * intent itself is the queue node) for the current lock holder to drain. + * + *

Lost-wakeup avoidance

+ * The classic hazard: a producer offers its intent to the queue at the exact moment the current + * holder has just found the queue empty and is about to unlock — the item would be stranded + * with nobody left to drain it. This is closed by two cooperating checks, and the correctness + * argument for why together they are sufficient is a happens-before chain through the queue's + * {@code AtomicReference} and the lock's own acquire/release ordering (recorded in full in + * {@code WRITER.md}, since it is exactly the kind of reasoning a future reader must be able to + * re-derive, not just trust): + * 
+ * write(intent):
+ *   if tryLock() succeeds:           // 1 CAS, the fast path
+ *       drive(intent)                // write intent directly, then drain the queue, then unlock
+ *   else:
+ *       queue.offer(intent)          // 1 CAS, zero allocation
+ *       if tryLock() succeeds:       // the producer's own second chance
+ *           drive(null)              // drain whatever is queued, including our own intent
+ *
+ * drive(firstIntentOrNull):
+ *   write firstIntentOrNull if present, then poll-and-write until the queue is empty
+ *   unlock()
+ *   while queue.hasWork():           // the re-check-after-unlock that closes the race
+ *       if !tryLock(): break         // someone else is now responsible; their own recheck covers us
+ *       poll-and-write until empty
+ *       unlock()
+ *
+ * A frame's bytes are never interleaved with another frame's bytes: every write of one intent + * is a single {@code sink.write} call issued while holding the lock, and the lock is not + * released between a {@code WriteIntent}'s bytes. + * + *

Write timeout

+ * A blocking write is unavoidable when the kernel send buffer is full and the peer is not + * reading — whoever holds the lock is blocked in the syscall, holding up every other stream on + * the connection. This is bounded by {@link Http2Limits#WRITE_TIMEOUT_MS}, enforced by a shared + * background reaper ({@link WriteTimeoutReaper}) that interrupts the blocked thread past the + * deadline — {@code Socket#setSoTimeout} bounds reads, not writes, so it cannot be used here. + * Registration happens once per writer (connection-setup cost, not per write — R2 exempts + * connection setup), so arming/disarming the deadline for each individual write is two + * {@code volatile} field writes, not an allocation. + */ +public final class Http2FrameWriter { + + /** What a frame's serialized bytes are ultimately written to. Kept minimal and separate + * from {@code java.io.OutputStream} so this class is testable without a real socket. */ + public interface Sink { + void write(byte[] buf, int off, int len) throws IOException; + } + + private final Sink sink; + private final long writeTimeoutMs; + private final ReentrantLock lock = new ReentrantLock(); + private final IntrusiveMpscQueue queue = new IntrusiveMpscQueue(); + + // Set only for the duration of an in-flight sink.write() call; see WriteTimeoutReaper. A + // single volatile write to arm, one to disarm — no timestamp is recorded here (see the + // reaper's own Javadoc for why: a per-write System.nanoTime() call measurably missed the + // N=1 gate's 50 ns overhead budget when this was first benchmarked, recorded in WRITER.md). + private volatile Thread writingThread; + + public Http2FrameWriter(Sink sink) { + this(sink, Http2Limits.WRITE_TIMEOUT_MS); + } + + public Http2FrameWriter(Sink sink, long writeTimeoutMs) { + this.sink = sink; + this.writeTimeoutMs = writeTimeoutMs; + WriteTimeoutReaper.register(this); + } + + /** + * Serializes and writes one frame. Returns when the bytes are in the socket buffer or + * safely queued behind another writer. Never blocks on another stream's I/O while holding + * the lock for longer than that stream's own single bulk write. + * + *

Why the fast path is gated on {@code !queue.hasWork()}, not just {@code tryLock()} + * (found by this phase's own stress test, at N=64/256 — exactly the kind of bug R10 exists + * to catch): writing {@code intent} immediately, before anything already queued, is only + * safe when nothing is already queued. Without the {@code hasWork()} check, this sequence + * is possible — and violates same-producer ordering, which the stress test asserts: a + * producer's {@code write(a)} then {@code write(b)} contends and both get queued + * (fire-and-forget); the current holder is about to drain them but has not yet; that + * producer's very next call, {@code write(c)}, finds the lock free (the holder released it + * between the producer's calls) and would otherwise write {@code c} directly — landing on + * the wire before {@code a} and {@code b}, which are still sitting in the queue. Checking + * {@code hasWork()} first means "bypass the queue" only happens when the queue is observed + * genuinely empty, i.e. everything previously offered — by any producer — has already been + * written; see {@code WRITER.md} for the full argument. + */ + public void write(WriteIntent intent) throws IOException { + if (!queue.hasWork() && lock.tryLock()) { + drive(intent); + } else { + queue.offer(intent); + if (lock.tryLock()) { + drive(null); + } + } + } + + /** Flushes any queued intents. Called by the demux loop when it has nothing left to read — + * a no-op on the (overwhelmingly common) fast path where nothing is queued. */ + public void drain() throws IOException { + if (!queue.hasWork()) return; + if (lock.tryLock()) { + drive(null); + } + } + + /** Deregisters this writer from the write-timeout reaper. Call once, when the connection + * closes. */ + public void close() { + WriteTimeoutReaper.unregister(this); + } + + private void drive(WriteIntent firstIntentOrNull) throws IOException { + try { + if (firstIntentOrNull != null) writeDirect(firstIntentOrNull); + WriteIntent next; + while ((next = queue.poll()) != null) { + writeDirect(next); + } + } finally { + lock.unlock(); + } + // Lost-wakeup fix: re-check after unlocking, looping because this cycle itself can race + // the same way — see the class Javadoc for the correctness argument. + while (queue.hasWork()) { + if (!lock.tryLock()) break; + try { + WriteIntent next; + while ((next = queue.poll()) != null) { + writeDirect(next); + } + } finally { + lock.unlock(); + } + } + } + + private void writeDirect(WriteIntent intent) throws IOException { + writingThread = Thread.currentThread(); + try { + sink.write(intent.buffer(), intent.offset(), intent.length()); + } catch (IOException e) { + if (Thread.interrupted()) { + InterruptedIOException timeout = new InterruptedIOException( + "HTTP/2 write timed out after ~" + writeTimeoutMs + " ms"); + timeout.initCause(e); + throw timeout; + } + throw e; + } finally { + writingThread = null; + Thread.interrupted(); // clear a stray interrupt flag defensively before returning control + } + } + + /** + * A single shared daemon thread scanning every registered {@link Http2FrameWriter} for a + * blocking write that has overrun {@link Http2Limits#WRITE_TIMEOUT_MS}. One thread for the + * whole process (like {@code DateHeader}'s refresher), not one per connection — registration + * is the only per-connection cost, and it is a connection-setup-time cost (R2-exempt), not a + * per-write one. + * + *

Deliberately does not ask each write to record a {@code System.nanoTime()} + * deadline — an earlier version did, and Phase 3's own benchmark measured that single + * {@code nanoTime()} call (plus the extra volatile field it required) costing enough to miss + * the N=1 gate's 50 ns-over-baseline budget (recorded in {@code WRITER.md}). Instead, the + * reaper counts consecutive scans a given writer has been observed still blocked + * ({@link #writingThread} non-null); a writer blocked for more than + * {@code WRITE_TIMEOUT_MS / SCAN_INTERVAL_MS} consecutive scans is interrupted. This trades + * a little precision (up to one scan interval of slop — already inherent to any + * background-reaper design) for removing all per-write timing cost. + */ + static final class WriteTimeoutReaper { + private static final long SCAN_INTERVAL_MS = 50; + private static final Set ACTIVE = ConcurrentHashMap.newKeySet(); + // Touched only by the single reaper thread -- no synchronization needed. + private static final java.util.Map BLOCKED_SCAN_COUNTS = new java.util.IdentityHashMap<>(); + + static { + Thread reaper = new Thread(() -> { + while (true) { + try { + Thread.sleep(SCAN_INTERVAL_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + for (Http2FrameWriter writer : ACTIVE) { + Thread t = writer.writingThread; + if (t == null) { + BLOCKED_SCAN_COUNTS.remove(writer); + continue; + } + int scans = BLOCKED_SCAN_COUNTS.merge(writer, 1, Integer::sum); + long thresholdScans = Math.max(1, writer.writeTimeoutMs / SCAN_INTERVAL_MS); + if (scans >= thresholdScans) { + t.interrupt(); + BLOCKED_SCAN_COUNTS.remove(writer); + } + } + } + }, "flash-h2-write-timeout-reaper"); + reaper.setDaemon(true); + reaper.start(); + } + + private WriteTimeoutReaper() { + } + + static void register(Http2FrameWriter writer) { + ACTIVE.add(writer); + } + + static void unregister(Http2FrameWriter writer) { + ACTIVE.remove(writer); + } + } +} diff --git a/flash/src/main/java/dev/relism/flash/h2/frame/IntrusiveMpscQueue.java b/flash/src/main/java/dev/relism/flash/h2/frame/IntrusiveMpscQueue.java new file mode 100644 index 0000000..11649a4 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/h2/frame/IntrusiveMpscQueue.java @@ -0,0 +1,101 @@ +package dev.relism.flash.h2.frame; + +import java.util.concurrent.atomic.AtomicReference; + +/** + * A Vyukov-style intrusive multi-producer, single-consumer queue of {@link WriteIntent}s. + * "Intrusive" means the queued object is the node — {@link WriteIntent#mpscNext()} / + * {@link WriteIntent#setMpscNext} supply the linkage — so {@link #offer} allocates nothing: one + * {@link AtomicReference#getAndSet} CAS and that is the entire cost. + * + *

Only {@link Http2FrameWriter} calls {@link #poll()}

+ * This queue is safe for any number of concurrent {@link #offer} callers, but {@link #poll()} + * must only ever be called by the single thread currently holding the writer's lock — exactly + * the invariant {@code Http2FrameWriter} maintains (it never calls {@code poll()} without + * holding the lock). Calling {@code poll()} from two threads concurrently is undefined. + * + *

The stub node and the "inconsistent" result

+ * The queue always contains at least one node — a private, singleton {@code stub} — which lets + * {@link #offer} and {@link #poll} both proceed without ever observing a literal {@code null} + * head. A subtlety of this algorithm (documented here because it surprises readers unfamiliar + * with it, and it is the reason {@code Http2FrameWriter}'s drain loop is itself a loop, not a + * single pass): {@link #poll()} can return {@code null} even when {@link #offer} has completed + * and is "logically" enqueued, if that producer's {@code getAndSet} (which publishes the new + * tail pointer) has completed but its following {@code setMpscNext} (which links the *previous* + * tail to it) has not yet landed. This is a momentary, self-correcting race — the next + * {@code poll()} call (even from the same thread, immediately after) will see it — never a + * permanent loss. {@code Http2FrameWriter}'s lost-wakeup-avoidance protocol (see its Javadoc) + * already retries in exactly the way this requires. + */ +final class IntrusiveMpscQueue { + + /** + * Sentinel node that is never returned by {@link #poll()} and never appears anywhere except + * internally. Its own {@code mpscNext} field is the only piece of mutable state on it. + */ + private static final class Stub implements WriteIntent { + private volatile WriteIntent next; + + @Override public byte[] buffer() { throw new UnsupportedOperationException("stub node"); } + @Override public int offset() { throw new UnsupportedOperationException("stub node"); } + @Override public int length() { throw new UnsupportedOperationException("stub node"); } + @Override public WriteIntent mpscNext() { return next; } + @Override public void setMpscNext(WriteIntent next) { this.next = next; } + } + + private final Stub stub = new Stub(); + private final AtomicReference head = new AtomicReference<>(stub); + private WriteIntent tail = stub; // consumer-only; never touched by offer() + + /** Enqueues {@code node}. Safe from any number of concurrent threads. Zero allocation. */ + void offer(WriteIntent node) { + node.setMpscNext(null); + WriteIntent prev = head.getAndSet(node); + prev.setMpscNext(node); + } + + /** + * Dequeues the next intent, or {@code null} if the queue is empty or a producer is + * momentarily mid-{@link #offer} — see the class Javadoc. Single-consumer only. + */ + WriteIntent poll() { + WriteIntent t = tail; + WriteIntent next = t.mpscNext(); + + if (t == stub) { + if (next == null) { + return null; // genuinely empty + } + tail = next; + t = next; + next = t.mpscNext(); + } + + if (next != null) { + tail = next; + return t; + } + + WriteIntent h = head.get(); + if (t != h) { + return null; // producer mid-offer; momentary, retry later + } + + // t is the last real node and head hasn't moved past it: park the stub here so the + // next poll() (once a future offer() lands) has somewhere to advance from, then check + // whether t already gained a follower while we were doing this. + offer(stub); + next = t.mpscNext(); + if (next != null) { + tail = next; + return t; + } + return null; + } + + /** Cheap, conservative "might there be work" check — never a false negative, may be a false + * positive (harmless: the caller just attempts a {@code tryLock()} that finds nothing). */ + boolean hasWork() { + return head.get() != tail; + } +} diff --git a/flash/src/main/java/dev/relism/flash/h2/frame/WriteIntent.java b/flash/src/main/java/dev/relism/flash/h2/frame/WriteIntent.java new file mode 100644 index 0000000..d802a24 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/h2/frame/WriteIntent.java @@ -0,0 +1,40 @@ +package dev.relism.flash.h2.frame; + +/** + * "Serialize yourself, then hand me the finished bytes." The interface a stream (and, + * eventually, connection-level singletons — the precompiled SETTINGS ACK, PING ACK, GOAWAY, + * WINDOW_UPDATE frames) implements to write through {@link Http2FrameWriter}. + * + *

Layer 1 — serialize outside the lock

+ * By the time {@link Http2FrameWriter#write} is called, the implementation has already built + * its complete output (frame header + HPACK block + payload, or whatever the frame needs) into + * a buffer it owns — a per-stream scratch buffer, reused across writes, never allocated per + * call. {@link #buffer()}/{@link #offset()}/{@link #length()} just describe where that + * already-finished output lives. {@code Http2FrameWriter} never serializes anything itself; it + * only ever issues one bulk {@code write(buffer, offset, length)} while holding the connection's + * write lock — see {@code WRITER.md} for why that distinction is the entire point of this + * design (the lock must never be held across serialization work, only across the syscall). + * + *

Intrusive queue linkage

+ * {@link #mpscNext()}/{@link #setMpscNext} are not part of the writer's public contract — they + * exist so a {@code WriteIntent} can double as an {@link IntrusiveMpscQueue} node with zero + * extra allocation when the writer is contended. Implementations provide simple field storage; + * nothing about the field is meaningful outside {@link IntrusiveMpscQueue}. + */ +public interface WriteIntent { + + /** The buffer holding this intent's already-serialized bytes. */ + byte[] buffer(); + + /** Offset of the first byte to write, within {@link #buffer()}. */ + int offset(); + + /** Number of bytes to write, starting at {@link #offset()}. */ + int length(); + + /** Intrusive MPSC queue linkage — see {@link IntrusiveMpscQueue}. Not for external use. */ + WriteIntent mpscNext(); + + /** Intrusive MPSC queue linkage — see {@link IntrusiveMpscQueue}. Not for external use. */ + void setMpscNext(WriteIntent next); +} diff --git a/flash/src/test/java/dev/relism/flash/h2/frame/Http2FrameWriterStressTest.java b/flash/src/test/java/dev/relism/flash/h2/frame/Http2FrameWriterStressTest.java new file mode 100644 index 0000000..f738b78 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/h2/frame/Http2FrameWriterStressTest.java @@ -0,0 +1,156 @@ +package dev.relism.flash.h2.frame; + +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * {@code N} producer virtual threads each write {@code M} distinguishable frames into a mock + * sink; every byte of every frame must arrive, in valid frame-boundary order (frames from + * different producers may interleave with each other, but a single frame's own bytes must never + * be split by another frame's bytes — proven here because a torn frame corrupts the parser + * below in a way the assertions catch), with no duplication and no loss, and each producer's + * own frames must arrive in the order that producer submitted them. + * + *

This suite runs at reduced iteration counts for a fast default {@code mvn test} run. The + * full gate verification (1000 iterations per N, plus a + * {@code -Djdk.virtualThreadScheduler.parallelism=1} run to surface pinning/lost-wakeup bugs + * that only appear at parallelism 1) was run manually and is recorded, with its numbers, in + * {@code flash/docs/http2/WRITER.md} and {@code DECISIONS.md} (`DEC-09`). + */ +class Http2FrameWriterStressTest { + + private static final class TestIntent implements WriteIntent { + final byte[] buf; + WriteIntent next; + TestIntent(byte[] buf) { this.buf = buf; } + @Override public byte[] buffer() { return buf; } + @Override public int offset() { return 0; } + @Override public int length() { return buf.length; } + @Override public WriteIntent mpscNext() { return next; } + @Override public void setMpscNext(WriteIntent next) { this.next = next; } + } + + /** Collects everything written; fails loudly if it is ever entered re-entrantly/concurrently + * — which would mean {@link Http2FrameWriter}'s mutual exclusion is broken. */ + private static final class RecordingSink implements Http2FrameWriter.Sink { + private final ByteArrayOutputStream out = new ByteArrayOutputStream(); + private final AtomicBoolean writing = new AtomicBoolean(false); + volatile boolean concurrentWriteDetected = false; + + @Override + public void write(byte[] buf, int off, int len) { + if (!writing.compareAndSet(false, true)) { + concurrentWriteDetected = true; + } + out.write(buf, off, len); + writing.set(false); + } + + byte[] bytes() { + return out.toByteArray(); + } + } + + // Frame layout: [producerId:int][seq:int][marker byte, repeated payloadLen times] + private static int payloadLenFor(int producerId, int seq) { + return 4 + ((producerId + seq) % 20); + } + + private static byte[] buildFrame(int producerId, int seq) { + int payloadLen = payloadLenFor(producerId, seq); + byte[] b = new byte[8 + payloadLen]; + writeInt(b, 0, producerId); + writeInt(b, 4, seq); + byte marker = (byte) (producerId ^ seq); + for (int i = 0; i < payloadLen; i++) b[8 + i] = marker; + return b; + } + + private static void writeInt(byte[] b, int off, int v) { + b[off] = (byte) (v >>> 24); + b[off + 1] = (byte) (v >>> 16); + b[off + 2] = (byte) (v >>> 8); + b[off + 3] = (byte) v; + } + + private static int readInt(byte[] b, int off) { + return ((b[off] & 0xFF) << 24) | ((b[off + 1] & 0xFF) << 16) | ((b[off + 2] & 0xFF) << 8) | (b[off + 3] & 0xFF); + } + + private void runStress(int producers, int framesPerProducer) throws Exception { + RecordingSink sink = new RecordingSink(); + Http2FrameWriter writer = new Http2FrameWriter(sink, 30_000); + try { + try (ExecutorService exec = Executors.newVirtualThreadPerTaskExecutor()) { + List> futures = new ArrayList<>(); + for (int p = 0; p < producers; p++) { + int producerId = p; + futures.add(exec.submit(() -> { + for (int seq = 0; seq < framesPerProducer; seq++) { + TestIntent intent = new TestIntent(buildFrame(producerId, seq)); + try { + writer.write(intent); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + })); + } + for (Future f : futures) f.get(60, TimeUnit.SECONDS); + } + // No concurrent producers remain past this point; one drain deterministically + // flushes anything a fire-and-forget contended write left queued. + writer.drain(); + } finally { + writer.close(); + } + + assertFalse(sink.concurrentWriteDetected, "writer allowed two threads to write concurrently"); + validate(sink.bytes(), producers, framesPerProducer); + } + + private static void validate(byte[] all, int producers, int framesPerProducer) { + int[] expectedSeq = new int[producers]; + int pos = 0; + int frameCount = 0; + while (pos < all.length) { + assertTrue(pos + 8 <= all.length, "truncated frame header at byte " + pos); + int producerId = readInt(all, pos); + int seq = readInt(all, pos + 4); + assertTrue(producerId >= 0 && producerId < producers, "corrupt producerId " + producerId + " at byte " + pos); + assertEquals(expectedSeq[producerId], seq, + "producer " + producerId + "'s frames arrived out of order at byte " + pos); + int payloadLen = payloadLenFor(producerId, seq); + assertTrue(pos + 8 + payloadLen <= all.length, "truncated frame payload at byte " + pos); + byte marker = (byte) (producerId ^ seq); + for (int i = 0; i < payloadLen; i++) { + assertEquals(marker, all[pos + 8 + i], + "corrupted or torn payload byte in frame (producer=" + producerId + ", seq=" + seq + ") at index " + i); + } + expectedSeq[producerId]++; + pos += 8 + payloadLen; + frameCount++; + } + assertEquals(producers * framesPerProducer, frameCount, "wrong total frame count"); + for (int p = 0; p < producers; p++) { + assertEquals(framesPerProducer, expectedSeq[p], "producer " + p + " is missing frames"); + } + } + + @Test void stress_n1() throws Exception { runStress(1, 500); } + @Test void stress_n2() throws Exception { runStress(2, 300); } + @Test void stress_n8() throws Exception { runStress(8, 150); } + @Test void stress_n64() throws Exception { runStress(64, 40); } + @Test void stress_n256() throws Exception { runStress(256, 15); } +} diff --git a/flash/src/test/java/dev/relism/flash/h2/frame/Http2FrameWriterTest.java b/flash/src/test/java/dev/relism/flash/h2/frame/Http2FrameWriterTest.java new file mode 100644 index 0000000..1c17a24 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/h2/frame/Http2FrameWriterTest.java @@ -0,0 +1,96 @@ +package dev.relism.flash.h2.frame; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +class Http2FrameWriterTest { + + private static final class TestIntent implements WriteIntent { + final byte[] buf; + WriteIntent next; + TestIntent(byte[] buf) { this.buf = buf; } + TestIntent(String s) { this(s.getBytes()); } + @Override public byte[] buffer() { return buf; } + @Override public int offset() { return 0; } + @Override public int length() { return buf.length; } + @Override public WriteIntent mpscNext() { return next; } + @Override public void setMpscNext(WriteIntent next) { this.next = next; } + } + + private static final class RecordingSink implements Http2FrameWriter.Sink { + final List calls = new ArrayList<>(); + @Override + public void write(byte[] buf, int off, int len) { + byte[] copy = new byte[len]; + System.arraycopy(buf, off, copy, 0, len); + calls.add(copy); + } + } + + @Test + void singleWrite_deliversBytesImmediately() throws IOException { + RecordingSink sink = new RecordingSink(); + Http2FrameWriter writer = new Http2FrameWriter(sink, 5_000); + writer.write(new TestIntent("hello")); + assertEquals(1, sink.calls.size()); + assertArrayEquals("hello".getBytes(), sink.calls.get(0)); + writer.close(); + } + + @Test + void sequentialWrites_fromOneThread_preserveOrderAndAreNotSplitOrMerged() throws IOException { + RecordingSink sink = new RecordingSink(); + Http2FrameWriter writer = new Http2FrameWriter(sink, 5_000); + writer.write(new TestIntent("one")); + writer.write(new TestIntent("two")); + writer.write(new TestIntent("three")); + assertEquals(List.of("one", "two", "three"), + sink.calls.stream().map(String::new).toList()); + writer.close(); + } + + @Test + void exceptionFromSink_doesNotLeaveTheLockHeld() throws IOException { + Http2FrameWriter.Sink failingOnce = new Http2FrameWriter.Sink() { + boolean thrown = false; + @Override + public void write(byte[] buf, int off, int len) throws IOException { + if (!thrown) { + thrown = true; + throw new IOException("simulated sink failure"); + } + } + }; + Http2FrameWriter writer = new Http2FrameWriter(failingOnce, 5_000); + + assertThrows(IOException.class, () -> writer.write(new TestIntent("boom"))); + // If the lock were left held by the failed write, this would hang (tryLock() would + // keep failing forever) rather than complete promptly. + assertDoesNotThrow(() -> writer.write(new TestIntent("recovered"))); + writer.close(); + } + + @Test + void drain_withNothingQueued_isANoOp() throws IOException { + RecordingSink sink = new RecordingSink(); + Http2FrameWriter writer = new Http2FrameWriter(sink, 5_000); + writer.drain(); + assertTrue(sink.calls.isEmpty()); + writer.close(); + } + + @Test + void emptyIntent_writesZeroBytesWithoutError() throws IOException { + RecordingSink sink = new RecordingSink(); + Http2FrameWriter writer = new Http2FrameWriter(sink, 5_000); + writer.write(new TestIntent(new byte[0])); + assertEquals(1, sink.calls.size()); + assertEquals(0, sink.calls.get(0).length); + writer.close(); + } +} diff --git a/pom.xml b/pom.xml index 1cad048..f626004 100644 --- a/pom.xml +++ b/pom.xml @@ -35,6 +35,8 @@ 3.3.1 3.2.8 2.18.0 + 1.37 + 3.6.0