diff --git a/include/pingcap/kv/Backoff.h b/include/pingcap/kv/Backoff.h index 4019767..5272c87 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 6905342..244a197 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 6b510c8..2b92dc3 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 is_bg_resolve); + 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 a5a10eb..00b0822 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 15cfa4c..b494b4b 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, false); +} + +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, false); +} + +int64_t LockResolver::resolveLocksImpl( + Backoffer & bo, + uint64_t caller_start_ts, + std::vector & locks, + std::vector & pushed, + bool for_write, + bool is_bg_resolve) { 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,20 @@ 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) { + // 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); + if (!is_bg_resolve && addPendingLocksForBgResolve(caller_start_ts, {lock})) + break; } bool exists = true; @@ -120,7 +197,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 +213,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 +221,6 @@ int64_t LockResolver::resolveLocks( { if (status.action != ::kvrpcpb::MinCommitTSPushed) { - push_fail = true; break; } pushed.push_back(lock->txn_id); @@ -155,10 +229,6 @@ int64_t LockResolver::resolveLocks( break; } } - if (push_fail) - { - pushed.clear(); - } return before_txn_expired.value(); } @@ -559,6 +629,87 @@ 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) + { + if (stopped.load()) + break; + + pingcap::kv::Backoffer bo(pingcap::kv::bgResolveLockMaxBackoff); + try + { + std::vector ignored; + resolveLocksImpl(bo, caller_start_ts, locks, ignored, false, true); + } + 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