[seekdb][transaction] Retry GTS acquisition when not ready

src/storage/tx/ob_trans_service.cpp

@@ -399,13 +399,14 @@ int ObTransService::register_mds_into_tx(ObTxDesc &tx_desc,
   if (OB_SUCC(ret)) {
     do {
       ret = register_mds_into_ctx_(tx_desc, type, buf, buf_len, seq_no, register_flag);
-      if (OB_EAGAIN == ret && ObTimeUtil::current_time() >= tx_desc.expire_ts_) {
+      if ((OB_EAGAIN == ret || is_id_not_ready_err(ret))
+          && ObTimeUtil::current_time() >= tx_desc.expire_ts_) {
         ret = OB_TIMEOUT;
         TRANS_LOG(WARN, "register tx data timeout", KR(ret), K(tx_desc), K(type));
-      } else if (OB_EAGAIN == ret) {
+      } else if (OB_EAGAIN == ret || is_id_not_ready_err(ret)) {
         ob_usleep(1000);
       }
-    } while (OB_EAGAIN == ret);
+    } while (OB_EAGAIN == ret || is_id_not_ready_err(ret));
 
     if (OB_TMP_FAIL(collect_tx_exec_result(tx_desc, tx_result))) {
     }

src/storage/tx/ob_trans_service_v4.cpp

@@ -707,6 +707,9 @@ int ObTransService::get_write_store_ctx(ObTxDesc &tx,
   const int16_t branch = store_ctx.branch_;
   ObTxSEQ data_scn = spec_seq_no; // for LOB aux table, spec_seq_no is valid
   ObTxSnapshot snap = snapshot.core_;
+  const int64_t snapshot_expire_ts = store_ctx.timeout_ > 0
+      ? store_ctx.timeout_
+      : tx.get_expire_ts();
   bool access_started = false;
   bool ctx_exist = false;
   ObTxTable *tx_table = nullptr;
@@ -727,7 +730,7 @@ int ObTransService::get_write_store_ctx(ObTxDesc &tx,
     ret = OB_INVALID_ARGUMENT;
     TRANS_LOG(WARN, "store_ctx's ls_ is invalid", K(ret), K(store_ctx), K(lbt()));
   } else if (snapshot.is_none_read()
-             && OB_FAIL(acquire_local_snapshot_(snap.version_))) {
+             && OB_FAIL(acquire_local_snapshot_with_retry_(snapshot_expire_ts, snap.version_))) {
     TRANS_LOG(WARN, "acquire ls snapshot for mvcc write fail", K(ret));
   } else if (OB_FAIL(acquire_tx_ctx(tx, tx_ctx, store_ctx.ls_, special, snapshot.read_elr(), ctx_exist))) {
   } else if (OB_FAIL(tx_ctx->start_access(tx, data_scn, branch))) {
@@ -995,6 +998,41 @@ int ObTransService::abort_write_state_(const ObTxDesc &tx_desc)
   return ret;
 }
 
+int ObTransService::sync_acquire_local_snapshot_(ObTxDesc &tx,
+                                                 const int64_t expire_ts,
+                                                 SCN &snapshot)
+{
+  int ret = acquire_local_snapshot_(snapshot);
+  if (OB_GTS_NOT_READY == ret) {
+    const uint64_t op_sn = tx.op_sn_;
+    tx.flags_.BLOCK_ = true;
+    tx.lock_.unlock();
+    ret = acquire_local_snapshot_with_retry_(expire_ts, snapshot);
+    tx.lock_.lock();
+
+    if (tx.flags_.INTERRUPTED_) {
+      ret = OB_ERR_INTERRUPTED;
+      TRANS_LOG(WARN, "acquiring local snapshot has been interrupted", KR(ret), K(tx));
+    }
+    tx.clear_interrupt();
+    tx.flags_.BLOCK_ = false;
+    if (op_sn != tx.op_sn_) {
+      if (tx.is_aborted()) {
+        ret = tx.abort_cause_ == OB_DEAD_LOCK ? OB_DEAD_LOCK : OB_TRANS_KILLED;
+        TRANS_LOG(WARN, "txn has been aborted", KR(ret), K(tx.abort_cause_));
+      } else if (tx.is_rollbacked()) {
+        ret = OB_TRANS_ROLLBACKED;
+        TRANS_LOG(WARN, "txn has been rollbacked", KR(ret), K(tx));
+      } else if (OB_FAIL(ret)) {
+      } else {
+        ret = OB_ERR_UNEXPECTED;
+        TRANS_LOG(WARN, "txn has been disturbed", KR(ret), K(tx));
+      }
+    }
+  }
+  return ret;
+}
+
 OB_NOINLINE int ObTransService::acquire_local_snapshot_(SCN &snapshot)
 {
   int ret = OB_SUCCESS;
@@ -1005,6 +1043,9 @@ OB_NOINLINE int ObTransService::acquire_local_snapshot_(SCN &snapshot)
   } else if (!snapshot0.is_valid_and_not_min()) {
     ret = OB_EAGAIN;
   } else if (OB_FAIL(ts_mgr_->get_gts(snapshot1))) {
+    if (OB_EAGAIN == ret) {
+      ret = OB_GTS_NOT_READY;
+    }
   } else {
     snapshot = SCN::max(snapshot0, snapshot1);
   }
@@ -1015,6 +1056,38 @@ OB_NOINLINE int ObTransService::acquire_local_snapshot_(SCN &snapshot)
   return ret;
 }
 
+int ObTransService::acquire_local_snapshot_with_retry_(const int64_t expire_ts,
+                                                       SCN &snapshot)
+{
+  int ret = acquire_local_snapshot_(snapshot);
+  if (OB_GTS_NOT_READY == ret) {
+    SCN snapshot0;
+    SCN snapshot1;
+    const bool can_elr = share::server_is_write_enabled();
+    const bool has_deadline = expire_ts > 0;
+    const int64_t current_time = ObClockGenerator::getClock();
+    const int64_t MAX_WAIT_TIME_US = 1 * 1000 * 1000;
+    const int64_t timeout_us = has_deadline && expire_ts > current_time
+        ? min(MAX_WAIT_TIME_US, expire_ts - current_time)
+        : MAX_WAIT_TIME_US;
+    if (has_deadline && current_time >= expire_ts) {
+      ret = OB_TIMEOUT;
+    } else if (OB_FAIL(ts_mgr_->get_gts_sync(timeout_us, snapshot1))) {
+      if (OB_TIMEOUT == ret) {
+        ret = has_deadline && ObClockGenerator::getClock() >= expire_ts
+            ? OB_TIMEOUT
+            : OB_GTS_NOT_READY;
+      }
+    } else if (FALSE_IT(snapshot0 = tx_version_mgr_.get_max_commit_ts(can_elr))) {
+    } else if (!snapshot0.is_valid_and_not_min()) {
+      ret = OB_EAGAIN;
+    } else {
+      snapshot = SCN::max(snapshot0, snapshot1);
+    }
+  }
+  return ret;
+}
+
 /********************************************************************
  *
  * RPC and Message Handle

src/storage/tx/ob_trans_service_v4.h

@@ -144,7 +144,12 @@ int validate_snapshot_version_(const share::SCN snapshot,
                                const int64_t expire_ts,
                                ObLS &ls);
 int abort_write_state_(const ObTxDesc &tx_desc);
+int sync_acquire_local_snapshot_(ObTxDesc &tx,
+                                 const int64_t expire_ts,
+                                 share::SCN &snapshot);
 int acquire_local_snapshot_(share::SCN &snapshot);
+int acquire_local_snapshot_with_retry_(const int64_t expire_ts,
+                                       share::SCN &snapshot);
 int abort_write_ctx_(const ObTxDesc &tx_desc);
 
 int update_max_read_ts_(const share::SCN ts);

src/storage/tx/ob_ts_mgr.cpp

@@ -54,12 +54,13 @@ int ObTsMgr::get_gts(const MonotonicTs stc,
 int ObTsMgr::get_gts_sync(const int64_t timeout_us, SCN &scn)
 {
   int ret = OB_EAGAIN;
+  const int64_t RETRY_INTERVAL_US = 10 * 1000;
   const int64_t expire_ts = ObClockGenerator::getClock() + timeout_us;
   while (OB_EAGAIN == ret) {
     if (ObClockGenerator::getClock() >= expire_ts) {
       ret = OB_TIMEOUT;
     } else if (OB_EAGAIN == (ret = get_gts(scn))) {
-      ob_usleep(500);
+      ob_usleep(RETRY_INTERVAL_US);
     }
   }
   return ret;

src/storage/tx/ob_tx_api.cpp

@@ -587,7 +587,7 @@ int ObTransService::get_read_snapshot(ObTxDesc &tx,
     if (tx.isolation_ != isolation /*change isolation*/ ||
         !tx.snapshot_version_.is_valid()/*version invalid*/) {
       SCN version;
-      if (OB_FAIL(acquire_local_snapshot_(version))) {
+      if (OB_FAIL(sync_acquire_local_snapshot_(tx, expire_ts, version))) {
       } else if (!tx.is_write_fenced()
                   && !tx.tx_id_.is_valid()
                  && OB_FAIL(tx_desc_mgr_.add(tx))) {
@@ -606,7 +606,7 @@ int ObTransService::get_read_snapshot(ObTxDesc &tx,
       snapshot.uncertain_bound_ = tx.snapshot_uncertain_bound_;
     }
   } else { // RC isolation level
-    if (OB_FAIL(acquire_local_snapshot_(snapshot.core_.version_))) {
+    if (OB_FAIL(sync_acquire_local_snapshot_(tx, expire_ts, snapshot.core_.version_))) {
     } else {
       snapshot.uncertain_bound_ = 0;
       adjust_tx_snapshot_(tx, snapshot);
@@ -648,8 +648,7 @@ int ObTransService::get_read_snapshot_version(const int64_t expire_ts,
                                               SCN &snapshot_version)
 {
   int ret = OB_SUCCESS;
-  UNUSED(expire_ts);
-  ret = acquire_local_snapshot_(snapshot_version);
+  ret = acquire_local_snapshot_with_retry_(expire_ts, snapshot_version);
   return ret;
 }