Bypass locks invisible to read snapshot (#243)

ci/build-test.sh

@@ -7,6 +7,36 @@ SCRIPTPATH="$( cd "$(dirname "$0")" ; pwd -P )"
 SRCPATH=$(cd $SCRIPTPATH/..; pwd -P)
 NPROC=$(nproc || grep -c ^processor /proc/cpuinfo)
 
+waitForMockTiKV() {
+    local max_wait_seconds=30
+
+    for ((i = 1; i <= max_wait_seconds; i++)); do
+        if ! kill -0 "$mock_kv_pid" 2>/dev/null; then
+            echo "mock-tikv exited before becoming ready"
+            wait "$mock_kv_pid" || true
+            return 1
+        fi
+
+        if (echo > /dev/tcp/127.0.0.1/2378) >/dev/null 2>&1; then
+            echo "mock-tikv is ready"
+            return 0
+        fi
+
+        sleep 1
+    done
+
+    echo "mock-tikv is not ready after ${max_wait_seconds}s"
+    return 1
+}
+
+mock_kv_pid=""
+cleanupMockTiKV() {
+    if [ -n "$mock_kv_pid" ]; then
+        kill -9 "$mock_kv_pid" 2>/dev/null || true
+    fi
+}
+trap cleanupMockTiKV EXIT
+
 build_dir="$SRCPATH/build"
 mkdir -p $build_dir && cd $build_dir
 cmake "$SRCPATH" \
@@ -15,9 +45,6 @@ make -j $NPROC
 
 nohup /mock-tikv/bin/mock-tikv &
 mock_kv_pid=$!
+waitForMockTiKV
 
 cd "$build_dir" && make test
-
-kill -9 $mock_kv_pid
-
-

include/pingcap/kv/LockResolver.h

@@ -20,25 +20,20 @@ struct TxnStatus
 {
     uint64_t ttl = 0;
     uint64_t commit_ts = 0;
-    ::kvrpcpb::Action action;
+    ::kvrpcpb::Action action = ::kvrpcpb::Action::NoAction;
     std::optional<::kvrpcpb::LockInfo> primary_lock;
     bool isCommitted() const { return ttl == 0 && commit_ts > 0; }
 
+    bool isRolledBack() const
+    {
+        return ttl == 0 && commit_ts == 0
+            && (action == kvrpcpb::Action::NoAction || action == kvrpcpb::Action::LockNotExistRollback
+                || action == kvrpcpb::Action::TTLExpireRollback);
+    }
+
     bool isCacheable() const
     {
-        if (isCommitted())
-        {
-            return true;
-        }
-        if (ttl == 0)
-        {
-            if (action == kvrpcpb::Action::NoAction || action == kvrpcpb::Action::LockNotExistRollback
-                || action == kvrpcpb::Action::TTLExpireRollback)
-            {
-                return true;
-            }
-        }
-        return false;
+        return isCommitted() || isRolledBack();
     }
 };
 

src/kv/LockResolver.cc

@@ -6,6 +6,24 @@ namespace pingcap
 {
 namespace kv
 {
+namespace
+{
+bool canBypassLockForRead(const TxnStatus & status, uint64_t caller_start_ts)
+{
+    if (status.ttl != 0)
+        return false;
+
+    if (status.isCommitted())
+        return status.commit_ts > caller_start_ts;
+
+    // Expired async-commit locks need resolveLockAsync to determine the final status.
+    if (status.primary_lock.has_value() && status.primary_lock->use_async_commit())
+        return false;
+
+    return status.isRolledBack();
+}
+} // namespace
+
 std::string Lock::toDebugString() const
 {
     return "key: " + Redact::keyToDebugString(key) + " primary: " + Redact::keyToDebugString(primary)
@@ -56,6 +74,11 @@ int64_t LockResolver::resolveLocks(
 
             if (status.ttl == 0)
             {
+                if (!for_write && canBypassLockForRead(status, caller_start_ts))
+                {
+                    pushed.push_back(lock->txn_id);
+                }
+
                 bool exists = true;
                 if (clean_txns.find(lock->txn_id) == clean_txns.end())
                 {

src/kv/Scanner.cc

@@ -73,6 +73,10 @@ void Scanner::getData(Backoffer & bo)
         auto * context = request.mutable_context();
         context->set_priority(::kvrpcpb::Normal);
         context->set_not_fill_cache(false);
+        for (auto ts : snap.min_commit_ts_pushed.getTimestamps())
+        {
+            context->add_resolved_locks(ts);
+        }
 
         kvrpcpb::ScanResponse response;
         try
@@ -94,6 +98,10 @@ void Scanner::getData(Backoffer & bo)
             std::vector<LockPtr> locks{lock};
             std::vector<uint64_t> pushed{};
             auto ms_before_expired = snap.cluster->lock_resolver->resolveLocks(bo, snap.version, locks, pushed);
+            if (!pushed.empty())
+            {
+                snap.min_commit_ts_pushed.addTimestamps(pushed);
+            }
             if (ms_before_expired > 0)
             {
                 bo.backoffWithMaxSleep(

src/test/lock_resolve_test.cc

@@ -37,6 +37,51 @@ class TestWithLockResolve : public testing::Test
     ClusterPtr control_cluster;
 };
 
+namespace
+{
+void writeRowsAndSplit(Cluster * test_cluster, Cluster * control_cluster)
+{
+    Txn txn(test_cluster);
+
+    txn.set("abc", "1");
+    txn.set("abd", "2");
+    txn.set("abe", "3");
+    txn.set("abf", "4");
+    txn.set("abg", "5");
+    txn.set("abz", "6");
+    txn.commit();
+    control_cluster->splitRegion("abf");
+}
+
+uint64_t leaveSecondaryLocksAfterPrimaryCommitted(Cluster * test_cluster)
+{
+    fiu_enable("rest commit fail", 1, nullptr, FIU_ONETIME);
+
+    Txn txn(test_cluster);
+    txn.set("abc", "6");
+    txn.set("abd", "5");
+    txn.set("abe", "4");
+    txn.set("abf", "3");
+    txn.set("abg", "2");
+    txn.set("abz", "1");
+    const auto txn_id = txn.start_ts;
+    txn.commit();
+    return txn_id;
+}
+
+LockPtr makeLock(const std::string & key, const std::string & primary, uint64_t txn_id, uint64_t ttl = defaultLockTTL, uint64_t txn_size = 1)
+{
+    kvrpcpb::LockInfo lock_info;
+    lock_info.set_key(key);
+    lock_info.set_primary_lock(primary);
+    lock_info.set_lock_version(txn_id);
+    lock_info.set_lock_ttl(ttl);
+    lock_info.set_txn_size(txn_size);
+    lock_info.set_lock_type(::kvrpcpb::Put);
+    return std::make_shared<Lock>(lock_info);
+}
+} // namespace
+
 TEST_F(TestWithLockResolve, testResolveLockGet)
 {
     // Write First Time and Split int two regions.
@@ -123,6 +168,82 @@ TEST_F(TestWithLockResolve, testResolveLockGet)
     }
 }
 
+TEST_F(TestWithLockResolve, testResolveLocksBypassesCommittedAfterRead)
+{
+    writeRowsAndSplit(test_cluster.get(), control_cluster.get());
+    const auto txn_id = leaveSecondaryLocksAfterPrimaryCommitted(test_cluster.get());
+
+    auto lock = makeLock("abf", "abc", txn_id, defaultLockTTL, 6);
+    std::vector<LockPtr> locks{lock};
+    std::vector<uint64_t> pushed;
+    Backoffer bo(kv::copNextMaxBackoff);
+
+    const auto before_expired = test_cluster->lock_resolver->resolveLocks(bo, txn_id, locks, pushed);
+
+    ASSERT_EQ(before_expired, 0);
+    ASSERT_EQ(pushed.size(), 1);
+    ASSERT_EQ(pushed[0], txn_id);
+
+    Snapshot snapshot(test_cluster.get(), txn_id);
+    snapshot.min_commit_ts_pushed.addTimestamps(pushed);
+    ASSERT_EQ(snapshot.Get("abf"), "4");
+}
+
+TEST_F(TestWithLockResolve, testScannerBypassesCommittedAfterRead)
+{
+    writeRowsAndSplit(test_cluster.get(), control_cluster.get());
+    const auto txn_id = leaveSecondaryLocksAfterPrimaryCommitted(test_cluster.get());
+
+    Snapshot snapshot(test_cluster.get(), txn_id);
+    auto scanner = snapshot.Scan("abf", "abz");
+
+    ASSERT_TRUE(scanner.valid);
+    ASSERT_EQ(scanner.key(), "abf");
+    ASSERT_EQ(scanner.value(), "4");
+
+    scanner.next();
+    ASSERT_TRUE(scanner.valid);
+    ASSERT_EQ(scanner.key(), "abg");
+    ASSERT_EQ(scanner.value(), "5");
+
+    scanner.next();
+    ASSERT_FALSE(scanner.valid);
+}
+
+TEST_F(TestWithLockResolve, testResolveLocksResolvesCommittedBeforeRead)
+{
+    writeRowsAndSplit(test_cluster.get(), control_cluster.get());
+    const auto txn_id = leaveSecondaryLocksAfterPrimaryCommitted(test_cluster.get());
+    const auto read_ts = test_cluster->pd_client->getTS();
+
+    auto lock = makeLock("abf", "abc", txn_id, defaultLockTTL, 6);
+    std::vector<LockPtr> locks{lock};
+    std::vector<uint64_t> pushed;
+    Backoffer bo(kv::copNextMaxBackoff);
+
+    const auto before_expired = test_cluster->lock_resolver->resolveLocks(bo, read_ts, locks, pushed);
+
+    ASSERT_EQ(before_expired, 0);
+    ASSERT_TRUE(pushed.empty());
+
+    Snapshot snapshot(test_cluster.get(), read_ts);
+    ASSERT_EQ(snapshot.Get("abf"), "3");
+}
+
+TEST_F(TestWithLockResolve, testResolveLocksBypassesRolledBackTxn)
+{
+    const uint64_t txn_id = 1;
+    auto lock = makeLock("rollback-key", "rollback-primary", txn_id, 1);
+    std::vector<LockPtr> locks{lock};
+    std::vector<uint64_t> pushed;
+    Backoffer bo(kv::copNextMaxBackoff);
+
+    const auto before_expired = test_cluster->lock_resolver->resolveLocks(bo, test_cluster->pd_client->getTS(), locks, pushed);
+
+    ASSERT_EQ(before_expired, 0);
+    ASSERT_EQ(pushed.size(), 1);
+    ASSERT_EQ(pushed[0], txn_id);
+}
 
 TEST_F(TestWithLockResolve, testResolveLockBase)
 {