chore: add more detailed timings for slow squash hops (#5679)

* chore: add more detailed timings for slow squash hops

We identify a callback that likely to contribute to the slowness of squashing and try to understand the reason.
The logging is tunable based on `log_squash_info_threshold_usec` parameter.

---------

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

Author: romange

helio

@@ -125,6 +125,10 @@ ABSL_FLAG(uint32_t, max_squashed_cmd_num, 100,
 
 ABSL_FLAG(uint32_t, max_busy_squash_usec, 1000,
           "Maximum time in microseconds to execute squashed commands before yielding.");
+
+ABSL_FLAG(uint32_t, log_squash_info_threshold_usec, 1 << 31,
+          "Threshold in microseconds above which to log squashing timings.");
+
 ABSL_FLAG(uint32_t, shard_thread_busy_polling_usec, 0,
           "If non-zero, overrides the busy polling parameter for shard threads.");
 
@@ -739,6 +743,11 @@ void SetMaxBusySquashUsec(uint32_t val) {
       [=](unsigned, auto*) { MultiCommandSquasher::SetMaxBusySquashUsec(val); });
 }
 
+void SetLogSquashInfoThreshold(uint32_t val) {
+  shard_set->pool()->AwaitBrief(
+      [=](unsigned, auto*) { MultiCommandSquasher::SetLogSquashThreshold(val); });
+}
+
 void SetShardThreadBusyPollingUsec(uint32_t val) {
   if (val == 0)
     return;
@@ -931,6 +940,8 @@ void Service::Init(util::AcceptServer* acceptor, std::vector<facade::Listener*>
                                            [](uint32_t val) { SetMaxBusySquashUsec(val); });
   config_registry.RegisterSetter<uint32_t>(
       "shard_thread_busy_polling_usec", [](uint32_t val) { SetShardThreadBusyPollingUsec(val); });
+  config_registry.RegisterSetter<uint32_t>("log_squash_info_threshold_usec",
+                                           [](uint32_t val) { SetLogSquashInfoThreshold(val); });
 
   config_registry.RegisterMutable("replica_partial_sync");
   config_registry.RegisterMutable("replication_timeout");
@@ -1019,6 +1030,7 @@ void Service::Init(util::AcceptServer* acceptor, std::vector<facade::Listener*>
   SetHuffmanTable(GetFlag(FLAGS_huffman_table));
   SetShardThreadBusyPollingUsec(GetFlag(FLAGS_shard_thread_busy_polling_usec));
   SetUringWakeMode(GetFlag(FLAGS_uring_wake_mode));
+  SetLogSquashInfoThreshold(GetFlag(FLAGS_log_squash_info_threshold_usec));
 
   // Requires that shard_set will be initialized before because server_family_.Init might
   // load the snapshot.

src/server/main_service.cc

@@ -27,6 +27,7 @@ using base::CycleClock;
 namespace {
 
 thread_local uint64_t max_busy_squash_cycles_cached = 1ULL << 32;
+thread_local uint32_t log_squash_threshold_cached = 1ULL << 31;
 
 void CheckConnStateClean(const ConnectionState& state) {
   DCHECK_EQ(state.exec_info.state, ConnectionState::ExecInfo::EXEC_INACTIVE);
@@ -245,6 +246,11 @@ bool MultiCommandSquasher::ExecuteSquashed(facade::RedisReplyBuilder* rb) {
   Transaction* tx = cntx_->transaction;
   ServerState::tlocal()->stats.squash_width_freq_arr[num_shards - 1]++;
   uint64_t start = CycleClock::Now();
+  atomic_uint64_t max_sched_cycles{0}, max_exec_cycles{0};
+  base::SpinLock lock;
+  uint64_t fiber_running_cycles{0}, proactor_running_cycles{0};
+  uint32_t max_sched_thread_id, max_sched_seq_num{0};
+  vector<string> past_fibers;
 
   // Atomic transactions (that have all keys locked) perform hops and run squashed commands via
   // stubs, non-atomic ones just run the commands in parallel.
@@ -258,19 +264,47 @@ bool MultiCommandSquasher::ExecuteSquashed(facade::RedisReplyBuilder* rb) {
     fb2::BlockingCounter bc(num_shards);
     DVLOG(1) << "Squashing " << num_shards << " " << tx->DebugId();
 
-    auto cb = [this, bc, rb]() mutable {
+    // Saves work in case logging is disable (i.e. log_squash_threshold_cached is high).
+    const uint64_t min_threshold_cycles = CycleClock::FromUsec(log_squash_threshold_cached / 5);
+    auto cb = [&, bc, rb]() mutable {
+      uint64_t sched_time = CycleClock::Now() - start;
+
+      // Update max_sched_cycles in lock-free fashion, to avoid contention
+      uint64_t current = max_sched_cycles.load(memory_order_relaxed);
+      while (sched_time > min_threshold_cycles && sched_time > current) {
+        if (max_sched_cycles.compare_exchange_weak(current, sched_time, memory_order_relaxed,
+                                                   memory_order_relaxed)) {
+          lock_guard<base::SpinLock> g(lock);
+
+          // If it is still the longest scheduling time
+          if (max_sched_cycles.load(memory_order_relaxed) == sched_time) {
+            // Store the stats from the callback with longest scheduling time.
+            fiber_running_cycles = ThisFiber::GetRunningTimeCycles();
+            proactor_running_cycles = ProactorBase::me()->GetCurrentBusyCycles();
+            max_sched_thread_id = ProactorBase::me()->GetPoolIndex();
+            max_sched_seq_num = fb2::GetFiberRunSeq();
+            past_fibers = fb2::GetPastFiberNames();
+          }
+          break;
+        }
+        // current is updated to the current value of max_sched_cycles, so the loop will retry
+        // with the new value if sched_time is still greater than it.
+      }
+
       if (ThisFiber::GetRunningTimeCycles() > max_busy_squash_cycles_cached) {
         ThisFiber::Yield();
         stats_.yields++;
       }
       this->SquashedHopCb(EngineShard::tlocal(), rb->GetRespVersion());
       bc->Dec();
+      uint64_t exec_time = CycleClock::Now() - start;
+      current = max_exec_cycles.load(memory_order_relaxed);
+      while (exec_time > current) {
+        if (max_exec_cycles.compare_exchange_weak(current, exec_time, memory_order_relaxed,
+                                                  memory_order_relaxed))
+          break;
+      }
     };
-    unsigned run_usec = CycleClock::ToUsec(ThisFiber::GetRunningTimeCycles());
-    if (run_usec > 5'000) {
-      LOG_EVERY_T(WARNING, 1) << "Fiber run " << run_usec << " usec, squashed " << cmds_.size()
-                              << " commands";
-    }
     for (unsigned i = 0; i < sharded_.size(); ++i) {
       if (!sharded_[i].dispatched.empty())
         shard_set->AddL2(i, cb);
@@ -297,14 +331,31 @@ bool MultiCommandSquasher::ExecuteSquashed(facade::RedisReplyBuilder* rb) {
     if (aborted)
       break;
   }
+
   uint64_t after_reply = CycleClock::Now();
   uint64_t total_usec = CycleClock::ToUsec(after_reply - start);
-
   stats_.hop_usec += total_usec;
   stats_.reply_usec += CycleClock::ToUsec(after_reply - after_hop);
   stats_.hops++;
   stats_.squashed_commands += order_.size();
 
+  if (total_usec > log_squash_threshold_cached) {
+    uint64_t max_sched_usec = CycleClock::ToUsec(max_sched_cycles.load());
+    uint64_t fiber_running_usec = CycleClock::ToUsec(fiber_running_cycles);
+    uint64_t proactor_running_usec = CycleClock::ToUsec(proactor_running_cycles);
+    uint64_t max_exec_usec = CycleClock::ToUsec(max_exec_cycles.load());
+
+    LOG_EVERY_T(INFO, 0.1)
+        << "Squashed " << order_.size() << " commands. "
+        << "Total/Fanout/MaxSchedTime/ThreadCbTime/ThreadId/FiberCbTime/FiberSeq/MaxExecTime: "
+        << total_usec << "/" << num_shards_ << "/" << max_sched_usec << "/" << proactor_running_usec
+        << "/" << max_sched_thread_id << "/" << fiber_running_usec << "/"
+        << "/" << max_sched_seq_num << "/" << max_exec_usec
+        << "\n past fibers: " << absl::StrJoin(past_fibers, ", ")
+        << "\ncoordinator thread running time: "
+        << CycleClock::ToUsec(ProactorBase::me()->GetCurrentBusyCycles());
+  }
+
   tl_facade_stats->reply_stats.squashing_current_reply_size.fetch_sub(total_reply_size,
                                                                       std::memory_order_release);
   for (auto& sinfo : sharded_) {
@@ -364,4 +415,8 @@ void MultiCommandSquasher::SetMaxBusySquashUsec(uint32_t usec) {
   max_busy_squash_cycles_cached = CycleClock::FromUsec(usec);
 }
 
+void MultiCommandSquasher::SetLogSquashThreshold(uint32_t usec) {
+  log_squash_threshold_cached = usec;
+}
+
 }  // namespace dfly

src/server/multi_command_squasher.cc

@@ -46,6 +46,7 @@ class MultiCommandSquasher {
   }
 
   static void SetMaxBusySquashUsec(uint32_t usec);
+  static void SetLogSquashThreshold(uint32_t usec);
 
  private:
   // Per-shard execution info.

src/server/multi_command_squasher.h

@@ -3887,7 +3887,7 @@ void ServerFamily::Register(CommandRegistry* registry) {
       << CI{"AUTH", CO::NOSCRIPT | CO::FAST | CO::LOADING, -2, 0, 0, acl::kAuth}.HFUNC(Auth)
       << CI{"BGSAVE", CO::ADMIN | CO::GLOBAL_TRANS, -1, 0, 0, acl::kBGSave}.HFUNC(BgSave)
       << CI{"CLIENT", CO::NOSCRIPT | CO::LOADING, -2, 0, 0, acl::kClient}.HFUNC(Client)
-      << CI{"CONFIG", CO::ADMIN | CO::DANGEROUS, -2, 0, 0, acl::kConfig}.HFUNC(Config)
+      << CI{"CONFIG", CO::ADMIN | CO::LOADING | CO::DANGEROUS, -2, 0, 0, acl::kConfig}.HFUNC(Config)
       << CI{"DBSIZE", CO::READONLY | CO::FAST | CO::LOADING, 1, 0, 0, acl::kDbSize}.HFUNC(DbSize)
       << CI{"DEBUG", CO::ADMIN | CO::LOADING, -2, 0, 0, acl::kDebug}.HFUNC(Debug)
       << CI{"FLUSHDB", CO::WRITE | CO::GLOBAL_TRANS | CO::DANGEROUS, 1, 0, 0, acl::kFlushDB}.HFUNC(