Avoid copying every push batch into a fresh byte[]

[1linkovdim]

What

The push write path buffers events for 200ms per connection, then concatenated the whole batch into a freshly allocated byte[totalBytes] before handing it to the channel. This wraps the event arrays in a composite ByteBuf instead, so nothing is copied to make the batch contiguous.

Why

The concatenation exists only to produce one contiguous array. The channel writes it with a gathering write regardless, so the contiguity buys nothing downstream — the copy is pure overhead. Building the batch also took two passes over bufferOfBuffers per connection (one to sum lengths, one to fill).

Replacing the copy with a composite trades an O(total bytes) memcpy for O(component count) bookkeeping. That is a win exactly when events are large enough that the memcpy dominates the per-component cost — see Benchmarks for where that crossover sits and why the payloads this path carries clear it comfortably. Below the crossover it is a regression, and the PR is honest about that boundary rather than hiding it.

How

ChannelWriter already exposes writeBytesAndFlush(ByteBuf), and DefaultChannelWriter performs no copy — both the byte[] and ByteBuf overloads just call writeOnChannel(msg). So PushServer now builds the batch as a composite:

batch shape	before	after
non-SSE	byte[totalBytes], all events copied in	wrapBare — one component per event
SSE, uncompressed	byte[totalBytes], prefix+event+suffix copied in	wrapDelimited — three components per event
SSE, compressed	byte[totalBytes] around the compressed blob	wrap(prefix, compressed, suffix)

Wrapping is safe: each byte[] comes from encoder.call(chunk), freshly allocated per chunk and never reused or retained elsewhere.

LengthFieldPrepender and the legacy TCP handler preserve the property. The prepender adds a separate header buffer and retains the message rather than copying it, so the zero-copy batch survives to the socket. (JdkZlibEncoder necessarily copies when compression is enabled — unchanged behaviour, and it makes the wrap a no-op for compressed SSE, which is expected.)

The legacy TCP handler gains more than the wrap

LegacyTcpPipelineConfigurator's outbound handler has two branches. The byte[] one is commented "handle heart beat writes" — but until this change all legacy TCP data took it, because PushServer emitted a byte[]. That branch allocated an unsized ctx.alloc().buffer() — Netty's 256-byte default — and grew it to the full batch size: roughly eleven doubling reallocations and two passes over the payload per batch, on top of PushServer's own concatenation.

Data now takes the ByteBuf branch, which prepends the header by wrapping rather than copying, and the comment is accurate for the first time. The heartbeat branch is now pre-sized too.

The explicit bytes.release() was dropped from that branch deliberately: Unpooled.wrappedBuffer(ByteBuf...) takes ownership of its components' refcounts, so releasing the composite releases the payload.

Two traps worth flagging for review

1. Unpooled.wrappedBuffer(byte[]...) defaults to maxNumComponents = 16 and silently consolidates — copies everything into a single array — once the composite exceeds it. A 30-event batch is 30 components bare and ~90 delimited, so using the varargs form without an explicit count would have made this entire change a no-op with no visible symptom. Every call site now passes the count explicitly, and normalBatchIsNotCopiedIntoOneBuffer asserts nioBufferCount() is 64 / 192 for a 64-event batch so a regression here fails loudly.

2. Unpooled.wrappedBuffer returns EMPTY_BUFFER if any array element is null. A pre-sized new byte[events][] would leave trailing nulls if a batch shrank between the counting pass and the fill pass — silent data loss. The helpers build a List<byte[]> and call toArray(new byte[0][]), which makes the event count a sizing hint only.

MAX_WRAPPED_COMPONENTS = 1024

A gathering write is capped at 1024 iovecs by the JDK, so past that point a composite stops buying anything — the channel makes several write syscalls regardless, and the per-component bookkeeping costs more than the copy it was avoiding. Batches that large fall back to copyBare/copyDelimited, which produce byte-identical output. Typical batches are tens of events per 200ms window, so this is a safety valve rather than a live path. The benchmark confirms the fallback rows are within noise of legacy — it never makes a large batch materially worse.

Retry semantics are unchanged

writeBytesAndFlush performs the write eagerly, as a side effect at call time, and returns only the flush() Observable. So .retry(writeRetryCount) retries the flush, never the write — identically for byte[] and ByteBuf. Worth noting separately that this makes the retry largely vacuous today; that's pre-existing and not touched here.

Testing

./gradlew :mantis-network:test — green (JDK 17).

PushServerBatchWrapTest (5 tests) compares the new path against the verbatim pre-change concatenation bodies, copied out of PushServer.startServer rather than reimplemented, so the comparison cannot drift into testing the new code against a restatement of itself:

