Support extended keepalive parameters in epoll and io_uring

vertx-core/src/main/java/io/vertx/core/impl/transports/EpollTransport.java

@@ -39,7 +39,7 @@ public class EpollTransport implements Transport {
 
   /**
    * Return the number of of pending TFO connections in SYN-RCVD state for TCP_FASTOPEN.
-   *
+   * <p>
    * {@see #setPendingFastOpenRequestsThreshold}
    */
   public static int getPendingFastOpenRequestsThreshold() {
@@ -138,17 +138,23 @@ public void configure(TcpConfig config, boolean domainSocket, ServerBootstrap bo
       configChildOption(bootstrap, config, TcpOption.USER_TIMEOUT, EpollChannelOption.TCP_USER_TIMEOUT);
       configChildOption(bootstrap, config, TcpOption.QUICKACK, EpollChannelOption.TCP_QUICKACK);
       configChildOption(bootstrap, config, TcpOption.CORK, EpollChannelOption.TCP_CORK);
+      configChildOption(bootstrap, config, TcpOption.KEEPCNT, EpollChannelOption.TCP_KEEPCNT);
+      configChildOption(bootstrap, config, TcpOption.KEEPINTVL, EpollChannelOption.TCP_KEEPINTVL);
+      configChildOption(bootstrap, config, TcpOption.KEEPIDLE, EpollChannelOption.TCP_KEEPIDLE);
     }
     Transport.super.configure(config, domainSocket, bootstrap);
   }
 
   @Override
   public void configure(TcpConfig config, boolean domainSocket, Bootstrap bootstrap) {
     if (!domainSocket) {
-      NioTransport.configOption(bootstrap, config, TcpOption.FASTOPEN_CONNECT, EpollChannelOption.TCP_FASTOPEN_CONNECT);
-      NioTransport.configOption(bootstrap, config, TcpOption.USER_TIMEOUT, EpollChannelOption.TCP_USER_TIMEOUT);
-      NioTransport.configOption(bootstrap, config, TcpOption.QUICKACK, EpollChannelOption.TCP_QUICKACK);
-      NioTransport.configOption(bootstrap, config, TcpOption.CORK, EpollChannelOption.TCP_CORK);
+      configOption(bootstrap, config, TcpOption.FASTOPEN_CONNECT, EpollChannelOption.TCP_FASTOPEN_CONNECT);
+      configOption(bootstrap, config, TcpOption.USER_TIMEOUT, EpollChannelOption.TCP_USER_TIMEOUT);
+      configOption(bootstrap, config, TcpOption.QUICKACK, EpollChannelOption.TCP_QUICKACK);
+      configOption(bootstrap, config, TcpOption.CORK, EpollChannelOption.TCP_CORK);
+      configOption(bootstrap, config, TcpOption.KEEPCNT, EpollChannelOption.TCP_KEEPCNT);
+      configOption(bootstrap, config, TcpOption.KEEPINTVL, EpollChannelOption.TCP_KEEPINTVL);
+      configOption(bootstrap, config, TcpOption.KEEPIDLE, EpollChannelOption.TCP_KEEPIDLE);
     }
     Transport.super.configure(config, domainSocket, bootstrap);
   }

vertx-core/src/main/java/io/vertx/core/impl/transports/IoUringTransport.java

@@ -142,6 +142,9 @@ public void configure(TcpConfig config, boolean domainSocket, ServerBootstrap bo
     configChildOption(bootstrap, config, TcpOption.USER_TIMEOUT, IoUringChannelOption.TCP_USER_TIMEOUT);
     configChildOption(bootstrap, config, TcpOption.QUICKACK, IoUringChannelOption.TCP_QUICKACK);
     configChildOption(bootstrap, config, TcpOption.CORK, IoUringChannelOption.TCP_CORK);
+    configChildOption(bootstrap, config, TcpOption.KEEPCNT, IoUringChannelOption.TCP_KEEPCNT);
+    configChildOption(bootstrap, config, TcpOption.KEEPINTVL, IoUringChannelOption.TCP_KEEPINTVL);
+    configChildOption(bootstrap, config, TcpOption.KEEPIDLE, IoUringChannelOption.TCP_KEEPIDLE);
     Transport.super.configure(config, false, bootstrap);
   }
 
@@ -150,10 +153,13 @@ public void configure(TcpConfig config, boolean domainSocket, Bootstrap bootstra
     if (domainSocket) {
       throw new IllegalArgumentException();
     }
-    NioTransport.configOption(bootstrap, config, TcpOption.FASTOPEN_CONNECT, IoUringChannelOption.TCP_FASTOPEN_CONNECT);
-    NioTransport.configOption(bootstrap, config, TcpOption.USER_TIMEOUT, IoUringChannelOption.TCP_USER_TIMEOUT);
-    NioTransport.configOption(bootstrap, config, TcpOption.QUICKACK, IoUringChannelOption.TCP_QUICKACK);
-    NioTransport.configOption(bootstrap, config, TcpOption.CORK, IoUringChannelOption.TCP_CORK);
+    configOption(bootstrap, config, TcpOption.FASTOPEN_CONNECT, IoUringChannelOption.TCP_FASTOPEN_CONNECT);
+    configOption(bootstrap, config, TcpOption.USER_TIMEOUT, IoUringChannelOption.TCP_USER_TIMEOUT);
+    configOption(bootstrap, config, TcpOption.QUICKACK, IoUringChannelOption.TCP_QUICKACK);
+    configOption(bootstrap, config, TcpOption.CORK, IoUringChannelOption.TCP_CORK);
+    configOption(bootstrap, config, TcpOption.KEEPCNT, IoUringChannelOption.TCP_KEEPCNT);
+    configOption(bootstrap, config, TcpOption.KEEPINTVL, IoUringChannelOption.TCP_KEEPINTVL);
+    configOption(bootstrap, config, TcpOption.KEEPIDLE, IoUringChannelOption.TCP_KEEPIDLE);
     Transport.super.configure(config, false, bootstrap);
   }
 }

