diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java index 320b8b0f5..f6f3453b2 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java @@ -128,11 +128,32 @@ public static void scanUnusedClient() { }); unusedClientMap.forEach((bConfig, value) -> value.forEach(e -> { try { + RpcClientProxy proxy = (RpcClientProxy) e; + // Double-check before closing: ensure the client is still idle to avoid closing a client in use. + // This prevents race condition where getOrCreateClient() gets a client right before it's closed. + if (!isIdleTimeout(bConfig, proxy)) { + // lastUsedNanos was updated, meaning a business thread is using it, try to put it back + Map clientMap = CLUSTER_MAP.get(bConfig); + if (clientMap != null) { + RpcClientProxy existing = clientMap.putIfAbsent( + proxy.getProtocolConfig().toUniqId(), proxy); + if (existing == null) { + // Successfully put back, do not close + logger.info("RpcClient {} rescued from closing due to recent usage", + proxy.getProtocolConfig().toSimpleString()); + return; + } + } + // Failed to put back (a new client already exists), still need to close the old one + } e.close(); - } finally { logger.warn("RpcClient in clusterName={}, naming={}, remove rpc client{}, due to unused time > {} ms", bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(), e.getProtocolConfig().toSimpleString(), bConfig.getIdleTimeout()); + } catch (Exception ex) { + logger.error("Failed to close RpcClient in clusterName={}, naming={}, client={}", + bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(), + e.getProtocolConfig().toSimpleString(), ex); } })); } @@ -140,7 +161,24 @@ public static void scanUnusedClient() { private static boolean isIdleTimeout(BackendConfig bConfig, RpcClientProxy clientProxy) { long unusedNanosLimit = TimeUnit.MILLISECONDS.toNanos(bConfig.getIdleTimeout()); long lastUsedNanos = clientProxy.getLastUsedNanos(); - return lastUsedNanos > 0 && unusedNanosLimit > 0 && (System.nanoTime() - lastUsedNanos) > unusedNanosLimit; + boolean idleTooLong = lastUsedNanos > 0 && unusedNanosLimit > 0 + && (System.nanoTime() - lastUsedNanos) > unusedNanosLimit; + if (!idleTooLong) { + return false; + } + // The client is idle for a long time, but there are still requests in flight on it. + // Skip cleaning in this round and re-check in the next one, otherwise closeClient() would + // forcibly fail all those in-flight requests with "Client(...) stop". + int pending = clientProxy.getPendingRequestCount(); + if (pending > 0) { + logger.warn("RpcClient in clusterName={}, naming={}, client={} idle > {} ms, " + + "but {} request(s) still in flight, skip cleaning this round", + bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(), + clientProxy.getProtocolConfig().toSimpleString(), + bConfig.getIdleTimeout(), pending); + return false; + } + return true; } /** @@ -306,6 +344,11 @@ public ProtocolConfig getProtocolConfig() { return delegate.getProtocolConfig(); } + @Override + public int getPendingRequestCount() { + return delegate.getPendingRequestCount(); + } + @Override public int hashCode() { return Objects.hash(delegate); @@ -327,4 +370,4 @@ public boolean equals(Object obj) { } } -} \ No newline at end of file +} diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java index bf3155f18..8fc2ca9bb 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java @@ -72,6 +72,9 @@ protected CompletionStage doInvoke(Request request, CompletionStage getInvoker(ServiceInstance instance) { String key = toUniqKey(instance); ConsumerInvokerProxy result = invokerCache.get(key); + // Keep consistent with createInvoker: the invoker must be rebuilt once its client is closed + // (or is being closed), otherwise the request would be sent to a client which is being torn + // down by RpcClusterClientManager#scanUnusedClient. if (result != null && result.isAvailable()) { return result; } diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java b/trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java index 04f3b6edf..c37189bcf 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java @@ -61,4 +61,20 @@ public interface RpcClient { */ ProtocolConfig getProtocolConfig(); + /** + * Get the number of requests which are still in flight on this client. + * + *

It is used by the idle client cleaner to skip a client which still has in-flight requests, + * otherwise closing the client would forcibly fail those requests with {@code Client(...) stop}.

+ * + *

A {@code default} implementation is provided to keep binary compatibility with the existing + * third-party implementations, which simply reports "no in-flight request" and thus keeps the old + * cleaning behavior.

+ * + * @return the number of in-flight requests, 0 if unknown + */ + default int getPendingRequestCount() { + return 0; + } + } diff --git a/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java b/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java index df1af9260..815e7be2e 100644 --- a/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java +++ b/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java @@ -79,6 +79,19 @@ public DefResponseFuture newFuture(RpcClientContext context, ConsumerInvoker return future; } + /** + * Get the number of requests which are still in flight, i.e. the {@link DefResponseFuture}s that have + * not been completed (by a response or by the timeout watcher) yet. + * + *

It is used by the idle client cleaner to avoid closing a client which still has in-flight + * requests, otherwise those requests would be forcibly failed with {@code Client(...) stop}.

+ * + * @return the number of in-flight requests + */ + public int getPendingCount() { + return futureMap.size(); + } + /** * Removes and force stops all {@link DefResponseFuture}s related to a tRPC client. * Should be called when a tRPC client closes. diff --git a/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java b/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java index 0a488da22..f483ba29f 100644 --- a/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java +++ b/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java @@ -85,6 +85,14 @@ public boolean isAvailable() { return super.isAvailable() && (transport != null && transport.isConnected()); } + /** + * {@inheritDoc} + */ + @Override + public int getPendingRequestCount() { + return futureManager.getPendingCount(); + } + /** * {@inheritDoc} *