• bareWrapMatchesLegacyConcatenation / delimitedWrapMatchesLegacyConcatenation — byte equality over 16 generated batch shapes (a singleton; an empty inner list; empty-then-nonempty-then-empty; zero-length payloads; 12 random shapes of 1–6 inner lists × 0–7 events × 0–511 byte payloads, fixed seed).
• compressedFrameWrapMatchesLegacyConcatenation — the SSE compressed frame.
• oversizedBatchFallsBackToCopyWithIdenticalBytes — MAX_WRAPPED_COMPONENTS + 5 events; asserts the fallback fired (nioBufferCount() == 1) and byte equality.
• normalBatchIsNotCopiedIntoOneBuffer — the 16-component consolidation guard described above.

LegacyTcpPipelineConfiguratorTest (4 tests) drives the configured pipeline through an EmbeddedChannel and asserts the emitted frame is exactly [PROTOCOL_VERSION, nameLength, ...nameBytes, ...payload] for a multi-component composite, a single ByteBuf, a byte[] heartbeat, and the null/empty-name case.

Benchmarks

This repo had no JMH source set, so this PR adds one to mantis-network (src/jmh, me.champeau.jmh applied module-locally so the root build is untouched) plus PushServerBatchWrapBenchmark.

The baseline arm holds the verbatim pre-change concatenation body, copied out of the pre-change PushServer rather than reimplemented, so the two arms cannot drift apart under review. @Setup asserts both arms produce byte-for-byte identical output before anything is measured.

Wrapping does not make bytes disappear — it moves who touches them. Measuring construction alone would flatter it, because the channel still has to walk the finished buffer's nioBuffers() on the way to a gathering write. So each path has a ...Gathered arm that also does that walk; the honest number is the ...Gathered arm, not the construction-only one.

JDK 17.0.17-zulu, @Fork(1), 5×1s warmup, 5×1s measurement, Mode.AverageTime, µs/op, -t 1:

Bare (non-SSE), 30 events/batch

bytes/event	legacy (verbatim)	this PR (+gather)	result
128	0.195	0.562	2.9× slower
256	0.280	0.564	2.0× slower
512	0.517	0.565	~even
1024	0.907	0.573	1.6× faster
2048	1.586	0.568	2.8× faster
4096	3.371	0.567	5.9× faster
16384	12.597	0.568	22× faster

Delimited (SSE, uncompressed), 30 events/batch

bytes/event	legacy (verbatim)	this PR (+gather)	result
128	0.341	1.916	5.6× slower
256	0.410	1.858	4.5× slower
512	0.621	1.920	3.1× slower
1024	0.907	1.810	2.0× slower
2048	1.913	1.966	~even
4096	3.535	1.933	1.8× faster
16384	12.520	1.935	6.5× faster

Read this honestly:

• The effect changes sign on bytes-per-event, not on event count. Legacy cost is O(total bytes) — the memcpy — and scales linearly with payload size. The wrap arm is dead flat in payload size (0.56 µs bare from 128 B to 16 KB) because its cost is O(component count). The 200-event rows have the identical shape, just scaled up.
• Measured crossover: ~575 B/event bare, ~2 KB/event delimited. Below that the composite bookkeeping and the nioBuffers() gather cost more than the copy they replace, and this is a regression. Delimited pays three components per event, so its crossover is higher.
• The events ≥ 1024 fallback rows (not shown) are within noise of legacy — the safety valve does not make large batches worse.
• These are single-threaded construction numbers with no allocator pressure. The eliminated byte[totalBytes] allocation (one per batch per connection) is not modelled here; under real allocation and GC pressure the win at large payloads is a lower bound.

Where the payloads actually sit

Whether this PR is net-positive depends entirely on real payload size, so I measured it — from Atlas metrics, not profiling: per-mantisJobName, tcpServer_..._bytesWritten (wire bytes/s) ÷ PushServer_numSuccessfulWrites (events/s), us-east-1, 6h window:

representative job	bytes/event
every significant push source	~5 KB – 22 KB
fleet weighted mean	~10.5 KB

Every job that moves meaningful push volume sits well above both crossovers, in the region where the bare path wins 3–22× and the delimited path 1.8–6.5×. Caveat: bytesWritten is post-framing, post-compression wire bytes, so for compressed SSE the pre-compression payload the wrap actually sees is larger still — but compression also defeats the wrap (the zlib encoder copies), so the relevant jobs for this change are the uncompressed ones.

Not addressed here

Two adjacent findings from the same measurement pass, deliberately out of scope:

• Push batches that are fully built and then dropped at the writability check (processedWrites.increment fires before the check, so that work is performed then discarded). This change makes each such batch cheaper to build but does not stop it being dropped.
• One 200ms OperatorBufferWithTime timer per connection, many of them near-idle.

Both are cheaper per batch after this change but neither goes away.