chore: expose the multi length in slowlog (#2339)

1. Fix AnalyzeTxQueue to stop crashing for various transaction types.
2. Pass exec command length to slowlog

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

src/server/conn_context.h

@@ -174,6 +174,9 @@ class ConnectionContext : public facade::ConnectionContext {
     uint32_t shards_count = 0;
     TxClock clock = 0;
     bool is_ooo = false;
+
+    // number of commands in the last exec body.
+    unsigned exec_body_len = 0;
   };
 
   DebugInfo last_command_debug;

src/server/engine_shard_set.cc

@@ -735,17 +735,20 @@ auto EngineShard::AnalyzeTxQueue() -> TxQueueInfo {
       if (trx->IsGlobal() || (trx->IsMulti() && trx->GetMultiMode() == Transaction::GLOBAL)) {
         info.tx_global++;
       } else {
-        KeyLockArgs lock_args = trx->GetLockArgs(sid);
         DbTable* table = db_slice().GetDBTable(trx->GetDbIndex());
         bool can_run = true;
-        for (size_t i = 0; i < lock_args.args.size(); i += lock_args.key_step) {
+
-          string_view s = KeyLockArgs::GetLockKey(lock_args.args[i]);
+        if (!trx->IsMulti()) {
-          auto it = table->trans_locks.find(s);
+          KeyLockArgs lock_args = trx->GetLockArgs(sid);
-          DCHECK(it != table->trans_locks.end());
+          for (size_t i = 0; i < lock_args.args.size(); i += lock_args.key_step) {
-          if (it != table->trans_locks.end()) {
+            string_view s = KeyLockArgs::GetLockKey(lock_args.args[i]);
-            if (it->second.IsContended()) {
+            auto it = table->trans_locks.find(s);
-              can_run = false;
+            DCHECK(it != table->trans_locks.end());
-              break;
+            if (it != table->trans_locks.end()) {
+              if (it->second.IsContended()) {
+                can_run = false;
+                break;
+              }
             }
           }
         }
@@ -760,6 +763,9 @@ auto EngineShard::AnalyzeTxQueue() -> TxQueueInfo {
   // Analyze locks
   for (unsigned i = 0; i <= max_db_id; ++i) {
     DbTable* table = db_slice().GetDBTable(i);
+    if (table == nullptr)
+      continue;
+
     info.total_locks += table->trans_locks.size();
     for (const auto& k_v : table->trans_locks) {
       if (k_v.second.IsContended()) {

src/server/main_service.cc

@@ -1250,6 +1250,15 @@ bool Service::InvokeCmd(const CommandId* cid, CmdArgList tail_args, ConnectionCo
   const auto* conn = cntx->conn();
   if (!(cid->opt_mask() & CO::BLOCKING) && conn != nullptr && etl.GetSlowLog().IsEnabled() &&
       invoke_time_usec > etl.log_slower_than_usec) {
+    vector<string> aux_params;
+    CmdArgVec aux_slices;
+
+    if (tail_args.empty() && cid->name() == "EXEC") {
+      // abuse tail_args to pass more information about the slow EXEC.
+      aux_params.emplace_back(StrCat("CMDCOUNT/", cntx->last_command_debug.exec_body_len));
+      aux_slices.emplace_back(aux_params.back());
+      tail_args = absl::MakeSpan(aux_slices);
+    }
     etl.GetSlowLog().Add(cid->name(), tail_args, conn->GetName(), conn->RemoteEndpointStr(),
                          invoke_time_usec, absl::GetCurrentTimeNanos() / 1000);
   }
@@ -2043,6 +2052,7 @@ void Service::Exec(CmdArgList args, ConnectionContext* cntx) {
     return rb->SendNull();
   }
 
+  cntx->last_command_debug.exec_body_len = exec_info.body.size();
   const CommandId* const exec_cid = cntx->cid;
   CmdArgVec arg_vec;
   ExecEvalState state = DetermineEvalPresense(exec_info.body);

src/server/transaction.cc

@@ -166,7 +166,6 @@ void Transaction::InitMultiData(KeyIndex key_index) {
   IntentLock::Mode mode = Mode();
 
   auto& tmp_uniques = tmp_space.uniq_keys;
-  tmp_uniques.clear();
 
   auto lock_key = [this, mode, &tmp_uniques](string_view key) {
     if (auto [_, inserted] = tmp_uniques.insert(KeyLockArgs::GetLockKey(key)); !inserted)
@@ -178,13 +177,16 @@ void Transaction::InitMultiData(KeyIndex key_index) {
   // With EVAL, we call this function for EVAL itself as well as for each command
   // for eval. currently, we lock everything only during the eval call.
   if (!multi_->locks_recorded) {
+    tmp_uniques.clear();
+
     for (size_t i = key_index.start; i < key_index.end; i += key_index.step)
       lock_key(ArgS(full_args_, i));
     if (key_index.bonus)
       lock_key(ArgS(full_args_, *key_index.bonus));
+
+    multi_->locks_recorded = true;
   }
 
-  multi_->locks_recorded = true;
   DCHECK(IsAtomicMulti());
   DCHECK(multi_->mode == GLOBAL || !multi_->lock_counts.empty());
 }
@@ -1167,6 +1169,8 @@ bool Transaction::CancelShardCb(EngineShard* shard) {
 
 // runs in engine-shard thread.
 ArgSlice Transaction::GetShardArgs(ShardId sid) const {
+  DCHECK(!multi_ || multi_->role != SQUASHER);
+
   // We can read unique_shard_cnt_  only because ShardArgsInShard is called after IsArmedInShard
   // barrier.
   if (unique_shard_cnt_ == 1) {