From cff4752609a38a1ee27d7ece4dc9156a82bb5e5d Mon Sep 17 00:00:00 2001 From: Dimitry Linkov Date: Mon, 17 Aug 2026 23:27:31 -0700 Subject: [PATCH 1/2] Stop shuffling the whole connection set on every routed batch RoundRobinRouter.route() copied the connection set into an ArrayList and Collections.shuffle()'d it on every call, only to then walk it in order. Three consequences, all paid per call: - Collections.shuffle(List) draws from a single private static Random owned by java.util.Collections and shared by the entire JVM (Collections.java, `private static Random r`). Random.nextInt is a CAS loop on one seed word, so the shuffle performs n contended CAS operations, contended not just across router threads but against every other caller of shuffle() in the process. - the ArrayList copy allocates an n-slot array, on top of the fresh HashSet that ConnectionManager.connections() and ConnectionGroup.getConnections() already allocate per call. - n element swaps whose only observable effect is which connection the round robin happens to start on. Only that start offset is observable, so take it directly: one ThreadLocalRandom draw, then advance the looping iterator over the set that was already handed in. Same uniform expectation per connection, no shared seed, no allocation, and on average half the element visits. This is per subscriber group, not per drain -- GroupChunkProcessor calls route() once for each group with the same chunk list -- so it multiplies by the group count. Two allocation fixes in the same path while here: - the per-destination buffer was a LinkedList, one Node object per event held purely to reference a byte[] an array slot would hold for free. Now a pre-sized ArrayList, matching ConsistentHashingRouter. - the writes map was unsized and rehashed past 12 entries. Sized to min(numConnections, chunks.size()): a batch cannot reach more destinations than it has chunks, and sizing on numConnections alone would allocate a million-entry table for a large fan-out group. The empty-batch and no-connection guards moved to the top as early returns, which is only hygiene -- TimedChunker.drain() and SingleThreadedChunker.drain() both check for a non-empty buffer before calling the processor, so chunks is never empty here in practice. Behaviour is unchanged, including the pre-existing rule that a chunk rejected by its connection's predicate is dropped rather than offered to the next connection. RoundRobinRouterTest pins that plus batching, distribution, and the start-offset randomisation. Co-Authored-By: Claude Opus 5 --- .../mantis/network/push/RoundRobinRouter.java | 70 +++--- .../network/push/RoundRobinRouterTest.java | 200 ++++++++++++++++++ 2 files changed, 242 insertions(+), 28 deletions(-) create mode 100644 mantis-network/src/test/java/io/reactivex/mantis/network/push/RoundRobinRouterTest.java diff --git a/mantis-network/src/main/java/io/reactivex/mantis/network/push/RoundRobinRouter.java b/mantis-network/src/main/java/io/reactivex/mantis/network/push/RoundRobinRouter.java index e2cbf3083..438c4f2c9 100644 --- a/mantis-network/src/main/java/io/reactivex/mantis/network/push/RoundRobinRouter.java +++ b/mantis-network/src/main/java/io/reactivex/mantis/network/push/RoundRobinRouter.java @@ -18,14 +18,13 @@ import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; import java.util.Iterator; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.atomic.AtomicReference; import rx.functions.Func1; @@ -38,36 +37,51 @@ public RoundRobinRouter(String name, Func1 encoder) { @Override public void route(Set> connections, List chunks) { - if (chunks != null && !chunks.isEmpty()) { - numEventsProcessed.increment(chunks.size()); + if (chunks == null || chunks.isEmpty()) { + return; } - List> randomOrder = new ArrayList<>(connections); - Collections.shuffle(randomOrder); - if (chunks != null && !chunks.isEmpty() && !randomOrder.isEmpty()) { - Iterator> iter = loopingIterator(randomOrder); - Map, List> writes = new HashMap<>(); - // process chunks - for (T chunk : chunks) { - AsyncConnection connection = iter.next(); - Func1 predicate = connection.getPredicate(); - if (predicate == null || predicate.call(chunk)) { - List buffer = writes.get(connection); - if (buffer == null) { - buffer = new LinkedList<>(); - writes.put(connection, buffer); - } - buffer.add(encoder.call(chunk)); - } - } - if (!writes.isEmpty()) { - for (Entry, List> entry : writes.entrySet()) { - AsyncConnection connection = entry.getKey(); - List toWrite = entry.getValue(); - connection.write(toWrite); - numEventsRouted.increment(toWrite.size()); + numEventsProcessed.increment(chunks.size()); + + int numConnections = connections.size(); + if (numConnections == 0) { + return; + } + + // Start the round robin at a random offset rather than shuffling the whole connection set. + // A shuffle costs one Random.nextInt() and one swap per connection, and Collections.shuffle + // (List) draws from a single private static Random shared by the whole JVM, so every draw is + // a contended CAS on that one seed -- shared with every other caller of shuffle() in the + // process. A single ThreadLocalRandom draw gives the same uniform expectation per + // connection with no shared state and no per-connection work, which matters here because + // route() is called once per subscriber group per drain. + Iterator> iter = loopingIterator(connections); + for (int i = ThreadLocalRandom.current().nextInt(numConnections); i > 0; i--) { + iter.next(); + } + + // assume even distribution + int bufferCapacity = (chunks.size() / numConnections) + 1; + Map, List> writes = + new HashMap<>(Math.min(numConnections, chunks.size())); + // process chunks + for (T chunk : chunks) { + AsyncConnection connection = iter.next(); + Func1 predicate = connection.getPredicate(); + if (predicate == null || predicate.call(chunk)) { + List buffer = writes.get(connection); + if (buffer == null) { + buffer = new ArrayList<>(bufferCapacity); + writes.put(connection, buffer); } + buffer.add(encoder.call(chunk)); } } + for (Entry, List> entry : writes.entrySet()) { + AsyncConnection connection = entry.getKey(); + List toWrite = entry.getValue(); + connection.write(toWrite); + numEventsRouted.increment(toWrite.size()); + } } private Iterator> loopingIterator(final Collection> connections) { diff --git a/mantis-network/src/test/java/io/reactivex/mantis/network/push/RoundRobinRouterTest.java b/mantis-network/src/test/java/io/reactivex/mantis/network/push/RoundRobinRouterTest.java new file mode 100644 index 000000000..713ad3d20 --- /dev/null +++ b/mantis-network/src/test/java/io/reactivex/mantis/network/push/RoundRobinRouterTest.java @@ -0,0 +1,200 @@ +/* + * Copyright 2026 Netflix, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.reactivex.mantis.network.push; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.junit.jupiter.api.Test; +import rx.Observer; +import rx.functions.Func1; + +/** + * Behavioural pins for {@link RoundRobinRouter#route}. The change under test replaced a full + * {@code Collections.shuffle} of the connection set with a single random start offset, and swapped + * the per-destination {@code LinkedList} for a pre-sized {@code ArrayList}; neither is observable + * from the outside, so what these tests assert is the routing contract that must not move. + */ +class RoundRobinRouterTest { + + /** Records what a connection was handed, in order. */ + private static final class Recorder implements Observer> { + + private final List> writes = new ArrayList<>(); + + @Override + public void onNext(List data) { + writes.add(data.stream() + .map(b -> new String(b, StandardCharsets.UTF_8)) + .collect(Collectors.toList())); + } + + @Override + public void onError(Throwable e) { + throw new AssertionError(e); + } + + @Override + public void onCompleted() { + } + + List received() { + return writes.stream().flatMap(List::stream).collect(Collectors.toList()); + } + } + + private static RoundRobinRouter router(String name) { + return new RoundRobinRouter<>(name, s -> s.getBytes(StandardCharsets.UTF_8)); + } + + /** Connections keyed by id, insertion-ordered so failures are readable. */ + private static Map connect( + Set> into, int count, Func1 predicate) { + Map recorders = new LinkedHashMap<>(); + for (int i = 0; i < count; i++) { + String id = "conn-" + i; + Recorder recorder = new Recorder(); + recorders.put(id, recorder); + into.add(new AsyncConnection<>("host", 1000 + i, id, id, "group", recorder, predicate)); + } + return recorders; + } + + private static List chunks(int count) { + return IntStream.range(0, count).mapToObj(i -> "event-" + i).collect(Collectors.toList()); + } + + @Test + void everyChunkIsDeliveredExactlyOnceAndSpreadEvenly() { + Set> connections = new LinkedHashSet<>(); + Map recorders = connect(connections, 4, null); + List chunks = chunks(10); + + RoundRobinRouter router = router("even-spread"); + router.route(connections, chunks); + + List delivered = recorders.values().stream() + .flatMap(r -> r.received().stream()) + .collect(Collectors.toList()); + assertEquals(chunks.size(), delivered.size(), "no chunk may be dropped or duplicated"); + assertEquals(new HashSet<>(chunks), new HashSet<>(delivered)); + + // 10 chunks over 4 connections is 3 or 2 each, whatever the start offset. + for (Map.Entry e : recorders.entrySet()) { + int size = e.getValue().received().size(); + assertTrue(size == 2 || size == 3, e.getKey() + " received " + size); + } + } + + /** + * Each destination gets one write holding all of its chunks -- not one write per chunk. This is + * what makes the downstream batching worthwhile, so it is worth pinning. + */ + @Test + void eachDestinationReceivesASingleBatchedWrite() { + Set> connections = new LinkedHashSet<>(); + Map recorders = connect(connections, 2, null); + + router("batched").route(connections, chunks(6)); + + for (Map.Entry e : recorders.entrySet()) { + assertEquals(1, e.getValue().writes.size(), e.getKey() + " should get one write"); + assertEquals(3, e.getValue().received().size()); + } + } + + /** Fewer chunks than connections: only the chunks.size() connections in line get a write. */ + @Test + void connectionsWithNoChunksAreNotWrittenTo() { + Set> connections = new LinkedHashSet<>(); + Map recorders = connect(connections, 10, null); + + router("sparse").route(connections, chunks(3)); + + long written = recorders.values().stream().filter(r -> !r.writes.isEmpty()).count(); + assertEquals(3, written, "exactly one write per chunk, to distinct connections"); + } + + /** + * A chunk assigned to a connection whose predicate rejects it is dropped, not handed on to the + * next connection. That was the pre-existing behaviour and this change does not alter it. + */ + @Test + void chunksRejectedByAPredicateAreDroppedNotReRouted() { + Set> connections = new LinkedHashSet<>(); + Map recorders = connect(connections, 3, s -> s.endsWith("0")); + List chunks = chunks(9); + + RoundRobinRouter router = router("predicate"); + router.route(connections, chunks); + + List delivered = recorders.values().stream() + .flatMap(r -> r.received().stream()) + .collect(Collectors.toList()); + assertEquals(List.of("event-0"), delivered); + } + + /** + * The start offset must keep moving: with a single chunk and many connections, load only spreads + * if successive calls begin somewhere else. Over 2000 calls across 50 connections the chance of + * any connection being missed by a uniform draw is about 1e-16, so this is not flaky. + */ + @Test + void startOffsetIsRandomisedAcrossCallsSoASingleChunkSpreads() { + Set> connections = new LinkedHashSet<>(); + Map recorders = connect(connections, 50, null); + + RoundRobinRouter router = router("spread"); + for (int i = 0; i < 2000; i++) { + router.route(connections, List.of("event-" + i)); + } + + long touched = recorders.values().stream().filter(r -> !r.writes.isEmpty()).count(); + assertEquals(50, touched, "every connection should have been the start at least once"); + } + + @Test + void emptyAndNullChunkBatchesAreNoOps() { + Set> connections = new LinkedHashSet<>(); + Map recorders = connect(connections, 3, null); + + RoundRobinRouter router = router("empty"); + router.route(connections, List.of()); + router.route(connections, null); + + assertTrue(recorders.values().stream().allMatch(r -> r.writes.isEmpty())); + } + + @Test + void noConnectionsDropsTheBatchWithoutThrowing() { + Set> connections = new LinkedHashSet<>(); + Map recorders = connect(connections, 2, null); + // deliberately route to an empty set, not to `connections` + router("no-connections").route(new HashSet<>(), chunks(5)); + assertTrue(recorders.values().stream().allMatch(r -> r.writes.isEmpty())); + } +} From 74872fb1b2a18104b71e2fd491e8a821b04a7d21 Mon Sep 17 00:00:00 2001 From: Dimitry Linkov Date: Tue, 18 Aug 2026 09:20:06 -0700 Subject: [PATCH 2/2] Add a JMH source set to mantis-network, with a RoundRobinRouter benchmark This repo had no JMH module. The plugin is applied module-locally, mirroring how other Netflix mantis modules do it, so the root build is untouched. RoundRobinRouterBenchmark compares route() against LegacyRoundRobinRouter, which holds the pre-change route() and loopingIterator() bodies copied verbatim out of git rather than reimplemented -- the comparison is only worth anything if the baseline cannot drift under review, so that file is explicitly marked do-not-clean-up. Deliveries are counted in a plain field on a per-connection sink rather than via Counter: CounterImpl.value() reads through to a no-op Spectator registry outside a running worker, so it always reads 0 here. Thread count is the whole point of this benchmark -- the largest cost removed is the contended CAS on Collections' JVM-wide private static Random -- and -t is not expressible as a @Param, so the results in the PR description come from driving the built jmhJar at -t 1 and -t 8. --- mantis-network/build.gradle | 25 +++ .../network/push/LegacyRoundRobinRouter.java | 106 +++++++++++ .../push/RoundRobinRouterBenchmark.java | 167 ++++++++++++++++++ 3 files changed, 298 insertions(+) create mode 100644 mantis-network/src/jmh/java/io/reactivex/mantis/network/push/LegacyRoundRobinRouter.java create mode 100644 mantis-network/src/jmh/java/io/reactivex/mantis/network/push/RoundRobinRouterBenchmark.java diff --git a/mantis-network/build.gradle b/mantis-network/build.gradle index 314091388..06e8df7dd 100644 --- a/mantis-network/build.gradle +++ b/mantis-network/build.gradle @@ -14,6 +14,20 @@ * limitations under the License. */ +buildscript { + repositories { + gradlePluginPortal() + } + dependencies { + classpath 'me.champeau.jmh:jmh-gradle-plugin:0.7.3' + } +} + +// JMH microbenchmarks live in src/jmh. Run with: ./gradlew :mantis-network:jmh +// or build the self-contained jar (./gradlew :mantis-network:jmhJar) and drive it with the +// standard JMH CLI, which is the only way to vary -t (thread count). +apply plugin: 'me.champeau.jmh' + ext { mqlVersion = '3.4.+' nettyVersion = '4.1.17.Final' @@ -30,8 +44,19 @@ dependencies { testImplementation libraries.junitJupiter testImplementation libraries.mockitoCore testImplementation libraries.slf4jLog4j12 + + // The benchmarks drive the routers and the push-server batch wrap directly, so they need the + // metrics implementation that main only compiles against. + jmhImplementation libraries.spectatorApi } test { useJUnitPlatform() } + +jmh { + jmhVersion = '1.37' + fork = 1 + warmupIterations = 5 + iterations = 5 +} diff --git a/mantis-network/src/jmh/java/io/reactivex/mantis/network/push/LegacyRoundRobinRouter.java b/mantis-network/src/jmh/java/io/reactivex/mantis/network/push/LegacyRoundRobinRouter.java new file mode 100644 index 000000000..c170503fc --- /dev/null +++ b/mantis-network/src/jmh/java/io/reactivex/mantis/network/push/LegacyRoundRobinRouter.java @@ -0,0 +1,106 @@ +/* + * Copyright 2019 Netflix, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.reactivex.mantis.network.push; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; +import rx.functions.Func1; + +/** + * The pre-change {@link RoundRobinRouter}, kept as the JMH baseline arm. + * + *

{@code route} and {@code loopingIterator} below are the verbatim bodies from before the + * change, copied out of git rather than reimplemented, precisely so the comparison cannot drift: + * + * 

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

Do not "clean this up" or apply review suggestions to it — its only value is being byte-for-byte + * what shipped. + */ +public class LegacyRoundRobinRouter extends Router { + + public LegacyRoundRobinRouter(String name, Func1 encoder) { + super("LegacyRoundRobinRouter_" + name, encoder); + } + + @Override + public void route(Set> connections, List chunks) { + if (chunks != null && !chunks.isEmpty()) { + numEventsProcessed.increment(chunks.size()); + } + List> randomOrder = new ArrayList<>(connections); + Collections.shuffle(randomOrder); + if (chunks != null && !chunks.isEmpty() && !randomOrder.isEmpty()) { + Iterator> iter = loopingIterator(randomOrder); + Map, List> writes = new HashMap<>(); + // process chunks + for (T chunk : chunks) { + AsyncConnection connection = iter.next(); + Func1 predicate = connection.getPredicate(); + if (predicate == null || predicate.call(chunk)) { + List buffer = writes.get(connection); + if (buffer == null) { + buffer = new LinkedList<>(); + writes.put(connection, buffer); + } + buffer.add(encoder.call(chunk)); + } + } + if (!writes.isEmpty()) { + for (Entry, List> entry : writes.entrySet()) { + AsyncConnection connection = entry.getKey(); + List toWrite = entry.getValue(); + connection.write(toWrite); + numEventsRouted.increment(toWrite.size()); + } + } + } + } + + private Iterator> loopingIterator(final Collection> connections) { + final AtomicReference>> iterRef = new AtomicReference<>(connections.iterator()); + return + new Iterator>() { + @Override + public boolean hasNext() { + return true; + } + + @Override + public AsyncConnection next() { + Iterator> iter = iterRef.get(); + if (iter.hasNext()) { + return iter.next(); + } else { + iterRef.set(connections.iterator()); + return iterRef.get().next(); + } + } + }; + } +} diff --git a/mantis-network/src/jmh/java/io/reactivex/mantis/network/push/RoundRobinRouterBenchmark.java b/mantis-network/src/jmh/java/io/reactivex/mantis/network/push/RoundRobinRouterBenchmark.java new file mode 100644 index 000000000..30a388b26 --- /dev/null +++ b/mantis-network/src/jmh/java/io/reactivex/mantis/network/push/RoundRobinRouterBenchmark.java @@ -0,0 +1,167 @@ +/* + * Copyright 2019 Netflix, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.reactivex.mantis.network.push; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; +import rx.Observer; +import rx.functions.Func1; + +/** + * Compares {@link RoundRobinRouter#route} against {@link LegacyRoundRobinRouter#route}, which is the + * verbatim pre-change body. + * + *

The two arms differ in three places: the legacy one copies the connection {@link Set} into an + * {@link ArrayList} and {@link java.util.Collections#shuffle(List) shuffle}s it, buffers per-destination + * writes in a {@link java.util.LinkedList}, and allocates an unsized {@link java.util.HashMap}. The new + * one advances a looping iterator to a single {@code ThreadLocalRandom} offset and sizes both + * collections. + * + *

Run multi-threaded. The largest claimed cost is not the per-connection work, it is that + * {@code Collections.shuffle(List)} draws from a single {@code private static Random} shared by the + * whole JVM, so every draw is a contended CAS on one seed. That only shows up with more than one + * router thread, and JMH's thread count is not expressible as a {@code @Param}, so drive it from the + * CLI: + * + * 

+ *   ./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
+ *
+ */ +@State(Scope.Benchmark) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.MICROSECONDS) +@Warmup(iterations = 5, time = 1) +@Measurement(iterations = 5, time = 1) +@Fork(1) +public class RoundRobinRouterBenchmark { + + private static final AtomicInteger NAMES = new AtomicInteger(); + + /** Subscribers on one group. Real groups run from a handful to hundreds of thousands. */ + @Param({"2", "8", "64", "512"}) + public int connections; + + /** Events in one drain -- what the 200ms chunker hands to route() at a time. */ + @Param({"30", "200"}) + public int chunkSize; + + private Router current; + private Router legacy; + private Set> connectionSet; + private List chunks; + + /** + * Counts deliveries into a plain field on a per-connection object. Not volatile and not + * synchronized on purpose: each connection has its own sink, so there is nothing to contend on, + * and the store still cannot be optimised away because the sink stays reachable from the + * benchmark state. Deliberately not a Counter -- {@code CounterImpl.value()} reads through to a + * Spectator registry that is a no-op outside a running worker, so it always reads zero here. + */ + static final class Sink implements Observer> { + long batches; + long events; + + @Override + public void onNext(List data) { + batches++; + events += data.size(); + } + + @Override + public void onCompleted() { + } + + @Override + public void onError(Throwable e) { + } + } + + private List sinks; + + @Setup(Level.Trial) + public void setup() { + String tag = "bench" + NAMES.incrementAndGet(); + Func1 encoder = b -> b; + current = new RoundRobinRouter<>(tag, encoder); + legacy = new LegacyRoundRobinRouter<>(tag, encoder); + + sinks = new ArrayList<>(connections); + connectionSet = new HashSet<>(connections * 2); + for (int i = 0; i < connections; i++) { + Sink sink = new Sink(); + sinks.add(sink); + connectionSet.add(new AsyncConnection<>( + "host" + i, 7000 + i, "id" + i, "slot" + i, "group", sink, null)); + } + + chunks = new ArrayList<>(chunkSize); + for (int i = 0; i < chunkSize; i++) { + byte[] event = new byte[256]; + event[0] = (byte) i; + chunks.add(event); + } + + // Cross-check the arms deliver the same volume before measuring anything. Content and + // destination differ run to run -- both arms randomise the starting connection -- so the + // invariant that holds is every chunk goes out exactly once. + long before = totalEvents(); + current.route(connectionSet, chunks); + long afterCurrent = totalEvents() - before; + legacy.route(connectionSet, chunks); + long afterLegacy = totalEvents() - before - afterCurrent; + if (afterCurrent != chunkSize || afterLegacy != chunkSize) { + throw new IllegalStateException("arms disagree: current=" + afterCurrent + + " legacy=" + afterLegacy + " expected=" + chunkSize); + } + } + + private long totalEvents() { + long total = 0; + for (Sink sink : sinks) { + total += sink.events; + } + return total; + } + + @Benchmark + public void route() { + current.route(connectionSet, chunks); + } + + @Benchmark + public void routeLegacyVerbatim() { + legacy.route(connectionSet, chunks); + } +}