pd_client: reduce PD reconnection (tikv#14954)

ref tikv/pd#6556, close tikv#14964

Signed-off-by: Ryan Leung <[email protected]>

Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>
Signed-off-by: tonyxuqqi <[email protected]>

components/pd_client/src/client.rs

@@ -103,6 +103,7 @@ impl RpcClient {
                             target,
                             tso.unwrap(),
                             cfg.enable_forwarding,
+                            cfg.retry_interval.0,
                         )),
                         monitor: monitor.clone(),
                     };

components/pd_client/src/client_v2.rs

@@ -413,7 +413,7 @@ async fn reconnect_loop(
             use tikv_util::config::ReadableDuration;
             ReadableDuration::from_str(&s.unwrap()).unwrap().0
         });
-        request_timeout()
+        cfg.retry_interval.0
     })();
     let mut last_connect = StdInstant::now();
     loop {

components/pd_client/src/config.rs

@@ -16,7 +16,7 @@ pub struct Config {
     ///
     /// Default is `"127.0.0.1:2379"`.
     pub endpoints: Vec<String>,
-    /// The interval at which to retry a PD connection initialization.
+    /// The interval at which to retry a PD connection.
     ///
     /// Default is 300ms.
     pub retry_interval: ReadableDuration,

components/pd_client/src/util.rs

@@ -49,7 +49,6 @@ const MAX_RETRY_TIMES: u64 = 5;
 const MAX_RETRY_DURATION: Duration = Duration::from_secs(10);
 
 // FIXME: Use a request-independent way to handle reconnection.
-const GLOBAL_RECONNECT_INTERVAL: Duration = Duration::from_millis(100); // 0.1s
 pub const REQUEST_RECONNECT_INTERVAL: Duration = Duration::from_secs(1); // 1s
 
 #[derive(Clone)]
@@ -160,6 +159,7 @@ pub struct Client {
     pub(crate) inner: RwLock<Inner>,
     pub feature_gate: FeatureGate,
     enable_forwarding: bool,
+    retry_interval: Duration,
 }
 
 impl Client {
@@ -171,6 +171,7 @@ impl Client {
         target: TargetInfo,
         tso: TimestampOracle,
         enable_forwarding: bool,
+        retry_interval: Duration,
     ) -> Client {
         if !target.direct_connected() {
             REQUEST_FORWARDED_GAUGE_VEC
@@ -206,6 +207,7 @@ impl Client {
             }),
             feature_gate: FeatureGate::default(),
             enable_forwarding,
+            retry_interval,
         }
     }
 
@@ -333,8 +335,7 @@ impl Client {
 
         let future = {
             let inner = self.inner.rl();
-            if start.saturating_duration_since(inner.last_try_reconnect) < GLOBAL_RECONNECT_INTERVAL
-            {
+            if start.saturating_duration_since(inner.last_try_reconnect) < self.retry_interval {
                 // Avoid unnecessary updating.
                 // Prevent a large number of reconnections in a short time.
                 PD_RECONNECT_COUNTER_VEC
@@ -360,8 +361,7 @@ impl Client {
 
         {
             let mut inner = self.inner.wl();
-            if start.saturating_duration_since(inner.last_try_reconnect) < GLOBAL_RECONNECT_INTERVAL
-            {
+            if start.saturating_duration_since(inner.last_try_reconnect) < self.retry_interval {
                 // There may be multiple reconnections that pass the read lock at the same time.
                 // Check again in the write lock to avoid unnecessary updating.
                 PD_RECONNECT_COUNTER_VEC

etc/config-template.toml

@@ -344,7 +344,7 @@
 ## PD endpoints.
 # endpoints = ["127.0.0.1:2379"]
 
-## The interval at which to retry a PD connection initialization.
+## The interval at which to retry a PD connection.
 ## Default is 300ms.
 # retry-interval = "300ms"
 

tests/failpoints/cases/test_pd_client.rs

@@ -119,8 +119,8 @@ fn test_slow_periodical_update() {
 
     fail::cfg(pd_client_reconnect_fp, "pause").unwrap();
     // Wait for the PD client thread blocking on the fail point.
-    // The GLOBAL_RECONNECT_INTERVAL is 0.1s so sleeps 0.2s here.
-    thread::sleep(Duration::from_millis(200));
+    // The retry interval is 300ms so sleeps 400ms here.
+    thread::sleep(Duration::from_millis(400));
 
     let (tx, rx) = mpsc::channel();
     let handle = thread::spawn(move || {

tests/failpoints/cases/test_pd_client_legacy.rs

@@ -220,8 +220,8 @@ fn test_slow_periodical_update() {
 
     fail::cfg(pd_client_reconnect_fp, "pause").unwrap();
     // Wait for the PD client thread blocking on the fail point.
-    // The GLOBAL_RECONNECT_INTERVAL is 0.1s so sleeps 0.2s here.
-    thread::sleep(Duration::from_millis(200));
+    // The retry interval is 300ms so sleeps 400ms here.
+    thread::sleep(Duration::from_millis(400));
 
     let (tx, rx) = mpsc::channel();
     let handle = thread::spawn(move || {
@@ -245,8 +245,8 @@ fn test_reconnect_limit() {
     let pd_client_reconnect_fp = "pd_client_reconnect";
     let (_server, client) = new_test_server_and_client(ReadableDuration::secs(100));
 
-    // The GLOBAL_RECONNECT_INTERVAL is 0.1s so sleeps 0.2s here.
-    thread::sleep(Duration::from_millis(200));
+    // The default retry interval is 300ms so sleeps 400ms here.
+    thread::sleep(Duration::from_millis(400));
 
     // The first reconnection will succeed, and the last_update will not be updated.
     fail::cfg(pd_client_reconnect_fp, "return").unwrap();

tests/integrations/pd/test_rpc_client.rs

@@ -366,8 +366,8 @@ fn restart_leader(mgr: SecurityManager) {
     server.stop();
     server.start(&mgr, eps);
 
-    // The GLOBAL_RECONNECT_INTERVAL is 0.1s so sleeps 0.2s here.
-    thread::sleep(Duration::from_millis(200));
+    // The default retry interval is 300ms so sleeps 400ms here.
+    thread::sleep(Duration::from_millis(400));
 
     let region = block_on(client.get_region_by_id(region.get_id())).unwrap();
     assert_eq!(region.unwrap().get_id(), region_id);
@@ -604,9 +604,9 @@ fn test_cluster_version() {
     assert!(feature_gate.can_enable(feature_b));
     assert!(!feature_gate.can_enable(feature_c));
 
-    // After reconnect the version should be still accessable.
-    // The GLOBAL_RECONNECT_INTERVAL is 0.1s so sleeps 0.2s here.
-    thread::sleep(Duration::from_millis(200));
+    // After reconnect the version should be still accessible.
+    // The default retry interval is 300ms so sleeps 400ms here.
+    thread::sleep(Duration::from_millis(400));
     client.reconnect().unwrap();
     assert!(feature_gate.can_enable(feature_b));
     assert!(!feature_gate.can_enable(feature_c));

tests/integrations/pd/test_rpc_client_legacy.rs

@@ -427,8 +427,8 @@ fn restart_leader(mgr: SecurityManager) {
     server.stop();
     server.start(&mgr, eps);
 
-    // The GLOBAL_RECONNECT_INTERVAL is 0.1s so sleeps 0.2s here.
-    thread::sleep(Duration::from_millis(200));
+    // The default retry interval is 300ms so sleeps 400ms here.
+    thread::sleep(Duration::from_millis(400));
 
     let region = block_on(client.get_region_by_id(region.get_id())).unwrap();
     assert_eq!(region.unwrap().get_id(), region_id);
@@ -518,7 +518,7 @@ fn test_pd_client_heartbeat_send_failed() {
             RegionStat::default(),
             None,
         ));
-        let rsp = rx.recv_timeout(Duration::from_millis(100));
+        let rsp = rx.recv_timeout(Duration::from_millis(300));
         if ok {
             assert!(rsp.is_ok());
             assert_eq!(rsp.unwrap().get_region_id(), 1);
@@ -677,9 +677,9 @@ fn test_cluster_version() {
     assert!(feature_gate.can_enable(feature_b));
     assert!(!feature_gate.can_enable(feature_c));
 
-    // After reconnect the version should be still accessable.
-    // The GLOBAL_RECONNECT_INTERVAL is 0.1s so sleeps 0.2s here.
-    thread::sleep(Duration::from_millis(200));
+    // After reconnect the version should be still accessible.
+    // The default retry interval is 300ms so sleeps 400ms here.
+    thread::sleep(Duration::from_millis(400));
     client.reconnect().unwrap();
     assert!(feature_gate.can_enable(feature_b));
     assert!(!feature_gate.can_enable(feature_c));