feat: support long link

Author: wardseptember

trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java

@@ -44,11 +44,15 @@
  *         scanner closes idle connections.</li>
  *     <li>A lightweight background timer periodically (every
  *         {@value #RECONNECT_CHECK_PERIOD_SECONDS}s) scans cached clients. If a client is found
- *         to be unavailable, its failure counter is incremented; once the counter reaches
- *         {@value #MAX_RECONNECT_FAILURES}, the client is closed and evicted, freeing memory and
- *         allowing the next request to rebuild a fresh long connection.</li>
- *     <li>When the underlying {@link RpcClient} closes itself (transport error or our own timer),
- *         the {@link RpcClient#closeFuture()} callback removes the cache entry.</li>
+ *         to be unavailable, its failure counter is incremented and a warning is logged. The
+ *         scanner <b>never actively closes</b> the underlying transport: doing so would tear
+ *         down its Netty {@code EventLoopGroup} (and any in-flight long-connection requests).
+ *         Recovery is delegated to the transport's existing lazy reconnect, which is triggered
+ *         by the request path ({@link com.tencent.trpc.core.transport.AbstractClientTransport
+ *         #ensureChannelActive}) or by Netty's {@code channelInactive} event.</li>
+ *     <li>When the underlying {@link RpcClient} closes itself (transport error or explicit
+ *         shutdown), the {@link RpcClient#closeFuture()} callback removes the cache entry so
+ *         the next request rebuilds a fresh long connection.</li>
  *     <li>{@link #shutdownBackendConfig(BackendConfig)} / {@link #close()} still release clients
  *         explicitly.</li>
  * </ul>
@@ -62,8 +66,9 @@ public class RpcClusterClientManager {
      */
     private static final int RECONNECT_CHECK_PERIOD_SECONDS = 30;
     /**
-     * Maximum number of consecutive reconnect-check failures tolerated before the client is
-     * closed and evicted from the cache.
+     * Maximum number of consecutive reconnect-check failures tolerated before a warning is
+     * escalated. The scanner will <b>not</b> close the client transport itself; recovery is
+     * delegated to the transport's lazy reconnect.
      */
     private static final int MAX_RECONNECT_FAILURES = 5;
 
@@ -180,12 +185,13 @@ private static void ensureReconnectCheckerStarted() {
 
     /**
      * Periodic check: walk every cached client; for each one
-     * that is currently unavailable, increment its failure counter; once the counter reaches
-     * {@link #MAX_RECONNECT_FAILURES} the client is closed (which triggers
-     * {@code closeFuture} → cache eviction). Healthy clients have their counter reset.
-     * <p>The check itself does not actively send a heartbeat: it simply observes the current
-     * connection state. The transport's existing lazy reconnect (triggered by request path or
-     * by Netty's channelInactive event) takes care of re-establishing the long connection.</p>
+     * that is currently unavailable, increment its failure counter and log a warning once the
+     * counter reaches {@link #MAX_RECONNECT_FAILURES}. Healthy clients have their counter reset.
+     * <p>The check itself does not actively send a heartbeat and <b>never closes the underlying
+     * transport</b>: closing would tear down the shared Netty {@code EventLoopGroup} and abort
+     * in-flight long-connection requests. The transport's existing lazy reconnect (triggered by
+     * request path or by Netty's {@code channelInactive} event) takes care of re-establishing
+     * the long connection.</p>
      */
     static void checkAndReconnect() {
         if (CLOSED_FLAG.get()) {
@@ -195,24 +201,29 @@ static void checkAndReconnect() {
             try {
                 if (proxy.isAvailable()) {
                     proxy.failureCount.set(0);
+                    proxy.warnedAtMaxFailures.set(false);
                     return;
                 }
-                int fails = proxy.failureCount.incrementAndGet();
+                // Cap the counter at MAX_RECONNECT_FAILURES to avoid unbounded growth (and the
+                // resulting integer wrap-around) when the backend stays unavailable for a very
+                // long time. Once capped, further iterations are silent — the warning has
+                // already been emitted at the threshold-crossing iteration below.
+                int fails = proxy.failureCount.updateAndGet(
+                        cur -> cur >= MAX_RECONNECT_FAILURES ? MAX_RECONNECT_FAILURES : cur + 1);
                 if (logger.isDebugEnabled()) {
                     logger.debug("Reconnect-check: client {} not available, failureCount={}",
                             proxy.getProtocolConfig().toSimpleString(), fails);
                 }
-                if (fails >= MAX_RECONNECT_FAILURES) {
+                if (fails == MAX_RECONNECT_FAILURES
+                        && proxy.warnedAtMaxFailures.compareAndSet(false, true)) {
+                    // Log only once per unavailability spell to avoid log spam. The flag is
+                    // reset as soon as the proxy becomes available again (see the healthy
+                    // branch above is reached via a separate state, so we reset there too).
                     logger.warn("Reconnect-check: client {} unavailable for {} consecutive checks "
-                                    + "(~{}s), closing and evicting from cache",
+                                    + "(~{}s); leaving the transport intact and relying on lazy "
+                                    + "reconnect on the request path",
                             proxy.getProtocolConfig().toSimpleString(), fails,
                             fails * RECONNECT_CHECK_PERIOD_SECONDS);
-                    try {
-                        proxy.close();
-                    } catch (Throwable ex) {
-                        logger.error("Close stale client {} failed",
-                                proxy.getProtocolConfig().toSimpleString(), ex);
-                    }
                 }
             } catch (Throwable ex) {
                 logger.error("Reconnect-check on client {} threw", key, ex);
@@ -308,9 +319,16 @@ private static class RpcClientProxy implements RpcClient {
         private final RpcClient delegate;
         /**
          * Consecutive failure count observed by the reconnect-check timer; reset to 0 whenever
-         * the client is observed available.
+         * the client is observed available. Capped at {@link #MAX_RECONNECT_FAILURES} to avoid
+         * unbounded growth.
          */
         final AtomicInteger failureCount = new AtomicInteger(0);
+        /**
+         * Whether a "stuck unavailable" warning has already been emitted for the current
+         * unavailability spell. Reset to {@code false} as soon as the proxy is observed
+         * available again, so a future incident produces a fresh warning.
+         */
+        final AtomicBoolean warnedAtMaxFailures = new AtomicBoolean(false);
 
         RpcClientProxy(RpcClient delegate) {
             this.delegate = delegate;

trpc-core/src/main/java/com/tencent/trpc/core/common/Constants.java

@@ -36,7 +36,11 @@ public class Constants {
 
     public static final String DEFAULT_KEEP_ALIVE = "true";
     /**
-     * Default shared IO thread pool true
+     * Default shared IO thread pool: true. The shared pool now supports both NIO and
+     * Epoll: each transport joins one of two reference-counted shared groups based on
+     * {@code Epoll.isAvailable() && config.useEpoll()}, so enabling epoll no longer forces
+     * the user to also flip {@code ioThreadGroupShare} off. Server-side transports do not
+     * consult this flag (they always own their own group).
      */
     public static final String DEFAULT_IO_THREAD_GROUPSHARE = "true";
     /**
@@ -119,9 +123,34 @@ public class Constants {
      */
     public static final String DEFAULT_BUFFER_SIZE = "16384";
     /**
-     * Default client idle timeout 3 minutes
+     * Default client idle timeout 3 minutes. Drives the read-idle close handler installed
+     * by {@code NettyTcpClientTransport}: a long connection that has not received any
+     * inbound bytes for this duration is recycled by the client (the next request goes
+     * through lazy reconnect). Half-dead detection in faster paths is delegated to the
+     * configurable TCP keepalive parameters below; the read-idle handler is the slow
+     * universal fallback that also covers macOS / Windows where TCP keepalive tuning is
+     * not available.
      */
     public static final String DEFAULT_IDLE_TIMEOUT = "180000";
+    /**
+     * Default TCP keepalive idle (Linux {@code TCP_KEEPIDLE}) in seconds: 30 s without
+     * traffic on the socket before the kernel starts emitting keepalive probes. Together
+     * with {@link #DEFAULT_TCP_KEEPALIVE_INTVL} and {@link #DEFAULT_TCP_KEEPALIVE_CNT}
+     * (Dubbo-style 30/10/3) this yields a half-dead detection window of roughly 60 s on
+     * Linux + epoll (30 + 10 × 3 = 60). Value 0 leaves the OS default (Linux: 7200 s).
+     */
+    public static final String DEFAULT_TCP_KEEPALIVE_IDLE = "30";
+    /**
+     * Default TCP keepalive interval (Linux {@code TCP_KEEPINTVL}) in seconds between
+     * consecutive keepalive probes after the idle window has elapsed.
+     */
+    public static final String DEFAULT_TCP_KEEPALIVE_INTVL = "10";
+    /**
+     * Default TCP keepalive probe count (Linux {@code TCP_KEEPCNT}): the number of
+     * unacknowledged keepalive probes after which the kernel marks the connection dead
+     * and emits a RST.
+     */
+    public static final String DEFAULT_TCP_KEEPALIVE_CNT = "3";
     /**
      * Default server idle timeout 4 minutes
      */

trpc-core/src/main/java/com/tencent/trpc/core/common/config/BackendConfig.java

@@ -497,6 +497,9 @@ public ProtocolConfig generateProtocolConfig(String host, int port, String netwo
         config.setPort(port);
         config.setNetwork(network);
         config.setIdleTimeout(idleTimeout);
+        config.setTcpKeepAliveIdle(tcpKeepAliveIdle);
+        config.setTcpKeepAliveIntvl(tcpKeepAliveIntvl);
+        config.setTcpKeepAliveCnt(tcpKeepAliveCnt);
         config.setCharset(charset);
         config.setLazyinit(lazyinit);
         config.setConnsPerAddr(connsPerAddr);

trpc-core/src/main/java/com/tencent/trpc/core/common/config/BaseProtocolConfig.java

@@ -94,6 +94,28 @@ public class BaseProtocolConfig implements Serializable, Cloneable {
      */
     @ConfigProperty(value = Constants.DEFAULT_IDLE_TIMEOUT, type = Integer.class, override = true, moreZero = false)
     protected Integer idleTimeout;
+    /**
+     * TCP keepalive idle in seconds (Linux {@code TCP_KEEPIDLE}). Applied only on
+     * Linux + epoll TCP client transports. Value 0 disables tuning and leaves the OS
+     * default in place (Linux ≈ 7200 s).
+     */
+    @ConfigProperty(value = Constants.DEFAULT_TCP_KEEPALIVE_IDLE, type = Integer.class, override = true,
+            moreZero = false)
+    protected Integer tcpKeepAliveIdle;
+    /**
+     * TCP keepalive probe interval in seconds (Linux {@code TCP_KEEPINTVL}). Applied only
+     * on Linux + epoll TCP client transports. Value 0 disables tuning.
+     */
+    @ConfigProperty(value = Constants.DEFAULT_TCP_KEEPALIVE_INTVL, type = Integer.class, override = true,
+            moreZero = false)
+    protected Integer tcpKeepAliveIntvl;
+    /**
+     * TCP keepalive probe count (Linux {@code TCP_KEEPCNT}). Applied only on Linux +
+     * epoll TCP client transports. Value 0 disables tuning.
+     */
+    @ConfigProperty(value = Constants.DEFAULT_TCP_KEEPALIVE_CNT, type = Integer.class, override = true,
+            moreZero = false)
+    protected Integer tcpKeepAliveCnt;
     /**
      * Whether to delay initialization.
      */
@@ -303,6 +325,33 @@ public void setIdleTimeout(Integer idleTimeout) {
         this.idleTimeout = idleTimeout;
     }
 
+    public Integer getTcpKeepAliveIdle() {
+        return tcpKeepAliveIdle;
+    }
+
+    public void setTcpKeepAliveIdle(Integer tcpKeepAliveIdle) {
+        checkFiledModifyPrivilege();
+        this.tcpKeepAliveIdle = tcpKeepAliveIdle;
+    }
+
+    public Integer getTcpKeepAliveIntvl() {
+        return tcpKeepAliveIntvl;
+    }
+
+    public void setTcpKeepAliveIntvl(Integer tcpKeepAliveIntvl) {
+        checkFiledModifyPrivilege();
+        this.tcpKeepAliveIntvl = tcpKeepAliveIntvl;
+    }
+
+    public Integer getTcpKeepAliveCnt() {
+        return tcpKeepAliveCnt;
+    }
+
+    public void setTcpKeepAliveCnt(Integer tcpKeepAliveCnt) {
+        checkFiledModifyPrivilege();
+        this.tcpKeepAliveCnt = tcpKeepAliveCnt;
+    }
+
     public Boolean getLazyinit() {
         return lazyinit;
     }

trpc-core/src/main/java/com/tencent/trpc/core/transport/AbstractClientTransport.java

@@ -257,21 +257,93 @@ protected void ensureChannelActive(int chIndex) {
         ChannelFutureItem curChannelItem = channels.get(chIndex);
         // initiate a new connection creation action when the connection is not yet established or the connection
         // is broken
-        boolean futureIsDoneAndNotConnected =
-                !curChannelItem.isAvailable() && !curChannelItem.isConnecting();
-        // not available and not establishing a connection
-        if (futureIsDoneAndNotConnected || curChannelItem.isNotYetConnect()) {
+        if (!needsReconnect(curChannelItem)) {
+            return;
+        }
+        connLock.lock();
+        try {
+            // Double-check inside the lock: another thread may have already replaced this slot
+            // with a fresh ChannelFutureItem (either still connecting or already connected).
+            // Without this check, a thundering-herd of requests arriving right after a
+            // disconnect would each rebuild the slot, producing a connect/disconnect storm
+            // against the peer and a burst of short-lived TIME_WAIT sockets.
+            ChannelFutureItem latest = channels.get(chIndex);
+            if (!needsReconnect(latest)) {
+                return;
+            }
+            channels.set(chIndex, new ChannelFutureItem(createChannel().toCompletableFuture(), config));
+            try {
+                latest.close();
+            } catch (Exception ex) {
+                logger.error("close " + latest + " exception", ex);
+            }
+        } finally {
+            connLock.unlock();
+        }
+    }
+
+    /**
+     * A slot needs a fresh connection when it is either uninitialized (lazy-init not yet
+     * triggered) or its previous future has finished without producing a connected channel.
+     * A slot whose future is still in flight ({@code isConnecting()}) is left alone — the
+     * in-flight {@code bootstrap.connect} will publish the result into the same item.
+     */
+    private static boolean needsReconnect(ChannelFutureItem item) {
+        if (item.isNotYetConnect()) {
+            return true;
+        }
+        return !item.isAvailable() && !item.isConnecting();
+    }
+
+    /**
+     * Invalidate the slot holding {@code target}: replace it with a blank placeholder
+     * ({@code channelFuture==null}, i.e. {@code isNotYetConnect=true}) so the next request
+     * unconditionally goes through {@link #ensureChannelActive} and rebuilds a fresh
+     * connection. The previous item is closed best-effort.
+     * <p>Called by the client-side idle handler so that <em>before</em> the actual
+     * {@code channel.close()} runs (asynchronously in the EventLoop), the request thread
+     * already sees the slot as needing a reconnect — eliminating the "request lands on a
+     * channel that is about to be closed" race window.</p>
+     */
+    @Override
+    public void invalidateChannel(Channel target) {
+        if (target == null || channels == null || channels.isEmpty()) {
+            return;
+        }
+        // The list is bounded by connsPerAddr; a linear scan is fine.
+        for (int i = 0; i < channels.size(); i++) {
+            ChannelFutureItem item = channels.get(i);
+            if (item == null || item.channelFuture == null || !item.channelFuture.isDone()
+                    || item.channelFuture.isCompletedExceptionally()) {
+                continue;
+            }
+            Channel ch;
+            try {
+                ch = item.channelFuture.join();
+            } catch (Throwable ignore) {
+                continue;
+            }
+            if (ch != target) {
+                continue;
+            }
             connLock.lock();
             try {
-                channels.set(chIndex, new ChannelFutureItem(createChannel().toCompletableFuture(), config));
+                // Re-read under the lock to avoid clobbering a slot another thread already
+                // refreshed (e.g. concurrent reconnect).
+                ChannelFutureItem latest = channels.get(i);
+                if (latest != item) {
+                    return;
+                }
+                channels.set(i, new ChannelFutureItem(null, config));
                 try {
-                    curChannelItem.close();
+                    item.close();
                 } catch (Exception ex) {
-                    logger.error("close " + curChannelItem + " exception", ex);
+                    logger.error("close invalidated " + item + " exception", ex);
                 }
             } finally {
                 connLock.unlock();
             }
+            return;
         }
     }
 

trpc-core/src/main/java/com/tencent/trpc/core/transport/ClientTransport.java

@@ -62,6 +62,22 @@ public interface ClientTransport {
      */
     ChannelHandler getChannelHandler();
 
+    /**
+     * Mark the slot currently holding {@code channel} as invalidated so that the very next
+     * request observes "needs reconnect" instead of routing onto the about-to-be-closed
+     * channel. The default no-op keeps the contract optional for transports that do not
+     * support a slot/pool model.
+     * <p>This is the hook used by client-side idle handlers: when the idle handler decides to
+     * tear down a long-idle channel, it calls {@code invalidateChannel} <b>before</b>
+     * {@code channel.close()} so that the request-thread side cannot read the stale slot any
+     * longer. The actual TCP close happens asynchronously in the EventLoop afterwards.</p>
+     *
+     * @param channel the channel about to be closed; may be {@code null} (no-op).
+     */
+    default void invalidateChannel(Channel channel) {
+        // default no-op
+    }
+
     /**
      * Get the remote address.
      */