perf(network): stop shuffling the whole connection set on every routed batch

[1linkovdim]

What

RoundRobinRouter.route() no longer copies and shuffles the connection set on every routed batch. It takes a single random start offset instead, and stops using a LinkedList per destination.

Why

route() is on the push server's fan-out path, once per subscriber group per drain — GroupChunkProcessor.process loops the groups and calls route() for each with the same chunk list, so everything below multiplies by the group count, once per drain.

The old body was:

List<AsyncConnection<T>> randomOrder = new ArrayList<>(connections);
Collections.shuffle(randomOrder);

and then walked randomOrder in order via a looping iterator. Three costs, all per call, and only one bit of the result is ever observed:

1. Collections.shuffle(List) draws from one Random shared by the whole JVM. From java.util.Collections:

private static Random r;

public static void shuffle(List<?> list) {
    Random rnd = r;
    if (rnd == null)
        r = rnd = new Random(); // harmless race.
    shuffle(list, rnd);
}

Random.nextInt is a CAS loop on a single seed word. So a shuffle of n connections is n contended CAS operations on one shared location — contended across every router thread, and against every other caller of Collections.shuffle anywhere in the process. This is the part that does not scale with core count.

2. The ArrayList copy allocates an n-slot array, on top of the fresh HashSet that ConnectionManager.connections() / ConnectionGroup.getConnections() already allocate per call. It exists only so the shuffle has something mutable to swap in.

3. n swaps whose only observable effect is which connection the round robin starts at.

Since the start offset is the only observable output, take it directly:

Iterator<AsyncConnection<T>> iter = loopingIterator(connections);
for (int i = ThreadLocalRandom.current().nextInt(numConnections); i > 0; i--) {
    iter.next();
}

Same uniform expectation per connection, one RNG draw instead of n, no shared seed, no allocation, and on average half the element visits. The looping iterator already wraps, so nothing else changes.

Note what this does not fix: advancing to a random offset is still O(n) in the worst case, so a group with a very large connection count still walks a large fraction of the set per drain. Making that O(chunks) needs a start cursor that persists across calls, which is a behaviour change (the current code re-randomises every drain) and is better argued separately. Left out deliberately.

Two allocation fixes in the same loop

	Before	After
per-destination buffer	new LinkedList<>()	new ArrayList<>((chunks.size() / numConnections) + 1)
writes map	new HashMap<>()	new HashMap<>(Math.min(numConnections, chunks.size()))

The LinkedList allocated one Node per event purely to hold a byte[] reference that an array slot holds for free — garbage on a per-batch hot path, with no purpose. The capacity expression is the same one ConsistentHashingRouter already uses.

The map sizing is deliberately not ConsistentHashingRouter's new HashMap<>(numConnections). A batch cannot reach more destinations than it has chunks, so numConnections alone would allocate a table sized to the whole fan-out — for a group with ~1e6 connections and a ~30-event batch, a million-entry table per route() call. min(numConnections, chunks.size()) is the correct bound. Worth a look at whether CHR should get the same clamp; not touched here.

Hygiene, not a saving

The empty-chunks and empty-connections checks moved to the top as early returns, so the router no longer does setup work for a batch it will not route. This reads like a win but isn't one in practice: TimedChunker.drain() and SingleThreadedChunker.drain() both check for a non-empty buffer before calling processor.process, so chunks is never empty by the time it reaches route(). Calling it out so nobody counts it twice.

Behaviour

Unchanged, including one pre-existing rule that is easy to misread as a bug: a chunk assigned to a connection whose predicate rejects it is dropped, not offered to the next connection. That is what the old code did and this change preserves it.

loopingIterator now takes the Set directly rather than the removed ArrayList copy. That is safe because both ConnectionManager.connections() and ConnectionGroup.getConnections() return a freshly built HashSet that no one else holds — the old code was copying a copy.

Testing

New RoundRobinRouterTest — 7 tests, pinning the contract rather than the internals:

• every chunk delivered exactly once, spread evenly (10 chunks / 4 connections → 3,3,2,2 for any offset)
• one batched write per destination, not one per chunk
• with fewer chunks than connections, exactly chunks.size() connections are written to
• predicate-rejected chunks are dropped, not re-routed
• the start offset really does move: 2000 single-chunk calls over 50 connections must touch all 50 (probability of a miss under a uniform draw is ~1e-16, so not flaky)
• empty and null chunk batches are no-ops; an empty connection set does not throw

