lizining1231 opened a new pull request, #3746:
URL: https://github.com/apache/dubbo-go/pull/3746
### Description
Fixes #3741
dubbo-go's streaming write path pays a synchronous `io.Pipe` hand-off on
every `Send`. Under the default gRPC wire encoding one message is split into
two `Write` calls (5-byte prefix + body), so a burst of 128 B messages costs
two hand-offs per message; and because `x/net/http2` flushes every DATA frame
immediately, each hand-off corresponds to a real `write(2)` syscall.
For small-payload, high-rate streaming workloads (log upload, batch
collection) this per-message constant cost dominates throughput: strace
measured about 2.1 `write(2)` calls per message, and end-to-end benchmarking
showed that raising concurrency from 32 to 128 decreased QPS rather than
increasing it, indicating the throughput ceiling is set by the number of
hand-offs, not by the amount of computation.
This change inserts an optional 32 KiB coalescing buffer,
`streamBufferWriter`, between the envelope writer and `duplexHTTPCall`. Small
messages accumulate in the buffer and are pushed down in one shot when the
watermark is reached or when `CloseRequest` seals the stream. Filling the 32
KiB watermark takes 256 messages of 128 B, so the number of syscalls on the
gRPC wire can drop by up to 256x. The `WithWriteBuffering()` switch defaults to
off, and with it off the write path is byte-for-byte identical to today's.
### Changes
`protocol/triple/triple_protocol/buffered_writer.go` (new):
- `streamBufferWriter` wraps a 32 KiB coalescing buffer on top of an
`io.Writer` (that is, `duplexHTTPCall`); the watermark matches the default
gRPC-Go write buffer.
- `Write` accumulates small messages; a payload whose length reaches or
exceeds the watermark first flushes the pending bytes and then writes straight
through, so a large message is not staged in memory twice.
- `Flush` writes the pending bytes in a single call; flushing an empty
buffer is a no-op.
- `Close` completes "flush + seal" atomically under one lock: a `Send`
racing with `CloseRequest` either lands in the final flush or returns `io.EOF`,
which is also what `duplexHTTPCall` returns after `CloseWrite` when buffering
is off.
- The first error freezes the buffer: every later call returns the same
sticky error.
`protocol/triple/triple_protocol/option.go`:
- Adds the `WithWriteBuffering()` ClientOption. Off by default, it trades a
small amount of first-packet latency for higher throughput on small-message
streaming. It covers every call that goes through `duplexHTTPCall` (both the
gRPC and the Triple wire, and both streaming and non-fast-path unary); the
unary fast path does not touch `io.Pipe` and is unaffected.
`protocol/triple/triple_protocol/protocol.go`,
`protocol/triple/triple_protocol/client.go`:
- Threads `WriteBuffering bool` through `protocolClientParams` and
`clientConfig` so the option reaches both protocol clients.
`protocol/triple/triple_protocol/protocol_grpc.go`,
`protocol/triple/triple_protocol/protocol_triple.go`:
- In `NewConn`, when `WriteBuffering` is on, wraps `duplexHTTPCall` with
`newStreamBufferWriter` and routes the envelope writer to the buffer. The
envelope writer and the conn-level `writeBuffer` field derive from the same
`streamWriter` variable, so they are always paired and cannot be mismatched.
- `CloseRequest` now flushes and seals the buffer first, then calls
`CloseWrite`. `CloseWrite` runs even when the flush fails, so the peer never
waits on a request body that never closes. A flush error takes precedence in
the returned error.
`tools/benchmark/server/dubbo-go/main.go`:
- Fixes the stream termination check: `err == io.EOF` becomes
`errors.Is(err, io.EOF)`. The termination error wraps `io.EOF` rather than
being equal to that sentinel.
`tools/benchmark/stream_ab/main.go` (new):
- A standalone streaming AB benchmark client. It depends only on
`triple_protocol.NewClient` and connects straight to a real Triple server over
h2c. Each worker opens one stream, sends N messages, closes the request side,
then drains every response. The two arms differ only in whether
`WithWriteBuffering()` is passed.
`tools/benchmark/client/clients/proto_fastpath_client.go` (new):
- A unary fast-path prototype client for baseline comparison: bare HTTP/2,
no `duplexHTTPCall`, no `io.Pipe`, with pooled request and response buffers
using `MarshalAppend`.
### Tests
| Test | Purpose |
|---|---|
| `TestStreamBufferWriterCoalesces` | Three small messages are coalesced
into one underlying Write |
| `TestStreamBufferWriterFlushesAtLimit` | The buffer flushes itself at the
32 KiB watermark, carrying the whole batch in exactly one flush |
| `TestStreamBufferWriterLargeMessageWritesDirectly` | A payload reaching or
exceeding the watermark bypasses the buffer and writes straight through |
| `TestStreamBufferWriterLargeMessageFlushesPendingFirst` | A direct write
does not jump ahead of already-buffered small messages; ordering holds |
| `TestStreamBufferWriterFlushEmptyIsNoOp` | Flushing an empty buffer
produces no underlying Write |
| `TestStreamBufferWriterWriteAfterError` | After a failed flush, later
Writes return the same sticky error |
| `TestStreamBufferWriterWriteReportsFlushFailure` | The Send that trips the
watermark receives the flush failure as `(len(p), err)` |
| `TestStreamBufferWriterLargeMessageErrorPropagates` | The direct-write
path reports the pending batch's flush failure instead of masking it with the
tail message |
| `TestStreamBufferWriterShortWriteIsFailure` | A downstream short write (n
< len(p) with err == nil) is treated as a failure and freezes the buffer |
| `TestStreamBufferWriterStickyErrorOnEveryExit` | After a failure, Write,
Flush and Close all return the same error |
| `TestStreamBufferWriterCloseFlushFailureReturnsError` | Close reports a
failed final flush instead of returning nil |
| `TestStreamBufferWriterCloseAfterCloseFailure` | A failed Close stays
sticky when called again |
| `TestStreamBufferWriterFlushAfterClose` | After a successful Close, Flush
returns nil and does not write again |
| `TestStreamBufferWriterWriteAfterClose` | A Send after Close returns
io.EOF, the tail message has been delivered, and Close is idempotent |
| `TestStreamBufferWriterCloseNeverDropsRacingWrite` | A Write racing with
Close either lands in the final flush or returns io.EOF; it is never accepted
and then dropped (200 rounds) |
| `TestStreamBufferWriterConcurrent` | 8 goroutines × 100 Sends plus
concurrent Flush: no data race, no lost data |
| `TestStreamBufferWriterConcurrentCloseAndFlush` | Close concurrent with
Send and Flush: no duplicate writes, Close stays idempotent, byte counts match |
| `TestStreamBufferWriterConcurrentBufferBound` | Under concurrent Sends no
batch exceeds the watermark plus one message |
| `TestWriteBufferingStreamingFlushOnClose` | End to end: 1000 messages of
100 B over a real duplexHTTPCall; the server receives every byte, including the
tail message that was never flushed |
| `TestWriteBufferingIsWiredIntoStreamingConns` | WithWriteBuffering reaches
both the gRPC and the Triple streaming conns, the envelope writer is routed to
the buffer, and a fast-path conn stays unbuffered |
| `TestWriteBufferingCoversUnaryNonFastPath` | A unary call that does not
take the fast path is buffered as well |
| `TestWriteBufferingEmptyStreamStillSendsRequest` | An empty stream that
sends no message at all still gets its request out at CloseRequest |
| `TestCloseRequestClosesWriteSideAfterFlushFailure` | Even when the final
flush fails, CloseRequest still closes the write side |
| `TestWritePathAggregationProbe` | io.Pipe segment-count probe: one segment
per message on the baseline, collapsing to a single segment once coalesced |
| `TestEnvelopeWritePathProbe` | Pins down the write-call structure: 2
writes on the gRPC wire (prefix + body), 1 on the Triple wire, a single
Content-Length body on the fast path |
| `BenchmarkStreamWritePerMessage` / `Buffered` | L1 micro-bench: os.Pipe
with a real fd, a 128 B to 4 KiB ladder, AB switching only the write strategy |
### Validation
**benchstat -count10 (L1 micro-benchmark, not end-to-end)**
`go test -run '^$' -bench BenchmarkStreamWrite -benchmem -count=10`, with
benchstat pairing PerMessage (before) against Buffered (after) per payload
size. This is a micro-benchmark writing to an os.Pipe with a real fd, switching
only the write strategy: it isolates the write path and does not include
serialization, HTTP/2 framing, the socket or the server.
| Payload | before sec/op | after sec/op | vs base |
|---|---|---|---|
| 128 B | 633.95n ±5% | 54.48n ±2% | **−91.41%** (p=0.000 n=10) |
| 256 B | 652.95n ±6% | 87.90n ±5% | −86.54% (p=0.000 n=10) |
| 512 B | 631.2n ±5% | 158.2n ±2% | −74.94% (p=0.000 n=10) |
| 1 KiB | 668.1n ±4% | 296.4n ±7% | −55.64% (p=0.000 n=10) |
| 4 KiB | 1.278µ ±4% | 1.102µ ±4% | −13.74% (p=0.000 n=10) |
| geomean | 740.7n | 190.0n | **−74.35%** |
**strace syscall count**
| per message | before | after | delta |
|---|---|---|---|
| DATA frame writes per message | **1.990** | **0.0078** | **−99.61%** |
| total write(2) per message | 2.105 | 0.105 | −95.0% |
```bash
strace -ff -ttt -e trace=write,writev -o st_before \
./client -addr 127.0.0.1:20000 -payload 128 -msgs 128 -concurrency 4
-warmup 20s -duration 60s
strace -ff -ttt -e trace=write,writev -o st_after \
./client -addr 127.0.0.1:20000 -buffering -payload 128 -msgs 128
-concurrency 4 -warmup 20s -duration 60s
```
128 B × 512 messages / concurrency 4 (512 messages ≈ 68 KiB, enough to trip
the 32 KiB watermark mid-stream): **0.00585** DATA-frame writes per message, a
**−99.71%** drop.
**End-to-end AB on the bottleneck scenario**
128 B × 128 messages / concurrency 32, A and B interleaved for 3 rounds each
(server restarted every round), medians reported:
| Metric | buffering off | buffering on | Δ |
|---|---|---|---|
| Stream QPS (3 rounds) | 189.25 / 202.32 / 195.55 | 243.28 / 241.98 /
247.18 | — |
| Stream QPS median | 195.55 | **243.28** | **+24.4%** |
| P50 median | 161.37 ms | 128.50 ms | −20.4% |
| P99 median | 221.60 ms | 174.42 ms | **−21.3%** |
128 B × 512 messages / concurrency 32 (crossing the watermark), same 3
rounds:
| Metric | buffering off | buffering on | Δ |
|---|---|---|---|
| Stream QPS median | 45.24 | **63.95** | **+41.4%** |
| P50 median | 706.50 ms | 496.70 ms | −29.7% |
| P99 median | 979.87 ms | 630.20 ms | **−35.7%** |
**pprof CPU diff**
<img width="1907" height="890" alt="image"
src="https://github.com/user-attachments/assets/ed5fa98e-3494-491c-9305-4ce9bbdfecf2";
/>
### Checklist
- [x] I confirm the target branch is `develop`
- [x] I have run `make fmt` to format my code
- [x] I have run `make test` to run local tests
- [x] I have added tests that prove my fix is effective or that my feature
works
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
