Introduce fmt dependency to simplify string formatting

CMakeLists.txt

@@ -116,6 +116,7 @@ include(cmake/zlib.cmake)
 include(cmake/zstd.cmake)
 include(cmake/rocksdb.cmake)
 include(cmake/libevent.cmake)
+include(cmake/fmt.cmake)
 
 if (USE_LUAJIT)
 include(cmake/luajit.cmake)
@@ -132,6 +133,7 @@ list(APPEND EXTERNAL_LIBS event_with_headers)
 list(APPEND EXTERNAL_LIBS lz4)
 list(APPEND EXTERNAL_LIBS zstd)
 list(APPEND EXTERNAL_LIBS zlib_with_headers)
+list(APPEND EXTERNAL_LIBS fmt)
 if (USE_LUAJIT)
 list(APPEND EXTERNAL_LIBS luajit)
 else()

cmake/fmt.cmake

@@ -0,0 +1,27 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+include_guard()
+
+include(cmake/utils.cmake)
+
+FetchContent_DeclareGitHubWithMirror(fmt
+  fmtlib/fmt 9.1.0
+  MD5=e6754011ff56bfc37631fcc90961e377
+)
+
+FetchContent_MakeAvailableWithArgs(fmt)

src/cluster/cluster.cc

@@ -26,6 +26,7 @@
 #include <memory>
 
 #include "commands/redis_cmd.h"
+#include "fmt/format.h"
 #include "parse_util.h"
 #include "replication.h"
 #include "server/server.h"
