diff --git a/tx_service/include/cc/cc_req_misc.h b/tx_service/include/cc/cc_req_misc.h index 51f4d050..75721903 100644 --- a/tx_service/include/cc/cc_req_misc.h +++ b/tx_service/include/cc/cc_req_misc.h @@ -871,11 +871,11 @@ struct WaitableCc : public RunOnTxProcessorCc void Reset(std::function task = {}, uint16_t core_cnt = 1) { - std::lock_guard lk(mux_); RunOnTxProcessorCc::Reset(std::move(task)); - unfinished_cnt_ = core_cnt; - error_code_ = CcErrorCode::NO_ERROR; + unfinished_cnt_.store(core_cnt, std::memory_order_relaxed); + error_code_.store(CcErrorCode::NO_ERROR, std::memory_order_relaxed); + waiting_.store(false, std::memory_order_relaxed); } void SetCoroCallbacks(const std::function *yield_fn, @@ -887,10 +887,13 @@ struct WaitableCc : public RunOnTxProcessorCc void Wait() { - std::unique_lock lk(mux_); - while (unfinished_cnt_) + uint64_t interval_us = 100; + constexpr uint64_t kMaxIntervalUs = 100000; + while (unfinished_cnt_.load(std::memory_order_acquire) > 0) { - cv_.wait(lk); + bthread_usleep(interval_us); + if ((interval_us << 1) < kMaxIntervalUs) + interval_us <<= 1; } } @@ -902,53 +905,47 @@ struct WaitableCc : public RunOnTxProcessorCc Wait(); return; } - std::unique_lock lk(mux_); - while (unfinished_cnt_) + while (unfinished_cnt_.load(std::memory_order_acquire) > 0) { waiting_.store(true, std::memory_order_release); - lk.unlock(); + if (unfinished_cnt_.load(std::memory_order_acquire) == 0) + { + waiting_.store(false, std::memory_order_release); + break; + } (*yield_fn)(); - lk.lock(); - waiting_.store(false, std::memory_order_release); } + waiting_.store(false, std::memory_order_release); } bool IsFinished() const { - std::lock_guard lk(mux_); - return unfinished_cnt_ == 0; + return unfinished_cnt_.load(std::memory_order_acquire) == 0; } bool IsError() const { - std::lock_guard lk(mux_); - return error_code_ != CcErrorCode::NO_ERROR; + return error_code_.load(std::memory_order_acquire) != + CcErrorCode::NO_ERROR; } CcErrorCode ErrorCode() const { - std::lock_guard lk(mux_); - return error_code_; + return error_code_.load(std::memory_order_acquire); } void AbortCcRequest(CcErrorCode error_code) override { - std::unique_lock lk(mux_); - unfinished_cnt_--; - error_code_ = error_code; - if (unfinished_cnt_ == 0) + error_code_.store(error_code, std::memory_order_release); + if (unfinished_cnt_.fetch_sub(1, std::memory_order_acq_rel) == 1) { - if (resume_fn_ != nullptr && - waiting_.load(std::memory_order_acquire)) + if (resume_fn_ != nullptr) { - waiting_.store(false, std::memory_order_release); auto *fn = resume_fn_; - lk.unlock(); - (*fn)(); - } - else if (resume_fn_ == nullptr) - { - cv_.notify_one(); + if (waiting_.exchange(false, std::memory_order_acq_rel)) + { + (*fn)(); + } } } } @@ -957,21 +954,15 @@ struct WaitableCc : public RunOnTxProcessorCc { if (RunOnTxProcessorCc::Execute(ccs)) { - std::unique_lock lk(mux_); - error_code_ = CcErrorCode::NO_ERROR; - if (--unfinished_cnt_ == 0) + if (unfinished_cnt_.fetch_sub(1, std::memory_order_acq_rel) == 1) { - if (resume_fn_ != nullptr && - waiting_.load(std::memory_order_acquire)) + if (resume_fn_ != nullptr) { - waiting_.store(false, std::memory_order_release); auto *fn = resume_fn_; - lk.unlock(); - (*fn)(); - } - else if (resume_fn_ == nullptr) - { - cv_.notify_one(); + if (waiting_.exchange(false, std::memory_order_acq_rel)) + { + (*fn)(); + } } } } @@ -989,11 +980,8 @@ struct WaitableCc : public RunOnTxProcessorCc } private: - mutable bthread::Mutex mux_; - bthread::ConditionVariable cv_; - - uint32_t unfinished_cnt_{0}; - CcErrorCode error_code_; + std::atomic unfinished_cnt_{0}; + std::atomic error_code_; // Coroutine yield/resume support const std::function *yield_fn_{nullptr}; diff --git a/tx_service/include/cc/cc_request.h b/tx_service/include/cc/cc_request.h index c6ef3a29..bdd9d30f 100644 --- a/tx_service/include/cc/cc_request.h +++ b/tx_service/include/cc/cc_request.h @@ -3159,11 +3159,7 @@ struct ActiveTxMaxTsCc : public CcRequestBase { public: ActiveTxMaxTsCc(size_t shard_cnt, NodeGroupId ng_id) - : active_tx_max_ts_(0), - mux_(), - cv_(), - unfinish_cnt_(shard_cnt), - cc_ng_id_(ng_id) + : active_tx_max_ts_(0), unfinish_cnt_(shard_cnt), cc_ng_id_(ng_id) { } @@ -3181,11 +3177,7 @@ struct ActiveTxMaxTsCc : public CcRequestBase old_val, shard_active_tx_max_ts, std::memory_order_acq_rel)) ; - std::unique_lock lk(mux_); - if (--unfinish_cnt_ == 0) - { - cv_.notify_one(); - } + unfinish_cnt_.fetch_sub(1, std::memory_order_acq_rel); // return false since ActiveTxMaxTsCc is not reused and does not need // to call CcRequestBase::Free @@ -3194,10 +3186,13 @@ struct ActiveTxMaxTsCc : public CcRequestBase void Wait() { - std::unique_lock lk(mux_); - while (unfinish_cnt_ > 0) + uint64_t interval_us = 100; + constexpr uint64_t kMaxIntervalUs = 100000; + while (unfinish_cnt_.load(std::memory_order_acquire) > 0) { - cv_.wait(lk); + bthread_usleep(interval_us); + if ((interval_us << 1) < kMaxIntervalUs) + interval_us <<= 1; } } @@ -3208,9 +3203,7 @@ struct ActiveTxMaxTsCc : public CcRequestBase private: std::atomic active_tx_max_ts_; - bthread::Mutex mux_; - bthread::ConditionVariable cv_; - size_t unfinish_cnt_; + std::atomic_size_t unfinish_cnt_; NodeGroupId cc_ng_id_; }; @@ -8410,8 +8403,9 @@ struct DbSizeCc : public CcRequestBase Clear(); table_names_ = table_names; - total_ref_cnt_ = local_ref_cnt + remote_ref_cnt; - remote_ref_cnt_ = remote_ref_cnt; + total_ref_cnt_.store(local_ref_cnt + remote_ref_cnt, + std::memory_order_relaxed); + remote_ref_cnt_.store(remote_ref_cnt, std::memory_order_relaxed); total_obj_sizes_.resize(table_names_->size(), 0); } @@ -8433,13 +8427,7 @@ struct DbSizeCc : public CcRequestBase } } - std::unique_lock lk(mux_); - if (--total_ref_cnt_ == 0) - { - cv_.notify_one(); - } - - return false; + return OnLocalRefFinished(); } std::vector GetTotalObjSizes() @@ -8473,13 +8461,7 @@ struct DbSizeCc : public CcRequestBase idx, total_obj_sizes[idx], std::memory_order_relaxed); } - std::unique_lock lk(mux_); - --remote_ref_cnt_; - --total_ref_cnt_; - if (total_ref_cnt_ == 0) - { - cv_.notify_one(); - } + OnRemoteRefFinished(); } int32_t GetTerm() @@ -8496,25 +8478,35 @@ struct DbSizeCc : public CcRequestBase total_obj_sizes_.clear(); total_obj_sizes_.shrink_to_fit(); - total_ref_cnt_ = 0; - remote_ref_cnt_ = 0; + total_ref_cnt_.store(0, std::memory_order_relaxed); + remote_ref_cnt_.store(0, std::memory_order_relaxed); table_names_ = nullptr; vct_ng_id_.clear(); } void Wait() { - const uint64_t MAX_WAIT_TS = 2000000; - std::unique_lock lk(mux_); + uint64_t remaining_wait_us = 2000000; + uint64_t interval_us = 100; + constexpr uint64_t kMaxIntervalUs = 100000; - while (total_ref_cnt_ > 0) + while (total_ref_cnt_.load(std::memory_order_acquire) > 0) { - int wait_res = cv_.wait_for(lk, MAX_WAIT_TS); - if (wait_res == ETIMEDOUT && total_ref_cnt_ <= remote_ref_cnt_) + bthread_usleep(interval_us); + if (total_ref_cnt_.load(std::memory_order_acquire) <= + remote_ref_cnt_.load(std::memory_order_acquire)) { - LOG(WARNING) << "Waitting timeout for dbsize"; - break; + remaining_wait_us = remaining_wait_us > interval_us + ? remaining_wait_us - interval_us + : 0; + if (remaining_wait_us == 0) + { + LOG(WARNING) << "Waiting timeout for dbsize"; + break; + } } + if ((interval_us << 1) < kMaxIntervalUs) + interval_us <<= 1; } } @@ -8538,15 +8530,23 @@ struct DbSizeCc : public CcRequestBase return total_obj_sizes_.size(); } - bthread::Mutex mux_; - bthread::ConditionVariable cv_; - private: std::vector total_obj_sizes_; protected: - size_t total_ref_cnt_{0}; - size_t remote_ref_cnt_{0}; + bool OnLocalRefFinished() + { + return total_ref_cnt_.fetch_sub(1, std::memory_order_acq_rel) == 1; + } + + bool OnRemoteRefFinished() + { + remote_ref_cnt_.fetch_sub(1, std::memory_order_acq_rel); + return total_ref_cnt_.fetch_sub(1, std::memory_order_acq_rel) == 1; + } + + std::atomic_size_t total_ref_cnt_{0}; + std::atomic_size_t remote_ref_cnt_{0}; int32_t term_{0}; std::vector vct_ng_id_; std::vector *table_names_{nullptr}; diff --git a/tx_service/src/remote/remote_cc_request.cpp b/tx_service/src/remote/remote_cc_request.cpp index 7b24630b..80916182 100644 --- a/tx_service/src/remote/remote_cc_request.cpp +++ b/tx_service/src/remote/remote_cc_request.cpp @@ -2317,7 +2317,7 @@ txservice::remote::RemoteDbSizeCc::RemoteDbSizeCc() post_lambda_ = [this]() { - assert(total_ref_cnt_ == 0); + assert(total_ref_cnt_.load(std::memory_order_acquire) == 0); output_msg_.set_handler_addr(input_msg_->handler_addr()); output_msg_.set_txm_addr(input_msg_->txm_addr()); @@ -2362,8 +2362,8 @@ void txservice::remote::RemoteDbSizeCc::Reset( DbSizeCc::Reset(&redis_table_names_, core_cnt, 0); assert(table_names_ == &redis_table_names_); - assert(total_ref_cnt_ == core_cnt); - assert(remote_ref_cnt_ == 0); + assert(total_ref_cnt_.load(std::memory_order_relaxed) == core_cnt); + assert(remote_ref_cnt_.load(std::memory_order_relaxed) == 0); AddLocalNodeGroupId(cmds_req.node_group_id()); @@ -2388,10 +2388,8 @@ bool txservice::remote::RemoteDbSizeCc::Execute(CcShard &ccs) } } - std::unique_lock lk(mux_); - assert(remote_ref_cnt_ == 0); - --total_ref_cnt_; - if (total_ref_cnt_ == 0) + assert(remote_ref_cnt_.load(std::memory_order_relaxed) == 0); + if (OnLocalRefFinished()) { table_names_ = nullptr; redis_table_names_.clear();