From d56c60824e723db10b5f2fe2203e9df1b8e7d1eb Mon Sep 17 00:00:00 2001 From: asmhatre Date: Fri, 4 Sep 2026 00:52:18 +0530 Subject: [PATCH] Add the protocol-comparison module --- protocol-comparison/README.md | 91 ++++++++++ .../docs/01-three-protocols.md | 83 ++++++++++ protocol-comparison/docs/02-benchmarks.md | 74 +++++++++ protocol-comparison/docs/03-payload.md | 56 +++++++ protocol-comparison/docs/04-backpressure.md | 112 +++++++++++++ protocol-comparison/docs/05-choosing.md | 63 +++++++ .../docs/output/backpressure.txt | 10 ++ .../docs/output/grpc-isready.txt | 4 + .../docs/output/payload-sizes.txt | 5 + .../docs/output/request-response.txt | 4 + .../docs/output/rsocket-codecs.txt | 4 + .../docs/output/stream-throughput.txt | 4 + protocol-comparison/docs/output/tests.txt | 6 + protocol-comparison/pom.xml | 95 +++++++++++ protocol-comparison/scripts/run-all.sh | 40 +++++ .../ankurm/protocols/GrpcQuoteService.java | 121 ++++++++++++++ .../ProtocolComparisonApplication.java | 22 +++ .../main/java/com/ankurm/protocols/Quote.java | 8 + .../com/ankurm/protocols/QuoteSource.java | 22 +++ .../protocols/RSocketQuoteController.java | 72 ++++++++ .../com/ankurm/protocols/WebSocketConfig.java | 27 +++ .../protocols/WebSocketQuoteHandler.java | 94 +++++++++++ .../src/main/proto/quotes.proto | 45 +++++ .../src/main/resources/application.yaml | 22 +++ .../ankurm/protocols/BackPressureTest.java | 156 ++++++++++++++++++ .../test/java/com/ankurm/protocols/Bench.java | 29 ++++ .../com/ankurm/protocols/GrpcIsReadyTest.java | 66 ++++++++ .../com/ankurm/protocols/PayloadSizeTest.java | 46 ++++++ .../ankurm/protocols/RSocketDefaultsTest.java | 38 +++++ .../RequestResponseBenchmarkTest.java | 111 +++++++++++++ .../StreamThroughputBenchmarkTest.java | 120 ++++++++++++++ .../java/com/ankurm/protocols/WsClient.java | 60 +++++++ 32 files changed, 1710 insertions(+) create mode 100644 protocol-comparison/README.md create mode 100644 protocol-comparison/docs/01-three-protocols.md create mode 100644 protocol-comparison/docs/02-benchmarks.md create mode 100644 protocol-comparison/docs/03-payload.md create mode 100644 protocol-comparison/docs/04-backpressure.md create mode 100644 protocol-comparison/docs/05-choosing.md create mode 100644 protocol-comparison/docs/output/backpressure.txt create mode 100644 protocol-comparison/docs/output/grpc-isready.txt create mode 100644 protocol-comparison/docs/output/payload-sizes.txt create mode 100644 protocol-comparison/docs/output/request-response.txt create mode 100644 protocol-comparison/docs/output/rsocket-codecs.txt create mode 100644 protocol-comparison/docs/output/stream-throughput.txt create mode 100644 protocol-comparison/docs/output/tests.txt create mode 100644 protocol-comparison/pom.xml create mode 100755 protocol-comparison/scripts/run-all.sh create mode 100644 protocol-comparison/src/main/java/com/ankurm/protocols/GrpcQuoteService.java create mode 100644 protocol-comparison/src/main/java/com/ankurm/protocols/ProtocolComparisonApplication.java create mode 100644 protocol-comparison/src/main/java/com/ankurm/protocols/Quote.java create mode 100644 protocol-comparison/src/main/java/com/ankurm/protocols/QuoteSource.java create mode 100644 protocol-comparison/src/main/java/com/ankurm/protocols/RSocketQuoteController.java create mode 100644 protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketConfig.java create mode 100644 protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java create mode 100644 protocol-comparison/src/main/proto/quotes.proto create mode 100644 protocol-comparison/src/main/resources/application.yaml create mode 100644 protocol-comparison/src/test/java/com/ankurm/protocols/BackPressureTest.java create mode 100644 protocol-comparison/src/test/java/com/ankurm/protocols/Bench.java create mode 100644 protocol-comparison/src/test/java/com/ankurm/protocols/GrpcIsReadyTest.java create mode 100644 protocol-comparison/src/test/java/com/ankurm/protocols/PayloadSizeTest.java create mode 100644 protocol-comparison/src/test/java/com/ankurm/protocols/RSocketDefaultsTest.java create mode 100644 protocol-comparison/src/test/java/com/ankurm/protocols/RequestResponseBenchmarkTest.java create mode 100644 protocol-comparison/src/test/java/com/ankurm/protocols/StreamThroughputBenchmarkTest.java create mode 100644 protocol-comparison/src/test/java/com/ankurm/protocols/WsClient.java diff --git a/protocol-comparison/README.md b/protocol-comparison/README.md new file mode 100644 index 0000000..2f720b8 --- /dev/null +++ b/protocol-comparison/README.md @@ -0,0 +1,91 @@ +# `protocol-comparison` — RSocket vs gRPC vs WebSocket, same use case, measured + +Companion project for +[**RSocket vs gRPC vs WebSocket on Spring Boot 4.1: When Each One Wins**](https://ankurm.com/rsocket-vs-grpc-vs-websocket-spring-boot-4-1/) +on ankurm.com. + +**One application serves the same two operations over all three protocols.** Same JVM, same heap, +same data generator, same five fields — so a difference in the numbers is a difference in the +transports rather than in three separately-written demos. +[`./scripts/run-all.sh`](scripts/run-all.sh) re-measures everything under +[`docs/output/`](docs/output/). + +## Versions + +| | Version | Notes | +|---|---|---| +| JDK | 25 (Temurin 25.0.4.1+1) | current LTS | +| Spring Boot | 4.1.1 | latest GA of the 4.1 line | +| Spring Framework | 7.0.9 | Boot-managed | +| grpc-java | **1.83.1** | Boot-managed. Boot 4.1.0 pinned 1.80.0 — it moved | +| protobuf-java | **4.35.1** | Boot-managed. Boot 4.1.0 pinned 4.34.2 | +| rsocket-java | 1.1.5 | Boot-managed | +| Spring gRPC | 1.1.1 | the programming model; separate version from Boot's | +| Jackson | 3.1.5 (`tools.jackson`) | plus `jackson-dataformat-cbor`, RSocket's default | + +Read from `repo1.maven.org/.../maven-metadata.xml` and Boot's own `spring-boot-dependencies` POM. + +## Quickstart + +```bash +./scripts/run-all.sh # six tests, then re-measure docs/output/ +mvn test # the same without the capture +mvn spring-boot:run # gRPC 9090, RSocket 7000, WebSocket ws://localhost:8080/quotes +``` + +Nothing to install: `protoc` and the gRPC codegen plugin resolve as Maven artifacts for your OS +and architecture, and all three servers start inside the test JVM. + +## The operations + +| | gRPC | RSocket route | WebSocket command | +|---|---|---|---| +| one quote | `QuoteService/GetQuote` | `quote` | `QUOTE ` | +| N quotes | `QuoteService/StreamQuotes` | `quotes` | `STREAM ` | +| unbounded | `QuoteService/StreamUnbounded` | `quotes.unbounded` | `UNBOUNDED ` | +| same, `isReady()`-gated | `QuoteService/StreamUnboundedReady` | — | — | +| server-side produced count | `QuoteService/Produced` | `produced` | `PRODUCED` | + +## Documentation + +1. [The same use case, three ways](docs/01-three-protocols.md) +2. [How they measure — and why a loopback benchmark flatters the wrong one](docs/02-benchmarks.md) +3. [Payload: the one comparison that needs no benchmark](docs/03-payload.md) +4. [Back-pressure: the measurement that decides it](docs/04-backpressure.md) +5. [Choosing](docs/05-choosing.md) + +## Captured output + +| File | What it shows | +|---|---| +| [`payload-sizes.txt`](docs/output/payload-sizes.txt) | one message as protobuf, JSON and CBOR, in hex | +| [`request-response.txt`](docs/output/request-response.txt) | 5 000 sequential calls, p50/p99/mean per protocol | +| [`stream-throughput.txt`](docs/output/stream-throughput.txt) | 50 000 messages on one connection | +| [`backpressure.txt`](docs/output/backpressure.txt) | how many the server produced for a client that wanted 100 | +| [`grpc-isready.txt`](docs/output/grpc-isready.txt) | the documented gRPC fix, tested — and it did not work here | +| [`rsocket-codecs.txt`](docs/output/rsocket-codecs.txt) | what Boot configures for RSocket, and why CBOR wins | +| [`tests.txt`](docs/output/tests.txt) | 6 tests | + +## Read the benchmark honestly + +Client and server are the same process on one loopback interface, two cores. **There is no +network.** That deletes protobuf's size advantage, and it flatters whichever protocol does the +least work per message — which is why raw WebSocket wins both speed benchmarks and why that is +not the recommendation. Treat the ratios as indicative and the absolutes as meaningless off this +box. + +The one result that *does* transfer is the back-pressure table, because it is a property of the +protocols rather than of the link. + +## Four things this module exists to prove + +1. **RSocket's `request(n)` is real and the others have no equivalent.** For a client that took + 100 messages then went quiet for two seconds, the servers produced **100**, **1 346 233** and + **311 617** respectively. +2. **`ServerCallStreamObserver.isReady()` did not bound a gRPC producer here.** The documented fix + was tested; the transcript is committed as measured, including its instability. +3. **Raw WebSocket is the fastest and has the worst tail.** p50 of 124 µs against a p99 of + 1 448 µs — a 12x spread, where gRPC's is 3.5x. +4. **CBOR keeps the field names, and it wins by list order.** It is binary JSON, not a schema + format: 67 bytes against protobuf's 35 and JSON's 83. Boot puts `JacksonCborEncoder` ahead of + `JacksonJsonEncoder`, and a bare `RSocketStrategies.create()` has neither. diff --git a/protocol-comparison/docs/01-three-protocols.md b/protocol-comparison/docs/01-three-protocols.md new file mode 100644 index 0000000..e67ce98 --- /dev/null +++ b/protocol-comparison/docs/01-three-protocols.md @@ -0,0 +1,83 @@ +# 1. The same use case, three ways + +Next: [2. How they measure](02-benchmarks.md) + +--- + +One application in this module serves the same two operations over gRPC, RSocket and a raw +WebSocket: fetch one quote, and stream N quotes. Same JVM, same heap, same +[`QuoteSource`](../src/main/java/com/ankurm/protocols/QuoteSource.java), same five fields. + +That constraint is the point. Cross-protocol benchmarks published on the internet almost always +compare three different applications, and end up measuring three different serialisation +libraries, three JIT states and three thread pools. + +## What each one is, in one sentence + +| | gRPC | RSocket | Raw WebSocket | +|---|---|---|---| +| Transport | HTTP/2 | TCP, WebSocket, or others | HTTP/1.1 Upgrade | +| Schema | `.proto`, mandatory, code-generated | none required | none | +| Encoding here | protobuf | CBOR (Spring's default) | JSON | +| Interaction models | 4, declared in the `.proto` | 4, chosen by return type | whatever you invent | +| Demand signalling | none in the API | `request(n)` on the wire | none at all | +| Deadlines | per call, absolute, propagating | none built in | none | +| Browser client | needs grpc-web + a proxy | yes, over the WebSocket transport | yes, natively | +| Boot 4 starters | `spring-boot-starter-grpc-server` / `-client` | `spring-boot-starter-rsocket` | `spring-boot-starter-websocket` | + +## The three server sides, side by side + +**gRPC** ([`GrpcQuoteService`](../src/main/java/com/ankurm/protocols/GrpcQuoteService.java)) is +generated-class inheritance. There is no registration code: the generated `ImplBase` is a +`BindableService` and Boot registers every such bean. + +```java +@Override +public void getQuote(QuoteRequest request, StreamObserver observer) { + observer.onNext(toProto(source.at(request.getSymbol(), 0))); + observer.onCompleted(); +} +``` + +**RSocket** ([`RSocketQuoteController`](../src/main/java/com/ankurm/protocols/RSocketQuoteController.java)) +is annotation routing, and the interaction model *is the return type*: + +```java +@MessageMapping("quote") +public Mono quote(String symbol) { ... } // request/response + +@MessageMapping("quotes") +public Flux quotes(StreamSpec spec) { ... } // request/stream +``` + +**Raw WebSocket** ([`WebSocketQuoteHandler`](../src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java)) +is a `switch` on a string, because there is nothing else: + +```java +switch (parts[0]) { + case "QUOTE" -> send(session, source.at(parts[1], 0)); + case "STREAM" -> { for (int i = 0; i < count; i++) { send(...); } } + ... +} +``` + +That third block is the honest summary of what a raw WebSocket gives you: a bidirectional pipe +for text or bytes. Everything above it — routing, correlation, errors, versioning, demand +— is yours to invent, and inventing it is how teams end up with a private, undocumented +protocol that only its authors can debug. STOMP exists to stop that, and is covered in +[`../sse-websocket`](../../sse-websocket/README.md). + +## Boot 4 version notes + +Boot 4.1.1's BOM pins **grpc-java 1.83.1** and **protobuf-java 4.35.1**, both deliberately older +than the newest releases on Maven Central. Overriding them independently is how you get a +`NoSuchMethodError` between grpc-java and its shaded Netty. Note that these have moved since Boot +4.1.0, which pinned 1.80.0 and 4.34.2 — so a `spring.grpc.*` guide written against 4.1.0 is +already describing different jars. + +RSocket is **rsocket-java 1.1.5**, and `spring-boot-starter-rsocket` brings +`jackson-dataformat-cbor` on purpose: CBOR, not JSON, is Spring's default RSocket data mime type. + +--- + +Next: [2. How they measure](02-benchmarks.md) diff --git a/protocol-comparison/docs/02-benchmarks.md b/protocol-comparison/docs/02-benchmarks.md new file mode 100644 index 0000000..4d2664b --- /dev/null +++ b/protocol-comparison/docs/02-benchmarks.md @@ -0,0 +1,74 @@ +# 2. How they measure + +Previous: [1. Three protocols](01-three-protocols.md) · Next: [3. Payload](03-payload.md) + +--- + +## Read this before the numbers + +Client and server are **the same process on one machine**, talking over loopback: two cores, +3.9 GB, JDK 25, Boot 4.1.1. That has three consequences, and ignoring them makes the numbers +say the opposite of the truth. + +1. **There is no network.** On a real link, a 35-byte message and an 83-byte message differ by + more than the framing cost, and protobuf's size advantage starts paying. Here it does not. +2. **A protocol that does less looks faster.** WebSocket wins both benchmarks below partly because + it has no flow control, no deadlines, no schema and no per-call metadata. That is a real cost + difference, and it is also exactly what you give up. +3. **Ratios, not absolutes.** Absolute microsecond figures from a two-core sandbox mean nothing + for your hardware. The ordering and the rough factors are what transfer. + +Re-measured on every `./scripts/run-all.sh`, and re-measured **in their own JVM invocation**. +That is not tidiness. The back-pressure tests in [chapter 4](04-backpressure.md) leave unbounded +producers spinning, and running the throughput benchmark after them on a two-core box halved +every number — gRPC fell from 44 801 msgs/s to 18 477. If you take one operational lesson from +this module rather than a protocol one, let it be that: a benchmark that shares a machine with +anything is measuring the machine. + +## Request/response + +5 000 sequential calls after 2 000 warm-up +([`request-response.txt`](output/request-response.txt)): + +``` +gRPC n=5000 p50= 291.3 us p99= 1015.3 us mean= 324.4 us ~3,083 calls/s +RSocket n=5000 p50= 300.9 us p99= 1337.4 us mean= 350.9 us ~2,850 calls/s +WebSocket n=5000 p50= 124.3 us p99= 1447.8 us mean= 172.8 us ~5,786 calls/s +``` + +**gRPC and RSocket are the same speed.** Within noise across runs the two swap places; treat them +as indistinguishable for request/response on a warm connection. + +**Raw WebSocket is about twice as fast at the median — and has the worst tail.** p50 of +124 µs against a p99 of 1 448 µs is a 12x spread. gRPC's is 3.5x. The +median is cheap because a text frame over an already-open socket is nearly free; the tail is bad +because nothing is scheduling anything, so a GC pause or a scheduling hiccup lands on the request +undiluted. + +If your SLO is a percentile rather than an average — and it should be — that table +does not say what a first glance suggests. + +## Server streaming + +50 000 messages on one connection +([`stream-throughput.txt`](output/stream-throughput.txt)): + +``` +gRPC 50,000 msgs in 1.116 s = 44,801 msgs/s ( 22.32 us each) +RSocket 50,000 msgs in 0.702 s = 71,175 msgs/s ( 14.05 us each) +WebSocket 50,000 msgs in 0.459 s = 109,022 msgs/s ( 9.17 us each) +``` + +Same ordering, wider gaps: WebSocket about 2.4x gRPC, RSocket about 1.6x. + +Now read the WebSocket row again with its harness in view. The client's inbox is an +`ArrayBlockingQueue` sized to hold **all fifty thousand messages**, because there was nothing else +to do with them — a raw WebSocket consumer that falls behind can buffer or drop, and those +are the only two options. WebSocket "won" this benchmark by buffering the entire stream in the +client's heap. + +That is not a rhetorical point. It is the next chapter. + +--- + +Previous: [1. Three protocols](01-three-protocols.md) · Next: [3. Payload](03-payload.md) diff --git a/protocol-comparison/docs/03-payload.md b/protocol-comparison/docs/03-payload.md new file mode 100644 index 0000000..0c3a9d9 --- /dev/null +++ b/protocol-comparison/docs/03-payload.md @@ -0,0 +1,56 @@ +# 3. Payload: the one comparison that needs no benchmark + +Previous: [2. Benchmarks](02-benchmarks.md) · Next: [4. Back-pressure](04-backpressure.md) + +--- + +Five fields — a symbol, a sequence number, two doubles and a timestamp — encoded three +ways ([`payload-sizes.txt`](output/payload-sizes.txt)): + +``` +protobuf : 35 bytes 0a044141504c102a197b14ae47e11a5940215c8fc2f5281c5940288080abb4fef39203 +JSON : 83 bytes {"symbol":"AAPL","seq":42,"bid":100.42,"ask":100.44,"epochMicros":1772000000000000} +CBOR : 67 bytes bf6673796d626f6c644141504c63736571182a63626964fb40591ae147ae147b6361736bfb40591c28f5c28f5c6b65706f63684d6963726f731b00064b9fe68ac000ff +JSON is 2.37x protobuf; CBOR is 1.91x protobuf +``` + +This is a property of the formats, not of a machine, so it is the one row of the comparison that +transfers to your hardware unchanged. + +**Protobuf is 35 bytes because field names are integers.** `0a04 41 41 50 4c` is field 1, length +4, `AAPL`. There is no `"symbol"` on the wire at all — the name lives in the `.proto` file +that both sides compiled against. That is where the size comes from, and it is also the whole +argument about schemas: the saving and the coupling are the same fact. + +**CBOR is 67 bytes and still writes the field names.** `6673796d626f6c` is a 6-character text +string, `symbol`. CBOR is binary JSON, not a schema format: it saves the punctuation and encodes +numbers compactly, and it keeps every key. So its advantage over JSON is real but modest, and it +does not require the coupling. + +Concretely, on a stream of a million quotes: 35 MB, 67 MB, 83 MB. On a loopback +that difference disappears into memory bandwidth — which is exactly why the benchmarks in +[chapter 2](02-benchmarks.md) do not reward it and a real link would. + +Two practical notes: + +- **Spring's RSocket default is CBOR**, not JSON, which is why + `spring-boot-starter-rsocket` brings `jackson-dataformat-cbor`. If you are debugging with + `tcpdump` and expecting to read your payloads, that is why you cannot. The mechanism is codec + ordering, printed by + [`RSocketDefaultsTest`](../src/test/java/com/ankurm/protocols/RSocketDefaultsTest.java): + + ``` + encoders : [CharSequenceEncoder, ByteBufferEncoder, ByteArrayEncoder, DataBufferEncoder, JacksonCborEncoder, JacksonJsonEncoder] + bare RSocketStrategies.create() : [CharSequenceEncoder, ByteBufferEncoder, ByteArrayEncoder, DataBufferEncoder] + ``` + + CBOR sits *ahead of* JSON in the Boot-configured list, which is the whole of why it wins. Note + also that a bare `RSocketStrategies.create()` carries neither — an `RSocketRequester` built + without injecting Boot's strategies cannot encode your objects at all. And the class names + follow the Boot 4 rule: `JacksonCborEncoder`, not `Jackson2…`. +- **You can put protobuf on RSocket.** The encoding and the protocol are independent choices; + the row above is what each stack does *by default*, not what it is capable of. + +--- + +Previous: [2. Benchmarks](02-benchmarks.md) · Next: [4. Back-pressure](04-backpressure.md) diff --git a/protocol-comparison/docs/04-backpressure.md b/protocol-comparison/docs/04-backpressure.md new file mode 100644 index 0000000..ad09c23 --- /dev/null +++ b/protocol-comparison/docs/04-backpressure.md @@ -0,0 +1,112 @@ +# 4. Back-pressure: the measurement that decides it + +Previous: [3. Payload](03-payload.md) · Next: [5. Choosing](05-choosing.md) + +--- + +Every protocol comparison eventually reduces to one question, and it is not throughput. It is: + +> **When the consumer stops keeping up, what happens?** + +[`BackPressureTest`](../src/test/java/com/ankurm/protocols/BackPressureTest.java) asks it the +simplest way there is. Each server offers an unbounded stream. Each client takes a hundred +messages and then does nothing. How many did the server produce? + +The sweep over quiet periods is what makes the answer conclusive. A number that stays flat as the +wait grows means something bounded the producer. A number that keeps climbing means nothing did. + +[`backpressure.txt`](output/backpressure.txt): + +``` +protocol quiet 0.5s quiet 1s quiet 2s +RSocket 100 100 100 +gRPC 103,425 449,318 1,208,219 +WebSocket 19,664 143,870 320,255 +``` + +RSocket produced **exactly one hundred**, at every quiet period. gRPC produced over a million in +two seconds. Raw WebSocket, three hundred thousand. These are re-measured on every run and the +absolute counts move by tens of percent; the three orders of magnitude between the rows do not. + +(The growth is faster than linear because the loop is still being JIT-compiled during the shortest +run. The shape is the finding, not the exponent.) + +## Why RSocket stops + +`request(n)` is a **frame on the wire**. When the client's subscriber calls `subscription.request(100)`, +a REQUEST_N frame carries that number to the server, and Reactor's `Flux.generate` is called +exactly a hundred times and then not again. Nothing blocks; there is no buffer to size and no +thread parked. The producer simply is not invoked. + +That is what "Reactive Streams over the network" means, and it is the entire reason RSocket +exists. The same property is why RSocket's streaming throughput in +[chapter 2](02-benchmarks.md) is respectable rather than the highest: it is doing real work per +batch that the others are not doing at all. + +## Why the other two do not + +**Raw WebSocket has no mechanism.** The frame protocol has no notion of demand. The server writes +until something underneath refuses — a socket buffer, Spring's `sendBufferSizeLimit`, the +heap. Every one of those 320 255 messages was serialised and written for a client that had +asked for a hundred. + +**gRPC has HTTP/2 flow control, and it did not help.** This is the result I did not expect. +`StreamObserver.onNext` on a gRPC server **does not block**, so a plain server-streaming loop keeps +producing regardless of what the client is doing. + +The documented answer is `ServerCallStreamObserver.isReady()`, and this module tests it rather than +citing it ([`grpc-isready.txt`](output/grpc-isready.txt)): + +``` +handler quiet 0.5s quiet 2s +plain onNext loop 258,145 1,421,778 +isReady() checked before write 667,259 1,044,018 +``` + +**It did not bound the producer.** The `isReady()` row is not lower than the plain one in any +useful sense — it produced 667 thousand messages in half a second for a client that +wanted a hundred — and across runs it has come out both far above and far below the plain +loop, which is its own kind of answer. I +have not established whether the surplus accumulates in the server's outbound queue, in the client +transport, or is discarded after the deframer, and I am not going to assert a mechanism I have not +read the source for. + +What the numbers do establish is narrower and still worth knowing: **`isReady()` is not a demand +signal**, and on a fast link with a stalled consumer it does not behave like one. If you are +streaming from gRPC to a consumer that can fall behind, the bound has to come from your own +design — a paged API, an explicit "send me more" message on a bidirectional stream, or a +`count` in the request as `StreamQuotes` uses. + +## The honest caveat + +A loopback connection is the worst case for this test in the fairest possible way: the wire drains +instantly, so nothing downstream ever pushes back. On a slow real link, TCP and the HTTP/2 window +*would* eventually stop the gRPC and WebSocket producers — by blocking a thread or by +filling a buffer, but they would stop. + +That does not rescue them, because the failure mode you actually meet is not "the network is +slow", it is "one consumer of forty is slow", and a shared server discovers that as latency for +everyone else or as a heap that will not shrink. The difference RSocket makes is that the slow +consumer's slowness is *expressed*, in a number, that the producer can act on before anything +fills up. + +## The other thing gRPC does not do for you + +Abandoning a gRPC stream produced this, repeatedly, in the server log: + +``` +java.lang.IllegalStateException: Failed to close the call +Caused by: java.lang.IllegalStateException: call already closed +``` + +That is the cancellation story from +[the Spring gRPC article](https://ankurm.com/spring-grpc-spring-boot-4/) showing up in practice: +gRPC does not interrupt your thread when a client leaves, it sets a flag on the `Context`, and a +handler that touches an already-closed stream throws. The loop in +[`GrpcQuoteService`](../src/main/java/com/ankurm/protocols/GrpcQuoteService.java) checks +`Context.current().isCancelled()` and returns without calling `onCompleted` for exactly this +reason. + +--- + +Previous: [3. Payload](03-payload.md) · Next: [5. Choosing](05-choosing.md) diff --git a/protocol-comparison/docs/05-choosing.md b/protocol-comparison/docs/05-choosing.md new file mode 100644 index 0000000..4f47cd5 --- /dev/null +++ b/protocol-comparison/docs/05-choosing.md @@ -0,0 +1,63 @@ +# 5. Choosing + +Previous: [4. Back-pressure](04-backpressure.md) + +--- + +## The decision table + +| If | Pick | Because | +|---|---|---| +| Service-to-service RPC, polyglot team, you want a schema | **gRPC** | The `.proto` is a contract other languages generate from. Deadlines are absolute and propagate. Boot 4 makes it a first-class starter. | +| A stream whose consumer can fall behind | **RSocket** | `request(n)` is the only demand signal that crosses the wire. 100 vs 1.3 million is not a tuning difference. | +| A browser is one of the peers | **WebSocket**, with STOMP on it | gRPC needs grpc-web and a proxy. RSocket over WebSocket works but the client ecosystem is thin. | +| Server pushes, client never speaks | **none of these** — use SSE | See [`../sse-websocket`](../../sse-websocket/README.md). | +| You want the lowest possible median latency and control both ends | **raw WebSocket** | It is fastest here because it does the least. Read the p99 before deciding that is what you want. | +| Long-lived bidirectional exchange with independent streams | **RSocket channel** or **gRPC bidi** | Both give you two independent streams; only RSocket gives each of them demand. | + +## What each one really costs + +**gRPC** costs a build step, a schema you must version, and an operational story for HTTP/2 through +your proxies. In exchange you get the best-supported cross-language RPC there is, and deadlines +— which is a bigger deal than it sounds, because an absolute deadline that propagates across +hops is the single most effective defence against one slow dependency taking down a system. + +**RSocket** costs ecosystem. It is the least deployed of the three, the client libraries outside +Java and JavaScript are thin, and if you are not already reactive, `Mono` and `Flux` in your +service signatures is a real commitment. In exchange you get the only demand signalling in the +list, plus resumption and leasing. + +**Raw WebSocket** costs everything you then invent: routing, correlation, error shape, versioning, +and the demand mechanism you will eventually need. That is the cost STOMP exists to prevent, and +if you find yourself designing a text protocol with a command word at the front — as +[`WebSocketQuoteHandler`](../src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java) +deliberately does — that is the signal to stop and use one. + +## Should you use any of them? + +Three cases where the answer is no: + +- **Request/response between two services that both speak HTTP already.** A REST endpoint with a + `RestClient` and a timeout is fine, and everything in your organisation already knows how to + observe, cache, proxy and debug it. Reach for gRPC when the schema or the deadline propagation + is the thing you need, not because it benchmarks faster. +- **Streaming that is really batching.** If the consumer processes a page at a time anyway, + paginated HTTP has back-pressure for free: the consumer asks for the next page when it is ready. + That is `request(n)` with a URL. +- **Events that outlive the connection.** None of these three persist anything. A dropped + connection loses the messages in flight. If that matters, the answer is a broker — + [`../kafka-basics`](../../kafka-basics/README.md), + [`../rabbitmq`](../../rabbitmq/README.md), and the comparison in + [`../broker-comparison`](../../broker-comparison/README.md). + +## And the benchmark caveat, one more time + +Every number in this module was measured with client and server in one JVM on one machine. That +deletes the network, which is where protobuf's 35 bytes would pay and where the tail latencies +would look completely different. The measurement that does transfer is +[chapter 4](04-backpressure.md), because "how many did the server produce for a consumer that +wanted a hundred" is a property of the protocol and not of the link. + +--- + +Previous: [4. Back-pressure](04-backpressure.md) diff --git a/protocol-comparison/docs/output/backpressure.txt b/protocol-comparison/docs/output/backpressure.txt new file mode 100644 index 0000000..bb63b03 --- /dev/null +++ b/protocol-comparison/docs/output/backpressure.txt @@ -0,0 +1,10 @@ +=== unbounded stream; client takes 100, then goes quiet === +protocol quiet 0.5s quiet 1s quiet 2s +RSocket 100 100 100 +gRPC 103,425 449,318 1,208,219 +WebSocket 19,664 143,870 320,255 +RSocket is flat: the server produced exactly what request(n) asked for. +gRPC grows 11.7x between 0.5 s and 2 s; WebSocket grows 16.3x. +Growth that tracks the wait, rather than settling at a buffer size, +means nothing bounded the producer. (Growth is faster than linear +because the loop is still being JIT-compiled during the first run.) diff --git a/protocol-comparison/docs/output/grpc-isready.txt b/protocol-comparison/docs/output/grpc-isready.txt new file mode 100644 index 0000000..d5d4dea --- /dev/null +++ b/protocol-comparison/docs/output/grpc-isready.txt @@ -0,0 +1,4 @@ +=== gRPC server streaming, client takes 100 then stops === +handler quiet 0.5s quiet 2s +plain onNext loop 258,145 1,421,778 +isReady() checked before write 667,259 1,044,018 diff --git a/protocol-comparison/docs/output/payload-sizes.txt b/protocol-comparison/docs/output/payload-sizes.txt new file mode 100644 index 0000000..8d9090c --- /dev/null +++ b/protocol-comparison/docs/output/payload-sizes.txt @@ -0,0 +1,5 @@ +=== one Quote, five fields, three encodings === +protobuf : 35 bytes 0a044141504c102a197b14ae47e11a5940215c8fc2f5281c5940288080abb4fef39203 +JSON : 83 bytes {"symbol":"AAPL","seq":42,"bid":100.42,"ask":100.44,"epochMicros":1772000000000000} +CBOR : 67 bytes bf6673796d626f6c644141504c63736571182a63626964fb40591ae147ae147b6361736bfb40591c28f5c28f5c6b65706f63684d6963726f731b00064b9fe68ac000ff +JSON is 2.37x protobuf; CBOR is 1.91x protobuf diff --git a/protocol-comparison/docs/output/request-response.txt b/protocol-comparison/docs/output/request-response.txt new file mode 100644 index 0000000..2c0a657 --- /dev/null +++ b/protocol-comparison/docs/output/request-response.txt @@ -0,0 +1,4 @@ +=== request/response, 5000 sequential calls after 2000 warm-up, loopback, JDK 25.0.4.1 === +gRPC n=5000 p50= 302.0 us p99= 994.7 us mean= 335.1 us ~2,984 calls/s +RSocket n=5000 p50= 265.9 us p99= 1166.2 us mean= 311.9 us ~3,206 calls/s +WebSocket n=5000 p50= 122.8 us p99= 1893.9 us mean= 184.8 us ~5,410 calls/s diff --git a/protocol-comparison/docs/output/rsocket-codecs.txt b/protocol-comparison/docs/output/rsocket-codecs.txt new file mode 100644 index 0000000..8a813a4 --- /dev/null +++ b/protocol-comparison/docs/output/rsocket-codecs.txt @@ -0,0 +1,4 @@ +=== Boot-configured RSocketStrategies === +encoders : [CharSequenceEncoder, ByteBufferEncoder, ByteArrayEncoder, DataBufferEncoder, JacksonCborEncoder, JacksonJsonEncoder] +decoders : [StringDecoder, ByteBufferDecoder, ByteArrayDecoder, DataBufferDecoder, JacksonCborDecoder, JacksonJsonDecoder] +bare RSocketStrategies.create() encoders : [CharSequenceEncoder, ByteBufferEncoder, ByteArrayEncoder, DataBufferEncoder] diff --git a/protocol-comparison/docs/output/stream-throughput.txt b/protocol-comparison/docs/output/stream-throughput.txt new file mode 100644 index 0000000..8481b1c --- /dev/null +++ b/protocol-comparison/docs/output/stream-throughput.txt @@ -0,0 +1,4 @@ +=== server streaming, 50000 messages on one connection, loopback === +gRPC 50,000 msgs in 1.341 s = 37,285 msgs/s ( 26.82 us each) +RSocket 50,000 msgs in 0.900 s = 55,580 msgs/s ( 17.99 us each) +WebSocket 50,000 msgs in 0.839 s = 59,630 msgs/s ( 16.77 us each) diff --git a/protocol-comparison/docs/output/tests.txt b/protocol-comparison/docs/output/tests.txt new file mode 100644 index 0000000..50ade27 --- /dev/null +++ b/protocol-comparison/docs/output/tests.txt @@ -0,0 +1,6 @@ +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.321 s -- in com.ankurm.protocols.PayloadSizeTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 11.76 s -- in com.ankurm.protocols.RequestResponseBenchmarkTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 11.29 s -- in com.ankurm.protocols.BackPressureTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 7.409 s -- in com.ankurm.protocols.GrpcIsReadyTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 6.474 s -- in com.ankurm.protocols.RSocketDefaultsTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 7.244 s -- in com.ankurm.protocols.StreamThroughputBenchmarkTest diff --git a/protocol-comparison/pom.xml b/protocol-comparison/pom.xml new file mode 100644 index 0000000..d4e73eb --- /dev/null +++ b/protocol-comparison/pom.xml @@ -0,0 +1,95 @@ + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 4.1.1 + + + + com.ankurm + protocol-comparison + 1.0 + jar + + + 25 + UTF-8 + + + + + + org.springframework.boot + spring-boot-starter-webmvc + + + org.springframework.boot + spring-boot-starter-websocket + + + + org.springframework.boot + spring-boot-starter-rsocket + + + + org.springframework.boot + spring-boot-starter-grpc-server + + + org.springframework.boot + spring-boot-starter-grpc-client + + + + org.springframework.boot + spring-boot-starter-jackson + + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + + + + + + io.github.ascopes + protobuf-maven-plugin + 3.9.0 + + ${protobuf-java.version} + + + io.grpc + protoc-gen-grpc-java + ${grpc-java.version} + + + + + generate + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/protocol-comparison/scripts/run-all.sh b/protocol-comparison/scripts/run-all.sh new file mode 100755 index 0000000..192d420 --- /dev/null +++ b/protocol-comparison/scripts/run-all.sh @@ -0,0 +1,40 @@ +#!/usr/bin/env bash +# Regenerate every file under docs/output/. Needs a JDK and nothing else: protoc and the gRPC +# codegen plugin resolve as Maven artifacts, and all three servers start inside the test JVM. +# +# The benchmark numbers are NOT deterministic. They are re-measured on every run and the ratios +# between the three protocols are what the documentation quotes. +set -eu +cd "$(dirname "$0")/.." +LOG=/tmp/protocol-comparison-test.log + +# Everything, for the test count and the non-timing transcripts. +MAVEN_OPTS=${MAVEN_OPTS:--Xmx900m} mvn -B test > "$LOG" 2>&1 + +# The two timing benchmarks are then re-run ALONE. This is not tidiness: the back-pressure and +# isReady tests leave unbounded producers spinning, and on a two-core box that halves every +# throughput number measured after them. Benchmarks share a JVM with nothing. +BENCH=/tmp/protocol-comparison-bench.log +MAVEN_OPTS=${MAVEN_OPTS:--Xmx900m} mvn -B test \ + -Dtest=RequestResponseBenchmarkTest,StreamThroughputBenchmarkTest > "$BENCH" 2>&1 + +# Interleaved log lines from other contexts have to come out, or the transcripts are unreadable. +clean() { grep -vE '^2026-|^\s*$|^WARNING|^[[:space:]]+at |^Caused by|^java\.|^org\.|Initializing|Completed initialization'; } + +sed -n '/=== one Quote, five fields/,/^JSON is /p' "$LOG" | clean > docs/output/payload-sizes.txt +sed -n '/=== request\/response/,/^WebSocket /p' "$BENCH" | clean > docs/output/request-response.txt +sed -n '/=== server streaming, 50000/,/^WebSocket /p' "$BENCH" | clean > docs/output/stream-throughput.txt +{ grep -m1 '=== unbounded stream' "$LOG"; grep -m1 '^protocol ' "$LOG" + grep -m1 '^RSocket *[0-9]' "$LOG"; grep -m1 '^gRPC *[0-9]' "$LOG" + grep -m1 '^WebSocket *[0-9]' "$LOG" + sed -n '/^RSocket is flat/,/JIT-compiled/p' "$LOG"; } > docs/output/backpressure.txt +{ grep -m1 '=== gRPC server streaming' "$LOG"; grep -m1 '^handler ' "$LOG" + grep -m1 '^plain onNext loop' "$LOG"; grep -m1 '^isReady() checked' "$LOG"; } \ + > docs/output/grpc-isready.txt + +sed -n '/=== Boot-configured RSocketStrategies/,/^bare RSocketStrategies/p' "$LOG" | clean \ + > docs/output/rsocket-codecs.txt + +grep -E 'Tests run:.*in com\.ankurm' "$LOG" | sed 's/^\[INFO\] //' > docs/output/tests.txt + +echo "regenerated:"; ls -1 docs/output/ diff --git a/protocol-comparison/src/main/java/com/ankurm/protocols/GrpcQuoteService.java b/protocol-comparison/src/main/java/com/ankurm/protocols/GrpcQuoteService.java new file mode 100644 index 0000000..7961487 --- /dev/null +++ b/protocol-comparison/src/main/java/com/ankurm/protocols/GrpcQuoteService.java @@ -0,0 +1,121 @@ +package com.ankurm.protocols; + +import com.ankurm.protocols.grpc.Quote; +import com.ankurm.protocols.grpc.QuoteRequest; +import com.ankurm.protocols.grpc.QuoteServiceGrpc; +import com.ankurm.protocols.grpc.StreamRequest; +import io.grpc.Context; +import io.grpc.stub.ServerCallStreamObserver; +import io.grpc.stub.StreamObserver; +import java.util.concurrent.atomic.AtomicLong; + +import org.springframework.stereotype.Service; + +/** + * The gRPC side. No registration code: the generated {@code ImplBase} is a + * {@code BindableService} and Boot registers every such bean with the server. + * + *