@@ -313,7 +314,7 @@ Status Cluster::ImportSlot(Redis::Connection *conn, int slot, int state) {
   switch (state) {
     case kImportStart:
       if (!svr_->slot_import_->Start(conn->GetFD(), slot)) {
-        return Status(Status::NotOK, "Can't start importing slot " + std::to_string(slot));
+        return {Status::NotOK, fmt::format("Can't start importing slot {}", slot)};
       }
       // Set link importing
       conn->SetImporting();
@@ -330,20 +331,20 @@ Status Cluster::ImportSlot(Redis::Connection *conn, int slot, int state) {
       if (!svr_->slot_import_->Success(slot)) {
         LOG(ERROR) << "[import] Failed to set slot importing success, maybe slot is wrong"
                    << ", received slot: " << slot << ", current slot: " << svr_->slot_import_->GetSlot();
-        return Status(Status::NotOK, "Failed to set slot " + std::to_string(slot) + " importing success");
+        return {Status::NotOK, fmt::format("Failed to set slot {} importing success", slot)};
       }
       LOG(INFO) << "[import] Succeed to import slot " << slot;
       break;
     case kImportFailed:
       if (!svr_->slot_import_->Fail(slot)) {
         LOG(ERROR) << "[import] Failed to set slot importing error, maybe slot is wrong"
                    << ", received slot: " << slot << ", current slot: " << svr_->slot_import_->GetSlot();
-        return Status(Status::NotOK, "Failed to set slot " + std::to_string(slot) + " importing error");
+        return {Status::NotOK, fmt::format("Failed to set slot {} importing error", slot)};
       }
       LOG(INFO) << "[import] Failed to import slot " << slot;
       break;
     default:
-      return Status(Status::NotOK, errInvalidImportState);
+      return {Status::NotOK, errInvalidImportState};
   }
   return Status::OK();
 }
@@ -469,9 +470,9 @@ std::string Cluster::GenNodesDescription() {
     // Generate slots info when occur different node with start or end of slot
     if (i == kClusterSlots || n != slots_nodes_[i]) {
       if (start == i - 1) {
-        n->slots_info_ += std::to_string(start) + " ";
+        n->slots_info_ += fmt::format("{} ", start);
       } else {
-        n->slots_info_ += std::to_string(start) + "-" + std::to_string(i - 1) + " ";
+        n->slots_info_ += fmt::format("{}-{} ", start, i - 1);
       }
       if (i == kClusterSlots) break;
       n = slots_nodes_[i];
@@ -486,8 +487,7 @@ std::string Cluster::GenNodesDescription() {
     std::string node_str;
     // ID, host, port
     node_str.append(n->id_ + " ");
-    node_str.append(n->host_ + ":" + std::to_string(n->port_) + "@" + std::to_string(n->port_ + kClusterPortIncr) +
-                    " ");
+    node_str.append(fmt::format("{}:{}@{} ", n->host_, n->port_, n->port_ + kClusterPortIncr));
 
     // Flags
     if (n->id_ == myid_) node_str.append("myself,");
@@ -499,8 +499,7 @@ std::string Cluster::GenNodesDescription() {
 
     // Ping sent, pong received, config epoch, link status
     auto now = Util::GetTimeStampMS();
-    node_str.append(std::to_string(now - 1) + " " + std::to_string(now) + " " + std::to_string(version_) + " " +
-                    "connected");
+    node_str.append(fmt::format("{} {} {} connected", now - 1, now, version_));
 
     // Slots
     if (n->slots_info_.size() > 0) n->slots_info_.pop_back();  // Trim space
@@ -642,24 +641,24 @@ Status Cluster::CanExecByMySelf(const Redis::CommandAttributes *attributes, cons
     int cur_slot = GetSlotNumFromKey(cmd_tokens[i]);
     if (slot == -1) slot = cur_slot;
     if (slot != cur_slot) {
-      return Status(Status::RedisExecErr, "CROSSSLOT Attempted to access keys that don't hash to the same slot");
+      return {Status::RedisExecErr, "CROSSSLOT Attempted to access keys that don't hash to the same slot"};
     }
   }
   if (slot == -1) return Status::OK();
 
   if (slots_nodes_[slot] == nullptr) {
-    return Status(Status::ClusterDown, "CLUSTERDOWN Hash slot not served");
+    return {Status::ClusterDown, "CLUSTERDOWN Hash slot not served"};
   } else if (myself_ && myself_ == slots_nodes_[slot]) {
     // We use central controller to manage the topology of the cluster.
     // Server can't change the topology directly, so we record the migrated slots
     // to move the requests of the migrated slots to the destination node.
     if (migrated_slots_.count(slot)) {  // I'm not serving the migrated slot
-      return Status(Status::RedisExecErr, "MOVED " + std::to_string(slot) + " " + migrated_slots_[slot]);
+      return {Status::RedisExecErr, fmt::format("MOVED {} {}", slot, migrated_slots_[slot])};
     }
     // To keep data consistency, slot will be forbidden write while sending the last incremental data.
     // During this phase, the requests of the migrating slot has to be rejected.
     if (attributes->is_write() && IsWriteForbiddenSlot(slot)) {
-      return Status(Status::RedisExecErr, "Can't write to slot being migrated which is in write forbidden phase");
+      return {Status::RedisExecErr, "Can't write to slot being migrated which is in write forbidden phase"};
     }
     return Status::OK();  // I'm serving this slot
   } else if (myself_ && myself_->importing_slot_ == slot && conn->IsImporting()) {
@@ -677,7 +676,7 @@ Status Cluster::CanExecByMySelf(const Redis::CommandAttributes *attributes, cons
              nodes_.find(myself_->master_id_) != nodes_.end() && nodes_[myself_->master_id_] == slots_nodes_[slot]) {
     return Status::OK();  // My mater is serving this slot
   } else {
-    std::string ip_port = slots_nodes_[slot]->host_ + ":" + std::to_string(slots_nodes_[slot]->port_);
-    return Status(Status::RedisExecErr, "MOVED " + std::to_string(slot) + " " + ip_port);
+    return {Status::RedisExecErr,
+            fmt::format("MOVED {} {}:{}", slot, slots_nodes_[slot]->host_, slots_nodes_[slot]->port_)};
   }
 }

src/cluster/replication.cc

@@ -34,6 +34,7 @@
 
 #include "event_util.h"
 #include "fd_util.h"
+#include "fmt/format.h"
 #include "rocksdb_crc32c.h"
 #include "server/redis_reply.h"
 #include "server/server.h"
@@ -797,13 +798,13 @@ Status ReplicationThread::fetchFile(int sock_fd, evbuffer *evbuf, const std::str
     UniqueEvbufReadln line(evbuf, EVBUFFER_EOL_CRLF_STRICT);
     if (!line) {
       if (evbuffer_read(evbuf, sock_fd, -1) <= 0) {
-        return Status(Status::NotOK, std::string("read size: ") + strerror(errno));
+        return {Status::NotOK, fmt::format("read size: {}", strerror(errno))};
       }
       continue;
     }
     if (line[0] == '-') {
       std::string msg(line.get());
-      return Status(Status::NotOK, msg);
+      return {Status::NotOK, msg};
     }
     file_size = line.length > 0 ? std::strtoull(line.get(), nullptr, 10) : 0;
     break;
@@ -812,7 +813,7 @@ Status ReplicationThread::fetchFile(int sock_fd, evbuffer *evbuf, const std::str
   // Write to tmp file
   auto tmp_file = Engine::Storage::ReplDataManager::NewTmpFile(storage_, dir, file);
   if (!tmp_file) {
-    return Status(Status::NotOK, "unable to create tmp file");
+    return {Status::NotOK, "unable to create tmp file"};
   }
 
   size_t remain = file_size;
@@ -823,22 +824,20 @@ Status ReplicationThread::fetchFile(int sock_fd, evbuffer *evbuf, const std::str
       auto data_len = evbuffer_remove(evbuf, data, remain > 16 * 1024 ? 16 * 1024 : remain);
       if (data_len == 0) continue;
       if (data_len < 0) {
-        return Status(Status::NotOK, "read sst file data error");
+        return {Status::NotOK, "read sst file data error"};
       }
       tmp_file->Append(rocksdb::Slice(data, data_len));
       tmp_crc = rocksdb::crc32c::Extend(tmp_crc, data, data_len);
       remain -= data_len;
     } else {
       if (evbuffer_read(evbuf, sock_fd, -1) <= 0) {
-        return Status(Status::NotOK, std::string("read sst file: ") + strerror(errno));
+        return {Status::NotOK, fmt::format("read sst file: {}", strerror(errno))};
       }
     }
   }
   // Verify file crc checksum if crc is not 0
   if (crc && crc != tmp_crc) {
-    char err_buf[64];
-    snprintf(err_buf, sizeof(err_buf), "CRC mismatched, %u was expected but got %u", crc, tmp_crc);
-    return Status(Status::NotOK, err_buf);
+    return {Status::NotOK, fmt::format("CRC mismatched, {} was expected but got {}", crc, tmp_crc)};
   }
   // File is OK, rename to formal name
   auto s = Engine::Storage::ReplDataManager::SwapTmpFile(storage_, dir, file);

src/cluster/slot_import.cc

@@ -153,5 +153,5 @@ void SlotImport::GetImportInfo(std::string *info) {
       break;
   }
 
-  *info = "importing_slot: " + std::to_string(import_slot_) + "\r\n" + "import_state: " + import_stat + "\r\n";
+  *info = fmt::format("importing_slot: {}\r\nimport_state: {}\r\n", import_slot_, import_stat);
 }

src/cluster/slot_migrate.cc

@@ -24,6 +24,7 @@
 #include <utility>
 
 #include "event_util.h"
+#include "fmt/format.h"
 #include "storage/batch_extractor.h"
 
 static std::map<RedisType, std::string> type_to_cmd = {
@@ -481,7 +482,7 @@ bool SlotMigrate::CheckResponseWithCounts(int sock_fd, int total) {
   while (true) {
     // Read response data from socket buffer to the event buffer
     if (evbuffer_read(evbuf.get(), sock_fd, -1) <= 0) {
-      LOG(ERROR) << "[migrate] Failed to read response, Err: " + std::string(strerror(errno));
+      LOG(ERROR) << "[migrate] Failed to read response, Err: " << strerror(errno);
       return false;
     }
 
@@ -499,8 +500,7 @@ bool SlotMigrate::CheckResponseWithCounts(int sock_fd, int total) {
           }
 
           if (line[0] == '-') {
-            LOG(ERROR) << "[migrate] Got invalid response: " + std::string(line.get())
-                       << ", line length: " << line.length;
+            LOG(ERROR) << "[migrate] Got invalid response: " << line.get() << ", line length: " << line.length;
             stat_ = Error;
           } else if (line[0] == '$') {
             auto parse_result = ParseInt<uint64_t>(std::string(line.get() + 1, line.length - 1), 10);
@@ -985,6 +985,6 @@ void SlotMigrate::GetMigrateInfo(std::string *info) {
       break;
   }
 
-  *info = "migrating_slot: " + std::to_string(slot) + "\r\n" + "destination_node: " + dst_node_ + "\r\n" +
-          "migrating_state: " + task_state + "\r\n";
+  *info =
+      fmt::format("migrating_slot: {}\r\ndestination_node: {}\r\nmigrating_state: \r\n", slot, dst_node_, task_state);
 }

src/common/util.cc

@@ -18,6 +18,7 @@
  *
  */
 
+#include "fmt/core.h"
 #define __STDC_FORMAT_MACROS
 
 #include <event2/buffer.h>
@@ -67,17 +68,14 @@
 
 namespace Util {
 Status SockConnect(const std::string &host, uint32_t port, int *fd) {
-  int rv;
-  char portstr[6]; /* strlen("65535") + 1; */
   addrinfo hints, *servinfo, *p;
 
-  snprintf(portstr, sizeof(portstr), "%u", port);
   memset(&hints, 0, sizeof(hints));
   hints.ai_family = AF_UNSPEC;
   hints.ai_socktype = SOCK_STREAM;
 
-  if ((rv = getaddrinfo(host.c_str(), portstr, &hints, &servinfo)) != 0) {
-    return Status(Status::NotOK, gai_strerror(rv));
+  if (int rv = getaddrinfo(host.c_str(), std::to_string(port).c_str(), &hints, &servinfo); rv != 0) {
+    return {Status::NotOK, gai_strerror(rv)};
   }
 
   auto exit = MakeScopeExit([servinfo] { freeaddrinfo(servinfo); });
@@ -108,22 +106,20 @@ const std::string Float2String(double d) {
     return d > 0 ? "inf" : "-inf";
   }
 
-  char buf[128];
-  snprintf(buf, sizeof(buf), "%.17g", d);
-  return buf;
+  return fmt::format("{.17g}", d);
 }
 
 Status SockSetTcpNoDelay(int fd, int val) {
   if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &val, sizeof(val)) == -1) {
-    return Status(Status::NotOK, strerror(errno));
+    return Status::FromErrno();
   }
   return Status::OK();
 }
 
 Status SockSetTcpKeepalive(int fd, int interval) {
   int val = 1;
   if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val)) == -1) {
-    return Status(Status::NotOK, strerror(errno));
+    return Status::FromErrno();
   }
 
 #ifdef __linux__
@@ -134,7 +130,7 @@ Status SockSetTcpKeepalive(int fd, int interval) {
   // Send first probe after interval.
   val = interval;
   if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &val, sizeof(val)) < 0) {
-    return Status(Status::NotOK, std::string("setsockopt TCP_KEEPIDLE: ") + strerror(errno));
+    return {Status::NotOK, fmt::format("setsockopt TCP_KEEPIDLE: {}", strerror(errno))};
   }
 
   // Send next probes after the specified interval. Note that we set the
@@ -143,14 +139,14 @@ Status SockSetTcpKeepalive(int fd, int interval) {
   val = interval / 3;
   if (val == 0) val = 1;
   if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &val, sizeof(val)) < 0) {
-    return Status(Status::NotOK, std::string("setsockopt TCP_KEEPINTVL: ") + strerror(errno));
+    return {Status::NotOK, fmt::format("setsockopt TCP_KEEPINTVL: {}", strerror(errno))};
   }
 
   // Consider the socket in error state after three we send three ACK
   // probes without getting a reply.
   val = 3;
   if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &val, sizeof(val)) < 0) {
-    return Status(Status::NotOK, std::string("setsockopt TCP_KEEPCNT: ") + strerror(errno));
+    return {Status::NotOK, fmt::format("setsockopt TCP_KEEPCNT: {}", strerror(errno))};
   }
 #else
   ((void)interval);  // Avoid unused var warning for non Linux systems.
@@ -540,28 +536,30 @@ std::string StringToHex(const std::string &input) {
   return output;
 }
 
+constexpr unsigned long long expTo1024(unsigned n) { return 1ULL << (n * 10); }
+
 void BytesToHuman(char *buf, size_t size, uint64_t n) {
-  double d;
-
-  if (n < 1024) {
-    snprintf(buf, size, "%" PRIu64 "B", n);
-  } else if (n < (1024 * 1024)) {
-    d = static_cast<double>(n) / (1024);
-    snprintf(buf, size, "%.2fK", d);
-  } else if (n < (1024LL * 1024 * 1024)) {
-    d = static_cast<double>(n) / (1024 * 1024);
-    snprintf(buf, size, "%.2fM", d);
-  } else if (n < (1024LL * 1024 * 1024 * 1024)) {
-    d = static_cast<double>(n) / (1024LL * 1024 * 1024);
-    snprintf(buf, size, "%.2fG", d);
-  } else if (n < (1024LL * 1024 * 1024 * 1024 * 1024)) {
-    d = static_cast<double>(n) / (1024LL * 1024 * 1024 * 1024);
-    snprintf(buf, size, "%.2fT", d);
-  } else if (n < (1024LL * 1024 * 1024 * 1024 * 1024 * 1024)) {
-    d = static_cast<double>(n) / (1024LL * 1024 * 1024 * 1024 * 1024);
-    snprintf(buf, size, "%.2fP", d);
+  double d = 0;
+
+  if (n < expTo1024(1)) {
+    fmt::format_to_n(buf, size, "{}B", n);
+  } else if (n < expTo1024(2)) {
+    d = static_cast<double>(n) / expTo1024(1);
+    fmt::format_to_n(buf, size, "{.2f}K", d);
+  } else if (n < expTo1024(3)) {
+    d = static_cast<double>(n) / expTo1024(2);
+    fmt::format_to_n(buf, size, "{.2f}M", d);
+  } else if (n < expTo1024(4)) {
+    d = static_cast<double>(n) / expTo1024(3);
+    fmt::format_to_n(buf, size, "{.2f}G", d);
+  } else if (n < expTo1024(5)) {
+    d = static_cast<double>(n) / expTo1024(4);
+    fmt::format_to_n(buf, size, "{.2f}T", d);
+  } else if (n < expTo1024(6)) {
+    d = static_cast<double>(n) / expTo1024(5);
+    fmt::format_to_n(buf, size, "{.2f}P", d);
   } else {
-    snprintf(buf, size, "%" PRIu64 "B", n);
+    fmt::format_to_n(buf, size, "{}B", n);
   }
 }