feat(server): add debug information about running transactions

Specifically, provide a total length over all shards of pending txs.
Also, provide number of "free" transactions pending in the queue -
those that could progress immediately but do not because
they are not first in the queue.

Signed-off-by: Roman Gershman <[email protected]>

src/server/debugcmd.cc

@@ -148,6 +148,10 @@ void DebugCmd::Run(CmdArgList args) {
     return Inspect(key);
   }
 
+  if (subcmd == "TRANSACTION") {
+    return TxAnalysis();
+  }
+
   string reply = UnknownSubCmd(subcmd, "DEBUG");
   return (*cntx_)->SendError(reply, kSyntaxErrType);
 }
@@ -430,4 +434,55 @@ void DebugCmd::Watched() {
   (*cntx_)->SendStringArr(watched_keys);
 }
 
+void DebugCmd::TxAnalysis() {
+  atomic_uint32_t queue_len{0};
+  atomic_uint32_t free_len{0};
+
+  using SvLockTable = absl::flat_hash_map<string_view, IntentLock>;
+  vector<SvLockTable> lock_table_arr(shard_set->size());
+
+  auto cb = [&](EngineShard* shard) {
+    ShardId sid = shard->shard_id();
+
+    TxQueue* queue = shard->txq();
+
+    if (queue->Empty())
+      return;
+
+    auto cur = queue->Head();
+    do {
+      auto value = queue->At(cur);
+      Transaction* trx = std::get<Transaction*>(value);
+
+      IntentLock::Mode mode = trx->Mode();
+      KeyLockArgs lock_args = trx->GetLockArgs(sid);
+      auto& lock_table = lock_table_arr[sid];
+
+      // We count locks ourselves and do not rely on the lock table inside dbslice.
+      // The reason for this - to account for ordering information.
+      // For example, consider T1, T2 both residing in the queue and both lock 'x' exclusively.
+      // DbSlice::CheckLock returns false for both transactions, but T1 in practice owns the lock.
+      bool can_take = true;
+      for (size_t i = 0; i < lock_args.args.size(); i += lock_args.key_step) {
+        string_view s = lock_args.args[i];
+        bool was_ack = lock_table[s].Acquire(mode);
+        if (!was_ack) {
+          can_take = false;
+        }
+      }
+
+      queue_len.fetch_add(1, std::memory_order_relaxed);
+      if (can_take) {
+        free_len.fetch_add(1, std::memory_order_relaxed);
+      }
+      cur = queue->Next(cur);
+    } while (cur != queue->Head());
+  };
+
+  shard_set->RunBriefInParallel(cb);
+
+  (*cntx_)->SendSimpleString(
+      absl::StrCat("queue_len:", queue_len.load(), " free_len:", free_len.load()));
+}
+
 }  // namespace dfly

src/server/debugcmd.h

@@ -26,6 +26,7 @@ class DebugCmd {
   void Load(std::string_view filename);
   void Inspect(std::string_view key);
   void Watched();
+  void TxAnalysis();
 
   ServerFamily& sf_;
   ConnectionContext* cntx_;