From f02f9c9950cc1caac885ab932b9b732944a54d54 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Mon, 2 Nov 2026 11:26:13 -0500 Subject: [PATCH 1/5] RATIS-2787. Spread appendEntries between two servers over more than one connection Generated-by: Claude Code (Claude Opus 5.5) --- .../org/apache/ratis/grpc/GrpcConfigKeys.java | 15 + .../ratis/grpc/server/GrpcLogAppender.java | 13 +- .../grpc/server/GrpcServerProtocolClient.java | 47 ++- .../ratis/grpc/server/GrpcServicesImpl.java | 7 +- .../GrpcLogAppenderConnectionsBenchmark.java | 292 ++++++++++++++++++ .../server/TestGrpcLogAppenderCallbacks.java | 2 +- .../TestGrpcLogAppenderConnections.java | 65 ++++ .../TestGrpcServerProtocolClientChannels.java | 84 +++++ 8 files changed, 511 insertions(+), 14 deletions(-) create mode 100644 ratis-test/src/test/java/org/apache/ratis/grpc/server/GrpcLogAppenderConnectionsBenchmark.java create mode 100644 ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java create mode 100644 ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcConfigKeys.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcConfigKeys.java index e8fbffe602..451ba19231 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcConfigKeys.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcConfigKeys.java @@ -362,6 +362,21 @@ static void setStubPoolSize(RaftProperties properties, int size) { setInt(properties::setInt, STUB_POOL_SIZE_KEY, size); } + /** + * The number of connections to each peer used for appendEntries. The groups shared with a peer are spread over + * them, each group always using the same one, so that the replication between two servers is not limited to one + * connection and so one event-loop thread at each end. Heartbeats keep their own channel. + */ + String LOG_APPENDER_CONNECTIONS_KEY = PREFIX + ".log-appender.connections"; + int LOG_APPENDER_CONNECTIONS_DEFAULT = 1; + static int logAppenderConnections(RaftProperties properties) { + return getInt(properties::getInt, LOG_APPENDER_CONNECTIONS_KEY, + LOG_APPENDER_CONNECTIONS_DEFAULT, getDefaultLog(), requireMin(1), requireMax(1024)); + } + static void setLogAppenderConnections(RaftProperties properties, int connections) { + setInt(properties::setInt, LOG_APPENDER_CONNECTIONS_KEY, connections); + } + String BOSS_GROUP_SIZE_KEY = PREFIX + ".boss-group.size"; int BOSS_GROUP_SIZE_DEFAULT = 0; static int bossGroupSize(RaftProperties properties) { diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java index 14b2d6405c..851f4c11c3 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java @@ -24,6 +24,7 @@ import org.apache.ratis.grpc.metrics.GrpcServerMetrics; import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.proto.RaftProtos.InstallSnapshotResult; +import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.retry.RetryPolicy; import org.apache.ratis.server.RaftServer; @@ -389,10 +390,10 @@ static class StreamObservers { private final TimeDuration waitForReady; private volatile boolean running = true; - StreamObservers(GrpcServerProtocolClient client, AppendLogResponseHandler handler, boolean separateHeartbeat, - TimeDuration waitTimeMin) { - this.appendLog = client.appendEntries(handler, false); - this.heartbeat = separateHeartbeat? client.appendEntries(handler, true): null; + StreamObservers(GrpcServerProtocolClient client, RaftGroupId groupId, AppendLogResponseHandler handler, + boolean separateHeartbeat, TimeDuration waitTimeMin) { + this.appendLog = client.appendEntries(groupId, handler, false); + this.heartbeat = separateHeartbeat? client.appendEntries(groupId, handler, true): null; this.waitForReady = waitTimeMin.isPositive()? waitTimeMin: TimeDuration.ONE_MILLISECOND; } @@ -448,8 +449,8 @@ void appendLog(boolean heartbeat) throws IOException { notifyAppendEntriesListener(l -> l.onRequest(pending)); increaseNextIndex(pending); if (appendLogRequestObserver == null) { - appendLogRequestObserver = new StreamObservers( - getClient(), new AppendLogResponseHandler(), useSeparateHBChannel, getWaitTimeMin()); + appendLogRequestObserver = new StreamObservers(getClient(), getServer().getMemberId().getGroupId(), + new AppendLogResponseHandler(), useSeparateHBChannel, getWaitTimeMin()); } } diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java index 6dc0d32527..3e515ed9e9 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java @@ -20,8 +20,10 @@ import org.apache.ratis.grpc.GrpcUtil; import org.apache.ratis.util.NettyUtils; import org.apache.ratis.grpc.util.StreamObserverWithTimeout; +import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.server.util.ServerStringUtils; +import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting; import org.apache.ratis.thirdparty.io.grpc.ManagedChannel; import org.apache.ratis.thirdparty.io.grpc.netty.NegotiationType; import org.apache.ratis.thirdparty.io.grpc.netty.NettyChannelBuilder; @@ -39,6 +41,9 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; /** * This is a RaftClient implementation that supports streaming data to the raft @@ -52,6 +57,9 @@ class GrpcServerProtocolClient implements Closeable { private ManagedChannel hbChannel; private RaftServerProtocolServiceStub hbAsyncStub; private final RaftServerProtocolServiceStub asyncStub; + // Channels and stubs for appendEntries, each group always using the same one; the first is the main channel + private final List appendEntriesChannels; + private final List appendEntriesStubs; private final RaftServerProtocolServiceBlockingStub blockingStub; private final boolean useSeparateHBChannel; @@ -61,7 +69,7 @@ class GrpcServerProtocolClient implements Closeable { private final RaftPeerId raftPeerId; private final EventLoopGroup eventLoopGroup; - GrpcServerProtocolClient(RaftPeer target, int connections, int flowControlWindow, + GrpcServerProtocolClient(RaftPeer target, int connections, int appendEntriesConnections, int flowControlWindow, TimeDuration requestTimeout, SslContext sslContext, boolean separateHBChannel, EventLoopGroup eventLoopGroup) { raftPeerId = target.getId(); @@ -71,6 +79,17 @@ class GrpcServerProtocolClient implements Closeable { channel = buildChannel(target, flowControlWindow, sslContext); blockingStub = RaftServerProtocolServiceGrpc.newBlockingStub(channel); asyncStub = RaftServerProtocolServiceGrpc.newStub(channel); + final List channels = new ArrayList<>(appendEntriesConnections); + final List stubs = new ArrayList<>(appendEntriesConnections); + channels.add(channel); + stubs.add(asyncStub); + for (int i = 1; i < appendEntriesConnections; i++) { + final ManagedChannel c = buildChannel(target, flowControlWindow, sslContext); + channels.add(c); + stubs.add(RaftServerProtocolServiceGrpc.newStub(c)); + } + appendEntriesChannels = Collections.unmodifiableList(channels); + appendEntriesStubs = Collections.unmodifiableList(stubs); if (useSeparateHBChannel) { hbChannel = buildChannel(target, flowControlWindow, sslContext); hbAsyncStub = RaftServerProtocolServiceGrpc.newStub(hbChannel); @@ -109,7 +128,8 @@ public void close() { if (useSeparateHBChannel) { GrpcUtil.shutdownManagedChannel(hbChannel); } - GrpcUtil.shutdownManagedChannel(channel); + // the first one is the main channel + appendEntriesChannels.forEach(GrpcUtil::shutdownManagedChannel); if (pool != null) { pool.close(); } @@ -171,15 +191,31 @@ public void onCompleted() { } } - CallStreamObserver appendEntries( + CallStreamObserver appendEntries(RaftGroupId groupId, StreamObserver responseHandler, boolean isHeartbeat) { if (isHeartbeat && useSeparateHBChannel) { return (CallStreamObserver) hbAsyncStub.appendEntries(responseHandler); } else { - return (CallStreamObserver) asyncStub.appendEntries(responseHandler); + final RaftServerProtocolServiceStub stub = appendEntriesStubs.get(getAppendEntriesIndex(groupId)); + return (CallStreamObserver) stub.appendEntries(responseHandler); } } + /** @return the index of the appendEntries channel for the given group, always the same one for a group. */ + int getAppendEntriesIndex(RaftGroupId groupId) { + return Math.floorMod(groupId.hashCode(), appendEntriesStubs.size()); + } + + @VisibleForTesting + List getAppendEntriesChannels() { + return appendEntriesChannels; + } + + @VisibleForTesting + ManagedChannel getMainChannel() { + return channel; + } + StreamObserver installSnapshot( String name, TimeDuration timeout, int limit, StreamObserver responseHandler) { return StreamObserverWithTimeout.newInstance(name, ServerStringUtils::toInstallSnapshotRequestString, @@ -191,6 +227,7 @@ public void resetConnectBackoff() { if (useSeparateHBChannel) { hbChannel.resetConnectBackoff(); } - channel.resetConnectBackoff(); + // the first one is the main channel + appendEntriesChannels.forEach(ManagedChannel::resetConnectBackoff); } } diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServicesImpl.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServicesImpl.java index cf3787b846..7c0cd9bd21 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServicesImpl.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServicesImpl.java @@ -113,6 +113,7 @@ public static final class Builder { private SslContext serverSslContextForServer; private SslContext serverSslContextForClient; private int serverStubPoolSize; + private int logAppenderConnections; private SizeInBytes messageSizeMax; private SizeInBytes flowControlWindow; @@ -140,6 +141,7 @@ public Builder setServer(RaftServer raftServer) { this.requestTimeoutDuration = RaftServerConfigKeys.Rpc.requestTimeout(properties); this.separateHeartbeatChannel = GrpcConfigKeys.Server.heartbeatChannel(properties); this.serverStubPoolSize = GrpcConfigKeys.Server.stubPoolSize(properties); + this.logAppenderConnections = GrpcConfigKeys.Server.logAppenderConnections(properties); final SizeInBytes appenderBufferSize = RaftServerConfigKeys.Log.Appender.bufferByteLimit(properties); final SizeInBytes gap = SizeInBytes.ONE_MB; @@ -165,8 +167,9 @@ public Builder setServerCredentials(ServerCredentials credentials) { } private GrpcServerProtocolClient newGrpcServerProtocolClient(RaftPeer target, EventLoopGroup eventLoopGroup) { - return new GrpcServerProtocolClient(target, serverStubPoolSize, flowControlWindow.getSizeInt(), - requestTimeoutDuration, serverSslContextForClient, separateHeartbeatChannel, eventLoopGroup); + return new GrpcServerProtocolClient(target, serverStubPoolSize, logAppenderConnections, + flowControlWindow.getSizeInt(), requestTimeoutDuration, serverSslContextForClient, separateHeartbeatChannel, + eventLoopGroup); } private ExecutorService newExecutor() { diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/GrpcLogAppenderConnectionsBenchmark.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/GrpcLogAppenderConnectionsBenchmark.java new file mode 100644 index 0000000000..517d5a4459 --- /dev/null +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/GrpcLogAppenderConnectionsBenchmark.java @@ -0,0 +1,292 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.ratis.grpc.server; + +import org.apache.ratis.BaseTest; +import org.apache.ratis.RaftTestUtil; +import org.apache.ratis.client.RaftClient; +import org.apache.ratis.conf.Parameters; +import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.grpc.GrpcConfigKeys; +import org.apache.ratis.grpc.GrpcTlsConfig; +import org.apache.ratis.grpc.MiniRaftClusterWithGrpc; +import org.apache.ratis.protocol.Message; +import org.apache.ratis.protocol.RaftClientReply; +import org.apache.ratis.protocol.RaftGroup; +import org.apache.ratis.protocol.RaftGroupId; +import org.apache.ratis.protocol.RaftPeer; +import org.apache.ratis.protocol.RaftPeerId; +import org.apache.ratis.security.SecurityTestUtils; +import org.apache.ratis.server.RaftServerConfigKeys; +import org.apache.ratis.server.impl.MiniRaftCluster; +import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; +import org.apache.ratis.thirdparty.io.netty.util.ResourceLeakDetector; +import org.apache.ratis.util.SizeInBytes; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.AbstractMap; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * RATIS-2787 benchmark: replication throughput between servers that share many groups, against the number of + * appendEntries connections per peer (raft.grpc.server.log-appender.connections). Not run by the build (the name + * doesn't match the surefire patterns); run with + * 

+ *   mvn -pl ratis-test -am test -Dtest=GrpcLogAppenderConnectionsBenchmark -Dsurefire.failIfNoSpecifiedTests=false \
+ *     -Djacoco.skip=true -DextraJavaTestArgs=-Xmx12g [-Dbench.connections=1,4 -Dbench.groups=12 \
+ *     -Dbench.messageKB=1024 -Dbench.messagesPerGroup=50 -Dbench.warmupPerGroup=20 -Dbench.window=8 \
+ *     -Dbench.rounds=1 -Dbench.tls=true -Dbench.workers=8 -Dbench.leaderOnOne=true -Dbench.flowControlWindowKB=0]
+ *
+ * JaCoCo (on by default in ratis-test) roughly halves the throughput, so skip it. Netty leak detection is disabled. + *

+ * Each arm starts a new 3-server gRPC cluster (mutual TLS by default, in-memory Raft log, so no disk), adds the groups + * on all three servers and, by default, moves every group's leader to the first server, so that all replication goes + * over its connections to the two others: that is the case where a single connection per peer is the limit. By + * default each server also gets its own event-loop groups (bench.workers threads each), as if in its own process; + * with bench.workers=0 all servers share gRPC's default event loops, as many as cores. It writes a warm-up, then the + * timed messages, to every group at once, with a window of outstanding async requests per group, until all are + * committed. All three servers run in this JVM and keep the log in memory, so the heap needs about + * 3 x groups x (messagesPerGroup + warmupPerGroup) x messageKB, in a fresh JVM per arm. + *

+ * It reports the replication rate, the hottest event-loop threads during the timed writes (CPU time / wall time; + * cwI is a client worker, i.e. outgoing connections, swI a server worker), how many were over half busy, and the + * cores used by all event loops and by the whole JVM. + */ +public class GrpcLogAppenderConnectionsBenchmark extends BaseTest { + private static final int GROUPS = Integer.getInteger("bench.groups", 12); + private static final int MESSAGE_KB = Integer.getInteger("bench.messageKB", 1024); + private static final int MESSAGES = Integer.getInteger("bench.messagesPerGroup", 50); + private static final int WARMUP = Integer.getInteger("bench.warmupPerGroup", 20); + private static final int WORKERS = Integer.getInteger("bench.workers", 8); + // 0 keeps Ratis's default (raft.grpc.flow.control.window, 1MB); Ozone sets 5MB for its 4MB chunks + private static final int FLOW_CONTROL_WINDOW_KB = Integer.getInteger("bench.flowControlWindowKB", 0); + private static final boolean LEADER_ON_ONE = Boolean.parseBoolean(System.getProperty("bench.leaderOnOne", "true")); + private static final int WINDOW = Integer.getInteger("bench.window", 8); + private static final int ROUNDS = Integer.getInteger("bench.rounds", 1); + private static final String CONNECTIONS = System.getProperty("bench.connections", "1,4"); + private static final boolean TLS = Boolean.parseBoolean(System.getProperty("bench.tls", "true")); + + static { + // Netty's default (SIMPLE) leak detection samples buffers and records a stack trace for each; not what is measured + ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.DISABLED); + } + + @Test + @Timeout(value = 60, unit = TimeUnit.MINUTES) // BaseTest's 100 s is for unit tests + public void benchmark() throws Exception { + final String[] ks = CONNECTIONS.split(","); + System.out.printf("%nRATIS-2787: %d groups x %d messages x %d KB (warm-up %d), window %d, TLS %s, workers %d," + + " leader on one server %s, flow control window %s%n", GROUPS, MESSAGES, MESSAGE_KB, WARMUP, WINDOW, TLS, + WORKERS, LEADER_ON_ONE, FLOW_CONTROL_WINDOW_KB > 0 ? FLOW_CONTROL_WINDOW_KB + "KB" : "default"); + System.out.printf("%-6s %-3s %10s %12s %40s %10s %10s %10s%n", + "round", "K", "seconds", "MB/s", "hottest loops (%CPU)", "loops>50%", "loop cores", "JVM cores"); + for (int round = 1; round <= ROUNDS; round++) { + for (int i = 0; i < ks.length; i++) { + // rotate the order, so no K always runs first + final int k = Integer.parseInt(ks[(i + round) % ks.length].trim()); + runArm(round, k); + System.gc(); + } + } + } + + private static Parameters newTlsParameters() throws Exception { + final GrpcTlsConfig server = new GrpcTlsConfig( + SecurityTestUtils.getKeyManager(SecurityTestUtils::getServerKeyStore), + SecurityTestUtils.getTrustManager(SecurityTestUtils::getTrustStore), true); + final GrpcTlsConfig client = new GrpcTlsConfig( + SecurityTestUtils.getKeyManager(SecurityTestUtils::getClientKeyStore), + SecurityTestUtils.getTrustManager(SecurityTestUtils::getTrustStore), true); + final Parameters parameters = new Parameters(); + GrpcConfigKeys.Server.setTlsConf(parameters, server); + GrpcConfigKeys.Admin.setTlsConf(parameters, client); + GrpcConfigKeys.Client.setTlsConf(parameters, client); + return parameters; + } + + private void runArm(int round, int connections) throws Exception { + final RaftProperties properties = new RaftProperties(); + RaftServerConfigKeys.Log.setUseMemory(properties, true); + GrpcConfigKeys.Server.setLogAppenderConnections(properties, connections); + if (FLOW_CONTROL_WINDOW_KB > 0) { + GrpcConfigKeys.setFlowControlWindow(properties, SizeInBytes.valueOf(FLOW_CONTROL_WINDOW_KB + "KB")); + } + if (WORKERS > 0) { + // each server gets its own event loops, as in separate processes, instead of sharing gRPC's default ones + GrpcConfigKeys.Server.setWorkerGroupSize(properties, WORKERS); + GrpcConfigKeys.Client.setWorkerGroupSize(properties, WORKERS); + } + final MiniRaftClusterWithGrpc cluster = new MiniRaftClusterWithGrpc(MiniRaftCluster.generateIds(3, 0), + new String[0], properties, TLS ? newTlsParameters() : null); + try { + final long setupStart = System.nanoTime(); + cluster.start(); + RaftTestUtil.waitForLeader(cluster); + final List peers = new ArrayList<>(cluster.getGroup().getPeers()); + final RaftPeerId leaderId = peers.get(0).getId(); + + // the groups, each on all three servers, so they share the connections between each pair of servers + final List groupIds = new ArrayList<>(); + final List clients = new ArrayList<>(); + try (RaftClient admin = cluster.createClient(cluster.getGroup())) { + for (int g = 0; g < GROUPS; g++) { + final RaftGroup group = RaftGroup.valueOf(RaftGroupId.randomId(), peers); + for (RaftPeer peer : peers) { + Assertions.assertTrue(admin.getGroupManagementApi(peer.getId()).add(group).isSuccess()); + } + groupIds.add(group.getGroupId()); + clients.add(cluster.createClient(group)); + } + } + // wait for a leader in every group + for (RaftClient client : clients) { + Assertions.assertTrue(client.io().send(Message.valueOf("first")).isSuccess()); + } + if (LEADER_ON_ONE) { + // all groups led by one server, so all replication goes over its connections to the two others + for (int g = 0; g < GROUPS; g++) { + if (!cluster.getDivision(leaderId, groupIds.get(g)).getInfo().isLeader()) { + Assertions.assertTrue(clients.get(g).admin().transferLeadership(leaderId, 30_000).isSuccess()); + } + } + for (int g = 0; g < GROUPS; g++) { + Assertions.assertTrue(cluster.getDivision(leaderId, groupIds.get(g)).getInfo().isLeader()); + } + } + + final byte[] bytes = new byte[MESSAGE_KB << 10]; + new Random(0).nextBytes(bytes); + final Message message = Message.valueOf(ByteString.copyFrom(bytes)); + // warm-up, so the JIT is mostly done before the timed writes + writeAll(clients, message, WARMUP); + System.out.printf(" K=%d: cluster and %d groups ready and warmed up in %.1f s%n", connections, GROUPS, + (System.nanoTime() - setupStart) / 1e9); + + final Map cpuBefore = loopCpuTimes(); + final long processCpuBefore = processCpuTime(); + final long startMillis = System.currentTimeMillis(); + final long start = System.nanoTime(); + writeAll(clients, message, MESSAGES); + final long elapsedNanos = System.nanoTime() - start; + final double jvmCores = (double) (processCpuTime() - processCpuBefore) / elapsedNanos; + final Map cpuAfter = loopCpuTimes(); + + final List> busy = new ArrayList<>(); + double loopCores = 0; + final ThreadMXBean threads = ManagementFactory.getThreadMXBean(); + for (Map.Entry e : cpuAfter.entrySet()) { + final long cpu = e.getValue() - cpuBefore.getOrDefault(e.getKey(), 0L); + final ThreadInfo info = threads.getThreadInfo(e.getKey()); + final String name = info == null ? "?" : shortName(info.getThreadName()); + busy.add(new AbstractMap.SimpleImmutableEntry<>(name, 100.0 * cpu / elapsedNanos)); + loopCores += (double) cpu / elapsedNanos; + } + busy.sort((a, b) -> Double.compare(b.getValue(), a.getValue())); + final StringBuilder hottest = new StringBuilder(); + for (int i = 0; i < Math.min(3, busy.size()); i++) { + hottest.append(i == 0 ? "" : " ").append(busy.get(i).getKey()).append('=') + .append(Math.round(busy.get(i).getValue())); + } + final long overHalf = busy.stream().filter(b -> b.getValue() > 50).count(); + final double seconds = elapsedNanos / 1e9; + final double mbPerSecond = (double) GROUPS * MESSAGES * MESSAGE_KB / 1024 / seconds; + System.out.printf("%-6d %-3d %10.2f %12.1f %40s %10d %10.1f %10.1f%n", round, connections, seconds, + mbPerSecond, hottest, overHalf, loopCores, jvmCores); + System.out.printf(" K=%d: writes from %d to %d ms (epoch)%n", connections, startMillis, + System.currentTimeMillis()); + for (RaftClient client : clients) { + client.close(); + } + } finally { + cluster.shutdown(); + } + } + + /** Writes the message count times to every group at once, with a window of outstanding requests per group. */ + private static void writeAll(List clients, Message message, int count) throws Exception { + final AtomicInteger failures = new AtomicInteger(); + final List> replies = new ArrayList<>(); + final List windows = new ArrayList<>(); + for (int g = 0; g < clients.size(); g++) { + windows.add(new Semaphore(WINDOW)); + } + for (int m = 0; m < count; m++) { + for (int g = 0; g < clients.size(); g++) { + final Semaphore window = windows.get(g); + window.acquire(); + replies.add(clients.get(g).async().send(message).whenComplete((reply, e) -> { + window.release(); + if (e != null || !reply.isSuccess()) { + failures.incrementAndGet(); + } + })); + } + } + CompletableFuture.allOf(replies.toArray(new CompletableFuture[0])).get(30, TimeUnit.MINUTES); + Assertions.assertEquals(0, failures.get(), "failed writes"); + } + + /** + * E.g. ...-client-workers--thread3 to cw3, ...-server-workers--thread3 to sw3, grpc-nio-worker-ELG-3-13 to elg13. + * The groups are named before the server knows its id, so the names don't tell the servers apart. + */ + private static String shortName(String name) { + final String id = name.replaceAll("^.*?(\\d+)$", "$1"); + if (name.contains("-client-workers")) { + return "cw" + id; + } else if (name.contains("-server-workers")) { + return "sw" + id; + } + return "elg" + id; + } + + private static long processCpuTime() { + return ((com.sun.management.OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean()) + .getProcessCpuTime(); + } + + /** CPU time of the Netty event-loop threads (gRPC's and Ratis's), by thread id. */ + private static Map loopCpuTimes() { + final ThreadMXBean threads = ManagementFactory.getThreadMXBean(); + final Map times = new HashMap<>(); + for (ThreadInfo info : threads.dumpAllThreads(false, false)) { + final String name = info.getThreadName(); + if (name.contains("grpc-default-worker") || name.contains("ELG") || name.contains("nioEventLoop") + || name.contains("epoll") || name.contains("-client-workers") || name.contains("-server-workers")) { + final long cpu = threads.getThreadCpuTime(info.getThreadId()); + if (cpu > 0) { + times.put(info.getThreadId(), cpu); + } + } + } + return times; + } +} diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java index 79ae4853bf..45e5e1c5f7 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java @@ -188,7 +188,7 @@ public void testStreamCreationFailure(boolean failProxy) throws Exception { when(serverRpc.getProxies().getProxy(destination)).thenThrow(error); } else { error = new IllegalStateException("Failed to create append stream"); - when(client.appendEntries(any(), eq(false))).thenThrow(error); + when(client.appendEntries(any(), any(), eq(false))).thenThrow(error); } Assertions.assertSame(error, Assertions.assertThrows(Exception.class, () -> appender.appendLog(false))); final InOrder order = inOrder(appendEntries); diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java new file mode 100644 index 0000000000..e7f23000a4 --- /dev/null +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java @@ -0,0 +1,65 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.ratis.grpc.server; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.ratis.BaseTest; +import org.apache.ratis.client.RaftClient; +import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.grpc.GrpcConfigKeys; +import org.apache.ratis.grpc.MiniRaftClusterWithGrpc; +import org.apache.ratis.protocol.RaftGroup; +import org.apache.ratis.protocol.RaftPeer; +import org.apache.ratis.server.impl.GroupManagementBaseTest; +import org.apache.ratis.server.impl.MiniRaftCluster; +import org.apache.ratis.server.RaftServer; +import org.apache.ratis.RaftTestUtil.SimpleMessage; +import org.apache.ratis.util.function.CheckedBiConsumer; +import org.junit.jupiter.api.Test; + +/** Several groups on the same servers replicate over more than one appendEntries connection per peer. */ +public class TestGrpcLogAppenderConnections extends BaseTest { + private static final int CONNECTIONS = 3; + + @Test + public void testMultiGroupWithLogAppenderConnections() throws Exception { + final RaftProperties properties = new RaftProperties(); + GrpcConfigKeys.Server.setLogAppenderConnections(properties, CONNECTIONS); + final MiniRaftCluster cluster = MiniRaftClusterWithGrpc.FACTORY.newCluster(0, properties); + + final CheckedBiConsumer checker = (c, group) -> { + for (RaftPeer peer : group.getPeers()) { + final RaftServer server = c.getServer(peer.getId()); + final GrpcServicesImpl rpc = (GrpcServicesImpl) server.getServerRpc(); + for (RaftPeer other : group.getPeers()) { + if (!other.getId().equals(peer.getId())) { + assertEquals(CONNECTIONS, rpc.getProxies().getProxy(other.getId()).getAppendEntriesChannels().size()); + } + } + } + try (RaftClient client = c.createClient(group)) { + for (int i = 0; i < 10; i++) { + assertTrue(client.io().send(new SimpleMessage(group.getGroupId() + "-" + i)).isSuccess()); + } + } + }; + GroupManagementBaseTest.runMultiGroupTest(cluster, new int[] {3, 6, 9}, -1, checker); + } +} diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java new file mode 100644 index 0000000000..f5b78b5ed1 --- /dev/null +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java @@ -0,0 +1,84 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.ratis.grpc.server; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import org.apache.ratis.BaseTest; +import org.apache.ratis.protocol.RaftGroupId; +import org.apache.ratis.protocol.RaftPeer; +import org.apache.ratis.thirdparty.io.grpc.ManagedChannel; +import org.apache.ratis.util.TimeDuration; +import org.junit.jupiter.api.Test; + +/** Tests the appendEntries channels of {@link GrpcServerProtocolClient}; channels connect lazily, no server needed. */ +public class TestGrpcServerProtocolClientChannels extends BaseTest { + private static final RaftPeer PEER = RaftPeer.newBuilder().setId("s1").setAddress("localhost:1").build(); + + private static GrpcServerProtocolClient newClient(int appendEntriesConnections) { + return new GrpcServerProtocolClient(PEER, 1, appendEntriesConnections, 1 << 20, + TimeDuration.valueOf(3, TimeUnit.SECONDS), null, true, null); + } + + @Test + public void testOneConnectionUsesMainChannel() { + final GrpcServerProtocolClient client = newClient(1); + try { + final List channels = client.getAppendEntriesChannels(); + assertEquals(1, channels.size()); + assertSame(client.getMainChannel(), channels.get(0)); + for (int i = 0; i < 10; i++) { + assertEquals(0, client.getAppendEntriesIndex(RaftGroupId.randomId())); + } + } finally { + client.close(); + } + } + + @Test + public void testGroupsSpreadOverConnections() { + final int connections = 4; + final GrpcServerProtocolClient client = newClient(connections); + final List channels = client.getAppendEntriesChannels(); + try { + assertEquals(connections, channels.size()); + assertSame(client.getMainChannel(), channels.get(0)); + assertEquals(connections, new HashSet<>(channels).size()); + + final Set used = new HashSet<>(); + for (int i = 0; i < 100; i++) { + final RaftGroupId groupId = RaftGroupId.randomId(); + final int index = client.getAppendEntriesIndex(groupId); + assertTrue(index >= 0 && index < connections, "index " + index); + // a group always uses the same connection, which keeps its appendEntries in order + assertEquals(index, client.getAppendEntriesIndex(RaftGroupId.valueOf(groupId.getUuid()))); + used.add(index); + } + assertEquals(connections, used.size()); + } finally { + client.close(); + } + channels.forEach(c -> assertTrue(c.isShutdown(), c + " is not shut down")); + } +} From 869440665c2be743a1e1b471b06d2bd1cd222438 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Mon, 2 Nov 2026 21:22:11 -0500 Subject: [PATCH 2/5] fixed checkstyle --- .../org/apache/ratis/grpc/server/GrpcServerProtocolClient.java | 1 + 1 file changed, 1 insertion(+) diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java index 3e515ed9e9..890de6ff8a 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java @@ -69,6 +69,7 @@ class GrpcServerProtocolClient implements Closeable { private final RaftPeerId raftPeerId; private final EventLoopGroup eventLoopGroup; + @SuppressWarnings("checkstyle:ParameterNumber") GrpcServerProtocolClient(RaftPeer target, int connections, int appendEntriesConnections, int flowControlWindow, TimeDuration requestTimeout, SslContext sslContext, boolean separateHBChannel, EventLoopGroup eventLoopGroup) { From 4bc69fe88be44179ed75a629eb95380c3222e618 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Mon, 2 Nov 2026 21:34:19 -0500 Subject: [PATCH 3/5] deleted benchmark --- .../GrpcLogAppenderConnectionsBenchmark.java | 292 ------------------ 1 file changed, 292 deletions(-) delete mode 100644 ratis-test/src/test/java/org/apache/ratis/grpc/server/GrpcLogAppenderConnectionsBenchmark.java diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/GrpcLogAppenderConnectionsBenchmark.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/GrpcLogAppenderConnectionsBenchmark.java deleted file mode 100644 index 517d5a4459..0000000000 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/GrpcLogAppenderConnectionsBenchmark.java +++ /dev/null @@ -1,292 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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 org.apache.ratis.grpc.server; - -import org.apache.ratis.BaseTest; -import org.apache.ratis.RaftTestUtil; -import org.apache.ratis.client.RaftClient; -import org.apache.ratis.conf.Parameters; -import org.apache.ratis.conf.RaftProperties; -import org.apache.ratis.grpc.GrpcConfigKeys; -import org.apache.ratis.grpc.GrpcTlsConfig; -import org.apache.ratis.grpc.MiniRaftClusterWithGrpc; -import org.apache.ratis.protocol.Message; -import org.apache.ratis.protocol.RaftClientReply; -import org.apache.ratis.protocol.RaftGroup; -import org.apache.ratis.protocol.RaftGroupId; -import org.apache.ratis.protocol.RaftPeer; -import org.apache.ratis.protocol.RaftPeerId; -import org.apache.ratis.security.SecurityTestUtils; -import org.apache.ratis.server.RaftServerConfigKeys; -import org.apache.ratis.server.impl.MiniRaftCluster; -import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; -import org.apache.ratis.thirdparty.io.netty.util.ResourceLeakDetector; -import org.apache.ratis.util.SizeInBytes; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.Timeout; - -import java.lang.management.ManagementFactory; -import java.lang.management.ThreadInfo; -import java.lang.management.ThreadMXBean; -import java.util.AbstractMap; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Random; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -/** - * RATIS-2787 benchmark: replication throughput between servers that share many groups, against the number of - * appendEntries connections per peer (raft.grpc.server.log-appender.connections). Not run by the build (the name - * doesn't match the surefire patterns); run with - * 

- *   mvn -pl ratis-test -am test -Dtest=GrpcLogAppenderConnectionsBenchmark -Dsurefire.failIfNoSpecifiedTests=false \
- *     -Djacoco.skip=true -DextraJavaTestArgs=-Xmx12g [-Dbench.connections=1,4 -Dbench.groups=12 \
- *     -Dbench.messageKB=1024 -Dbench.messagesPerGroup=50 -Dbench.warmupPerGroup=20 -Dbench.window=8 \
- *     -Dbench.rounds=1 -Dbench.tls=true -Dbench.workers=8 -Dbench.leaderOnOne=true -Dbench.flowControlWindowKB=0]
- *
- * JaCoCo (on by default in ratis-test) roughly halves the throughput, so skip it. Netty leak detection is disabled. - *

- * Each arm starts a new 3-server gRPC cluster (mutual TLS by default, in-memory Raft log, so no disk), adds the groups - * on all three servers and, by default, moves every group's leader to the first server, so that all replication goes - * over its connections to the two others: that is the case where a single connection per peer is the limit. By - * default each server also gets its own event-loop groups (bench.workers threads each), as if in its own process; - * with bench.workers=0 all servers share gRPC's default event loops, as many as cores. It writes a warm-up, then the - * timed messages, to every group at once, with a window of outstanding async requests per group, until all are - * committed. All three servers run in this JVM and keep the log in memory, so the heap needs about - * 3 x groups x (messagesPerGroup + warmupPerGroup) x messageKB, in a fresh JVM per arm. - *

- * It reports the replication rate, the hottest event-loop threads during the timed writes (CPU time / wall time; - * cwI is a client worker, i.e. outgoing connections, swI a server worker), how many were over half busy, and the - * cores used by all event loops and by the whole JVM. - */ -public class GrpcLogAppenderConnectionsBenchmark extends BaseTest { - private static final int GROUPS = Integer.getInteger("bench.groups", 12); - private static final int MESSAGE_KB = Integer.getInteger("bench.messageKB", 1024); - private static final int MESSAGES = Integer.getInteger("bench.messagesPerGroup", 50); - private static final int WARMUP = Integer.getInteger("bench.warmupPerGroup", 20); - private static final int WORKERS = Integer.getInteger("bench.workers", 8); - // 0 keeps Ratis's default (raft.grpc.flow.control.window, 1MB); Ozone sets 5MB for its 4MB chunks - private static final int FLOW_CONTROL_WINDOW_KB = Integer.getInteger("bench.flowControlWindowKB", 0); - private static final boolean LEADER_ON_ONE = Boolean.parseBoolean(System.getProperty("bench.leaderOnOne", "true")); - private static final int WINDOW = Integer.getInteger("bench.window", 8); - private static final int ROUNDS = Integer.getInteger("bench.rounds", 1); - private static final String CONNECTIONS = System.getProperty("bench.connections", "1,4"); - private static final boolean TLS = Boolean.parseBoolean(System.getProperty("bench.tls", "true")); - - static { - // Netty's default (SIMPLE) leak detection samples buffers and records a stack trace for each; not what is measured - ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.DISABLED); - } - - @Test - @Timeout(value = 60, unit = TimeUnit.MINUTES) // BaseTest's 100 s is for unit tests - public void benchmark() throws Exception { - final String[] ks = CONNECTIONS.split(","); - System.out.printf("%nRATIS-2787: %d groups x %d messages x %d KB (warm-up %d), window %d, TLS %s, workers %d," - + " leader on one server %s, flow control window %s%n", GROUPS, MESSAGES, MESSAGE_KB, WARMUP, WINDOW, TLS, - WORKERS, LEADER_ON_ONE, FLOW_CONTROL_WINDOW_KB > 0 ? FLOW_CONTROL_WINDOW_KB + "KB" : "default"); - System.out.printf("%-6s %-3s %10s %12s %40s %10s %10s %10s%n", - "round", "K", "seconds", "MB/s", "hottest loops (%CPU)", "loops>50%", "loop cores", "JVM cores"); - for (int round = 1; round <= ROUNDS; round++) { - for (int i = 0; i < ks.length; i++) { - // rotate the order, so no K always runs first - final int k = Integer.parseInt(ks[(i + round) % ks.length].trim()); - runArm(round, k); - System.gc(); - } - } - } - - private static Parameters newTlsParameters() throws Exception { - final GrpcTlsConfig server = new GrpcTlsConfig( - SecurityTestUtils.getKeyManager(SecurityTestUtils::getServerKeyStore), - SecurityTestUtils.getTrustManager(SecurityTestUtils::getTrustStore), true); - final GrpcTlsConfig client = new GrpcTlsConfig( - SecurityTestUtils.getKeyManager(SecurityTestUtils::getClientKeyStore), - SecurityTestUtils.getTrustManager(SecurityTestUtils::getTrustStore), true); - final Parameters parameters = new Parameters(); - GrpcConfigKeys.Server.setTlsConf(parameters, server); - GrpcConfigKeys.Admin.setTlsConf(parameters, client); - GrpcConfigKeys.Client.setTlsConf(parameters, client); - return parameters; - } - - private void runArm(int round, int connections) throws Exception { - final RaftProperties properties = new RaftProperties(); - RaftServerConfigKeys.Log.setUseMemory(properties, true); - GrpcConfigKeys.Server.setLogAppenderConnections(properties, connections); - if (FLOW_CONTROL_WINDOW_KB > 0) { - GrpcConfigKeys.setFlowControlWindow(properties, SizeInBytes.valueOf(FLOW_CONTROL_WINDOW_KB + "KB")); - } - if (WORKERS > 0) { - // each server gets its own event loops, as in separate processes, instead of sharing gRPC's default ones - GrpcConfigKeys.Server.setWorkerGroupSize(properties, WORKERS); - GrpcConfigKeys.Client.setWorkerGroupSize(properties, WORKERS); - } - final MiniRaftClusterWithGrpc cluster = new MiniRaftClusterWithGrpc(MiniRaftCluster.generateIds(3, 0), - new String[0], properties, TLS ? newTlsParameters() : null); - try { - final long setupStart = System.nanoTime(); - cluster.start(); - RaftTestUtil.waitForLeader(cluster); - final List peers = new ArrayList<>(cluster.getGroup().getPeers()); - final RaftPeerId leaderId = peers.get(0).getId(); - - // the groups, each on all three servers, so they share the connections between each pair of servers - final List groupIds = new ArrayList<>(); - final List clients = new ArrayList<>(); - try (RaftClient admin = cluster.createClient(cluster.getGroup())) { - for (int g = 0; g < GROUPS; g++) { - final RaftGroup group = RaftGroup.valueOf(RaftGroupId.randomId(), peers); - for (RaftPeer peer : peers) { - Assertions.assertTrue(admin.getGroupManagementApi(peer.getId()).add(group).isSuccess()); - } - groupIds.add(group.getGroupId()); - clients.add(cluster.createClient(group)); - } - } - // wait for a leader in every group - for (RaftClient client : clients) { - Assertions.assertTrue(client.io().send(Message.valueOf("first")).isSuccess()); - } - if (LEADER_ON_ONE) { - // all groups led by one server, so all replication goes over its connections to the two others - for (int g = 0; g < GROUPS; g++) { - if (!cluster.getDivision(leaderId, groupIds.get(g)).getInfo().isLeader()) { - Assertions.assertTrue(clients.get(g).admin().transferLeadership(leaderId, 30_000).isSuccess()); - } - } - for (int g = 0; g < GROUPS; g++) { - Assertions.assertTrue(cluster.getDivision(leaderId, groupIds.get(g)).getInfo().isLeader()); - } - } - - final byte[] bytes = new byte[MESSAGE_KB << 10]; - new Random(0).nextBytes(bytes); - final Message message = Message.valueOf(ByteString.copyFrom(bytes)); - // warm-up, so the JIT is mostly done before the timed writes - writeAll(clients, message, WARMUP); - System.out.printf(" K=%d: cluster and %d groups ready and warmed up in %.1f s%n", connections, GROUPS, - (System.nanoTime() - setupStart) / 1e9); - - final Map cpuBefore = loopCpuTimes(); - final long processCpuBefore = processCpuTime(); - final long startMillis = System.currentTimeMillis(); - final long start = System.nanoTime(); - writeAll(clients, message, MESSAGES); - final long elapsedNanos = System.nanoTime() - start; - final double jvmCores = (double) (processCpuTime() - processCpuBefore) / elapsedNanos; - final Map cpuAfter = loopCpuTimes(); - - final List> busy = new ArrayList<>(); - double loopCores = 0; - final ThreadMXBean threads = ManagementFactory.getThreadMXBean(); - for (Map.Entry e : cpuAfter.entrySet()) { - final long cpu = e.getValue() - cpuBefore.getOrDefault(e.getKey(), 0L); - final ThreadInfo info = threads.getThreadInfo(e.getKey()); - final String name = info == null ? "?" : shortName(info.getThreadName()); - busy.add(new AbstractMap.SimpleImmutableEntry<>(name, 100.0 * cpu / elapsedNanos)); - loopCores += (double) cpu / elapsedNanos; - } - busy.sort((a, b) -> Double.compare(b.getValue(), a.getValue())); - final StringBuilder hottest = new StringBuilder(); - for (int i = 0; i < Math.min(3, busy.size()); i++) { - hottest.append(i == 0 ? "" : " ").append(busy.get(i).getKey()).append('=') - .append(Math.round(busy.get(i).getValue())); - } - final long overHalf = busy.stream().filter(b -> b.getValue() > 50).count(); - final double seconds = elapsedNanos / 1e9; - final double mbPerSecond = (double) GROUPS * MESSAGES * MESSAGE_KB / 1024 / seconds; - System.out.printf("%-6d %-3d %10.2f %12.1f %40s %10d %10.1f %10.1f%n", round, connections, seconds, - mbPerSecond, hottest, overHalf, loopCores, jvmCores); - System.out.printf(" K=%d: writes from %d to %d ms (epoch)%n", connections, startMillis, - System.currentTimeMillis()); - for (RaftClient client : clients) { - client.close(); - } - } finally { - cluster.shutdown(); - } - } - - /** Writes the message count times to every group at once, with a window of outstanding requests per group. */ - private static void writeAll(List clients, Message message, int count) throws Exception { - final AtomicInteger failures = new AtomicInteger(); - final List> replies = new ArrayList<>(); - final List windows = new ArrayList<>(); - for (int g = 0; g < clients.size(); g++) { - windows.add(new Semaphore(WINDOW)); - } - for (int m = 0; m < count; m++) { - for (int g = 0; g < clients.size(); g++) { - final Semaphore window = windows.get(g); - window.acquire(); - replies.add(clients.get(g).async().send(message).whenComplete((reply, e) -> { - window.release(); - if (e != null || !reply.isSuccess()) { - failures.incrementAndGet(); - } - })); - } - } - CompletableFuture.allOf(replies.toArray(new CompletableFuture[0])).get(30, TimeUnit.MINUTES); - Assertions.assertEquals(0, failures.get(), "failed writes"); - } - - /** - * E.g. ...-client-workers--thread3 to cw3, ...-server-workers--thread3 to sw3, grpc-nio-worker-ELG-3-13 to elg13. - * The groups are named before the server knows its id, so the names don't tell the servers apart. - */ - private static String shortName(String name) { - final String id = name.replaceAll("^.*?(\\d+)$", "$1"); - if (name.contains("-client-workers")) { - return "cw" + id; - } else if (name.contains("-server-workers")) { - return "sw" + id; - } - return "elg" + id; - } - - private static long processCpuTime() { - return ((com.sun.management.OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean()) - .getProcessCpuTime(); - } - - /** CPU time of the Netty event-loop threads (gRPC's and Ratis's), by thread id. */ - private static Map loopCpuTimes() { - final ThreadMXBean threads = ManagementFactory.getThreadMXBean(); - final Map times = new HashMap<>(); - for (ThreadInfo info : threads.dumpAllThreads(false, false)) { - final String name = info.getThreadName(); - if (name.contains("grpc-default-worker") || name.contains("ELG") || name.contains("nioEventLoop") - || name.contains("epoll") || name.contains("-client-workers") || name.contains("-server-workers")) { - final long cpu = threads.getThreadCpuTime(info.getThreadId()); - if (cpu > 0) { - times.put(info.getThreadId(), cpu); - } - } - } - return times; - } -} From d421fe6d47700d906c506b0def91e00a9dac1456 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 4 Nov 2026 02:10:33 -0500 Subject: [PATCH 4/5] Addressed review comments --- .../java/org/apache/ratis/grpc/GrpcUtil.java | 38 +++++++ .../ratis/grpc/server/GrpcLogAppender.java | 13 ++- .../grpc/server/GrpcServerProtocolClient.java | 52 +++++++-- .../TestGrpcLogAppenderConnections.java | 100 +++++++++++++++--- .../TestGrpcServerProtocolClientChannels.java | 37 ++++++- 5 files changed, 212 insertions(+), 28 deletions(-) diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcUtil.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcUtil.java index 49ccca6b35..bdb629c773 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcUtil.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcUtil.java @@ -50,10 +50,13 @@ import javax.net.ssl.TrustManager; import java.io.IOException; import java.security.Provider; +import java.util.Collection; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.function.Supplier; +import java.util.stream.Collectors; import static org.apache.ratis.thirdparty.io.netty.handler.ssl.SslProvider.OPENSSL; import static org.apache.ratis.util.NettyUtils.getJsseProvider; @@ -279,6 +282,41 @@ static void shutdownManagedChannel(ManagedChannel managedChannel, } } + /** + * Shut down the given channels as {@link #shutdownManagedChannel(ManagedChannel)} does, but in parallel: + * all of them are shut down before any is awaited, so the total wait is bounded by one graceful and one forceful + * timeout instead of one of each per channel. + */ + static void shutdownManagedChannels(Collection channels) { + // Close the gRPC managed-channels if not shut down already. + final List open = channels.stream().filter(c -> !c.isShutdown()).collect(Collectors.toList()); + open.forEach(ManagedChannel::shutdown); + awaitTermination(open, GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_GRACEFUL_TIMEOUT_DEFAULT, "gracefully"); + + // Forceful shut down if still not terminated. + final List remaining = channels.stream().filter(c -> !c.isTerminated()) + .collect(Collectors.toList()); + remaining.forEach(ManagedChannel::shutdownNow); + awaitTermination(remaining, GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_FORCEFUL_TIMEOUT_DEFAULT, "forcefully"); + } + + /** Wait for the given channels to terminate, up to the given timeout for all of them together. */ + static void awaitTermination(List channels, TimeDuration timeout, String mode) { + final long deadline = System.nanoTime() + timeout.toLong(TimeUnit.NANOSECONDS); + for (ManagedChannel channel : channels) { + try { + if (!channel.awaitTermination(Math.max(0, deadline - System.nanoTime()), TimeUnit.NANOSECONDS)) { + LOG.warn("Timed out {} shutting down connection: {}. ", mode, channel); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } catch (Exception e) { + LOG.error("Unexpected exception while waiting for channel termination", e); + } + } + } + static SslContextBuilder initSslContextBuilderForServer(KeyManagerConf keyManagerConfig) { final KeyManager keyManager = keyManagerConfig.getKeyManager(); if (keyManager != null) { diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java index 851f4c11c3..b4ee888629 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java @@ -164,6 +164,8 @@ synchronized int process(Event event) { private final TimeoutExecutor scheduler = TimeoutExecutor.getInstance(); @SuppressWarnings({"squid:S3077"}) // Suppress volatile for generic type private volatile StreamObservers appendLogRequestObserver; + // the client whose appendEntries channel this group uses, to release the channel when stopped + private volatile GrpcServerProtocolClient appendEntriesClient; private final boolean useSeparateHBChannel; private final GrpcServerMetrics grpcServerMetrics; @@ -358,7 +360,13 @@ private void sleepForErrors() throws InterruptedException { @Override public CompletableFuture stopAsync() { grpcServerMetrics.unregister(); - return super.stopAsync(); + return super.stopAsync().whenComplete((state, e) -> { + // when CLOSED, no more appendEntries for this group to the follower; otherwise, a restarted appender takes over + final GrpcServerProtocolClient client = appendEntriesClient; + if (state == LifeCycle.State.CLOSED && client != null) { + client.releaseAppendEntriesIndex(getServer().getMemberId().getGroupId()); + } + }); } @Override @@ -449,7 +457,8 @@ void appendLog(boolean heartbeat) throws IOException { notifyAppendEntriesListener(l -> l.onRequest(pending)); increaseNextIndex(pending); if (appendLogRequestObserver == null) { - appendLogRequestObserver = new StreamObservers(getClient(), getServer().getMemberId().getGroupId(), + appendEntriesClient = getClient(); + appendLogRequestObserver = new StreamObservers(appendEntriesClient, getServer().getMemberId().getGroupId(), new AppendLogResponseHandler(), useSeparateHBChannel, getWaitTimeMin()); } } diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java index 890de6ff8a..238a1c9116 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java @@ -43,7 +43,9 @@ import java.io.Closeable; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; /** * This is a RaftClient implementation that supports streaming data to the raft @@ -60,6 +62,9 @@ class GrpcServerProtocolClient implements Closeable { // Channels and stubs for appendEntries, each group always using the same one; the first is the main channel private final List appendEntriesChannels; private final List appendEntriesStubs; + // The appendEntries channel of each group, and the number of groups on each channel; guarded by this + private final Map appendEntriesIndices = new HashMap<>(); + private final int[] appendEntriesGroupCounts; private final RaftServerProtocolServiceBlockingStub blockingStub; private final boolean useSeparateHBChannel; @@ -91,6 +96,7 @@ class GrpcServerProtocolClient implements Closeable { } appendEntriesChannels = Collections.unmodifiableList(channels); appendEntriesStubs = Collections.unmodifiableList(stubs); + appendEntriesGroupCounts = new int[appendEntriesConnections]; if (useSeparateHBChannel) { hbChannel = buildChannel(target, flowControlWindow, sslContext); hbAsyncStub = RaftServerProtocolServiceGrpc.newStub(hbChannel); @@ -126,11 +132,12 @@ private ManagedChannel buildChannel(RaftPeer target, int flowControlWindow, SslC @Override public void close() { LOG.info("{} Close channels", raftPeerId); + // the first one is the main channel + final List channels = new ArrayList<>(appendEntriesChannels); if (useSeparateHBChannel) { - GrpcUtil.shutdownManagedChannel(hbChannel); + channels.add(hbChannel); } - // the first one is the main channel - appendEntriesChannels.forEach(GrpcUtil::shutdownManagedChannel); + GrpcUtil.shutdownManagedChannels(channels); if (pool != null) { pool.close(); } @@ -202,9 +209,42 @@ CallStreamObserver appendEntries(RaftGroupId groupId, } } - /** @return the index of the appendEntries channel for the given group, always the same one for a group. */ - int getAppendEntriesIndex(RaftGroupId groupId) { - return Math.floorMod(groupId.hashCode(), appendEntriesStubs.size()); + /** + * @return the index of the appendEntries channel for the given group, always the same one for a group + * until {@link #releaseAppendEntriesIndex(RaftGroupId)}. + * A new group gets the channel with the fewest groups, so that the groups are spread evenly. + */ + synchronized int getAppendEntriesIndex(RaftGroupId groupId) { + if (appendEntriesGroupCounts.length == 1) { + return 0; + } + return appendEntriesIndices.computeIfAbsent(groupId, g -> { + int least = 0; + for (int i = 1; i < appendEntriesGroupCounts.length; i++) { + if (appendEntriesGroupCounts[i] < appendEntriesGroupCounts[least]) { + least = i; + } + } + appendEntriesGroupCounts[least]++; + return least; + }); + } + + /** + * The given group no longer sends appendEntries to this peer, so its channel can be given to another group. + * Any channel is correct for any group, so a race with {@link #getAppendEntriesIndex(RaftGroupId)} only affects + * how evenly the groups are spread. + */ + synchronized void releaseAppendEntriesIndex(RaftGroupId groupId) { + final Integer index = appendEntriesIndices.remove(groupId); + if (index != null) { + appendEntriesGroupCounts[index]--; + } + } + + @VisibleForTesting + synchronized Map getAppendEntriesIndices() { + return new HashMap<>(appendEntriesIndices); } @VisibleForTesting diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java index e7f23000a4..5474baac4f 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java @@ -20,46 +20,114 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; import org.apache.ratis.BaseTest; +import org.apache.ratis.RaftTestUtil; import org.apache.ratis.client.RaftClient; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.grpc.GrpcConfigKeys; import org.apache.ratis.grpc.MiniRaftClusterWithGrpc; import org.apache.ratis.protocol.RaftGroup; +import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeer; -import org.apache.ratis.server.impl.GroupManagementBaseTest; import org.apache.ratis.server.impl.MiniRaftCluster; import org.apache.ratis.server.RaftServer; import org.apache.ratis.RaftTestUtil.SimpleMessage; -import org.apache.ratis.util.function.CheckedBiConsumer; +import org.apache.ratis.util.JavaUtils; +import org.apache.ratis.util.TimeDuration; import org.junit.jupiter.api.Test; /** Several groups on the same servers replicate over more than one appendEntries connection per peer. */ public class TestGrpcLogAppenderConnections extends BaseTest { private static final int CONNECTIONS = 3; + private static final int NEW_GROUPS = 5; @Test - public void testMultiGroupWithLogAppenderConnections() throws Exception { + public void testGroupsOnSameServersShareLogAppenderConnections() throws Exception { final RaftProperties properties = new RaftProperties(); GrpcConfigKeys.Server.setLogAppenderConnections(properties, CONNECTIONS); - final MiniRaftCluster cluster = MiniRaftClusterWithGrpc.FACTORY.newCluster(0, properties); + final MiniRaftCluster cluster = MiniRaftClusterWithGrpc.FACTORY.newCluster(3, properties); + try { + cluster.start(); + RaftTestUtil.waitForLeader(cluster); + final Collection peers = cluster.getGroup().getPeers(); - final CheckedBiConsumer checker = (c, group) -> { - for (RaftPeer peer : group.getPeers()) { - final RaftServer server = c.getServer(peer.getId()); - final GrpcServicesImpl rpc = (GrpcServicesImpl) server.getServerRpc(); - for (RaftPeer other : group.getPeers()) { - if (!other.getId().equals(peer.getId())) { - assertEquals(CONNECTIONS, rpc.getProxies().getProxy(other.getId()).getAppendEntriesChannels().size()); + // add more groups on the same three servers + final List groups = new ArrayList<>(); + groups.add(cluster.getGroup()); + for (int i = 0; i < NEW_GROUPS; i++) { + final RaftGroup group = RaftGroup.valueOf(RaftGroupId.randomId(), peers); + for (RaftPeer peer : peers) { + try (RaftClient client = cluster.createClient(peer.getId(), group)) { + assertTrue(client.getGroupManagementApi(peer.getId()).add(group).isSuccess()); } } + groups.add(group); } - try (RaftClient client = c.createClient(group)) { - for (int i = 0; i < 10; i++) { - assertTrue(client.io().send(new SimpleMessage(group.getGroupId() + "-" + i)).isSuccess()); + + for (RaftGroup group : groups) { + RaftTestUtil.waitForLeader(cluster, group.getGroupId()); + try (RaftClient client = cluster.createClient(group)) { + for (int i = 0; i < 10; i++) { + assertTrue(client.io().send(new SimpleMessage(group.getGroupId() + "-" + i)).isSuccess()); + } + } + } + JavaUtils.attempt(() -> assertConnections(cluster, groups), 10, TimeDuration.ONE_SECOND, + "assertConnections", LOG); + + // a removed group releases its connections + final List removed = groups.subList(1, groups.size()); + for (RaftGroup group : removed) { + for (RaftPeer peer : peers) { + try (RaftClient client = cluster.createClient(peer.getId(), group)) { + assertTrue(client.getGroupManagementApi(peer.getId()).remove(group.getGroupId(), true, false) + .isSuccess()); + } + } + } + final List remaining = groups.subList(0, 1); + JavaUtils.attempt(() -> assertConnections(cluster, remaining), 10, TimeDuration.ONE_SECOND, + "assertConnections after remove", LOG); + } finally { + cluster.shutdown(); + } + } + + /** + * On each server, the groups it leads use the appendEntries connections to each follower, + * spread evenly over them; no other group uses them. + */ + private static void assertConnections(MiniRaftCluster cluster, List groups) throws Exception { + for (RaftServer server : cluster.getServers()) { + final Set led = new HashSet<>(); + for (RaftGroup group : groups) { + if (cluster.getDivision(server.getId(), group.getGroupId()).getInfo().isLeader()) { + led.add(group.getGroupId()); + } + } + final GrpcServicesImpl rpc = (GrpcServicesImpl) server.getServerRpc(); + for (RaftPeer follower : groups.get(0).getPeers()) { + if (follower.getId().equals(server.getId())) { + continue; + } + final GrpcServerProtocolClient proxy = rpc.getProxies().getProxy(follower.getId()); + assertEquals(CONNECTIONS, proxy.getAppendEntriesChannels().size()); + final Map indices = proxy.getAppendEntriesIndices(); + assertEquals(led, indices.keySet(), server.getId() + " -> " + follower.getId()); + + final int[] groupCounts = new int[CONNECTIONS]; + indices.values().forEach(i -> groupCounts[i]++); + for (int count : groupCounts) { + assertTrue(Math.abs(count * CONNECTIONS - led.size()) < CONNECTIONS, + () -> server.getId() + " -> " + follower.getId() + ": " + indices); } } - }; - GroupManagementBaseTest.runMultiGroupTest(cluster, new int[] {3, 6, 9}, -1, checker); + } } } diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java index f5b78b5ed1..ee8aff9bec 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java @@ -23,7 +23,6 @@ import java.util.HashSet; import java.util.List; -import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.ratis.BaseTest; import org.apache.ratis.protocol.RaftGroupId; @@ -51,6 +50,7 @@ public void testOneConnectionUsesMainChannel() { for (int i = 0; i < 10; i++) { assertEquals(0, client.getAppendEntriesIndex(RaftGroupId.randomId())); } + assertTrue(client.getAppendEntriesIndices().isEmpty()); } finally { client.close(); } @@ -66,19 +66,48 @@ public void testGroupsSpreadOverConnections() { assertSame(client.getMainChannel(), channels.get(0)); assertEquals(connections, new HashSet<>(channels).size()); - final Set used = new HashSet<>(); + final int[] groupCounts = new int[connections]; for (int i = 0; i < 100; i++) { final RaftGroupId groupId = RaftGroupId.randomId(); final int index = client.getAppendEntriesIndex(groupId); assertTrue(index >= 0 && index < connections, "index " + index); // a group always uses the same connection, which keeps its appendEntries in order assertEquals(index, client.getAppendEntriesIndex(RaftGroupId.valueOf(groupId.getUuid()))); - used.add(index); + groupCounts[index]++; + } + // each new group goes to the connection with the fewest groups + for (int count : groupCounts) { + assertEquals(100 / connections, count); } - assertEquals(connections, used.size()); } finally { client.close(); } channels.forEach(c -> assertTrue(c.isShutdown(), c + " is not shut down")); } + + @Test + public void testReleasedConnectionIsReused() { + final int connections = 3; + final GrpcServerProtocolClient client = newClient(connections); + try { + final RaftGroupId[] groups = new RaftGroupId[2 * connections]; + for (int i = 0; i < groups.length; i++) { + groups[i] = RaftGroupId.randomId(); + assertEquals(i % connections, client.getAppendEntriesIndex(groups[i])); + } + + // the connection of a released group goes to the next new group + client.releaseAppendEntriesIndex(groups[4]); + assertEquals(groups.length - 1, client.getAppendEntriesIndices().size()); + assertEquals(1, client.getAppendEntriesIndex(RaftGroupId.randomId())); + + // releasing a group again, or a group without a connection, changes nothing + client.releaseAppendEntriesIndex(groups[4]); + client.releaseAppendEntriesIndex(RaftGroupId.randomId()); + assertEquals(0, client.getAppendEntriesIndex(RaftGroupId.randomId())); + assertEquals(2, client.getAppendEntriesIndex(groups[2])); + } finally { + client.close(); + } + } } From 0776bb2c31f49ec21ad3433051cb70ec410d4aca Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 4 Nov 2026 12:45:13 -0500 Subject: [PATCH 5/5] simplified round robin implementation --- .../ratis/grpc/server/GrpcLogAppender.java | 34 ++++--- .../grpc/server/GrpcServerProtocolClient.java | 53 ++--------- .../server/TestGrpcLogAppenderCallbacks.java | 3 +- .../TestGrpcLogAppenderConnections.java | 92 ++++++++----------- .../TestGrpcServerProtocolClientChannels.java | 44 +-------- 5 files changed, 73 insertions(+), 153 deletions(-) diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java index b4ee888629..e5eb57d0e1 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java @@ -24,7 +24,6 @@ import org.apache.ratis.grpc.metrics.GrpcServerMetrics; import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.proto.RaftProtos.InstallSnapshotResult; -import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.retry.RetryPolicy; import org.apache.ratis.server.RaftServer; @@ -35,6 +34,7 @@ import org.apache.ratis.server.protocol.TermIndex; import org.apache.ratis.server.raftlog.RaftLog; import org.apache.ratis.server.util.ServerStringUtils; +import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting; import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException; import org.apache.ratis.thirdparty.io.grpc.stub.CallStreamObserver; import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; @@ -164,8 +164,8 @@ synchronized int process(Event event) { private final TimeoutExecutor scheduler = TimeoutExecutor.getInstance(); @SuppressWarnings({"squid:S3077"}) // Suppress volatile for generic type private volatile StreamObservers appendLogRequestObserver; - // the client whose appendEntries channel this group uses, to release the channel when stopped - private volatile GrpcServerProtocolClient appendEntriesClient; + // the appendEntries channel of this appender, taken when it creates its first stream; guarded by the write lock + private int appendEntriesIndex = -1; private final boolean useSeparateHBChannel; private final GrpcServerMetrics grpcServerMetrics; @@ -360,13 +360,7 @@ private void sleepForErrors() throws InterruptedException { @Override public CompletableFuture stopAsync() { grpcServerMetrics.unregister(); - return super.stopAsync().whenComplete((state, e) -> { - // when CLOSED, no more appendEntries for this group to the follower; otherwise, a restarted appender takes over - final GrpcServerProtocolClient client = appendEntriesClient; - if (state == LifeCycle.State.CLOSED && client != null) { - client.releaseAppendEntriesIndex(getServer().getMemberId().getGroupId()); - } - }); + return super.stopAsync(); } @Override @@ -398,10 +392,10 @@ static class StreamObservers { private final TimeDuration waitForReady; private volatile boolean running = true; - StreamObservers(GrpcServerProtocolClient client, RaftGroupId groupId, AppendLogResponseHandler handler, + StreamObservers(GrpcServerProtocolClient client, int appendEntriesIndex, AppendLogResponseHandler handler, boolean separateHeartbeat, TimeDuration waitTimeMin) { - this.appendLog = client.appendEntries(groupId, handler, false); - this.heartbeat = separateHeartbeat? client.appendEntries(groupId, handler, true): null; + this.appendLog = client.appendEntries(appendEntriesIndex, handler, false); + this.heartbeat = separateHeartbeat? client.appendEntries(appendEntriesIndex, handler, true): null; this.waitForReady = waitTimeMin.isPositive()? waitTimeMin: TimeDuration.ONE_MILLISECOND; } @@ -431,6 +425,13 @@ void onCompleted() { } } + @VisibleForTesting + int getAppendEntriesIndex() { + try (AutoCloseableLock readLock = lock.readLock(caller, LOG::trace)) { + return appendEntriesIndex; + } + } + @Override public long getCallId() { return callId.get(); @@ -457,8 +458,11 @@ void appendLog(boolean heartbeat) throws IOException { notifyAppendEntriesListener(l -> l.onRequest(pending)); increaseNextIndex(pending); if (appendLogRequestObserver == null) { - appendEntriesClient = getClient(); - appendLogRequestObserver = new StreamObservers(appendEntriesClient, getServer().getMemberId().getGroupId(), + final GrpcServerProtocolClient client = getClient(); + if (appendEntriesIndex < 0) { + appendEntriesIndex = client.nextAppendEntriesIndex(); + } + appendLogRequestObserver = new StreamObservers(client, appendEntriesIndex, new AppendLogResponseHandler(), useSeparateHBChannel, getWaitTimeMin()); } } diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java index 238a1c9116..915cb30316 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java @@ -20,7 +20,6 @@ import org.apache.ratis.grpc.GrpcUtil; import org.apache.ratis.util.NettyUtils; import org.apache.ratis.grpc.util.StreamObserverWithTimeout; -import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.server.util.ServerStringUtils; import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting; @@ -43,9 +42,8 @@ import java.io.Closeable; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.List; -import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; /** * This is a RaftClient implementation that supports streaming data to the raft @@ -62,9 +60,8 @@ class GrpcServerProtocolClient implements Closeable { // Channels and stubs for appendEntries, each group always using the same one; the first is the main channel private final List appendEntriesChannels; private final List appendEntriesStubs; - // The appendEntries channel of each group, and the number of groups on each channel; guarded by this - private final Map appendEntriesIndices = new HashMap<>(); - private final int[] appendEntriesGroupCounts; + // The appendEntries channel for the next new log appender, so that the appenders take the channels in turn + private final AtomicInteger nextAppendEntriesIndex = new AtomicInteger(); private final RaftServerProtocolServiceBlockingStub blockingStub; private final boolean useSeparateHBChannel; @@ -96,7 +93,6 @@ class GrpcServerProtocolClient implements Closeable { } appendEntriesChannels = Collections.unmodifiableList(channels); appendEntriesStubs = Collections.unmodifiableList(stubs); - appendEntriesGroupCounts = new int[appendEntriesConnections]; if (useSeparateHBChannel) { hbChannel = buildChannel(target, flowControlWindow, sslContext); hbAsyncStub = RaftServerProtocolServiceGrpc.newStub(hbChannel); @@ -199,52 +195,23 @@ public void onCompleted() { } } - CallStreamObserver appendEntries(RaftGroupId groupId, + /** @param appendEntriesIndex the appendEntries channel, from {@link #nextAppendEntriesIndex()} */ + CallStreamObserver appendEntries(int appendEntriesIndex, StreamObserver responseHandler, boolean isHeartbeat) { if (isHeartbeat && useSeparateHBChannel) { return (CallStreamObserver) hbAsyncStub.appendEntries(responseHandler); } else { - final RaftServerProtocolServiceStub stub = appendEntriesStubs.get(getAppendEntriesIndex(groupId)); + final RaftServerProtocolServiceStub stub = appendEntriesStubs.get(appendEntriesIndex); return (CallStreamObserver) stub.appendEntries(responseHandler); } } /** - * @return the index of the appendEntries channel for the given group, always the same one for a group - * until {@link #releaseAppendEntriesIndex(RaftGroupId)}. - * A new group gets the channel with the fewest groups, so that the groups are spread evenly. + * @return the index of the appendEntries channel for a new log appender. The appenders take the channels in turn; + * an appender keeps its channel, so the appendEntries of a group to this peer use one channel. */ - synchronized int getAppendEntriesIndex(RaftGroupId groupId) { - if (appendEntriesGroupCounts.length == 1) { - return 0; - } - return appendEntriesIndices.computeIfAbsent(groupId, g -> { - int least = 0; - for (int i = 1; i < appendEntriesGroupCounts.length; i++) { - if (appendEntriesGroupCounts[i] < appendEntriesGroupCounts[least]) { - least = i; - } - } - appendEntriesGroupCounts[least]++; - return least; - }); - } - - /** - * The given group no longer sends appendEntries to this peer, so its channel can be given to another group. - * Any channel is correct for any group, so a race with {@link #getAppendEntriesIndex(RaftGroupId)} only affects - * how evenly the groups are spread. - */ - synchronized void releaseAppendEntriesIndex(RaftGroupId groupId) { - final Integer index = appendEntriesIndices.remove(groupId); - if (index != null) { - appendEntriesGroupCounts[index]--; - } - } - - @VisibleForTesting - synchronized Map getAppendEntriesIndices() { - return new HashMap<>(appendEntriesIndices); + int nextAppendEntriesIndex() { + return Math.floorMod(nextAppendEntriesIndex.getAndIncrement(), appendEntriesStubs.size()); } @VisibleForTesting diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java index 45e5e1c5f7..3285893017 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java @@ -62,6 +62,7 @@ import java.util.concurrent.TimeUnit; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.RETURNS_DEEP_STUBS; @@ -188,7 +189,7 @@ public void testStreamCreationFailure(boolean failProxy) throws Exception { when(serverRpc.getProxies().getProxy(destination)).thenThrow(error); } else { error = new IllegalStateException("Failed to create append stream"); - when(client.appendEntries(any(), any(), eq(false))).thenThrow(error); + when(client.appendEntries(anyInt(), any(), eq(false))).thenThrow(error); } Assertions.assertSame(error, Assertions.assertThrows(Exception.class, () -> appender.appendLog(false))); final InOrder order = inOrder(appendEntries); diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java index 5474baac4f..bf817c37bb 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java @@ -21,11 +21,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.ArrayList; -import java.util.Collection; -import java.util.HashSet; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import org.apache.ratis.BaseTest; import org.apache.ratis.RaftTestUtil; import org.apache.ratis.client.RaftClient; @@ -35,8 +37,11 @@ import org.apache.ratis.protocol.RaftGroup; import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeer; -import org.apache.ratis.server.impl.MiniRaftCluster; +import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.server.RaftServer; +import org.apache.ratis.server.impl.MiniRaftCluster; +import org.apache.ratis.server.impl.RaftServerTestUtil; +import org.apache.ratis.server.leader.LogAppender; import org.apache.ratis.RaftTestUtil.SimpleMessage; import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.TimeDuration; @@ -45,7 +50,7 @@ /** Several groups on the same servers replicate over more than one appendEntries connection per peer. */ public class TestGrpcLogAppenderConnections extends BaseTest { private static final int CONNECTIONS = 3; - private static final int NEW_GROUPS = 5; + private static final int NEW_GROUPS = 2 * CONNECTIONS; @Test public void testGroupsOnSameServersShareLogAppenderConnections() throws Exception { @@ -55,14 +60,19 @@ public void testGroupsOnSameServersShareLogAppenderConnections() throws Exceptio try { cluster.start(); RaftTestUtil.waitForLeader(cluster); - final Collection peers = cluster.getGroup().getPeers(); + // the first server has the highest priority in the new groups, so that it becomes the leader of all of them + final List peers = new ArrayList<>(cluster.getGroup().getPeers()); + final RaftPeerId leaderId = peers.get(0).getId(); + final List prioritized = new ArrayList<>(); + for (RaftPeer peer : peers) { + prioritized.add(RaftPeer.newBuilder(peer).setPriority(peer.getId().equals(leaderId) ? 1 : 0).build()); + } // add more groups on the same three servers final List groups = new ArrayList<>(); - groups.add(cluster.getGroup()); for (int i = 0; i < NEW_GROUPS; i++) { - final RaftGroup group = RaftGroup.valueOf(RaftGroupId.randomId(), peers); - for (RaftPeer peer : peers) { + final RaftGroup group = RaftGroup.valueOf(RaftGroupId.randomId(), prioritized); + for (RaftPeer peer : prioritized) { try (RaftClient client = cluster.createClient(peer.getId(), group)) { assertTrue(client.getGroupManagementApi(peer.getId()).add(group).isSuccess()); } @@ -71,63 +81,39 @@ public void testGroupsOnSameServersShareLogAppenderConnections() throws Exceptio } for (RaftGroup group : groups) { - RaftTestUtil.waitForLeader(cluster, group.getGroupId()); + JavaUtils.attempt(() -> assertTrue(cluster.getDivision(leaderId, group.getGroupId()).getInfo().isLeader()), + 30, TimeDuration.ONE_SECOND, leaderId + " leads " + group.getGroupId(), LOG); try (RaftClient client = cluster.createClient(group)) { for (int i = 0; i < 10; i++) { assertTrue(client.io().send(new SimpleMessage(group.getGroupId() + "-" + i)).isSuccess()); } } } - JavaUtils.attempt(() -> assertConnections(cluster, groups), 10, TimeDuration.ONE_SECOND, - "assertConnections", LOG); - - // a removed group releases its connections - final List removed = groups.subList(1, groups.size()); - for (RaftGroup group : removed) { - for (RaftPeer peer : peers) { - try (RaftClient client = cluster.createClient(peer.getId(), group)) { - assertTrue(client.getGroupManagementApi(peer.getId()).remove(group.getGroupId(), true, false) - .isSuccess()); - } - } - } - final List remaining = groups.subList(0, 1); - JavaUtils.attempt(() -> assertConnections(cluster, remaining), 10, TimeDuration.ONE_SECOND, - "assertConnections after remove", LOG); + assertConnections(cluster, leaderId, groups); } finally { cluster.shutdown(); } } - /** - * On each server, the groups it leads use the appendEntries connections to each follower, - * spread evenly over them; no other group uses them. - */ - private static void assertConnections(MiniRaftCluster cluster, List groups) throws Exception { - for (RaftServer server : cluster.getServers()) { - final Set led = new HashSet<>(); - for (RaftGroup group : groups) { - if (cluster.getDivision(server.getId(), group.getGroupId()).getInfo().isLeader()) { - led.add(group.getGroupId()); - } - } - final GrpcServicesImpl rpc = (GrpcServicesImpl) server.getServerRpc(); - for (RaftPeer follower : groups.get(0).getPeers()) { - if (follower.getId().equals(server.getId())) { - continue; - } - final GrpcServerProtocolClient proxy = rpc.getProxies().getProxy(follower.getId()); - assertEquals(CONNECTIONS, proxy.getAppendEntriesChannels().size()); - final Map indices = proxy.getAppendEntriesIndices(); - assertEquals(led, indices.keySet(), server.getId() + " -> " + follower.getId()); - - final int[] groupCounts = new int[CONNECTIONS]; - indices.values().forEach(i -> groupCounts[i]++); - for (int count : groupCounts) { - assertTrue(Math.abs(count * CONNECTIONS - led.size()) < CONNECTIONS, - () -> server.getId() + " -> " + follower.getId() + ": " + indices); - } + /** The leader's appenders of all the groups use every appendEntries connection to each follower. */ + private static void assertConnections(MiniRaftCluster cluster, RaftPeerId leaderId, List groups) + throws Exception { + final RaftServer leader = cluster.getServer(leaderId); + final GrpcServicesImpl rpc = (GrpcServicesImpl) leader.getServerRpc(); + final Map> indices = new HashMap<>(); + for (RaftGroup group : groups) { + final List appenders = RaftServerTestUtil.getLogAppenders( + cluster.getDivision(leaderId, group.getGroupId())).collect(Collectors.toList()); + assertEquals(group.getPeers().size() - 1, appenders.size()); + for (LogAppender appender : appenders) { + final RaftPeerId follower = appender.getFollowerId(); + assertEquals(CONNECTIONS, rpc.getProxies().getProxy(follower).getAppendEntriesChannels().size()); + final int index = ((GrpcLogAppender) appender).getAppendEntriesIndex(); + indices.computeIfAbsent(follower, k -> new TreeSet<>()).add(index); } } + final Set all = IntStream.range(0, CONNECTIONS).boxed().collect(Collectors.toCollection(TreeSet::new)); + assertEquals(2, indices.size()); + indices.forEach((follower, used) -> assertEquals(all, used, leaderId + " -> " + follower)); } } diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java index ee8aff9bec..e863ca7792 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java @@ -25,7 +25,6 @@ import java.util.List; import java.util.concurrent.TimeUnit; import org.apache.ratis.BaseTest; -import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeer; import org.apache.ratis.thirdparty.io.grpc.ManagedChannel; import org.apache.ratis.util.TimeDuration; @@ -48,16 +47,15 @@ public void testOneConnectionUsesMainChannel() { assertEquals(1, channels.size()); assertSame(client.getMainChannel(), channels.get(0)); for (int i = 0; i < 10; i++) { - assertEquals(0, client.getAppendEntriesIndex(RaftGroupId.randomId())); + assertEquals(0, client.nextAppendEntriesIndex()); } - assertTrue(client.getAppendEntriesIndices().isEmpty()); } finally { client.close(); } } @Test - public void testGroupsSpreadOverConnections() { + public void testAppendersTakeConnectionsInTurn() { final int connections = 4; final GrpcServerProtocolClient client = newClient(connections); final List channels = client.getAppendEntriesChannels(); @@ -66,48 +64,12 @@ public void testGroupsSpreadOverConnections() { assertSame(client.getMainChannel(), channels.get(0)); assertEquals(connections, new HashSet<>(channels).size()); - final int[] groupCounts = new int[connections]; for (int i = 0; i < 100; i++) { - final RaftGroupId groupId = RaftGroupId.randomId(); - final int index = client.getAppendEntriesIndex(groupId); - assertTrue(index >= 0 && index < connections, "index " + index); - // a group always uses the same connection, which keeps its appendEntries in order - assertEquals(index, client.getAppendEntriesIndex(RaftGroupId.valueOf(groupId.getUuid()))); - groupCounts[index]++; - } - // each new group goes to the connection with the fewest groups - for (int count : groupCounts) { - assertEquals(100 / connections, count); + assertEquals(i % connections, client.nextAppendEntriesIndex()); } } finally { client.close(); } channels.forEach(c -> assertTrue(c.isShutdown(), c + " is not shut down")); } - - @Test - public void testReleasedConnectionIsReused() { - final int connections = 3; - final GrpcServerProtocolClient client = newClient(connections); - try { - final RaftGroupId[] groups = new RaftGroupId[2 * connections]; - for (int i = 0; i < groups.length; i++) { - groups[i] = RaftGroupId.randomId(); - assertEquals(i % connections, client.getAppendEntriesIndex(groups[i])); - } - - // the connection of a released group goes to the next new group - client.releaseAppendEntriesIndex(groups[4]); - assertEquals(groups.length - 1, client.getAppendEntriesIndices().size()); - assertEquals(1, client.getAppendEntriesIndex(RaftGroupId.randomId())); - - // releasing a group again, or a group without a connection, changes nothing - client.releaseAppendEntriesIndex(groups[4]); - client.releaseAppendEntriesIndex(RaftGroupId.randomId()); - assertEquals(0, client.getAppendEntriesIndex(RaftGroupId.randomId())); - assertEquals(2, client.getAppendEntriesIndex(groups[2])); - } finally { - client.close(); - } - } }