feat: add max age (#65)

jiacai2050 · web-flow · commit 7e80e0a89ddd · 2023-08-10T11:11:26.000+08:00
* feat: add max age

* fix CI

* fix format

* update pr

* add backoff

* fix compile

* fix CR
diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md
@@ -1,31 +1,7 @@
-# Which issue does this PR close?
+## Rationale
 
-Closes #
 
-# Rationale for this change
- 
-<!---
- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
- Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
--->
+## Detailed Changes
 
-# What changes are included in this PR?
 
-<!---
-There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR to help reviewer understand the structure.
--->
-
-# Are there any user-facing changes?
-
-<!---
-Please mention if:
-
-- there are user-facing changes that needs to update the documentation or configuration.
-- this is a breaking change to public APIs
--->
-
-# How does this change test
-
-<!-- 
-Please describe how you test this change (like by unit test case, integration test or some other ways) if this change has touched the code.
--->
+## Test Plan
diff --git a/Makefile b/Makefile
@@ -0,0 +1,3 @@
+
+fmt:
+	mvn formatter:format
diff --git a/ceresdb-grpc/src/main/java/io/ceresdb/rpc/GrpcClient.java b/ceresdb-grpc/src/main/java/io/ceresdb/rpc/GrpcClient.java
@@ -5,6 +5,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
@@ -68,7 +69,6 @@
 
 /**
  * Grpc client implementation.
- *
  */
 public class GrpcClient implements RpcClient {
 
@@ -596,11 +596,29 @@ private Channel getCheckedChannel(final Endpoint endpoint, final Consumer<Throwa
     }
 
     private ManagedChannel getChannel(final Endpoint endpoint, final boolean createIfAbsent) {
+        IdChannel ch = null;
         if (createIfAbsent) {
-            return this.managedChannelPool.computeIfAbsent(endpoint, this::newChannel);
+            ch = this.managedChannelPool.computeIfAbsent(endpoint, this::newChannel);
         } else {
-            return this.managedChannelPool.get(endpoint);
+            ch = this.managedChannelPool.get(endpoint);
         }
+
+        long maxAge = this.opts.getConnectionMaxAgeMs();
+        if (maxAge != 0 && ch != null) {
+            long createTime = ch.getCreateTime();
+            long now = System.currentTimeMillis();
+            Random rand = new Random();
+            // Add backoff here to avoid multiple connections retry at same time.
+            long backoff = rand.nextInt(20_000); // max backoff 20s
+            if (now - createTime > maxAge + backoff) {
+                ch.shutdown();
+                IdChannel newChannel = this.newChannel(endpoint);
+                this.managedChannelPool.put(endpoint, newChannel);
+                return newChannel;
+            }
+        }
+
+        return ch;
     }
 
     private IdChannel newChannel(final Endpoint endpoint) {
diff --git a/ceresdb-grpc/src/main/java/io/ceresdb/rpc/IdChannel.java b/ceresdb-grpc/src/main/java/io/ceresdb/rpc/IdChannel.java
@@ -14,11 +14,11 @@
 
 /**
  * A managed channel that has a channel id.
- *
  */
 public class IdChannel extends ManagedChannel {
 
-    private static final AtomicLong ID_ALLOC = new AtomicLong();
+    private static final AtomicLong ID_ALLOC   = new AtomicLong();
+    private final long              createTime = System.currentTimeMillis();
 
     private final long           channelId;
     private final ManagedChannel channel;
@@ -32,6 +32,10 @@ public IdChannel(ManagedChannel channel) {
         this.channel = channel;
     }
 
+    public long getCreateTime() {
+        return createTime;
+    }
+
     public long getChannelId() {
         return channelId;
     }
@@ -98,9 +102,7 @@ public void enterIdle() {
 
     @Override
     public String toString() {
-        return "IdChannel{" + //
-               "channelId=" + channelId + //
-               ", channel=" + channel + //
-               '}';
+        return "IdChannel{" + "createTime=" + createTime + ", channelId=" + channelId + '}';
     }
+
 }
diff --git a/ceresdb-rpc/src/main/java/io/ceresdb/rpc/RpcOptions.java b/ceresdb-rpc/src/main/java/io/ceresdb/rpc/RpcOptions.java
@@ -91,6 +91,11 @@ public class RpcOptions implements Copiable<RpcOptions> {
 
     private boolean logOnLimitChange = true;
 
+    /**
+     * Max time in milliseconds a connection can live, 0 means forever.
+     */
+    private long connectionMaxAgeMs = 0;
+
     public int getDefaultRpcTimeout() {
         return defaultRpcTimeout;
     }
@@ -99,6 +104,14 @@ public void setDefaultRpcTimeout(int defaultRpcTimeout) {
         this.defaultRpcTimeout = defaultRpcTimeout;
     }
 
+    public long getConnectionMaxAgeMs() {
+        return connectionMaxAgeMs;
+    }
+
+    public void setConnectionMaxAgeMs(long connectionMaxAgeMs) {
+        this.connectionMaxAgeMs = connectionMaxAgeMs;
+    }
+
     public int getRpcThreadPoolSize() {
         return rpcThreadPoolSize;
     }
@@ -238,29 +251,20 @@ public RpcOptions copy() {
         opts.smoothing = this.smoothing;
         opts.blockOnLimit = this.blockOnLimit;
         opts.logOnLimitChange = this.logOnLimitChange;
+        opts.connectionMaxAgeMs = this.connectionMaxAgeMs;
         return opts;
     }
 
     @Override
     public String toString() {
-        return "RpcOptions{" + //
-               "defaultRpcTimeout=" + defaultRpcTimeout + //
-               ", rpcThreadPoolSize=" + rpcThreadPoolSize + //
-               ", rpcThreadPoolQueueSize=" + rpcThreadPoolQueueSize + //
-               ", maxInboundMessageSize=" + maxInboundMessageSize + //
-               ", flowControlWindow=" + flowControlWindow + //
-               ", idleTimeoutSeconds=" + idleTimeoutSeconds + //
-               ", keepAliveTimeSeconds=" + keepAliveTimeSeconds + //
-               ", keepAliveTimeoutSeconds=" + keepAliveTimeoutSeconds + //
-               ", keepAliveWithoutCalls=" + keepAliveWithoutCalls + //
-               ", limitKind=" + limitKind + //
-               ", initialLimit=" + initialLimit + //
-               ", maxLimit=" + maxLimit + //
-               ", longRttWindow=" + longRttWindow + //
-               ", smoothing=" + smoothing + //
-               ", blockOnLimit=" + blockOnLimit + //
-               ", logOnLimitChange=" + logOnLimitChange + //
-               '}';
+        return "RpcOptions{" + "defaultRpcTimeout=" + defaultRpcTimeout + ", rpcThreadPoolSize=" + rpcThreadPoolSize
+               + ", rpcThreadPoolQueueSize=" + rpcThreadPoolQueueSize + ", maxInboundMessageSize="
+               + maxInboundMessageSize + ", flowControlWindow=" + flowControlWindow + ", idleTimeoutSeconds="
+               + idleTimeoutSeconds + ", keepAliveTimeSeconds=" + keepAliveTimeSeconds + ", keepAliveTimeoutSeconds="
+               + keepAliveTimeoutSeconds + ", keepAliveWithoutCalls=" + keepAliveWithoutCalls + ", limitKind="
+               + limitKind + ", initialLimit=" + initialLimit + ", maxLimit=" + maxLimit + ", longRttWindow="
+               + longRttWindow + ", smoothing=" + smoothing + ", blockOnLimit=" + blockOnLimit + ", logOnLimitChange="
+               + logOnLimitChange + ", connectionMaxAge=" + connectionMaxAgeMs + '}';
     }
 
     public static RpcOptions newDefault() {
diff --git a/pom.xml b/pom.xml
@@ -77,7 +77,7 @@
         <proto-internal.version>1.0.0</proto-internal.version>
         <protobuf.version>3.21.7</protobuf.version>
         <!-- according to https://maven.apache.org/maven-ci-friendly.html -->
-        <revision>1.0.3</revision>
+        <revision>1.0.4</revision>
         <slf4j.version>1.7.21</slf4j.version>
     </properties>
 

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+`
	`2`	`+fmt:`
	`3`	`+ mvn formatter:format`