./gradlew :mantis-network:test is green (JDK 17). Counter assertions were removed from the tests on purpose — the default Spectator registry in this module is a no-op registry, so Counter.value() always reads 0 and asserting on it measures nothing.

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 RoundRobinRouterBenchmark.

The baseline arm is LegacyRoundRobinRouter, which holds the verbatim pre-change route() and loopingIterator() bodies, copied out of git rather than reimplemented:

git show master:mantis-network/src/main/java/io/reactivex/mantis/network/push/RoundRobinRouter.java

so the two arms cannot drift apart under review. @Setup cross-checks that both deliver exactly chunkSize events before anything is measured. Deliveries are counted in a plain field on a per-connection sink object, not via Counter — CounterImpl.value() reads through to a no-op Spectator registry outside a running worker, so it would always read 0.

JDK 17.0.17-zulu, @Fork(1), 5x1s warmup, 5x1s measurement, Mode.AverageTime. Thread count is not expressible as a @Param, so both runs are driven from the built jar:

./gradlew :mantis-network:jmhJar
java -jar mantis-network/build/libs/mantis-network-*-jmh.jar RoundRobinRouterBenchmark -t 1
java -jar mantis-network/build/libs/mantis-network-*-jmh.jar RoundRobinRouterBenchmark -t 8

Single-threaded (-t 1) - the allocation and O(n) work only

chunks	connections	this PR	legacy (verbatim)	speedup
30	2	0.345 +/- 0.021 us	0.577 +/- 0.016 us	1.67x
30	8	0.400 +/- 0.013	0.612 +/- 0.087	1.53x
30	64	0.666 +/- 0.035	1.425 +/- 0.032	2.14x
30	512	1.700 +/- 0.139	4.171 +/- 0.126	2.45x
200	2	2.040 +/- 0.061	3.139 +/- 0.095	1.54x
200	8	2.034 +/- 0.249	3.207 +/- 0.127	1.58x
200	64	2.548 +/- 0.062	4.463 +/- 0.129	1.75x
200	512	6.298 +/- 0.173	12.525 +/- 0.341	1.99x

The gap widens with connection count and not with chunk count, which is what the analysis predicts: the removed work is O(connections) per call, and the per-chunk loop is unchanged apart from LinkedList -> ArrayList.

8 threads (-t 8) - the shared-Random CAS

chunks	connections	this PR	legacy (verbatim)	speedup
30	2	0.954 +/- 0.014 us	1.134 +/- 0.056 us	1.19x
30	8	1.242 +/- 0.020	8.681 +/- 1.326	7.0x
30	64	1.967 +/- 0.044	138.143 +/- 108.994	70x
30	512	2.633 +/- 0.086	686.425 +/- 175.100	261x
200	2	2.199 +/- 0.017	3.922 +/- 0.399	1.78x
200	8	2.549 +/- 0.033	4.844 +/- 0.109	1.90x
200	64	6.085 +/- 0.391	80.328 +/- 29.623	13x
200	512	11.737 +/- 0.439	967.883 +/- 340.559	82x

Read those numbers with the right caveats:

• The huge ratios are contention, and contention is superlinear and environment-dependent. At 512 connections the legacy arm goes from 4.2 us single-threaded to 686 us at 8 threads - 164x slower in absolute terms for the same work. That is the signature of every thread CAS-looping on one cache line. The wide error bars (+/- 175, +/- 340) are part of the finding, not noise to be averaged away: under contention the variance is the behaviour.
• Do not read "261x" as a fleet speedup. It is the speedup of this one method on a machine where 8 threads do nothing but call it. Real routers interleave I/O, and the number of concurrently-shuffling threads in a JVM is a property of the deployment. What generalises is the shape: the old cost grows with both thread count and connection count, the new cost barely grows with either.
• The single-threaded table is the conservative claim. 1.5-2.5x on route() with zero contention assumed, which holds regardless of deployment.
• The new arm is not free either (0.35 -> 0.95 us at 2 connections across the thread step) - that is the shared Counter increments and sink cache lines, present in both arms.