Make keepalive pings bidirectional and optimizable (#35441)

This is related to #34405 and a follow-up to #34753. It makes a number
of changes to our current keepalive pings.

The ping interval configuration is moved to the ConnectionProfile.

The server channel now responds to pings. This makes the keepalive
pings bidirectional.

On the client-side, the pings can now be optimized away. What this
means is that if the channel has received a message or sent a message
since the last pinging round, the ping is not sent for this round.

Author: Tim-Brooks

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpChannel.java

@@ -35,12 +35,15 @@
 public class Netty4TcpChannel implements TcpChannel {
 
     private final Channel channel;
+    private final boolean isServer;
     private final String profile;
     private final CompletableContext<Void> connectContext;
     private final CompletableContext<Void> closeContext = new CompletableContext<>();
+    private final ChannelStats stats = new ChannelStats();
 
-    Netty4TcpChannel(Channel channel, String profile, @Nullable ChannelFuture connectFuture) {
+    Netty4TcpChannel(Channel channel, boolean isServer, String profile, @Nullable ChannelFuture connectFuture) {
         this.channel = channel;
+        this.isServer = isServer;
         this.profile = profile;
         this.connectContext = new CompletableContext<>();
         this.channel.closeFuture().addListener(f -> {
@@ -77,6 +80,11 @@ public void close() {
         channel.close();
     }
 
+    @Override
+    public boolean isServerChannel() {
+        return isServer;
+    }
+
     @Override
     public String getProfile() {
         return profile;
@@ -92,6 +100,11 @@ public void addConnectListener(ActionListener<Void> listener) {
         connectContext.addListener(ActionListener.toBiConsumer(listener));
     }
 
+    @Override
+    public ChannelStats getChannelStats() {
+        return stats;
+    }
+
     @Override
     public boolean isOpen() {
         return channel.isOpen();

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java

@@ -232,7 +232,7 @@ protected Netty4TcpChannel initiateChannel(DiscoveryNode node) throws IOExceptio
         }
         addClosedExceptionLogger(channel);
 
-        Netty4TcpChannel nettyChannel = new Netty4TcpChannel(channel, "default", connectFuture);
+        Netty4TcpChannel nettyChannel = new Netty4TcpChannel(channel, false, "default", connectFuture);
         channel.attr(CHANNEL_KEY).set(nettyChannel);
 
         return nettyChannel;
@@ -246,14 +246,6 @@ protected Netty4TcpServerChannel bind(String name, InetSocketAddress address) {
         return esChannel;
     }
 
-    long successfulPingCount() {
-        return successfulPings.count();
-    }
-
-    long failedPingCount() {
-        return failedPings.count();
-    }
-
     @Override
     @SuppressForbidden(reason = "debug")
     protected void stopInternal() {
@@ -297,7 +289,7 @@ protected ServerChannelInitializer(String name) {
         @Override
         protected void initChannel(Channel ch) throws Exception {
             addClosedExceptionLogger(ch);
-            Netty4TcpChannel nettyTcpChannel = new Netty4TcpChannel(ch, name, ch.newSucceededFuture());
+            Netty4TcpChannel nettyTcpChannel = new Netty4TcpChannel(ch, true, name, ch.newSucceededFuture());
             ch.attr(CHANNEL_KEY).set(nettyTcpChannel);
             ch.pipeline().addLast("logging", new ESLoggingHandler());
             ch.pipeline().addLast("size", new Netty4SizeHeaderFrameDecoder());

modules/transport-netty4/src/test/java/org/elasticsearch/transport/netty4/Netty4ScheduledPingTests.java

@@ -28,17 +28,25 @@
 
 public class NioTcpChannel extends NioSocketChannel implements TcpChannel {
 
+    private final boolean isServer;
     private final String profile;
+    private final ChannelStats stats = new ChannelStats();
 
-    public NioTcpChannel(String profile, SocketChannel socketChannel) {
+    public NioTcpChannel(boolean isServer, String profile, SocketChannel socketChannel) {
         super(socketChannel);
+        this.isServer = isServer;
         this.profile = profile;
     }
 
     public void sendMessage(BytesReference reference, ActionListener<Void> listener) {
         getContext().sendMessage(BytesReference.toByteBuffers(reference), ActionListener.toBiConsumer(listener));
     }
 
+    @Override
+    public boolean isServerChannel() {
+        return isServer;
+    }
+
     @Override
     public String getProfile() {
         return profile;
@@ -54,6 +62,11 @@ public void addConnectListener(ActionListener<Void> listener) {
         addConnectListener(ActionListener.toBiConsumer(listener));
     }
 
+    @Override
+    public ChannelStats getChannelStats() {
+        return stats;
+    }
+
     @Override
     public void close() {
         getContext().closeChannel();

plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/NioTcpChannel.java

@@ -58,6 +58,7 @@
 import static org.elasticsearch.common.util.concurrent.EsExecutors.daemonThreadFactory;
 
 public class NioTransport extends TcpTransport {
+
     private static final Logger logger = LogManager.getLogger(NioTransport.class);
 
     public static final Setting<Integer> NIO_WORKER_COUNT =
@@ -135,11 +136,11 @@ protected void acceptChannel(NioSocketChannel channel) {
     }
 
     protected TcpChannelFactory serverChannelFactory(ProfileSettings profileSettings) {
-        return new TcpChannelFactoryImpl(profileSettings);
+        return new TcpChannelFactoryImpl(profileSettings, false);
     }
 
     protected Function<DiscoveryNode, TcpChannelFactory> clientChannelFactoryFunction(ProfileSettings profileSettings) {
-        return (n) -> new TcpChannelFactoryImpl(profileSettings);
+        return (n) -> new TcpChannelFactoryImpl(profileSettings, true);
     }
 
     protected abstract class TcpChannelFactory extends ChannelFactory<NioTcpServerChannel, NioTcpChannel> {
@@ -151,20 +152,22 @@ protected TcpChannelFactory(RawChannelFactory rawChannelFactory) {
 
     private class TcpChannelFactoryImpl extends TcpChannelFactory {
 
+        private final boolean isClient;
         private final String profileName;
 
-        private TcpChannelFactoryImpl(ProfileSettings profileSettings) {
+        private TcpChannelFactoryImpl(ProfileSettings profileSettings, boolean isClient) {
             super(new RawChannelFactory(profileSettings.tcpNoDelay,
                 profileSettings.tcpKeepAlive,
                 profileSettings.reuseAddress,
                 Math.toIntExact(profileSettings.sendBufferSize.getBytes()),
                 Math.toIntExact(profileSettings.receiveBufferSize.getBytes())));
+            this.isClient = isClient;
             this.profileName = profileSettings.profileName;
         }
 
         @Override
-        public NioTcpChannel createChannel(NioSelector selector, SocketChannel channel) throws IOException {
-            NioTcpChannel nioChannel = new NioTcpChannel(profileName, channel);
+        public NioTcpChannel createChannel(NioSelector selector, SocketChannel channel) {
+            NioTcpChannel nioChannel = new NioTcpChannel(isClient == false, profileName, channel);
             Supplier<InboundChannelBuffer.Page> pageSupplier = () -> {
                 Recycler.V<byte[]> bytes = pageCacheRecycler.bytePage(false);
                 return new InboundChannelBuffer.Page(ByteBuffer.wrap(bytes.v()), bytes::close);
@@ -178,7 +181,7 @@ public NioTcpChannel createChannel(NioSelector selector, SocketChannel channel)
         }
 
         @Override
-        public NioTcpServerChannel createServerChannel(NioSelector selector, ServerSocketChannel channel) throws IOException {
+        public NioTcpServerChannel createServerChannel(NioSelector selector, ServerSocketChannel channel) {
             NioTcpServerChannel nioChannel = new NioTcpServerChannel(profileName, channel);
             Consumer<Exception> exceptionHandler = (e) -> onServerException(nioChannel, e);
             Consumer<NioSocketChannel> acceptor = NioTransport.this::acceptChannel;

plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/NioTransport.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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.elasticsearch.common;
+
+import org.elasticsearch.action.ActionListener;
+
+/**
+ * A {@link java.util.function.BiFunction}-like interface designed to be used with asynchronous executions.
+ */
+public interface AsyncBiFunction<T,U,C> {
+
+    void apply(T t, U u, ActionListener<C> listener);
+}