1aca60c089
Summary: PointLockManager manages point lock per key. The old implementation partition the per key lock into 16 stripes. Each stripe handles the point lock for a subset of keys. Each stripe have only one conditional variable. This conditional variable is used by all the transactions that are waiting for its turn to acquire a lock of a key that belongs to this stripe. In production, we notified that when there are multiple transactions trying to write to the same key, all of them will wait on the same conditional variables. When the previous lock holder released the key, all of the transactions are woken up, but only one of them could proceed, and the rest goes back to sleep. This wasted a lot of CPU cycles. In addition, when there are other keys being locked/unlocked on the same lock stripe, the problem becomes even worse. In order to solve this issue, we implemented a new PerKeyPointLockManager that keeps a transaction waiter queue at per key level. When a transaction could not acquire a lock immediately, it joins the waiter queue of the key and waits on a dedicated conditional variable. When previous lock holder released the lock, it wakes up the next set of transactions that are eligible to acquire the lock from the waiting queue. The queue respect FIFO order, except it prioritizes lock upgrade/downgrade operation. However, this waiter queue change increases the deadlock detection cost, because the transaction waiting in the queue also needs to be considered during deadlock detection. To resolve this issue, a new deadlock_timeout_us (microseconds) configuration is introduced in transaction option. Essentially, when a transaction is waiting on a lock, it will join the wait queue and wait for the duration configured by deadlock_timeout_us without perform deadlock detection. If the transaction didn't get the lock after the deadlock_timeout_us timeout is reached, it will then perform deadlock detection and wait until lock_timeout is reached. This optimization takes the heuristic where majority of the transaction would be able to get the lock without perform deadlock detection. The deadlock_timeout_us configuration needs to be tuned for different workload, if the likelihood of deadlock is very low, the deadlock_timeout_us could be configured close to a big higher than the average transaction execution time, so that majority of the transaction would be able to acquire the lock without performing deadlock detection. If the likelihood of deadlock is high, deadlock_timeout_us could be configured with lower value, so that deadlock would get detected faster. The new PerKeyPointLockManager is disabled by default. It can be enabled by TransactionDBOptions.use_per_key_point_lock_mgr. The deadlock_timeout_us is only effective when PerKeyPointLockManager is used. When deadlock_timeout_us is set to 0, transaction will perform deadlock detection immediately before wait. Pull Request resolved: https://github.com/facebook/rocksdb/pull/13731 Test Plan: Unit test. Stress unit test that validates deadlock detection and exclusive, shared lock guarantee. A new point_lock_bench binary is created to help perform performance test. Reviewed By: pdillinger Differential Revision: D77353607 Pulled By: xingbowang fbshipit-source-id: 21cf93354f9a367a78c8666596ed14013ac7240b
102 lines
3.3 KiB
C++
102 lines
3.3 KiB
C++
// Copyright (c) Meta Platforms, Inc. and affiliates.
|
|
//
|
|
// This source code is licensed under both the GPLv2 (found in the
|
|
// COPYING file in the root directory) and Apache 2.0 License
|
|
// (found in the LICENSE.Apache file in the root directory).
|
|
|
|
#pragma once
|
|
|
|
#include "file/file_util.h"
|
|
#include "rocksdb/utilities/transaction_db.h"
|
|
#include "test_util/testharness.h"
|
|
#include "utilities/transactions/lock/point/point_lock_manager.h"
|
|
#include "utilities/transactions/lock/point/point_lock_manager_test_common.h"
|
|
#include "utilities/transactions/pessimistic_transaction_db.h"
|
|
|
|
namespace ROCKSDB_NAMESPACE {
|
|
|
|
class PointLockManagerTest : public testing::Test {
|
|
public:
|
|
void init() {
|
|
env_ = Env::Default();
|
|
db_dir_ = test::PerThreadDBPath("point_lock_manager_test");
|
|
ASSERT_OK(env_->CreateDir(db_dir_));
|
|
|
|
Options opt;
|
|
opt.create_if_missing = true;
|
|
// Reduce the number of stripes to 4 to increase contention in test
|
|
txndb_opt_.num_stripes = 4;
|
|
txndb_opt_.transaction_lock_timeout = 0;
|
|
|
|
ASSERT_OK(TransactionDB::Open(opt, txndb_opt_, db_dir_, &db_));
|
|
|
|
wait_sync_point_name_ = "PointLockManager::AcquireWithTimeout:WaitingTxn";
|
|
}
|
|
|
|
void SetUp() override {
|
|
init();
|
|
// CAUTION: This test creates a separate lock manager object (right, NOT
|
|
// the one that the TransactionDB is using!), and runs tests on it.
|
|
locker_.reset(new PointLockManager(
|
|
static_cast<PessimisticTransactionDB*>(db_), txndb_opt_));
|
|
}
|
|
|
|
void TearDown() override {
|
|
std::string errmsg;
|
|
auto no_lock_held = verifyNoLocksHeld(locker_, errmsg);
|
|
ASSERT_TRUE(no_lock_held) << errmsg;
|
|
delete db_;
|
|
EXPECT_OK(DestroyDir(env_, db_dir_));
|
|
}
|
|
|
|
PessimisticTransaction* NewTxn(
|
|
TransactionOptions txn_opt = TransactionOptions()) {
|
|
// override deadlock_timeout_us;
|
|
txn_opt.deadlock_timeout_us = deadlock_timeout_us;
|
|
Transaction* txn = db_->BeginTransaction(WriteOptions(), txn_opt);
|
|
return static_cast<PessimisticTransaction*>(txn);
|
|
}
|
|
|
|
int64_t deadlock_timeout_us = 0;
|
|
|
|
void UsePerKeyPointLockManager() {
|
|
locker_.reset(new PerKeyPointLockManager(
|
|
static_cast<PessimisticTransactionDB*>(db_), txndb_opt_));
|
|
}
|
|
|
|
protected:
|
|
Env* env_;
|
|
TransactionDBOptions txndb_opt_;
|
|
std::shared_ptr<LockManager> locker_;
|
|
const char* wait_sync_point_name_;
|
|
friend void PointLockManagerTestExternalSetup(PointLockManagerTest*);
|
|
|
|
std::string db_dir_;
|
|
TransactionDB* db_;
|
|
};
|
|
|
|
void BlockUntilWaitingTxn(const char* sync_point_name, port::Thread& t,
|
|
std::function<void()> f) {
|
|
std::atomic<bool> reached(false);
|
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
|
|
sync_point_name, [&](void* /*arg*/) { reached.store(true); });
|
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
|
|
|
|
t = port::Thread(f);
|
|
|
|
// timeout after 30 seconds, so test does not hang forever
|
|
// 30 seconds should be enough for the test to reach the expected state
|
|
// without causing too much flakiness
|
|
for (int i = 0; i < 3000; i++) {
|
|
if (reached.load()) {
|
|
break;
|
|
}
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
|
}
|
|
|
|
ASSERT_TRUE(reached.load());
|
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
|
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
|
|
}
|
|
|
|
} // namespace ROCKSDB_NAMESPACE
|