Expose locktree's wait count in RangeLockManagerHandle::Counters (#9289)

Summary:
locktree is a module providing Range Locking. It has a counter for
the number of times a lock acquisition request was blocked by an
existing conflicting lock and had to wait for it to be released.

Expose this counter in RangeLockManagerHandle::Counters::lock_wait_count.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/9289

Reviewed By: jay-zhuang

Differential Revision: D33079182

Pulled By: riversand963

fbshipit-source-id: 25b1a362d9da247536ab5007bd15900b319f139e

include/rocksdb/utilities/transaction_db.h

@@ -104,6 +104,11 @@ class RangeLockManagerHandle : public LockManagerHandle {
     // Number of times lock escalation was triggered (for all column families)
     uint64_t escalation_count;
 
+    // Number of times lock acquisition had to wait for a conflicting lock
+    // to be released. This counts both successful waits (where the desired
+    // lock was acquired) and waits that timed out or got other error.
+    uint64_t lock_wait_count;
+
     // How much memory is currently used for locks (total for all column
     // families)
     uint64_t current_lock_memory;

utilities/transactions/lock/range/range_locking_test.cc

@@ -288,6 +288,32 @@ TEST_F(RangeLockingTest, BasicLockEscalation) {
 }
 #endif
 
+TEST_F(RangeLockingTest, LockWaitCount) {
+  TransactionOptions txn_options;
+  auto cf = db->DefaultColumnFamily();
+  txn_options.lock_timeout = 50;
+  Transaction* txn0 = db->BeginTransaction(WriteOptions(), txn_options);
+  Transaction* txn1 = db->BeginTransaction(WriteOptions(), txn_options);
+
+  // Get a range lock
+  ASSERT_OK(txn0->GetRangeLock(cf, Endpoint("a"), Endpoint("c")));
+
+  uint64_t lock_waits1 = range_lock_mgr->GetStatus().lock_wait_count;
+  // Attempt to get a conflicting lock
+  auto s = txn1->GetRangeLock(cf, Endpoint("b"), Endpoint("z"));
+  ASSERT_TRUE(s.IsTimedOut());
+
+  // Check that the counter was incremented
+  uint64_t lock_waits2 = range_lock_mgr->GetStatus().lock_wait_count;
+  ASSERT_EQ(lock_waits1 + 1, lock_waits2);
+
+  txn0->Rollback();
+  txn1->Rollback();
+
+  delete txn0;
+  delete txn1;
+}
+
 void PointLockManagerTestExternalSetup(PointLockManagerTest* self) {
   self->env_ = Env::Default();
   self->db_dir_ = test::PerThreadDBPath("point_lock_manager_test");

utilities/transactions/lock/range/range_tree/range_tree_lock_manager.cc

@@ -329,6 +329,10 @@ RangeLockManagerHandle::Counters RangeTreeLockManager::GetStatus() {
       res.escalation_count = status->value.num;
       continue;
     }
+    if (strcmp(status->keyname, "LTM_WAIT_COUNT") == 0) {
+      res.lock_wait_count = status->value.num;
+      continue;
+    }
     if (strcmp(status->keyname, "LTM_SIZE_CURRENT") == 0) {
       res.current_lock_memory = status->value.num;
     }