Support bypassing resolved locks for read

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<Backoff>;

include/pingcap/kv/Cluster.h

@@ -15,6 +15,8 @@ namespace pingcap
 namespace kv
 {
 constexpr int oracle_update_interval = 2000;
+constexpr int mock_cluster_background_workers = 3; // RPC maintenance, MPP prober, and lock resolver.
+constexpr int cluster_background_workers = 4; // RPC maintenance, MPP prober, region cache, and lock resolver.
 // Cluster represents a tikv+pd cluster.
 
 struct Cluster
@@ -35,7 +37,9 @@ struct Cluster
     Cluster()
         : pd_client(std::make_shared<pd::MockPDClient>())
         , rpc_client(std::make_unique<RpcClient>(pd_client, ClusterConfig{}))
-        , thread_pool(std::make_unique<pingcap::common::FixedThreadPool>(2))
+        , oracle(std::make_unique<pd::Oracle>(pd_client, std::chrono::milliseconds(oracle_update_interval)))
+        , lock_resolver(std::make_unique<LockResolver>(this))
+        , thread_pool(std::make_unique<pingcap::common::FixedThreadPool>(mock_cluster_background_workers))
         , mpp_prober(std::make_unique<common::MPPProber>(this))
     {
         startBackgroundTasks();
@@ -48,7 +52,7 @@ struct Cluster
         , oracle(std::make_unique<pd::Oracle>(pd_client, std::chrono::milliseconds(oracle_update_interval)))
         , lock_resolver(std::make_unique<LockResolver>(this))
         , api_version(config.api_version)
-        , thread_pool(std::make_unique<pingcap::common::FixedThreadPool>(3))
+        , thread_pool(std::make_unique<pingcap::common::FixedThreadPool>(cluster_background_workers))
         , mpp_prober(std::make_unique<common::MPPProber>(this))
     {
         startBackgroundTasks();
@@ -64,6 +68,8 @@ struct Cluster
     // (e.g. background threads) that cluster object holds so as to exit elegantly.
     ~Cluster()
     {
+        if (lock_resolver)
+            lock_resolver->stopBgResolve();
         rpc_client->stop();
         mpp_prober->stop();
         if (region_cache)

include/pingcap/kv/LockResolver.h

@@ -5,9 +5,17 @@
 #include <pingcap/Log.h>
 #include <pingcap/kv/RegionCache.h>
 
+#include <atomic>
+#include <condition_variable>
+#include <memory>
+#include <mutex>
 #include <optional>
 #include <queue>
+#include <shared_mutex>
 #include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
 
 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<AsyncResolveData>;
 
+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<LockPtr> & 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<LockPtr> & locks, std::vector<uint64_t> & 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<uint64_t, std::vector<LockPtr>> & locks,
+        std::vector<uint64_t> & 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<LockPtr> & locks);
 
 private:
+    int64_t resolveLocksImpl(
+        Backoffer & bo,
+        uint64_t caller_start_ts,
+        std::vector<LockPtr> & locks,
+        std::vector<uint64_t> & pushed,
+        bool for_write,
+        bool is_bg_resolve);
+
     void saveResolved(uint64_t txn_id, const TxnStatus & status)
     {
         std::unique_lock<std::shared_mutex> lk(mu);
@@ -291,6 +327,13 @@ class LockResolver
     std::unordered_map<int64_t, TxnStatus> resolved;
     std::queue<int64_t> cached;
 
+    std::mutex bg_mutex;
+    std::condition_variable bg_cv;
+    std::atomic<bool> stopped{false};
+    std::vector<std::pair<uint64_t, std::vector<LockPtr>>> pending_locks;
+    size_t pending_lock_count = 0;
+    bool pending_locks_full_logged = false;
+
     Logger * log;
 };
 

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

src/kv/LockResolver.cc

@@ -1,4 +1,5 @@
 #include <pingcap/RedactHelpers.h>
+#include <pingcap/kv/Backoff.h>
 #include <pingcap/kv/LockResolver.h>
 #include <pingcap/kv/RegionClient.h>
 
@@ -34,7 +35,68 @@ std::string Lock::toDebugString() const
 
 int64_t LockResolver::resolveLocks(Backoffer & bo, uint64_t caller_start_ts, std::vector<LockPtr> & locks, std::vector<uint64_t> & 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<uint64_t, std::vector<LockPtr>> & locks,
+    std::vector<uint64_t> & 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<LockPtr> & locks,
     std::vector<uint64_t> & 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<LockPtr> & locks,
+    std::vector<uint64_t> & pushed,
+    bool for_write,
+    bool is_bg_resolve)
 {
     TxnExpireTime before_txn_expired;
     if (locks.empty())
         return before_txn_expired.value();
     std::unordered_map<uint64_t, std::unordered_set<RegionVerID>> 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<std::pair<uint64_t, std::vector<LockPtr>>> 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<uint64_t> 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<LockPtr> & 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