From 12181a557f8a8504132732448851d35481d11443 Mon Sep 17 00:00:00 2001 From: xufei Date: Mon, 20 Jul 2026 21:06:02 +0800 Subject: [PATCH 1/4] Add try bypass lock helper (#245) Signed-off-by: xufei --- include/pingcap/kv/Backoff.h | 1 + include/pingcap/kv/Cluster.h | 2 + include/pingcap/kv/LockResolver.h | 43 ++++++++ src/kv/Cluster.cc | 6 ++ src/kv/LockResolver.cc | 167 ++++++++++++++++++++++++++++-- 5 files changed, 208 insertions(+), 11 deletions(-) diff --git a/include/pingcap/kv/Backoff.h b/include/pingcap/kv/Backoff.h index 40197676..5272c875 100644 --- a/include/pingcap/kv/Backoff.h +++ b/include/pingcap/kv/Backoff.h @@ -100,6 +100,7 @@ constexpr int splitRegionBackoff = 20000; constexpr int cleanupMaxBackoff = 20000; constexpr int copBuildTaskMaxBackoff = 5000; constexpr int copNextMaxBackoff = 60000; +constexpr int bgResolveLockMaxBackoff = 5000; constexpr int pessimisticLockMaxBackoff = 20000; using BackoffPtr = std::shared_ptr; diff --git a/include/pingcap/kv/Cluster.h b/include/pingcap/kv/Cluster.h index 69053423..244a1974 100644 --- a/include/pingcap/kv/Cluster.h +++ b/include/pingcap/kv/Cluster.h @@ -68,6 +68,8 @@ struct Cluster mpp_prober->stop(); if (region_cache) region_cache->stop(); + if (lock_resolver) + lock_resolver->stopBgResolve(); thread_pool->stop(); } diff --git a/include/pingcap/kv/LockResolver.h b/include/pingcap/kv/LockResolver.h index 6b510c8d..d1ef67aa 100644 --- a/include/pingcap/kv/LockResolver.h +++ b/include/pingcap/kv/LockResolver.h @@ -5,9 +5,17 @@ #include #include +#include +#include +#include +#include #include #include +#include #include +#include +#include +#include namespace pingcap { @@ -38,6 +46,7 @@ struct TxnStatus }; constexpr size_t resolvedCacheSize = 2048; +constexpr size_t maxPendingLocksForBgResolve = 4096; constexpr int bigTxnThreshold = 16; const uint64_t defaultLockTTL = 3000; @@ -194,6 +203,12 @@ struct AsyncResolveData using AsyncResolveDataPtr = std::shared_ptr; +struct TryGetBypassLockResult +{ + bool has_pending_resolve = false; + bool need_wait_bg_resolve = false; +}; + // LockResolver resolves locks and also caches resolved txn status. class LockResolver { @@ -208,6 +223,10 @@ class LockResolver cluster = cluster_; } + void backgroundResolve(); + bool addPendingLocksForBgResolve(uint64_t caller_start_ts, const std::vector & locks); + void stopBgResolve(); + // resolveLocks tries to resolve Locks. The resolving process is in 3 steps: // 1) Use the `lockTTL` to pick up all expired locks. Only locks that are too // old are considered orphan locks and will be handled later. If all locks @@ -220,6 +239,15 @@ class LockResolver int64_t resolveLocks(Backoffer & bo, uint64_t caller_start_ts, std::vector & locks, std::vector & pushed); + // tryGetBypassLock checks the status of the transactions which own the locks in `locks`, and collect the txn ids which can be bypassed. + // It is a best-effort optimization and will not synchronously resolve locks in caller's thread. + // The result tells the caller whether background resolve is scheduled and whether it is worth waiting for. + TryGetBypassLockResult tryGetBypassLock( + Backoffer & bo, + uint64_t caller_start_ts, + const std::unordered_map> & locks, + std::vector & bypass_lock_ts); + int64_t resolveLocks( Backoffer & bo, uint64_t caller_start_ts, @@ -230,6 +258,14 @@ class LockResolver int64_t resolveLocksForWrite(Backoffer & bo, uint64_t caller_start_ts, std::vector & locks); private: + int64_t resolveLocksImpl( + Backoffer & bo, + uint64_t caller_start_ts, + std::vector & locks, + std::vector & pushed, + bool for_write, + bool allow_bypass); + void saveResolved(uint64_t txn_id, const TxnStatus & status) { std::unique_lock lk(mu); @@ -291,6 +327,13 @@ class LockResolver std::unordered_map resolved; std::queue cached; + std::mutex bg_mutex; + std::condition_variable bg_cv; + std::atomic stopped{false}; + std::vector>> pending_locks; + size_t pending_lock_count = 0; + bool pending_locks_full_logged = false; + Logger * log; }; diff --git a/src/kv/Cluster.cc b/src/kv/Cluster.cc index a5a10eb0..00b08228 100644 --- a/src/kv/Cluster.cc +++ b/src/kv/Cluster.cc @@ -45,6 +45,12 @@ void Cluster::startBackgroundTasks() region_cache->updateCachePeriodically(); }); } + if (lock_resolver) + { + thread_pool->enqueue([this] { + lock_resolver->backgroundResolve(); + }); + } } } // namespace kv diff --git a/src/kv/LockResolver.cc b/src/kv/LockResolver.cc index 15cfa4cd..00434969 100644 --- a/src/kv/LockResolver.cc +++ b/src/kv/LockResolver.cc @@ -1,4 +1,5 @@ #include +#include #include #include @@ -34,7 +35,68 @@ std::string Lock::toDebugString() const int64_t LockResolver::resolveLocks(Backoffer & bo, uint64_t caller_start_ts, std::vector & locks, std::vector & pushed) { - return resolveLocks(bo, caller_start_ts, locks, pushed, false); + return resolveLocksImpl(bo, caller_start_ts, locks, pushed, false, true); +} + +TryGetBypassLockResult LockResolver::tryGetBypassLock( + Backoffer & bo, + uint64_t caller_start_ts, + const std::unordered_map> & locks, + std::vector & bypass_lock_ts) +{ + TryGetBypassLockResult result; + try + { + bypass_lock_ts.reserve(bypass_lock_ts.size() + locks.size()); + for (const auto & [txn_id, txn_locks] : locks) + { + if (txn_locks.empty()) + continue; + + TxnStatus status; + try + { + status = getTxnStatusFromLock(bo, txn_locks.front(), caller_start_ts, false); + } + catch (Exception & e) + { + log->warning("get txn status failed: " + e.displayText()); + continue; + } + + if (status.ttl == 0) + { + const bool is_async_commit + = status.primary_lock.has_value() && status.primary_lock->use_async_commit(); + const bool can_bypass = canBypassLockForRead(status, caller_start_ts); + if (can_bypass) + { + bypass_lock_ts.push_back(txn_id); + } + + if (!is_async_commit && (status.isCommitted() || status.isRolledBack()) + && addPendingLocksForBgResolve(caller_start_ts, txn_locks)) + { + result.has_pending_resolve = true; + if (!can_bypass) + result.need_wait_bg_resolve = true; + } + } + else if (status.action == ::kvrpcpb::MinCommitTSPushed) + { + bypass_lock_ts.push_back(txn_id); + } + } + } + catch (Exception & e) + { + log->warning("tryGetBypassLock failed: " + e.displayText()); + } + catch (...) + { + log->warning("tryGetBypassLock failed"); + } + return result; } int64_t LockResolver::resolveLocks( @@ -43,12 +105,22 @@ int64_t LockResolver::resolveLocks( std::vector & locks, std::vector & pushed, bool for_write) +{ + return resolveLocksImpl(bo, caller_start_ts, locks, pushed, for_write, true); +} + +int64_t LockResolver::resolveLocksImpl( + Backoffer & bo, + uint64_t caller_start_ts, + std::vector & locks, + std::vector & pushed, + bool for_write, + bool allow_bypass) { TxnExpireTime before_txn_expired; if (locks.empty()) return before_txn_expired.value(); std::unordered_map> clean_txns; - bool push_fail = false; if (!for_write) { pushed.reserve(locks.size()); @@ -68,15 +140,17 @@ int64_t LockResolver::resolveLocks( { log->warning("get txn status failed: " + e.displayText()); before_txn_expired.update(0); - pushed.clear(); return before_txn_expired.value(); } if (status.ttl == 0) { - if (!for_write && canBypassLockForRead(status, caller_start_ts)) + if (!for_write && allow_bypass && canBypassLockForRead(status, caller_start_ts)) { pushed.push_back(lock->txn_id); + // Let current reads bypass the lock while cleaning it up asynchronously. + if (addPendingLocksForBgResolve(caller_start_ts, {lock})) + break; } bool exists = true; @@ -120,7 +194,6 @@ int64_t LockResolver::resolveLocks( { log->warning("resolve txn failed: " + e.displayText()); before_txn_expired.update(0); - pushed.clear(); return before_txn_expired.value(); } } @@ -137,7 +210,6 @@ int64_t LockResolver::resolveLocks( if (lock->lock_type != ::kvrpcpb::PessimisticLock && lock->txn_id > caller_start_ts) { log->warning("write conflict detected"); - pushed.clear(); // TODO: throw write conflict exception throw Exception("write conflict", ErrorCodes::UnknownError); } @@ -146,7 +218,6 @@ int64_t LockResolver::resolveLocks( { if (status.action != ::kvrpcpb::MinCommitTSPushed) { - push_fail = true; break; } pushed.push_back(lock->txn_id); @@ -155,10 +226,6 @@ int64_t LockResolver::resolveLocks( break; } } - if (push_fail) - { - pushed.clear(); - } return before_txn_expired.value(); } @@ -559,6 +626,84 @@ TxnStatus LockResolver::getTxnStatusFromLock(Backoffer & bo, LockPtr lock, uint6 } } +void LockResolver::backgroundResolve() +{ + while (!stopped.load()) + { + std::vector>> to_resolve; + { + std::unique_lock lk(bg_mutex); + bg_cv.wait(lk, [this] { return !pending_locks.empty() || stopped.load(); }); + if (stopped.load()) + return; + pending_locks.swap(to_resolve); + pending_lock_count = 0; + pending_locks_full_logged = false; + } + + for (auto & [caller_start_ts, locks] : to_resolve) + { + pingcap::kv::Backoffer bo(pingcap::kv::bgResolveLockMaxBackoff); + try + { + std::vector ignored; + resolveLocksImpl(bo, caller_start_ts, locks, ignored, false, false); + } + catch (Exception & e) + { + log->warning("background resolve lock failed: " + e.displayText()); + } + catch (...) + { + log->warning("background resolve lock failed"); + } + } + } +} + +bool LockResolver::addPendingLocksForBgResolve(uint64_t caller_start_ts, const std::vector & locks) +{ + if (locks.empty()) + return true; + + bool should_log = false; + size_t pending_count = 0; + { + std::unique_lock lk(bg_mutex); + if (stopped.load()) + return false; + + if (locks.size() > maxPendingLocksForBgResolve - pending_lock_count) + { + should_log = !pending_locks_full_logged; + pending_locks_full_logged = true; + pending_count = pending_lock_count; + } + else + { + pending_locks.push_back({caller_start_ts, locks}); + pending_lock_count += locks.size(); + bg_cv.notify_one(); + return true; + } + } + + if (should_log) + { + log->warning( + "background resolve lock queue is full, drop pending locks, queue_limit=" + + std::to_string(maxPendingLocksForBgResolve) + " pending_lock_count=" + std::to_string(pending_count) + + " dropped_lock_count=" + std::to_string(locks.size())); + } + return false; +} + +void LockResolver::stopBgResolve() +{ + std::unique_lock lk(bg_mutex); + stopped.store(true); + bg_cv.notify_all(); +} } // namespace kv } // namespace pingcap From 4aa7ef349199dc5521928ce9df964e418b1a5c0e Mon Sep 17 00:00:00 2001 From: xufei Date: Wed, 22 Jul 2026 13:26:03 +0800 Subject: [PATCH 2/4] save work Signed-off-by: xufei --- src/kv/LockResolver.cc | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/kv/LockResolver.cc b/src/kv/LockResolver.cc index 00434969..47d8a9de 100644 --- a/src/kv/LockResolver.cc +++ b/src/kv/LockResolver.cc @@ -643,6 +643,9 @@ void LockResolver::backgroundResolve() for (auto & [caller_start_ts, locks] : to_resolve) { + if (stopped.load()) + break; + pingcap::kv::Backoffer bo(pingcap::kv::bgResolveLockMaxBackoff); try { From 31e2ce81845956fb2c51658ac311c2c0e8cab626 Mon Sep 17 00:00:00 2001 From: xufei Date: Wed, 22 Jul 2026 13:45:32 +0800 Subject: [PATCH 3/4] Refine bypass lock background resolve behavior Signed-off-by: xufei --- include/pingcap/kv/LockResolver.h | 2 +- src/kv/LockResolver.cc | 13 ++++++------- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/include/pingcap/kv/LockResolver.h b/include/pingcap/kv/LockResolver.h index d1ef67aa..2b92dc3a 100644 --- a/include/pingcap/kv/LockResolver.h +++ b/include/pingcap/kv/LockResolver.h @@ -264,7 +264,7 @@ class LockResolver std::vector & locks, std::vector & pushed, bool for_write, - bool allow_bypass); + bool is_bg_resolve); void saveResolved(uint64_t txn_id, const TxnStatus & status) { diff --git a/src/kv/LockResolver.cc b/src/kv/LockResolver.cc index 47d8a9de..30f7b6e8 100644 --- a/src/kv/LockResolver.cc +++ b/src/kv/LockResolver.cc @@ -35,7 +35,7 @@ std::string Lock::toDebugString() const int64_t LockResolver::resolveLocks(Backoffer & bo, uint64_t caller_start_ts, std::vector & locks, std::vector & pushed) { - return resolveLocksImpl(bo, caller_start_ts, locks, pushed, false, true); + return resolveLocksImpl(bo, caller_start_ts, locks, pushed, false, false); } TryGetBypassLockResult LockResolver::tryGetBypassLock( @@ -106,7 +106,7 @@ int64_t LockResolver::resolveLocks( std::vector & pushed, bool for_write) { - return resolveLocksImpl(bo, caller_start_ts, locks, pushed, for_write, true); + return resolveLocksImpl(bo, caller_start_ts, locks, pushed, for_write, false); } int64_t LockResolver::resolveLocksImpl( @@ -115,7 +115,7 @@ int64_t LockResolver::resolveLocksImpl( std::vector & locks, std::vector & pushed, bool for_write, - bool allow_bypass) + bool is_bg_resolve) { TxnExpireTime before_txn_expired; if (locks.empty()) @@ -145,11 +145,10 @@ int64_t LockResolver::resolveLocksImpl( if (status.ttl == 0) { - if (!for_write && allow_bypass && canBypassLockForRead(status, caller_start_ts)) + if (!for_write && canBypassLockForRead(status, caller_start_ts)) { pushed.push_back(lock->txn_id); - // Let current reads bypass the lock while cleaning it up asynchronously. - if (addPendingLocksForBgResolve(caller_start_ts, {lock})) + if (!is_bg_resolve && addPendingLocksForBgResolve(caller_start_ts, {lock})) break; } @@ -650,7 +649,7 @@ void LockResolver::backgroundResolve() try { std::vector ignored; - resolveLocksImpl(bo, caller_start_ts, locks, ignored, false, false); + resolveLocksImpl(bo, caller_start_ts, locks, ignored, false, true); } catch (Exception & e) { From 584b1ffa4188c4eb444c2f5c049db806f8f9ae24 Mon Sep 17 00:00:00 2001 From: xufei Date: Wed, 22 Jul 2026 13:50:18 +0800 Subject: [PATCH 4/4] Document bypass lock resolve behavior Signed-off-by: xufei --- src/kv/LockResolver.cc | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/kv/LockResolver.cc b/src/kv/LockResolver.cc index 30f7b6e8..b494b4b1 100644 --- a/src/kv/LockResolver.cc +++ b/src/kv/LockResolver.cc @@ -145,6 +145,10 @@ int64_t LockResolver::resolveLocksImpl( if (status.ttl == 0) { + // Only reads can bypass locks that are invisible to the current snapshot. The txn_id is always + // returned in `pushed` for later requests to skip the same lock. Foreground reads try to hand the + // actual cleanup to the background worker; if enqueue fails, they fall through and resolve it + // synchronously. Background cleanup must not enqueue again, so it always falls through here. if (!for_write && canBypassLockForRead(status, caller_start_ts)) { pushed.push_back(lock->txn_id);