Fix timeout setting in location cache

src/rootserver/ob_root_utils.cpp

@@ -19,6 +19,7 @@
 #include "lib/string/ob_sql_string.h"
 #include "lib/hash/ob_hashset.h"
 #include "share/ob_rpc_struct.h"
+#include "share/ob_share_util.h"
 #include "share/ob_common_rpc_proxy.h"
 #include "share/schema/ob_table_schema.h"
 #include "share/schema/ob_schema_struct.h"
@@ -2655,33 +2656,8 @@ int ObRootUtils::get_rs_default_timeout_ctx(ObTimeoutCtx& ctx)
 {
   int ret = OB_SUCCESS;
   const int64_t DEFAULT_TIMEOUT_US = 2 * 1000 * 1000;  // 2s
-  int64_t abs_timeout_us = ctx.get_abs_timeout();
-  int64_t worker_timeout_us = THIS_WORKER.get_timeout_ts();
-
-  if (0 < abs_timeout_us) {
-    // nothing
-    // ctx was setted, no need to set again
-  } else if (INT64_MAX == worker_timeout_us) {
-    // is backgroup thread, set timeout is 2s.
-    abs_timeout_us = ObTimeUtility::current_time() + DEFAULT_TIMEOUT_US;
-  } else if (0 < worker_timeout_us) {
-    // if work has timeouts, set timeout equal to work's
-    abs_timeout_us = worker_timeout_us;
-  } else {
-    // if work has no timeout, it is not possible, but ignore error, set timeout to 2s
-    abs_timeout_us = ObTimeUtility::current_time() + DEFAULT_TIMEOUT_US;
-  }
-
-  if (OB_FAIL(ctx.set_abs_timeout(abs_timeout_us))) {
-    LOG_WARN("set timeout failed", K(ret), K(abs_timeout_us));
-  } else if (ctx.is_timeouted()) {
-    ret = OB_TIMEOUT;
-    LOG_WARN("is timeout",
-        K(ret),
-        "abs_timeout",
-        ctx.get_abs_timeout(),
-        "this worker timeout ts",
-        THIS_WORKER.get_timeout_ts());
+  if (OB_FAIL(ObShareUtil::set_default_timeout_ctx(ctx, DEFAULT_TIMEOUT_US))) {
+    LOG_WARN("fail to set default_timeout_ctx", KR(ret));
   }
   return ret;
 }

src/share/CMakeLists.txt

@@ -128,6 +128,7 @@ ob_set_subtarget(ob_share common
   ob_sstable_checksum_iterator.cpp
   ob_priv_common.cpp
   ob_bg_thread_monitor.cpp
+  ob_share_util.cpp
 )
 
 ob_set_subtarget(ob_share object

src/share/ob_share_util.cpp

+/**
+ * Copyright (c) 2021 OceanBase
+ * OceanBase CE is licensed under Mulan PubL v2.
+ * You can use this software according to the terms and conditions of the Mulan PubL v2.
+ * You may obtain a copy of Mulan PubL v2 at:
+ *          http://license.coscl.org.cn/MulanPubL-2.0
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PubL v2 for more details.
+ */
+
+#define USING_LOG_PREFIX SHARE
+#include "share/ob_share_util.h"
+#include "common/ob_timeout_ctx.h"
+#include "share/ob_worker.h"
+#include "lib/time/ob_time_utility.h"
+#include "lib/oblog/ob_log_module.h"
+namespace oceanbase {
+using namespace common;
+namespace share {
+int ObShareUtil::set_default_timeout_ctx(ObTimeoutCtx &ctx, const int64_t default_timeout)
+{
+  int ret = OB_SUCCESS;
+  int64_t abs_timeout_ts = OB_INVALID_TIMESTAMP;
+  int64_t ctx_timeout_ts = ctx.get_abs_timeout();
+  int64_t worker_timeout_ts = THIS_WORKER.get_timeout_ts();
+  if (0 < ctx_timeout_ts) {
+    // ctx has already been set, use it
+    abs_timeout_ts = ctx_timeout_ts;
+  } else if (INT64_MAX == worker_timeout_ts) {
+    // if worker's timeout_ts has not been set，change to default_timeout
+    abs_timeout_ts = ObTimeUtility::current_time() + default_timeout;
+  } else if (0 < worker_timeout_ts) {
+    // use worker's timeout if only it is valid
+    abs_timeout_ts = worker_timeout_ts;
+  } else {
+    // worker's timeout_ts is invalid, set to default timeout
+    abs_timeout_ts = ObTimeUtility::current_time() + default_timeout;
+  }
+  if (OB_FAIL(ctx.set_abs_timeout(abs_timeout_ts))) {
+    LOG_WARN(
+        "set timeout failed", KR(ret), K(abs_timeout_ts), K(ctx_timeout_ts), K(worker_timeout_ts), K(default_timeout));
+  } else if (ctx.is_timeouted()) {
+    ret = OB_TIMEOUT;
+    LOG_WARN("timeout", KR(ret), K(abs_timeout_ts), K(ctx_timeout_ts), K(worker_timeout_ts), K(default_timeout));
+  } else {
+    LOG_DEBUG("set_default_timeout_ctx success",
+        K(abs_timeout_ts),
+        K(ctx_timeout_ts),
+        K(worker_timeout_ts),
+        K(default_timeout));
+  }
+  return ret;
+}
+}  // end namespace share
+}  // end namespace oceanbase
\ No newline at end of file

src/share/ob_share_util.h

+/**
+ * Copyright (c) 2021 OceanBase
+ * OceanBase CE is licensed under Mulan PubL v2.
+ * You can use this software according to the terms and conditions of the Mulan PubL v2.
+ * You may obtain a copy of Mulan PubL v2 at:
+ *          http://license.coscl.org.cn/MulanPubL-2.0
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PubL v2 for more details.
+ */
+
+#ifndef OCEANBASE_SHARE_OB_SHARE_UTIL_H_
+#define OCEANBASE_SHARE_OB_SHARE_UTIL_H_
+#include "share/ob_define.h"
+namespace oceanbase {
+namespace common {
+class ObTimeoutCtx;
+}
+namespace share {
+class ObShareUtil {
+public:
+  // Order of setting timeout_ctx: ctx > worker > default_timeout
+  static int set_default_timeout_ctx(common::ObTimeoutCtx &ctx, const int64_t default_timeout);
+};
+}  // end namespace share
+}  // end namespace oceanbase
+#endif  // OCEANBASE_SHARE_OB_SHARE_UTIL_H_
\ No newline at end of file

src/share/partition_table/ob_partition_location_cache.cpp

@@ -27,6 +27,7 @@
 #include "share/inner_table/ob_inner_table_schema.h"
 #include "share/ob_common_rpc_proxy.h"
 #include "share/ob_rs_mgr.h"
+#include "share/ob_share_util.h"
 #include "share/partition_table/ob_location_update_task.h"
 #include "share/partition_table/ob_partition_table_operator.h"
 #include "share/partition_table/ob_remote_partition_table_operator.h"
@@ -4134,28 +4135,11 @@ int ObPartitionLocationCache::add_update_task(const ObLocationAsyncUpdateTask& t
 int ObPartitionLocationCache::set_timeout_ctx(common::ObTimeoutCtx& ctx)
 {
   int ret = OB_SUCCESS;
-  int64_t abs_timeout_us = ctx.get_abs_timeout();
-
-  if (abs_timeout_us < 0) {
-    abs_timeout_us = ObTimeUtility::current_time() + GCONF.location_cache_refresh_rpc_timeout +
-                     GCONF.location_cache_refresh_sql_timeout;
-  }
-  if (THIS_WORKER.get_timeout_ts() > 0 && THIS_WORKER.get_timeout_ts() < abs_timeout_us) {
-    abs_timeout_us = THIS_WORKER.get_timeout_ts();
-  }
-
-  if (OB_FAIL(ctx.set_abs_timeout(abs_timeout_us))) {
-    LOG_WARN("set timeout failed", K(ret), K(abs_timeout_us));
-  } else if (ctx.is_timeouted()) {
-    ret = OB_TIMEOUT;
-    LOG_WARN("is timeout",
-        K(ret),
-        "abs_timeout",
-        ctx.get_abs_timeout(),
-        "this worker timeout ts",
-        THIS_WORKER.get_timeout_ts());
+  int64_t default_timeout = GCONF.location_cache_refresh_rpc_timeout +
+                            GCONF.location_cache_refresh_sql_timeout;
+  if (OB_FAIL(ObShareUtil::set_default_timeout_ctx(ctx, default_timeout))) {
+    LOG_WARN("fail to set default_timeout_ctx", KR(ret));
   }
-
   return ret;
 }
 
@@ -4658,7 +4642,6 @@ int ObPartitionLocationCache::set_batch_timeout_ctx(
         "this worker timeout ts",
         THIS_WORKER.get_timeout_ts());
   }
-
   return ret;
 }
 /*-----batch async renew location end -----*/