chore: get rid of cmdstats_map (#1687)

cmdstats_map were on the hotpath and are needed only to update the command stats.
Instead, I introduced the stats withing the CommandId itself that we lookup anyways.
Also, it removes fragile dependency on naked command name char* pointers.

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

src/server/command_registry.cc

@@ -54,8 +54,7 @@ void CommandId::Invoke(CmdArgList args, ConnectionContext* cntx) const {
   uint64_t before = absl::GetCurrentTimeNanos();
   handler_(std::move(args), cntx);
   uint64_t after = absl::GetCurrentTimeNanos();
-
-  auto& ent = ServerState::tlocal()->cmd_stats_map[cntx->cid->name().data()];
+  auto& ent = command_stats_[ServerState::tlocal()->thread_index()];
   ++ent.first;
   ent.second += (after - before) / 1000;
 }
@@ -89,6 +88,12 @@ CommandRegistry::CommandRegistry() {
   }
 }
 
+void CommandRegistry::Init(unsigned int thread_count) {
+  for (auto& [_, cmd] : cmd_map_) {
+    cmd.Init(thread_count);
+  }
+}
+
 CommandRegistry& CommandRegistry::operator<<(CommandId cmd) {
   string_view k = cmd.name();
   auto it = cmd_rename_map_.find(k);

src/server/command_registry.h

@@ -54,6 +54,9 @@ static_assert(!IsEvalKind(""));
 
 };  // namespace CO
 
