Claude
09631dcaab
Fixes found during self-correction before publishing: - /stream used Flux.interval(), which ticks on its own wall-clock schedule independent of downstream demand and threw OverflowException under a slow subscriber; switched to Flux.range(), which has no independent production schedule and can never outrun demand. - Single-trial HTTP load tests on this shared sandbox swung by more than 50% run to run (795ms-1247ms observed on the identical /io endpoint back to back) -- large enough to flip which threading model looked faster. Fixed by taking the median of 5 independent trials for the I/O-bound benchmark and the median of 3 for the event-loop-starvation benchmark, rather than reporting a single noisy run as if it were precise. - The event-loop-starvation test's first cut used only 8 concurrent /cpu requests as background load, which drained through the 4 event-loop threads well inside the /io measurement window and produced an inconsistent, sometimes-inverted result across runs; raising to 60 fixed the under-loading problem but still flaked once during verification (372ms vs 374ms p99, a real tie). Final fix: 150 concurrent requests plus the median-of-3 trials above. Also adds StreamBackpressureTest, a StepVerifier proof that the /stream endpoint never emits ahead of its subscriber's outstanding requests, and updates the module's docs to report the de-noised numbers with an explicit methodology note on how they compare to the single-trial platform/virtual- thread numbers reused from a different post.
51 lines
2.3 KiB
Java
51 lines
2.3 KiB
Java
package com.ankurm.vthreadswebflux;
|
|
|
|
import org.junit.jupiter.api.Test;
|
|
import reactor.test.StepVerifier;
|
|
|
|
import java.time.Duration;
|
|
|
|
/**
|
|
* Proves the backpressure claim in docs/01-webflux-benchmark-methodology.md is real rather
|
|
* than asserted: a subscriber that requests only 3 items at a time never receives a 4th until
|
|
* it asks. StepVerifier.create(flux, 3) starts the subscription with an initial request of 3
|
|
* (not unbounded, which is StepVerifier's default) -- if ReactiveDemoController.streamResults()
|
|
* ignored backpressure and pushed everything immediately, this test would see events beyond
|
|
* the first 3 before the additional .thenRequest(...) calls run, and StepVerifier would fail
|
|
* the sequence.
|
|
*
|
|
* Output: docs/output/04-stream-backpressure.txt.
|
|
*/
|
|
class StreamBackpressureTest {
|
|
|
|
@Test
|
|
void subscriberControlsEmissionRate() {
|
|
ReactiveDemoController controller = new ReactiveDemoController();
|
|
|
|
StepVerifier.create(controller.streamResults(), 3)
|
|
.expectNext("event-0", "event-1", "event-2")
|
|
.expectNoEvent(Duration.ofMillis(80)) // no 4th item until we ask for one
|
|
.thenRequest(2)
|
|
.expectNext("event-3", "event-4")
|
|
.thenRequest(45)
|
|
.expectNextCount(45)
|
|
.expectComplete()
|
|
.verify(Duration.ofSeconds(10));
|
|
|
|
Transcript t = Transcript.start("04-stream-backpressure.txt",
|
|
"Flux backpressure proof: /stream, requested in batches of 3, 2, then 45");
|
|
t.line("StepVerifier.create(controller.streamResults(), 3)");
|
|
t.line(" .expectNext(\"event-0\", \"event-1\", \"event-2\")");
|
|
t.line(" .expectNoEvent(Duration.ofMillis(80)) -- no 4th item arrives without a request");
|
|
t.line(" .thenRequest(2).expectNext(\"event-3\", \"event-4\")");
|
|
t.line(" .thenRequest(45).expectNextCount(45)");
|
|
t.line(" .expectComplete()");
|
|
t.blank();
|
|
t.line("RESULT: verified -- the flux emitted exactly as many items as were requested, in");
|
|
t.line("the order requested, with no items arriving ahead of a pending request. This is");
|
|
t.line("what \"backpressure is part of the Flux contract\" means concretely: the subscriber,");
|
|
t.line("not the producer, controls the emission rate.");
|
|
t.save();
|
|
}
|
|
}
|