vertx-core/src/main/java/io/vertx/core/net/TCPSSLOptions.java

@@ -97,6 +97,27 @@ public abstract class TCPSSLOptions extends NetworkOptions {
    */
   public static final int DEFAULT_TCP_USER_TIMEOUT = 0;
 
+  /**
+   * Default value for tcp keepalive idle time.
+   * <p>
+   * {@code -1} defaults to OS settings
+   */
+  public static final int DEFAULT_TCP_KEEAPLIVE_IDLE_SECONDS = -1;
+
+  /**
+   * Default value for tcp keepalive count.
+   * <p>
+   * {@code -1} defaults to OS settings
+   */
+  public static final int DEFAULT_TCP_KEEAPLIVE_COUNT = -1;
+
+  /**
+   * Default value for tcp keepalive interval.
+   * <p>
+   * {@code -1} defaults to OS settings
+   */
+  public static final int DEFAULT_TCP_KEEAPLIVE_INTERVAL_SECONDS = -1;
+
   private TcpConfig transportOptions;
   private int idleTimeout;
   private int readIdleTimeout;
@@ -765,6 +786,65 @@ public TCPSSLOptions setTcpUserTimeout(int tcpUserTimeout) {
     return this;
   }
 
+  /**
+   * @return the time in seconds the connection needs to remain idle before TCP starts sending keepalive probes
+   */
+  public int getTcpKeepAliveIdleSeconds() {
+    return getOrDefaultOption(TcpOption.KEEPIDLE);
+  }
+
+  /**
+   * The time in seconds the connection needs to remain idle before TCP starts sending keepalive probes,
+   * if the socket option keepalive has been set.
+   * <p>
+   * Only works with linux native support (EPoll, IoUring).
+   *
+   * @param tcpKeepAliveIdleSeconds idle time in seconds
+   * @return a reference to this, so the API can be used fluently
+   */
+  public TCPSSLOptions setTcpKeepAliveIdleSeconds(int tcpKeepAliveIdleSeconds) {
+    transportOptions.setOption(TcpOption.KEEPIDLE, tcpKeepAliveIdleSeconds);
+    return this;
+  }
+
+  /**
+   * @return the maximum number of keepalive probes TCP should send before dropping the connection.
+   */
+  public int getTcpKeepAliveCount() {
+    return getOrDefaultOption(TcpOption.KEEPCNT);
+  }
+
+  /**
+   * The maximum number of keepalive probes TCP should send before dropping the connection.
+   * <p>
+   * Only works with linux native support (EPoll, IoUring).
+   * @param tcpKeepAliveCount number of probes
+   * @return a reference to this, so the API can be used fluently
+   */
+  public TCPSSLOptions setTcpKeepAliveCount(int tcpKeepAliveCount) {
+    transportOptions.setOption(TcpOption.KEEPCNT, tcpKeepAliveCount);
+    return this;
+  }
+
+  /**
+   * @return the time in seconds between individual keepalive probes (while the channel is idle).
+   */
+  public int getTcpKeepAliveIntervalSeconds() {
+    return getOrDefaultOption(TcpOption.KEEPINTVL);
+  }
+
+  /**
+   * The time in seconds between individual keepalive probes (while the channel is idle).
+   * <p>
+   * Only works with linux native support (EPoll, IoUring).
+   * @param tcpKeepAliveIntervalSeconds interval in seconds
+   * @return a reference to this, so the API can be used fluently
+   */
+  public TCPSSLOptions setTcpKeepAliveIntervalSeconds(int tcpKeepAliveIntervalSeconds) {
+    transportOptions.setOption(TcpOption.KEEPINTVL, tcpKeepAliveIntervalSeconds);
+    return this;
+  }
+
   /**
    * Returns the enabled SSL/TLS protocols
    * @return the enabled protocols

vertx-core/src/main/java/io/vertx/core/net/TcpConfig.java

@@ -53,6 +53,7 @@ void init() {
     reuseAddress = NetworkOptions.DEFAULT_REUSE_ADDRESS;
     trafficClass = NetworkOptions.DEFAULT_TRAFFIC_CLASS;
     soReusePort = NetworkOptions.DEFAULT_REUSE_PORT;
+    soKeepAlive = TCPSSLOptions.DEFAULT_TCP_KEEP_ALIVE;
     soLinger = DEFAULT_SO_LINGER;
     options = null;
   }

vertx-core/src/main/java/io/vertx/core/net/TcpOption.java

@@ -52,6 +52,48 @@ protected void validate(Integer value) {
    */
   public static final TcpOption<Integer> FASTOPEN = new TcpOption<>(Integer.class, 0);
 
+  /**
+   * The {@code TCP_KEEPCNT} option - only with Linux native transport.
+   * <p>
+   * The maximum number of keepalive probes TCP should send before dropping the connection.
+   */
+  public static final TcpOption<Integer> KEEPCNT = new TcpOption<>(Integer.class, TCPSSLOptions.DEFAULT_TCP_KEEAPLIVE_COUNT) {
+    @Override
+    protected void validate(final Integer value) {
+      if (value < -1) {
+        throw new IllegalArgumentException("KEEPCNT must be >= -1");
+      }
+    }
+  };
+
+  /**
+   * The {@code TCP_KEEPIDLE} option - only with Linux native transport.
+   * <p>
+   * The time (in seconds) the connection needs to remain idle before TCP starts sending keepalive probes, if enabled.
+   */
+  public static final TcpOption<Integer> KEEPIDLE = new TcpOption<>(Integer.class, TCPSSLOptions.DEFAULT_TCP_KEEAPLIVE_IDLE_SECONDS) {
+    @Override
+    protected void validate(final Integer value) {
+      if (value < -1) {
+        throw new IllegalArgumentException("KEEPIDLE must be >= -1");
+      }
+    }
+  };
+
+  /**
+   * The {@code TCP_KEEPINTVL} option - only with Linux native transport.
+   * <p>
+   * The time (in seconds) between individual keepalive probes.
+   */
+  public static final TcpOption<Integer> KEEPINTVL = new TcpOption<>(Integer.class, TCPSSLOptions.DEFAULT_TCP_KEEAPLIVE_INTERVAL_SECONDS) {
+    @Override
+    protected void validate(final Integer value) {
+      if (value < -1) {
+        throw new IllegalArgumentException("KEEPINTVL must be >= -1");
+      }
+    }
+  };
+
   final Class<T> type;
   final T defaultValue;
 

vertx-core/src/test/java/io/vertx/tests/net/NetTest.java

@@ -296,6 +296,27 @@ public void testClientOptions() {
     assertEquals(options, options.setSslHandshakeTimeout(randLong));
     assertEquals(randLong, options.getSslHandshakeTimeout());
     assertIllegalArgumentException(() -> options.setSslHandshakeTimeout(-123));
+
+    assertEquals(NetClientOptions.DEFAULT_TCP_KEEAPLIVE_IDLE_SECONDS, options.getTcpKeepAliveIdleSeconds());
+    rand = TestUtils.randomPositiveInt();
+    assertEquals(options, options.setTcpKeepAliveIdleSeconds(rand));
+    assertEquals(rand, options.getTcpKeepAliveIdleSeconds());
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveIdleSeconds(0));
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveIdleSeconds(-123));
+
+    assertEquals(NetClientOptions.DEFAULT_TCP_KEEAPLIVE_COUNT, options.getTcpKeepAliveCount());
+    rand = TestUtils.randomPositiveInt();
+    assertEquals(options, options.setTcpKeepAliveCount(rand));
+    assertEquals(rand, options.getTcpKeepAliveCount());
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveCount(0));
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveCount(-123));
+
+    assertEquals(NetClientOptions.DEFAULT_TCP_KEEAPLIVE_INTERVAL_SECONDS, options.getTcpKeepAliveIntervalSeconds());
+    rand = TestUtils.randomPositiveInt();
+    assertEquals(options, options.setTcpKeepAliveIntervalSeconds(rand));
+    assertEquals(rand, options.getTcpKeepAliveIntervalSeconds());
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveIntervalSeconds(0));
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveIntervalSeconds(-123));
   }
 
   @Test