The cancellation check in {@link #streamQuotes} is not decoration. gRPC does not interrupt + * your thread when a client goes away — it sets a flag on the {@link Context} — so a + * loop that never looks keeps producing into a stream nobody is reading. That is covered at + * length in the Spring gRPC article; + * here it matters because the back-pressure benchmark deliberately abandons a stream. + * + * @see docs/04-backpressure.md + */ +@Service +public class GrpcQuoteService extends QuoteServiceGrpc.QuoteServiceImplBase { + + /** Counts what the SERVER produced, comparable with the RSocket and WebSocket counters. */ + static final AtomicLong PRODUCED = new AtomicLong(); + + private final QuoteSource source; + + GrpcQuoteService(QuoteSource source) { + this.source = source; + } + + @Override + public void getQuote(QuoteRequest request, StreamObserver observer) { + observer.onNext(toProto(source.at(request.getSymbol(), 0))); + observer.onCompleted(); + } + + /** + * An effectively unbounded stream, for the back-pressure comparison. A consumer that stops + * calling {@code next()} closes the HTTP/2 flow-control window, and this loop blocks inside + * {@code onNext} — so the server stops producing, but it stops by blocking a thread + * rather than by being asked to stop. + */ + @Override + public void streamUnbounded(QuoteRequest request, StreamObserver observer) { + PRODUCED.set(0); + while (!Context.current().isCancelled()) { + observer.onNext(toProto(source.at(request.getSymbol(), PRODUCED.get()))); + PRODUCED.incrementAndGet(); + } + } + + /** + * The fix, and the reason the unfixed version is worth measuring. + * + *

{@code StreamObserver.onNext} on a gRPC server never blocks. If the + * client is not reading, the message is queued in the server's outbound buffer and the loop + * keeps going — HTTP/2 flow control governs the wire, not your code. The only thing + * that connects the two is {@link ServerCallStreamObserver#isReady()}, and using it means + * restructuring the handler around {@code setOnReadyHandler} rather than writing a loop. + * + *

The spin here is deliberately the simplest possible demonstration rather than the + * shape you would ship; a real implementation registers an on-ready handler and returns. + */ + @Override + public void streamUnboundedReady(QuoteRequest request, StreamObserver observer) { + PRODUCED.set(0); + ServerCallStreamObserver ready = (ServerCallStreamObserver) observer; + while (!ready.isCancelled()) { + if (!ready.isReady()) { + Thread.onSpinWait(); + continue; + } + ready.onNext(toProto(source.at(request.getSymbol(), PRODUCED.get()))); + PRODUCED.incrementAndGet(); + } + } + + @Override + public void produced(QuoteRequest request, StreamObserver observer) { + observer.onNext(com.ankurm.protocols.grpc.ProducedCount.newBuilder() + .setCount(PRODUCED.get()).build()); + observer.onCompleted(); + } + + @Override + public void streamQuotes(StreamRequest request, StreamObserver observer) { + for (int i = 0; i < request.getCount(); i++) { + if (Context.current().isCancelled()) { + // Do NOT call onCompleted/onError here: the stream is already closed and + // touching it throws IllegalStateException. Just return. + return; + } + observer.onNext(toProto(source.at(request.getSymbol(), i))); + if (request.getDelayMs() > 0) { + try { + Thread.sleep(request.getDelayMs()); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + return; + } + } + } + observer.onCompleted(); + } + + static Quote toProto(com.ankurm.protocols.Quote q) { + return Quote.newBuilder() + .setSymbol(q.symbol()).setSeq(q.seq()) + .setBid(q.bid()).setAsk(q.ask()) + .setEpochMicros(q.epochMicros()) + .build(); + } +} diff --git a/protocol-comparison/src/main/java/com/ankurm/protocols/ProtocolComparisonApplication.java b/protocol-comparison/src/main/java/com/ankurm/protocols/ProtocolComparisonApplication.java new file mode 100644 index 0000000..68cd11a --- /dev/null +++ b/protocol-comparison/src/main/java/com/ankurm/protocols/ProtocolComparisonApplication.java @@ -0,0 +1,22 @@ +package com.ankurm.protocols; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * One application, three protocols, one set of data. + * + *

gRPC on 9090, RSocket on 7000, WebSocket on the servlet port. Serving the same two + * operations from the same JVM is the only way to make a benchmark mean anything: the heap, the + * JIT state, the CPU and the data generator are shared, so a difference in the numbers is a + * difference in the transports. + * + * @see docs/01-three-protocols.md + */ +@SpringBootApplication +public class ProtocolComparisonApplication { + + public static void main(String[] args) { + SpringApplication.run(ProtocolComparisonApplication.class, args); + } +} diff --git a/protocol-comparison/src/main/java/com/ankurm/protocols/Quote.java b/protocol-comparison/src/main/java/com/ankurm/protocols/Quote.java new file mode 100644 index 0000000..795fd59 --- /dev/null +++ b/protocol-comparison/src/main/java/com/ankurm/protocols/Quote.java @@ -0,0 +1,8 @@ +package com.ankurm.protocols; + +/** + * The JSON/CBOR shape, field-for-field identical to the protobuf {@code Quote} message, so the + * encoded sizes in {@code docs/output/payload-sizes.txt} compare like with like. + */ +public record Quote(String symbol, long seq, double bid, double ask, long epochMicros) { +} diff --git a/protocol-comparison/src/main/java/com/ankurm/protocols/QuoteSource.java b/protocol-comparison/src/main/java/com/ankurm/protocols/QuoteSource.java new file mode 100644 index 0000000..ad2be85 --- /dev/null +++ b/protocol-comparison/src/main/java/com/ankurm/protocols/QuoteSource.java @@ -0,0 +1,22 @@ +package com.ankurm.protocols; + +import java.time.Instant; + +import org.springframework.stereotype.Component; + +/** + * Deterministic quote generation, shared by all three protocol adapters. + * + *

Deliberately cheap and deliberately allocation-free apart from the record: a benchmark of + * three transports must not spend its time in the thing behind them. + */ +@Component +public class QuoteSource { + + public Quote at(String symbol, long seq) { + double base = 100.0 + (seq % 97) * 0.01; + Instant now = Instant.now(); + return new Quote(symbol, seq, base, base + 0.02, + now.getEpochSecond() * 1_000_000L + now.getNano() / 1_000L); + } +} diff --git a/protocol-comparison/src/main/java/com/ankurm/protocols/RSocketQuoteController.java b/protocol-comparison/src/main/java/com/ankurm/protocols/RSocketQuoteController.java new file mode 100644 index 0000000..19095f1 --- /dev/null +++ b/protocol-comparison/src/main/java/com/ankurm/protocols/RSocketQuoteController.java @@ -0,0 +1,72 @@ +package com.ankurm.protocols; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicLong; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.stereotype.Controller; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * The RSocket side. The same two operations, and one extra that exists only to make + * back-pressure visible. + * + *

Note what the signatures say that the other two protocols' cannot: {@code Mono} means + * request/response, {@code Flux} means request/stream, and the choice of interaction model is + * the return type rather than a separate declaration. RSocket has four (fire-and-forget, + * request/response, request/stream, channel) and Spring picks from the method shape. + * + * @see docs/04-backpressure.md + */ +@Controller +public class RSocketQuoteController { + + private static final Logger log = LoggerFactory.getLogger(RSocketQuoteController.class); + + /** Counts what the SERVER produced, which is the number the back-pressure test cares about. */ + static final AtomicLong PRODUCED = new AtomicLong(); + + private final QuoteSource source; + + RSocketQuoteController(QuoteSource source) { + this.source = source; + } + + @MessageMapping("quote") + public Mono quote(String symbol) { + return Mono.just(source.at(symbol, 0)); + } + + @MessageMapping("quotes") + public Flux quotes(StreamSpec spec) { + Flux flux = Flux.range(0, spec.count()).map(i -> source.at(spec.symbol(), i)); + return spec.delayMs() > 0 ? flux.delayElements(Duration.ofMillis(spec.delayMs())) : flux; + } + + /** + * An unbounded generator. It emits only when the transport asks, so the count it reaches is + * a direct read-out of how much the consumer requested — which is the whole argument + * for RSocket. + */ + @MessageMapping("quotes.unbounded") + public Flux unbounded(String symbol) { + PRODUCED.set(0); + return Flux.generate(() -> 0L, (seq, sink) -> { + sink.next(source.at(symbol, seq)); + PRODUCED.incrementAndGet(); + return seq + 1; + }); + } + + @MessageMapping("produced") + public Mono produced() { + return Mono.just(PRODUCED.get()); + } + + /** Request payload for {@code quotes}. */ + public record StreamSpec(String symbol, int count, int delayMs) { + } +} diff --git a/protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketConfig.java b/protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketConfig.java new file mode 100644 index 0000000..348594b --- /dev/null +++ b/protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketConfig.java @@ -0,0 +1,27 @@ +package com.ankurm.protocols; + +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.config.annotation.EnableWebSocket; +import org.springframework.web.socket.config.annotation.WebSocketConfigurer; +import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; + +/** + * {@code @EnableWebSocket}, not {@code @EnableWebSocketMessageBroker}: this module wants the raw + * transport, with no broker in the path, so the benchmark measures a socket rather than a routing + * layer. The {@code sse-websocket} module in this repository is the STOMP version. + */ +@Configuration +@EnableWebSocket +public class WebSocketConfig implements WebSocketConfigurer { + + private final WebSocketQuoteHandler handler; + + WebSocketConfig(WebSocketQuoteHandler handler) { + this.handler = handler; + } + + @Override + public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { + registry.addHandler(handler, "/quotes").setAllowedOriginPatterns("*"); + } +} diff --git a/protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java b/protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java new file mode 100644 index 0000000..e5ed2a4 --- /dev/null +++ b/protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java @@ -0,0 +1,94 @@ +package com.ankurm.protocols; + +import java.util.concurrent.atomic.AtomicLong; + +import org.springframework.stereotype.Component; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.TextWebSocketHandler; +import tools.jackson.databind.ObjectMapper; + +/** + * The WebSocket side, with no STOMP and no framework on top: a two-word text protocol, so the + * benchmark measures the transport and not a broker. + * + * 

+ *   QUOTE <symbol>                       -> one JSON quote
+ *   STREAM <symbol> <count> <delayMs>    -> count JSON quotes, then "END"
+ *   UNBOUNDED <symbol>                   -> quotes until the session dies
+ *
+ * + *

The absence worth noticing is in {@code UNBOUNDED}: there is nowhere for a consumer to say + * how many it wants. {@code sendMessage} either succeeds, blocks, or eventually throws when + * Spring's send buffer limit is exceeded. That is not a gap in this handler — it is what a + * raw WebSocket offers. + * + * @see docs/04-backpressure.md + */ +@Component +public class WebSocketQuoteHandler extends TextWebSocketHandler { + + /** Counts what the SERVER wrote, comparable with RSocket's PRODUCED. */ + static final AtomicLong PRODUCED = new AtomicLong(); + + private final QuoteSource source; + private final ObjectMapper mapper; + + WebSocketQuoteHandler(QuoteSource source, ObjectMapper mapper) { + this.source = source; + this.mapper = mapper; + } + + @Override + protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { + String[] parts = message.getPayload().trim().split("\\s+"); + switch (parts[0]) { + case "QUOTE" -> send(session, source.at(parts[1], 0)); + case "STREAM" -> { + int count = Integer.parseInt(parts[2]); + int delay = parts.length > 3 ? Integer.parseInt(parts[3]) : 0; + for (int i = 0; i < count; i++) { + send(session, source.at(parts[1], i)); + if (delay > 0) { + Thread.sleep(delay); + } + } + session.sendMessage(new TextMessage("END")); + } + case "UNBOUNDED" -> { + PRODUCED.set(0); + // No requestN. Nothing here can be told to slow down: the loop writes until the + // socket, the send buffer or the send-time limit stops it. Catching the write + // failure keeps an abandoned stream from filling the log with one stack trace + // per run -- and note that catching it is the ONLY notification that arrives. + try { + while (session.isOpen()) { + send(session, source.at(parts[1], PRODUCED.get())); + PRODUCED.incrementAndGet(); + } + } + catch (Exception ex) { + // Broken pipe. The client left; there was no other way to find out. + } + } + case "PRODUCED" -> session.sendMessage(new TextMessage(Long.toString(PRODUCED.get()))); + default -> session.sendMessage(new TextMessage("ERR unknown command")); + } + } + + @Override + public void handleTransportError(WebSocketSession session, Throwable exception) { + // Where an over-buffered slow consumer surfaces: as a transport error on the writer, + // long after the consumer stopped reading. + PRODUCED.addAndGet(0); + } + + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { + } + + private void send(WebSocketSession session, Quote quote) throws Exception { + session.sendMessage(new TextMessage(mapper.writeValueAsString(quote))); + } +} diff --git a/protocol-comparison/src/main/proto/quotes.proto b/protocol-comparison/src/main/proto/quotes.proto new file mode 100644 index 0000000..c2dbbe0 --- /dev/null +++ b/protocol-comparison/src/main/proto/quotes.proto @@ -0,0 +1,45 @@ +syntax = "proto3"; + +package quotes; + +option java_multiple_files = true; +option java_package = "com.ankurm.protocols.grpc"; + +// The same two operations the RSocket and WebSocket sides expose, so the comparison is between +// transports rather than between designs. +service QuoteService { + // Request/response. + rpc GetQuote (QuoteRequest) returns (Quote); + // Server streaming: count quotes, as fast as the transport allows. + rpc StreamQuotes (StreamRequest) returns (stream Quote); + // Never completes on its own. Used to show what stops a gRPC server producing. + rpc StreamUnbounded (QuoteRequest) returns (stream Quote); + // The same unbounded stream, but the server checks ServerCallStreamObserver.isReady() + // before every write. Same API, different memory profile. + rpc StreamUnboundedReady (QuoteRequest) returns (stream Quote); + // How many messages the server has produced on the current unbounded stream. + rpc Produced (QuoteRequest) returns (ProducedCount); +} + +message ProducedCount { + int64 count = 1; +} + +message QuoteRequest { + string symbol = 1; +} + +message StreamRequest { + string symbol = 1; + int32 count = 2; + // Milliseconds to sleep between emissions; 0 means "as fast as possible". + int32 delay_ms = 3; +} + +message Quote { + string symbol = 1; + int64 seq = 2; + double bid = 3; + double ask = 4; + int64 epoch_micros = 5; +} diff --git a/protocol-comparison/src/main/resources/application.yaml b/protocol-comparison/src/main/resources/application.yaml new file mode 100644 index 0000000..a56e0c8 --- /dev/null +++ b/protocol-comparison/src/main/resources/application.yaml @@ -0,0 +1,22 @@ +spring: + application: + name: protocol-comparison + rsocket: + server: + # RSocket gets its own port and its own Reactor Netty server. TCP here; "websocket" is the + # other transport and is what you use when the client is a browser. + port: 7000 + transport: tcp + grpc: + server: + # spring.grpc.server.port -- note the spring. prefix and that "port: -1" means in-process + # only. See the property table in ankurm.com/spring-grpc-spring-boot-4/ + port: 9090 + +server: + port: 8080 + +logging: + level: + io.grpc: WARN + io.rsocket: WARN diff --git a/protocol-comparison/src/test/java/com/ankurm/protocols/BackPressureTest.java b/protocol-comparison/src/test/java/com/ankurm/protocols/BackPressureTest.java new file mode 100644 index 0000000..4bbb449 --- /dev/null +++ b/protocol-comparison/src/test/java/com/ankurm/protocols/BackPressureTest.java @@ -0,0 +1,156 @@ +package com.ankurm.protocols; + +import java.util.Iterator; +import java.util.concurrent.atomic.AtomicLong; + +import com.ankurm.protocols.grpc.QuoteRequest; +import com.ankurm.protocols.grpc.QuoteServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscription; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.server.LocalServerPort; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import reactor.core.publisher.BaseSubscriber; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * The measurement that decides the article. + * + *

Each protocol serves an unbounded stream. Each client takes a hundred messages and then does + * nothing for a while. The question is: how many did the server produce? + * + *

The sweep over quiet periods is what makes the answer conclusive. A number that stays flat + * as the quiet period grows means something bounded the producer. A number that grows in + * proportion to the wait means nothing did — the producer is running flat out and + * the only limit is the clock. + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = {"spring.rsocket.server.port=7003", "spring.grpc.server.port=9093"}) +class BackPressureTest { + + private static final int WANTED = 100; + private static final long[] QUIET_MS = {500, 1_000, 2_000}; + + @LocalServerPort + int httpPort; + + @Autowired + RSocketStrategies strategies; + + @Test + void howManyDidTheServerProduce() throws Exception { + System.out.println("=== unbounded stream; client takes " + WANTED + ", then goes quiet ==="); + System.out.printf("%-14s %10s %10s %10s%n", "protocol", "quiet 0.5s", "quiet 1s", "quiet 2s"); + + long[] rs = new long[QUIET_MS.length]; + long[] gr = new long[QUIET_MS.length]; + long[] ws = new long[QUIET_MS.length]; + for (int i = 0; i < QUIET_MS.length; i++) { + rs[i] = rsocket(QUIET_MS[i]); + gr[i] = grpc(QUIET_MS[i]); + ws[i] = websocket(QUIET_MS[i]); + } + row("RSocket", rs); + row("gRPC", gr); + row("WebSocket", ws); + + System.out.println(); + System.out.println("RSocket is flat: the server produced exactly what request(n) asked for."); + System.out.printf("gRPC grows %.1fx between 0.5 s and 2 s; WebSocket grows %.1fx.%n", + gr[2] / (double) gr[0], ws[2] / (double) ws[0]); + System.out.println("Growth that tracks the wait, rather than settling at a buffer size,"); + System.out.println("means nothing bounded the producer. (Growth is faster than linear"); + System.out.println("because the loop is still being JIT-compiled during the first run.)"); + + // RSocket delivered exactly the demand, at every quiet period. + assertThat(rs).containsOnly(WANTED); + // The other two grew with the wait rather than settling at a buffer size. + assertThat(gr[2]).isGreaterThan(gr[0] * 2); + assertThat(ws[2]).isGreaterThan(ws[0] * 2); + } + + private void row(String label, long[] v) { + System.out.printf("%-14s %,10d %,10d %,10d%n", label, v[0], v[1], v[2]); + } + + /** + * RSocket: {@code request(n)} is a frame on the wire. The server is told the number, produces + * that many, and stops. Nothing blocks and nothing buffers — the producer is simply not + * called again. + */ + private long rsocket(long quietMs) throws Exception { + RSocketRequester requester = RSocketRequester.builder() + .rsocketStrategies(strategies).tcp("localhost", 7003); + try { + AtomicLong received = new AtomicLong(); + BaseSubscriber subscriber = new BaseSubscriber<>() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + subscription.request(WANTED); // and never again + } + + @Override + protected void hookOnNext(Quote value) { + received.incrementAndGet(); + } + }; + requester.route("quotes.unbounded").data("AAPL") + .retrieveFlux(Quote.class).subscribe(subscriber); + Thread.sleep(quietMs); + long produced = requester.route("produced").retrieveMono(Long.class).block(); + subscriber.dispose(); + assertThat(received.get()).isEqualTo(WANTED); + return produced; + } + finally { + requester.dispose(); + } + } + + /** + * gRPC: there is no application-level demand signal. The blocking stub's {@code Iterator} + * requests one message per {@code next()}, and the handler writes with + * {@code StreamObserver.onNext}, which never blocks. + */ + private long grpc(long quietMs) throws Exception { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9093) + .usePlaintext().build(); + try { + var stub = QuoteServiceGrpc.newBlockingStub(channel); + QuoteRequest req = QuoteRequest.newBuilder().setSymbol("AAPL").build(); + Iterator it = stub.streamUnbounded(req); + long received = 0; + while (received < WANTED && it.hasNext()) { + it.next(); + received++; + } + Thread.sleep(quietMs); + return stub.produced(req).getCount(); + } + finally { + channel.shutdownNow(); + } + } + + /** Raw WebSocket: the frame protocol has no notion of demand at all. */ + private long websocket(long quietMs) throws Exception { + try (WsClient client = new WsClient(httpPort, WANTED)) { + client.send("UNBOUNDED AAPL"); + long received = 0; + while (received < WANTED && client.take(5_000) != null) { + received++; + } + Thread.sleep(quietMs); + // Ask on a second connection: the first one's server-side thread is busy writing. + try (WsClient asker = new WsClient(httpPort, 8)) { + asker.send("PRODUCED"); + return Long.parseLong(asker.take(5_000)); + } + } + } +} diff --git a/protocol-comparison/src/test/java/com/ankurm/protocols/Bench.java b/protocol-comparison/src/test/java/com/ankurm/protocols/Bench.java new file mode 100644 index 0000000..d1eefd9 --- /dev/null +++ b/protocol-comparison/src/test/java/com/ankurm/protocols/Bench.java @@ -0,0 +1,29 @@ +package com.ankurm.protocols; + +import java.util.Arrays; + +/** Latency bookkeeping, kept in one place so the three protocols are measured identically. */ +final class Bench { + + private final long[] nanos; + private int i; + + Bench(int n) { + this.nanos = new long[n]; + } + + void record(long ns) { + nanos[i++] = ns; + } + + /** p50, p99 and mean in microseconds, plus calls per second derived from the mean. */ + String summary(String label) { + long[] sorted = Arrays.copyOf(nanos, i); + Arrays.sort(sorted); + double p50 = sorted[(int) (sorted.length * 0.50)] / 1000.0; + double p99 = sorted[(int) (sorted.length * 0.99)] / 1000.0; + double mean = Arrays.stream(sorted).average().orElse(0) / 1000.0; + return String.format("%-12s n=%-6d p50=%8.1f us p99=%9.1f us mean=%8.1f us ~%,.0f calls/s", + label, sorted.length, p50, p99, mean, 1_000_000.0 / mean); + } +} diff --git a/protocol-comparison/src/test/java/com/ankurm/protocols/GrpcIsReadyTest.java b/protocol-comparison/src/test/java/com/ankurm/protocols/GrpcIsReadyTest.java new file mode 100644 index 0000000..aba9a04 --- /dev/null +++ b/protocol-comparison/src/test/java/com/ankurm/protocols/GrpcIsReadyTest.java @@ -0,0 +1,66 @@ +package com.ankurm.protocols; + +import java.util.Iterator; + +import com.ankurm.protocols.grpc.QuoteRequest; +import com.ankurm.protocols.grpc.QuoteServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.SpringBootTest; + +/** + * The documented gRPC answer to over-production, tested rather than assumed. + * + *

Every guide to gRPC flow control says the same thing: {@code StreamObserver.onNext} does not + * block, so a server-streaming handler must consult + * {@code ServerCallStreamObserver.isReady()} before writing. This test runs the identical + * scenario against a handler that does exactly that. + * + *

It did not help. The transcript is committed as measured — on a loopback connection + * with a client that has stopped reading, the ready flag stays set and the loop keeps producing + * at the same rate. I have not established whether the messages accumulate in the server's + * outbound queue, in the client transport, or are discarded after the deframer, and I am not + * going to assert a mechanism I have not read the source for. What the numbers do establish is + * narrower and still useful: {@code isReady()} is not a substitute for a demand + * signal. + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = {"spring.rsocket.server.port=7004", "spring.grpc.server.port=9094"}) +class GrpcIsReadyTest { + + private static final int WANTED = 100; + + @Test + void isReadyDoesNotBoundTheProducerHere() throws Exception { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9094) + .usePlaintext().build(); + try { + var stub = QuoteServiceGrpc.newBlockingStub(channel); + QuoteRequest req = QuoteRequest.newBuilder().setSymbol("AAPL").build(); + + System.out.println("=== gRPC server streaming, client takes 100 then stops ==="); + System.out.printf("%-28s %10s %10s%n", "handler", "quiet 0.5s", "quiet 2s"); + System.out.printf("%-28s %,10d %,10d%n", "plain onNext loop", + run(stub, req, false, 500), run(stub, req, false, 2_000)); + System.out.printf("%-28s %,10d %,10d%n", "isReady() checked before write", + run(stub, req, true, 500), run(stub, req, true, 2_000)); + } + finally { + channel.shutdownNow(); + } + } + + private long run(QuoteServiceGrpc.QuoteServiceBlockingStub stub, QuoteRequest req, + boolean ready, long quietMs) throws Exception { + Iterator it = + ready ? stub.streamUnboundedReady(req) : stub.streamUnbounded(req); + long received = 0; + while (received < WANTED && it.hasNext()) { + it.next(); + received++; + } + Thread.sleep(quietMs); + return stub.produced(req).getCount(); + } +} diff --git a/protocol-comparison/src/test/java/com/ankurm/protocols/PayloadSizeTest.java b/protocol-comparison/src/test/java/com/ankurm/protocols/PayloadSizeTest.java new file mode 100644 index 0000000..c783027 --- /dev/null +++ b/protocol-comparison/src/test/java/com/ankurm/protocols/PayloadSizeTest.java @@ -0,0 +1,46 @@ +package com.ankurm.protocols; + +import java.util.HexFormat; + +import com.ankurm.protocols.grpc.Quote; +import org.junit.jupiter.api.Test; +import tools.jackson.databind.ObjectMapper; +import tools.jackson.dataformat.cbor.CBORMapper; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * The same five fields, encoded three ways. + * + *

This is the one comparison that needs no server, no warm-up and no statistics: it is a + * property of the wire formats. It is also the honest place to start, because on a loopback + * benchmark the encoding is most of what separates the three protocols. + */ +class PayloadSizeTest { + + private static final com.ankurm.protocols.Quote POJO = + new com.ankurm.protocols.Quote("AAPL", 42, 100.42, 100.44, 1_772_000_000_000_000L); + + @Test + void protobufVersusJsonVersusCbor() { + byte[] proto = Quote.newBuilder() + .setSymbol("AAPL").setSeq(42) + .setBid(100.42).setAsk(100.44) + .setEpochMicros(1_772_000_000_000_000L) + .build().toByteArray(); + + byte[] json = new ObjectMapper().writeValueAsBytes(POJO); + byte[] cbor = new CBORMapper().writeValueAsBytes(POJO); + + System.out.println("=== one Quote, five fields, three encodings ==="); + System.out.printf("protobuf : %3d bytes %s%n", proto.length, HexFormat.of().formatHex(proto)); + System.out.printf("JSON : %3d bytes %s%n", json.length, new String(json)); + System.out.printf("CBOR : %3d bytes %s%n", cbor.length, HexFormat.of().formatHex(cbor)); + System.out.printf("JSON is %.2fx protobuf; CBOR is %.2fx protobuf%n", + json.length / (double) proto.length, cbor.length / (double) proto.length); + + // Protobuf is smallest because field names are numbers and there is no framing text. + assertThat(proto.length).isLessThan(cbor.length); + assertThat(cbor.length).isLessThan(json.length); + } +} diff --git a/protocol-comparison/src/test/java/com/ankurm/protocols/RSocketDefaultsTest.java b/protocol-comparison/src/test/java/com/ankurm/protocols/RSocketDefaultsTest.java new file mode 100644 index 0000000..a4dc429 --- /dev/null +++ b/protocol-comparison/src/test/java/com/ankurm/protocols/RSocketDefaultsTest.java @@ -0,0 +1,38 @@ +package com.ankurm.protocols; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.messaging.rsocket.RSocketStrategies; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * What Boot actually configures for RSocket, printed rather than quoted. + * + *

A bare {@code RSocketStrategies.create()} carries no JSON and no CBOR at all — only + * the string and buffer codecs. Everything interesting comes from Boot's + * {@code RSocketStrategiesAutoConfiguration}, which is also where the default data mime type is + * decided, and it is not the one most people assume. + */ +@SpringBootTest(properties = {"spring.rsocket.server.port=7005", "spring.grpc.server.port=9095"}) +class RSocketDefaultsTest { + + @Autowired + RSocketStrategies strategies; + + @Test + void bootConfiguredCodecs() { + var encoders = strategies.encoders().stream().map(e -> e.getClass().getSimpleName()).toList(); + var decoders = strategies.decoders().stream().map(d -> d.getClass().getSimpleName()).toList(); + + System.out.println("=== Boot-configured RSocketStrategies ==="); + System.out.println("encoders : " + encoders); + System.out.println("decoders : " + decoders); + System.out.println("bare RSocketStrategies.create() encoders : " + + RSocketStrategies.create().encoders().stream() + .map(e -> e.getClass().getSimpleName()).toList()); + + assertThat(encoders).isNotEmpty(); + } +} diff --git a/protocol-comparison/src/test/java/com/ankurm/protocols/RequestResponseBenchmarkTest.java b/protocol-comparison/src/test/java/com/ankurm/protocols/RequestResponseBenchmarkTest.java new file mode 100644 index 0000000..5c7d119 --- /dev/null +++ b/protocol-comparison/src/test/java/com/ankurm/protocols/RequestResponseBenchmarkTest.java @@ -0,0 +1,111 @@ +package com.ankurm.protocols; + +import com.ankurm.protocols.grpc.QuoteRequest; +import com.ankurm.protocols.grpc.QuoteServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.server.LocalServerPort; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; + +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Sequential request/response round trips, all three protocols, same JVM, same data source. + * + *

Read the numbers as a comparison of the three, not as absolutes. Client and server are the + * same process on one loopback interface with two cores, so the network — the thing that + * dominates every real deployment — is absent. That deletes the advantage a smaller + * encoding would have on a real link and leaves framing and dispatch cost, which is exactly the + * part a loopback measures well. + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = {"spring.rsocket.server.port=7001", "spring.grpc.server.port=9091"}) +class RequestResponseBenchmarkTest { + + private static final int WARMUP = 2_000; + private static final int MEASURED = 5_000; + + @LocalServerPort + int httpPort; + + @Autowired + RSocketStrategies strategies; + + @Test + void threeProtocolsOneOperation() throws Exception { + System.out.println("=== request/response, " + MEASURED + " sequential calls after " + + WARMUP + " warm-up, loopback, JDK " + System.getProperty("java.version") + " ==="); + System.out.println(grpc()); + System.out.println(rsocket()); + System.out.println(websocket()); + } + + private String grpc() { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9091) + .usePlaintext().build(); + try { + var stub = QuoteServiceGrpc.newBlockingStub(channel); + QuoteRequest req = QuoteRequest.newBuilder().setSymbol("AAPL").build(); + for (int i = 0; i < WARMUP; i++) { + stub.getQuote(req); + } + Bench bench = new Bench(MEASURED); + for (int i = 0; i < MEASURED; i++) { + long t0 = System.nanoTime(); + var quote = stub.getQuote(req); + bench.record(System.nanoTime() - t0); + assertThat(quote.getSymbol()).isEqualTo("AAPL"); + } + return bench.summary("gRPC"); + } + finally { + channel.shutdownNow(); + } + } + + private String rsocket() { + RSocketRequester requester = RSocketRequester.builder() + .rsocketStrategies(strategies).tcp("localhost", 7001); + try { + for (int i = 0; i < WARMUP; i++) { + requester.route("quote").data("AAPL").retrieveMono(Quote.class).block(); + } + Bench bench = new Bench(MEASURED); + for (int i = 0; i < MEASURED; i++) { + long t0 = System.nanoTime(); + Quote quote = requester.route("quote").data("AAPL") + .retrieveMono(Quote.class).block(); + bench.record(System.nanoTime() - t0); + assertThat(quote).isNotNull(); + } + return bench.summary("RSocket"); + } + finally { + requester.dispose(); + } + } + + private String websocket() throws Exception { + try (WsClient client = new WsClient(httpPort, 64)) { + for (int i = 0; i < WARMUP; i++) { + client.send("QUOTE AAPL"); + assertThat(client.take(5_000)).isNotNull(); + } + Bench bench = new Bench(MEASURED); + for (int i = 0; i < MEASURED; i++) { + long t0 = System.nanoTime(); + client.send("QUOTE AAPL"); + String reply = client.take(5_000); + bench.record(System.nanoTime() - t0); + assertThat(reply).contains("AAPL"); + } + return bench.summary("WebSocket"); + } + } +} diff --git a/protocol-comparison/src/test/java/com/ankurm/protocols/StreamThroughputBenchmarkTest.java b/protocol-comparison/src/test/java/com/ankurm/protocols/StreamThroughputBenchmarkTest.java new file mode 100644 index 0000000..798b62e --- /dev/null +++ b/protocol-comparison/src/test/java/com/ankurm/protocols/StreamThroughputBenchmarkTest.java @@ -0,0 +1,120 @@ +package com.ankurm.protocols; + +import java.util.Iterator; + +import com.ankurm.protocols.grpc.StreamRequest; +import com.ankurm.protocols.grpc.QuoteServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.server.LocalServerPort; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Server streaming: how fast can one connection carry N messages one way? + * + *

This is the shape most of these protocols are actually chosen for — a price feed, a + * log tail, a progress stream — and it separates them differently from request/response, + * because the per-message cost stops being dominated by a round trip. + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = {"spring.rsocket.server.port=7002", "spring.grpc.server.port=9092"}) +class StreamThroughputBenchmarkTest { + + private static final int WARMUP = 5_000; + private static final int N = 50_000; + + @LocalServerPort + int httpPort; + + @Autowired + RSocketStrategies strategies; + + @Test + void fiftyThousandMessagesEachWay() throws Exception { + System.out.println("=== server streaming, " + N + " messages on one connection, loopback ==="); + System.out.println(grpc()); + System.out.println(rsocket()); + System.out.println(websocket()); + } + + private String grpc() { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9092) + .usePlaintext().build(); + try { + var stub = QuoteServiceGrpc.newBlockingStub(channel); + drain(stub, WARMUP); + long t0 = System.nanoTime(); + int seen = drain(stub, N); + return report("gRPC", seen, System.nanoTime() - t0); + } + finally { + channel.shutdownNow(); + } + } + + private int drain(QuoteServiceGrpc.QuoteServiceBlockingStub stub, int count) { + // The blocking stub returns an Iterator. Errors surface mid-iteration rather than at the + // call, which is the first thing that surprises people about gRPC server streaming. + Iterator it = stub.streamQuotes( + StreamRequest.newBuilder().setSymbol("AAPL").setCount(count).build()); + int seen = 0; + while (it.hasNext()) { + it.next(); + seen++; + } + return seen; + } + + private String rsocket() { + RSocketRequester requester = RSocketRequester.builder() + .rsocketStrategies(strategies).tcp("localhost", 7002); + try { + var warm = new RSocketQuoteController.StreamSpec("AAPL", WARMUP, 0); + requester.route("quotes").data(warm).retrieveFlux(Quote.class).blockLast(); + + var spec = new RSocketQuoteController.StreamSpec("AAPL", N, 0); + long t0 = System.nanoTime(); + long seen = requester.route("quotes").data(spec) + .retrieveFlux(Quote.class).count().block(); + return report("RSocket", (int) seen, System.nanoTime() - t0); + } + finally { + requester.dispose(); + } + } + + private String websocket() throws Exception { + // A generous inbox, because there is nothing else to do with messages that arrive faster + // than they are read. This IS the WebSocket back-pressure story, in one constant. + try (WsClient client = new WsClient(httpPort, N + WARMUP + 16)) { + client.send("STREAM AAPL " + WARMUP + " 0"); + int warm = 0; + String m; + while ((m = client.take(20_000)) != null && !"END".equals(m)) { + warm++; + } + assertThat(warm).isEqualTo(WARMUP); + + long t0 = System.nanoTime(); + client.send("STREAM AAPL " + N + " 0"); + int seen = 0; + while ((m = client.take(20_000)) != null && !"END".equals(m)) { + seen++; + } + return report("WebSocket", seen, System.nanoTime() - t0); + } + } + + private String report(String label, int seen, long elapsedNanos) { + assertThat(seen).isEqualTo(N); + double seconds = elapsedNanos / 1e9; + return String.format("%-12s %,7d msgs in %6.3f s = %,10.0f msgs/s (%6.2f us each)", + label, seen, seconds, seen / seconds, elapsedNanos / 1000.0 / seen); + } +} diff --git a/protocol-comparison/src/test/java/com/ankurm/protocols/WsClient.java b/protocol-comparison/src/test/java/com/ankurm/protocols/WsClient.java new file mode 100644 index 0000000..bf53021 --- /dev/null +++ b/protocol-comparison/src/test/java/com/ankurm/protocols/WsClient.java @@ -0,0 +1,60 @@ +package com.ankurm.protocols; + +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import jakarta.websocket.ContainerProvider; +import jakarta.websocket.WebSocketContainer; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.client.standard.StandardWebSocketClient; +import org.springframework.web.socket.handler.TextWebSocketHandler; + +/** + * A minimal raw-WebSocket client for the benchmark. + * + *

The queue is the point of interest. A WebSocket client has no way to tell the server how + * many messages it is ready for, so the only thing it can do with an over-eager producer is + * buffer them — here, in a bounded queue that starts dropping. That is the shape of every + * real WebSocket consumer, and it is what {@code BackPressureTest} contrasts with RSocket. + */ +final class WsClient implements AutoCloseable { + + private final WebSocketSession session; + private final BlockingQueue inbox; + + WsClient(int port, int inboxCapacity) throws Exception { + this.inbox = new ArrayBlockingQueue<>(inboxCapacity); + WebSocketContainer container = ContainerProvider.getWebSocketContainer(); + container.setDefaultMaxTextMessageBufferSize(1024 * 1024); + this.session = new StandardWebSocketClient(container) + .execute(new TextWebSocketHandler() { + @Override + protected void handleTextMessage(WebSocketSession s, TextMessage message) { + inbox.offer(message.getPayload()); // drops when full; nothing else to do + } + }, "ws://localhost:" + port + "/quotes").get(); + } + + void send(String command) throws Exception { + session.sendMessage(new TextMessage(command)); + } + + String take(long timeoutMs) throws Exception { + return inbox.poll(timeoutMs, TimeUnit.MILLISECONDS); + } + + int drained() { + return inbox.size(); + } + + boolean isOpen() { + return session.isOpen(); + } + + @Override + public void close() throws Exception { + session.close(); + } +}