+// Per thread vector of command stats. Each entry is {cmd_calls, cmd_sum}.
+using CmdCallStats = std::pair<uint64_t, uint64_t>;
+
 class CommandId : public facade::CommandId {
  public:
   // NOTICE: name must be a literal string, otherwise metrics break! (see cmd_stats_map in
@@ -61,6 +64,12 @@ class CommandId : public facade::CommandId {
   CommandId(const char* name, uint32_t mask, int8_t arity, int8_t first_key, int8_t last_key,
             int8_t step);
 
+  CommandId(CommandId&&) = default;
+
+  void Init(unsigned thread_count) {
+    command_stats_ = std::make_unique<CmdCallStats[]>(thread_count);
+  }
+
   using Handler =
       fu2::function_base<true /*owns*/, true /*copyable*/, fu2::capacity_default,
                          false /* non-throwing*/, false /* strong exceptions guarantees*/,
@@ -78,21 +87,30 @@ class CommandId : public facade::CommandId {
 
   static const char* OptName(CO::CommandOpt fl);
 
-  CommandId& SetHandler(Handler f) {
+  CommandId&& SetHandler(Handler f) && {
     handler_ = std::move(f);
-    return *this;
+    return std::move(*this);
   }
 
-  CommandId& SetValidator(ArgValidator f) {
+  CommandId&& SetValidator(ArgValidator f) && {
     validator_ = std::move(f);
-    return *this;
+    return std::move(*this);
   }
 
   bool is_multi_key() const {
     return (last_key_ != first_key_) || (opt_mask_ & CO::VARIADIC_KEYS);
   }
 
+  void ResetStats(unsigned thread_index) {
+    command_stats_[thread_index] = {0, 0};
+  }
+
+  CmdCallStats GetStats(unsigned thread_index) const {
+    return command_stats_[thread_index];
+  }
+
  private:
+  std::unique_ptr<CmdCallStats[]> command_stats_;
   Handler handler_;
   ArgValidator validator_;
 };
@@ -104,6 +122,8 @@ class CommandRegistry {
  public:
   CommandRegistry();
 
+  void Init(unsigned thread_count);
+
   CommandRegistry& operator<<(CommandId cmd);
 
   const CommandId* Find(std::string_view cmd) const {
@@ -123,6 +143,22 @@ class CommandRegistry {
       cb(k_v.first, k_v.second);
     }
   }
+
+  void ResetCallStats(unsigned thread_index) {
+    for (auto& k_v : cmd_map_) {
+      k_v.second.ResetStats(thread_index);
+    }
+  }
+
+  void MergeCallStats(unsigned thread_index,
+                      std::function<void(std::string_view, const CmdCallStats&)> cb) const {
+    for (const auto& k_v : cmd_map_) {
+      auto src = k_v.second.GetStats(thread_index);
+      if (src.first == 0)
+        continue;
+      cb(k_v.first, src);
+    }
+  }
 };
 
 }  // namespace dfly

src/server/main_service.cc

@@ -2001,6 +2001,9 @@ void Service::RegisterCommands() {
   server_family_.Register(&registry_);
   cluster_family_.Register(&registry_);
 
+  // Only after all the commands are registered
+  registry_.Init(pp_.size());
+
   if (VLOG_IS_ON(1)) {
     LOG(INFO) << "Multi-key commands are: ";
     registry_.Traverse([](std::string_view key, const CI& cid) {

src/server/main_service.h

@@ -68,6 +68,10 @@ class Service : public facade::ServiceInterface {
     return registry_.Find(cmd);
   }
 
+  CommandRegistry* mutable_registry() {
+    return &registry_;
+  }
+
   facade::ErrorReply ReportUnknownCmd(std::string_view cmd_name);
 
   // Returns: the new state.

src/server/server_family.cc

@@ -1006,17 +1006,11 @@ void PrintPrometheusMetrics(const Metrics& m, StringResponse* resp) {
 
   // Command stats
   {
-    vector<pair<const char*, pair<uint64_t, uint64_t>>> commands(m.cmd_stats_map.cbegin(),
-                                                                 m.cmd_stats_map.cend());
-
-    sort(commands.begin(), commands.end(),
-         [](const auto& s1, const auto& s2) { return std::strcmp(s1.first, s2.first) < 0; });
-
     string command_metrics;
 
     AppendMetricHeader("commands", "Metrics for all commands ran", MetricType::COUNTER,
                        &command_metrics);
-    for (const auto& [name, stat] : commands) {
+    for (const auto& [name, stat] : m.cmd_stats_map) {
       const auto calls = stat.first;
       const auto duration_seconds = stat.second * 0.001;
       AppendMetricValue("commands_total", calls, {"cmd"}, {name}, &command_metrics);
@@ -1597,15 +1591,15 @@ void ServerFamily::Config(CmdArgList args, ConnectionContext* cntx) {
 
     return (*cntx)->SendStringArr(res, RedisReplyBuilder::MAP);
   } else if (sub_cmd == "RESETSTAT") {
-    shard_set->pool()->Await([](auto*) {
+    shard_set->pool()->Await([registry = service_.mutable_registry()](unsigned index, auto*) {
+      registry->ResetCallStats(index);
       auto& sstate = *ServerState::tlocal();
-      sstate.cmd_stats_map.clear();
-
       auto& stats = sstate.connection_stats;
       stats.err_count_map.clear();
       stats.command_cnt = 0;
       stats.async_writes_cnt = 0;
     });
+
     return (*cntx)->SendOk();
   } else {
     return (*cntx)->SendError(UnknownSubCmd(sub_cmd, "CONFIG"), kSyntaxErrType);
@@ -1676,7 +1670,7 @@ Metrics ServerFamily::GetMetrics() const {
 
   Mutex mu;
 
-  auto cb = [&](ProactorBase* pb) {
+  auto cb = [&](unsigned index, ProactorBase* pb) {
     EngineShard* shard = EngineShard::tlocal();
     ServerState* ss = ServerState::tlocal();
 
@@ -1686,11 +1680,13 @@ Metrics ServerFamily::GetMetrics() const {
     result.conn_stats += ss->connection_stats;
     result.qps += uint64_t(ss->MovingSum6());
     result.ooo_tx_transaction_cnt += ss->stats.ooo_tx_cnt;
-    for (const auto& [k, v] : ss->cmd_stats_map) {
-      auto& ent = result.cmd_stats_map[k];
-      ent.first += v.first;
-      ent.second += v.second;
-    }
+
+    service_.mutable_registry()->MergeCallStats(
+        index, [&dest_map = result.cmd_stats_map](string_view name, const CmdCallStats& src) {
+          auto& ent = dest_map[string{name}];
+          ent.first += src.first;
+          ent.second += src.second;
+        });
 
     if (shard) {
       MergeInto(shard->db_slice().GetStats(), &result);

src/server/server_family.h

@@ -64,8 +64,9 @@ struct Metrics {
   bool is_master = true;
 
   facade::ConnectionStats conn_stats;
-  absl::flat_hash_map<const char*, std::pair<uint64_t, uint64_t>>
-      cmd_stats_map;  // command statistics; see ServerState
+
+  // command statistics; see CommandId.
+  std::map<std::string, std::pair<uint64_t, uint64_t>> cmd_stats_map;
 
   std::vector<ReplicaRoleInfo> replication_metrics;
 };

src/server/server_state.h

@@ -205,10 +205,6 @@ class ServerState {  // public struct - to allow initialization.
 
   facade::ConnectionStats connection_stats;
 
-  absl::flat_hash_map<const char*, std::pair<uint64_t, uint64_t>> cmd_stats_map;
-  // command statistics. the pair holds {cmd_calls, cmd_sum}.
-  // IMPORTANT: assumes command names are constant literals. (the pointer is being hashed)
-
  private:
   int64_t live_transactions_ = 0;
   mi_heap_t* data_heap_;