RATIS-2787. Spread appendEntries between two servers over more than one connection

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) {

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<ManagedChannel> channels) {
+    // Close the gRPC managed-channels if not shut down already.
+    final List<ManagedChannel> 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<ManagedChannel> 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<ManagedChannel> 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) {

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());
         }
       }
 

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<ManagedChannel> appendEntriesChannels;
+  private final List<RaftServerProtocolServiceStub> 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<ManagedChannel> channels = new ArrayList<>(appendEntriesConnections);
+    final List<RaftServerProtocolServiceStub> 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<ManagedChannel> 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<AppendEntriesRequestProto> appendEntries(
+  /** @param appendEntriesIndex the appendEntries channel, from {@link #nextAppendEntriesIndex()} */
+  CallStreamObserver<AppendEntriesRequestProto> appendEntries(int appendEntriesIndex,
       StreamObserver<AppendEntriesReplyProto> responseHandler, boolean isHeartbeat) {
     if (isHeartbeat && useSeparateHBChannel) {
       return (CallStreamObserver<AppendEntriesRequestProto>) hbAsyncStub.appendEntries(responseHandler);
     } else {
-      return (CallStreamObserver<AppendEntriesRequestProto>) asyncStub.appendEntries(responseHandler);
+      final RaftServerProtocolServiceStub stub = appendEntriesStubs.get(appendEntriesIndex);
+      return (CallStreamObserver<AppendEntriesRequestProto>) 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<ManagedChannel> getAppendEntriesChannels() {
+    return appendEntriesChannels;
+  }
+
+  @VisibleForTesting
+  ManagedChannel getMainChannel() {
+    return channel;
+  }
+
   StreamObserver<InstallSnapshotRequestProto> installSnapshot(
       String name, TimeDuration timeout, int limit, StreamObserver<InstallSnapshotReplyProto> 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);
   }
 }

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() {

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);