feat(core): HTTP/2 Phase 3 — serialized frame writer (GO/NO-GO gate)
Implements the connection-level serialized frame writer per the plan's go/no-go gate: tryLock() fast path with an intrusive Vyukov-style MPSC fallback under contention, ReentrantLock throughout (never synchronized), and a scan-based write-timeout reaper. All four gate criteria met and measured: N=1 0 B/op and 42.6 ns overhead (<=50 ns budget); N=64 65.5% throughput retention (>=60%) and 11.8-14.2 us p999 (<1 ms); no carrier pinning; stress test 10,000/10,000 green across 1000 iterations x 5 concurrency levels x 2 scheduler configs. Compared against plain-lock and dedicated-thread designs with real benchmark numbers, not assertion. Full methodology and results in WRITER.md, DEC-09. Also fixes a real regression found while resuming this work: the JMH benchmark broke plain `mvn test` (no -Pjmh) because it lived in src/test/java, which Surefire's test discovery loads regardless of whether a class is ultimately selected as a test. Moved to a dedicated src/jmh/java source root registered only under the jmh profile (build-helper-maven-plugin), per DEC-17. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Zakaria El Orche
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
a315e1df8b
commit
2bf261e4e2
@@ -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 <b>virtual-thread</b> writers:
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@code trylock_mpsc} — the design that ships as {@link Http2FrameWriter}: {@code tryLock()}
|
||||
* fast path, intrusive MPSC fallback.</li>
|
||||
* <li>{@code plain_lock} — every write blocks on {@code ReentrantLock#lock()}, unconditionally.</li>
|
||||
* <li>{@code dedicated_thread} — every write hands off to a single dedicated platform thread
|
||||
* via the same {@link IntrusiveMpscQueue}, parked/unparked (never a busy poll).</li>
|
||||
* </ul>
|
||||
*
|
||||
* <h3>Why this benchmark drives its own concurrency instead of JMH's {@code @Threads}</h3>
|
||||
* {@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.
|
||||
*
|
||||
* <h3>Why {@code runBurst} waits on a write counter, not just thread completion</h3>
|
||||
* {@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.
|
||||
*
|
||||
* <p>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.
|
||||
*
|
||||
* <p>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<T> {
|
||||
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<WriteIntent> 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<String> 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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
*
|
||||
* <h2>The design, three layers</h2>
|
||||
*
|
||||
* <p><b>Layer 1 — serialize outside the lock.</b> 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."
|
||||
*
|
||||
* <p><b>Layer 2 — {@link ReentrantLock}, never {@code synchronized}.</b> 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()}.
|
||||
*
|
||||
* <p><b>Layer 3 — {@code tryLock()} fast path, intrusive MPSC fallback.</b> 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.
|
||||
*
|
||||
* <h3>Lost-wakeup avoidance</h3>
|
||||
* 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):
|
||||
* <pre>
|
||||
* 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()
|
||||
* </pre>
|
||||
* 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.
|
||||
*
|
||||
* <h3>Write timeout</h3>
|
||||
* 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.
|
||||
*
|
||||
* <p><b>Why the fast path is gated on {@code !queue.hasWork()}, not just {@code tryLock()}</b>
|
||||
* (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.
|
||||
*
|
||||
* <p>Deliberately does <em>not</em> 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 <em>consecutive scans</em> 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<Http2FrameWriter> ACTIVE = ConcurrentHashMap.newKeySet();
|
||||
// Touched only by the single reaper thread -- no synchronization needed.
|
||||
private static final java.util.Map<Http2FrameWriter, Integer> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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 <em>is</em> 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.
|
||||
*
|
||||
* <h3>Only {@link Http2FrameWriter} calls {@link #poll()}</h3>
|
||||
* 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.
|
||||
*
|
||||
* <h3>The stub node and the "inconsistent" result</h3>
|
||||
* 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<WriteIntent> 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 <em>or</em> 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;
|
||||
}
|
||||
}
|
||||
@@ -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}.
|
||||
*
|
||||
* <h3>Layer 1 — serialize outside the lock</h3>
|
||||
* 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).
|
||||
*
|
||||
* <h3>Intrusive queue linkage</h3>
|
||||
* {@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);
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <p>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<Future<?>> 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); }
|
||||
}
|
||||
@@ -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<byte[]> 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();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user