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/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 14b2d6405c..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 @@ -34,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; @@ -163,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 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; @@ -389,10 +392,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, int appendEntriesIndex, AppendLogResponseHandler handler, + boolean separateHeartbeat, TimeDuration waitTimeMin) { + this.appendLog = client.appendEntries(appendEntriesIndex, handler, false); + this.heartbeat = separateHeartbeat? client.appendEntries(appendEntriesIndex, handler, true): null; this.waitForReady = waitTimeMin.isPositive()? waitTimeMin: TimeDuration.ONE_MILLISECOND; } @@ -422,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(); @@ -448,8 +458,12 @@ void appendLog(boolean heartbeat) throws IOException { notifyAppendEntriesListener(l -> l.onRequest(pending)); increaseNextIndex(pending); if (appendLogRequestObserver == null) { - appendLogRequestObserver = new StreamObservers( - getClient(), new AppendLogResponseHandler(), useSeparateHBChannel, getWaitTimeMin()); + 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 6dc0d32527..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 @@ -22,6 +22,7 @@ import org.apache.ratis.grpc.util.StreamObserverWithTimeout; 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 +40,10 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; /** * This is a RaftClient implementation that supports streaming data to the raft @@ -52,6 +57,11 @@ 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; + // 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; @@ -61,7 +71,8 @@ class GrpcServerProtocolClient implements Closeable { private final RaftPeerId raftPeerId; private final EventLoopGroup eventLoopGroup; - GrpcServerProtocolClient(RaftPeer target, int connections, int flowControlWindow, + @SuppressWarnings("checkstyle:ParameterNumber") + GrpcServerProtocolClient(RaftPeer target, int connections, int appendEntriesConnections, int flowControlWindow, TimeDuration requestTimeout, SslContext sslContext, boolean separateHBChannel, EventLoopGroup eventLoopGroup) { raftPeerId = target.getId(); @@ -71,6 +82,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); @@ -106,10 +128,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); } - GrpcUtil.shutdownManagedChannel(channel); + GrpcUtil.shutdownManagedChannels(channels); if (pool != null) { pool.close(); } @@ -171,15 +195,35 @@ public void onCompleted() { } } - CallStreamObserver appendEntries( + /** @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 { - return (CallStreamObserver) asyncStub.appendEntries(responseHandler); + final RaftServerProtocolServiceStub stub = appendEntriesStubs.get(appendEntriesIndex); + return (CallStreamObserver) stub.appendEntries(responseHandler); } } + /** + * @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. + */ + int nextAppendEntriesIndex() { + return Math.floorMod(nextAppendEntriesIndex.getAndIncrement(), 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 +235,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/TestGrpcLogAppenderCallbacks.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java index 79ae4853bf..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(), 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 new file mode 100644 index 0000000000..bf817c37bb --- /dev/null +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java @@ -0,0 +1,119 @@ +/** + * 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 java.util.ArrayList; +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; +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.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; +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 = 2 * CONNECTIONS; + + @Test + public void testGroupsOnSameServersShareLogAppenderConnections() throws Exception { + final RaftProperties properties = new RaftProperties(); + GrpcConfigKeys.Server.setLogAppenderConnections(properties, CONNECTIONS); + final MiniRaftCluster cluster = MiniRaftClusterWithGrpc.FACTORY.newCluster(3, properties); + try { + cluster.start(); + RaftTestUtil.waitForLeader(cluster); + // 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<>(); + for (int i = 0; i < NEW_GROUPS; i++) { + 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()); + } + } + groups.add(group); + } + + for (RaftGroup group : groups) { + 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()); + } + } + } + assertConnections(cluster, leaderId, groups); + } finally { + cluster.shutdown(); + } + } + + /** 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 new file mode 100644 index 0000000000..e863ca7792 --- /dev/null +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java @@ -0,0 +1,75 @@ +/** + * 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.concurrent.TimeUnit; +import org.apache.ratis.BaseTest; +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.nextAppendEntriesIndex()); + } + } finally { + client.close(); + } + } + + @Test + public void testAppendersTakeConnectionsInTurn() { + 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()); + + for (int i = 0; i < 100; i++) { + assertEquals(i % connections, client.nextAppendEntriesIndex()); + } + } finally { + client.close(); + } + channels.forEach(c -> assertTrue(c.isShutdown(), c + " is not shut down")); + } +}