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

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
+}

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.
+ *
+ * <p>{@code route} and {@code loopingIterator} below are the <b>verbatim</b> bodies from before the
+ * change, copied out of git rather than reimplemented, precisely so the comparison cannot drift:
+ *
+ * <pre>
+ *   git show master:mantis-network/src/main/java/io/reactivex/mantis/network/push/RoundRobinRouter.java
+ * </pre>
+ *
+ * <p>Do not "clean this up" or apply review suggestions to it — its only value is being byte-for-byte
+ * what shipped.
+ */
+public class LegacyRoundRobinRouter<T> extends Router<T> {
+
+    public LegacyRoundRobinRouter(String name, Func1<T, byte[]> encoder) {
+        super("LegacyRoundRobinRouter_" + name, encoder);
+    }
+
+    @Override
+    public void route(Set<AsyncConnection<T>> connections, List<T> chunks) {
+        if (chunks != null && !chunks.isEmpty()) {
+            numEventsProcessed.increment(chunks.size());
+        }
+        List<AsyncConnection<T>> randomOrder = new ArrayList<>(connections);
+        Collections.shuffle(randomOrder);
+        if (chunks != null && !chunks.isEmpty() && !randomOrder.isEmpty()) {
+            Iterator<AsyncConnection<T>> iter = loopingIterator(randomOrder);
+            Map<AsyncConnection<T>, List<byte[]>> writes = new HashMap<>();
+            // process chunks
+            for (T chunk : chunks) {
+                AsyncConnection<T> connection = iter.next();
+                Func1<T, Boolean> predicate = connection.getPredicate();
+                if (predicate == null || predicate.call(chunk)) {
+                    List<byte[]> buffer = writes.get(connection);
+                    if (buffer == null) {
+                        buffer = new LinkedList<>();
+                        writes.put(connection, buffer);
+                    }
+                    buffer.add(encoder.call(chunk));
+                }
+            }
+            if (!writes.isEmpty()) {
+                for (Entry<AsyncConnection<T>, List<byte[]>> entry : writes.entrySet()) {
+                    AsyncConnection<T> connection = entry.getKey();
+                    List<byte[]> toWrite = entry.getValue();
+                    connection.write(toWrite);
+                    numEventsRouted.increment(toWrite.size());
+                }
+            }
+        }
+    }
+
+    private Iterator<AsyncConnection<T>> loopingIterator(final Collection<AsyncConnection<T>> connections) {
+        final AtomicReference<Iterator<AsyncConnection<T>>> iterRef = new AtomicReference<>(connections.iterator());
+        return
+                new Iterator<AsyncConnection<T>>() {
+                    @Override
+                    public boolean hasNext() {
+                        return true;
+                    }
+
+                    @Override
+                    public AsyncConnection<T> next() {
+                        Iterator<AsyncConnection<T>> iter = iterRef.get();
+                        if (iter.hasNext()) {
+                            return iter.next();
+                        } else {
+                            iterRef.set(connections.iterator());
+                            return iterRef.get().next();
+                        }
+                    }
+                };
+    }
+}

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.
+ *
+ * <p>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.
+ *
+ * <p><b>Run multi-threaded.</b> 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:
+ *
+ * <pre>
+ *   ./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
+ * </pre>
+ */
+@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<byte[]> current;
+    private Router<byte[]> legacy;
+    private Set<AsyncConnection<byte[]>> connectionSet;
+    private List<byte[]> 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<List<byte[]>> {
+        long batches;
+        long events;
+
+        @Override
+        public void onNext(List<byte[]> data) {
+            batches++;
+            events += data.size();
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+
+        @Override
+        public void onError(Throwable e) {
+        }
+    }
+
+    private List<Sink> sinks;
+
+    @Setup(Level.Trial)
+    public void setup() {
+        String tag = "bench" + NAMES.incrementAndGet();
+        Func1<byte[], byte[]> 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);
+    }
+}