@@ -418,6 +439,27 @@ public void testServerOptions() {
     assertEquals(options, options.setProxyProtocolTimeout(randomProxyTimeout));
     assertEquals(randomProxyTimeout, options.getProxyProtocolTimeout());
     assertIllegalArgumentException(() -> options.setProxyProtocolTimeout(-123));
+
+    assertEquals(NetServerOptions.DEFAULT_TCP_KEEAPLIVE_IDLE_SECONDS, options.getTcpKeepAliveIdleSeconds());
+    rand = TestUtils.randomPositiveInt();
+    assertEquals(options, options.setTcpKeepAliveIdleSeconds(rand));
+    assertEquals(rand, options.getTcpKeepAliveIdleSeconds());
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveIdleSeconds(0));
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveIdleSeconds(-123));
+
+    assertEquals(NetServerOptions.DEFAULT_TCP_KEEAPLIVE_COUNT, options.getTcpKeepAliveCount());
+    rand = TestUtils.randomPositiveInt();
+    assertEquals(options, options.setTcpKeepAliveCount(rand));
+    assertEquals(rand, options.getTcpKeepAliveCount());
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveCount(0));
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveCount(-123));
+
+    assertEquals(NetServerOptions.DEFAULT_TCP_KEEAPLIVE_INTERVAL_SECONDS, options.getTcpKeepAliveIntervalSeconds());
+    rand = TestUtils.randomPositiveInt();
+    assertEquals(options, options.setTcpKeepAliveIntervalSeconds(rand));
+    assertEquals(rand, options.getTcpKeepAliveIntervalSeconds());
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveIntervalSeconds(0));
+    assertIllegalArgumentException(() -> options.setTcpKeepAliveIntervalSeconds(-123));
   }
 
   @Test