diff --git a/ci/build-test.sh b/ci/build-test.sh index 87c26f15..fdf7a1db 100755 --- a/ci/build-test.sh +++ b/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 - - diff --git a/include/pingcap/kv/LockResolver.h b/include/pingcap/kv/LockResolver.h index 13763476..6b510c8d 100644 --- a/include/pingcap/kv/LockResolver.h +++ b/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(); } }; diff --git a/src/kv/LockResolver.cc b/src/kv/LockResolver.cc index 40dc6ef9..15cfa4cd 100644 --- a/src/kv/LockResolver.cc +++ b/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()) { diff --git a/src/kv/Scanner.cc b/src/kv/Scanner.cc index 8ab5b34f..050ad1c8 100644 --- a/src/kv/Scanner.cc +++ b/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 locks{lock}; std::vector 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( diff --git a/src/test/lock_resolve_test.cc b/src/test/lock_resolve_test.cc index bc155862..2ef1a35a 100644 --- a/src/test/lock_resolve_test.cc +++ b/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_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 locks{lock}; + std::vector 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 locks{lock}; + std::vector 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 locks{lock}; + std::vector 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) {