fix: merge branch 'main' into bobik/rename_in_cluster_mode_crash

Author: vyavdoshenko

.github/workflows/ci.yml

@@ -140,7 +140,7 @@ jobs:
           echo "disk space is:"
           df -h
 
-      - name: C++ Unit Tests
+      - name: C++ Unit Tests - IoUring
         run: |
           cd ${GITHUB_WORKSPACE}/build
           echo Run ctest -V -L DFLY
@@ -151,7 +151,14 @@ jobs:
           # Run allocation tracker test separately without alsologtostderr because it generates a TON of logs.
           FLAGS_fiber_safety_margin=4096 timeout 5m ./allocation_tracker_test
 
-          echo "Running tests with --force_epoll"
+          timeout 5m ./dragonfly_test
+          timeout 5m ./json_family_test --jsonpathv2=false
+          timeout 5m ./tiered_storage_test --vmodule=db_slice=2 --logtostderr
+
+
+      - name: C++ Unit Tests - Epoll
+        run: |
+          cd ${GITHUB_WORKSPACE}/build
 
           # Create a rule that automatically prints stacktrace upon segfault
           cat > ./init.gdb <<EOF
@@ -166,18 +173,14 @@ jobs:
 
           FLAGS_fiber_safety_margin=4096 FLAGS_force_epoll=true timeout 5m ./allocation_tracker_test
 
-          echo "Finished running tests with --force_epoll"
-
-          echo "Running tests with --cluster_mode=emulated"
+      - name: C++ Unit Tests - IoUring with cluster mode
+        run: |
           FLAGS_fiber_safety_margin=4096 FLAGS_cluster_mode=emulated timeout 20m ctest -V -L DFLY
 
-          echo "Running tests with both --cluster_mode=emulated & --lock_on_hashtags"
+      - name: C++ Unit Tests - IoUring with cluster mode and FLAGS_lock_on_hashtags
+        run: |
           FLAGS_fiber_safety_margin=4096 FLAGS_cluster_mode=emulated FLAGS_lock_on_hashtags=true timeout 20m ctest -V -L DFLY
 
-          timeout 5m ./dragonfly_test
-          timeout 5m ./json_family_test --jsonpathv2=false
-          timeout 5m ./tiered_storage_test --vmodule=db_slice=2 --logtostderr
-
       - name: Upload unit logs on failure
         if: failure()
         uses: actions/upload-artifact@v4

src/redis/lua/struct/lua_struct.c

@@ -82,14 +82,12 @@ typedef struct Header {
 } Header;
 
 
-static int getnum (lua_State *L, const char **fmt, int df) {
+static int getnum (const char **fmt, int df) {
   if (!isdigit(**fmt))  /* no number? */
     return df;  /* return default value */
   else {
     int a = 0;
     do {
-      if (a > (INT_MAX / 10) || a * 10 > (INT_MAX - (**fmt - '0')))
-        luaL_error(L, "integral size overflow");
       a = a*10 + *((*fmt)++) - '0';
     } while (isdigit(**fmt));
     return a;
@@ -110,9 +108,9 @@ static size_t optsize (lua_State *L, char opt, const char **fmt) {
     case 'f':  return sizeof(float);
     case 'd':  return sizeof(double);
     case 'x': return 1;
-    case 'c': return getnum(L, fmt, 1);
+    case 'c': return  getnum(fmt, 1);
     case 'i': case 'I': {
-      int sz = getnum(L, fmt, sizeof(int));
+      int sz = getnum(fmt, sizeof(int));
       if (sz > MAXINTSIZE)
         luaL_error(L, "integral size %d is larger than limit of %d",
                        sz, MAXINTSIZE);
@@ -145,7 +143,7 @@ static void controloptions (lua_State *L, int opt, const char **fmt,
     case '>': h->endian = BIG; return;
     case '<': h->endian = LITTLE; return;
     case '!': {
-      int a = getnum(L, fmt, MAXALIGN);
+      int a = getnum(fmt, MAXALIGN);
       if (!isp2(a))
         luaL_error(L, "alignment %d is not a power of 2", a);
       h->align = a;

src/server/cluster_support.cc

@@ -17,6 +17,10 @@ ABSL_FLAG(string, cluster_mode, "",
           "Cluster mode supported. Possible values are "
           "'emulated', 'yes' or ''");
 
+ABSL_FLAG(bool, experimental_cluster_shard_by_slot, false,
+          "If true, cluster mode is enabled and sharding is done by slot. "
+          "Otherwise, sharding is done by hash tag.");
+
 namespace dfly {
 
 void UniqueSlotChecker::Add(std::string_view key) {
@@ -43,16 +47,13 @@ optional<SlotId> UniqueSlotChecker::GetUniqueSlotId() const {
   return slot_id_ > kMaxSlotNum ? optional<SlotId>() : slot_id_;
 }
 
-namespace {
-enum class ClusterMode {
-  kUninitialized,
-  kNoCluster,
-  kEmulatedCluster,
-  kRealCluster,
-};
-
+namespace detail {
 ClusterMode cluster_mode = ClusterMode::kUninitialized;
-}  // namespace
+bool cluster_shard_by_slot = false;
+
+}  // namespace detail
+
+using namespace detail;
 
 void InitializeCluster() {
   string cluster_mode_str = absl::GetFlag(FLAGS_cluster_mode);
@@ -67,25 +68,17 @@ void InitializeCluster() {
     LOG(ERROR) << "Invalid value for flag --cluster_mode. Exiting...";
     exit(1);
   }
-}
 
-bool IsClusterEnabled() {
-  return cluster_mode == ClusterMode::kRealCluster;
-}
-
-bool IsClusterEmulated() {
-  return cluster_mode == ClusterMode::kEmulatedCluster;
+  if (cluster_mode != ClusterMode::kNoCluster) {
+    cluster_shard_by_slot = absl::GetFlag(FLAGS_experimental_cluster_shard_by_slot);
+  }
 }
 
 SlotId KeySlot(std::string_view key) {
   string_view tag = LockTagOptions::instance().Tag(key);
   return crc16(tag.data(), tag.length()) & kMaxSlotNum;
 }
 
-bool IsClusterEnabledOrEmulated() {
-  return IsClusterEnabled() || IsClusterEmulated();
-}
-
 bool IsClusterShardedByTag() {
   return IsClusterEnabledOrEmulated() || LockTagOptions::instance().enabled;
 }

src/server/cluster_support.h

@@ -10,6 +10,20 @@
 
 namespace dfly {
 
+namespace detail {
+
+enum class ClusterMode {
+  kUninitialized,
+  kNoCluster,
+  kEmulatedCluster,
+  kRealCluster,
+};
+
+extern ClusterMode cluster_mode;
+extern bool cluster_shard_by_slot;
+
+};  // namespace detail
+
 using SlotId = std::uint16_t;
 constexpr SlotId kMaxSlotNum = 0x3FFF;
 
@@ -42,9 +56,23 @@ class UniqueSlotChecker {
 SlotId KeySlot(std::string_view key);
 
 void InitializeCluster();
-bool IsClusterEnabled();
-bool IsClusterEmulated();
-bool IsClusterEnabledOrEmulated();
+
+inline bool IsClusterEnabled() {
+  return detail::cluster_mode == detail::ClusterMode::kRealCluster;
+}
+
+inline bool IsClusterEmulated() {
+  return detail::cluster_mode == detail::ClusterMode::kEmulatedCluster;
+}
+
+inline bool IsClusterEnabledOrEmulated() {
+  return IsClusterEnabled() || IsClusterEmulated();
+}
+
+inline bool IsClusterShardedBySlot() {
+  return detail::cluster_shard_by_slot;
+}
+
 bool IsClusterShardedByTag();
 
 }  // namespace dfly

src/server/debugcmd.cc

@@ -28,6 +28,7 @@ extern "C" {
 #include "core/sorted_map.h"
 #include "core/string_map.h"
 #include "core/string_set.h"
+#include "facade/cmd_arg_parser.h"
 #include "server/blocking_controller.h"
 #include "server/container_utils.h"
 #include "server/engine_shard_set.h"
@@ -38,7 +39,6 @@ extern "C" {
 #include "server/server_state.h"
 #include "server/string_family.h"
 #include "server/transaction.h"
-
 using namespace std;
 
 ABSL_DECLARE_FLAG(string, dir);
@@ -712,106 +712,57 @@ void DebugCmd::Migration(CmdArgList args, facade::SinkReplyBuilder* builder) {
   return builder->SendError(UnknownSubCmd("MIGRATION", "DEBUG"));
 }
 
+enum PopulateFlag { FLAG_RAND, FLAG_TYPE, FLAG_ELEMENTS, FLAG_SLOT, FLAG_EXPIRE, FLAG_UNKNOWN };
+
 // Populate arguments format:
 // required: (total count) (key prefix) (val size)
 // optional: [RAND | TYPE typename | ELEMENTS element num | SLOTS (key value)+ | EXPIRE start end]
 optional<DebugCmd::PopulateOptions> DebugCmd::ParsePopulateArgs(CmdArgList args,
                                                                 facade::SinkReplyBuilder* builder) {
-  if (args.size() < 2) {
-    builder->SendError(UnknownSubCmd("populate", "DEBUG"));
-    return nullopt;
-  }
-
+  CmdArgParser parser(args.subspan(1));
   PopulateOptions options;
-  if (!absl::SimpleAtoi(ArgS(args, 1), &options.total_count)) {
-    builder->SendError(kUintErr);
-    return nullopt;
-  }
-
-  if (args.size() > 2) {
-    options.prefix = ArgS(args, 2);
-  }
 
-  if (args.size() > 3) {
-    if (!absl::SimpleAtoi(ArgS(args, 3), &options.val_size)) {
-      builder->SendError(kUintErr);
-      return nullopt;
-    }
-  }
+  options.total_count = parser.Next<uint64_t>();
+  options.prefix = parser.NextOrDefault<string_view>("key");
+  options.val_size = parser.NextOrDefault<uint32_t>(16);
 
-  for (size_t index = 4; args.size() > index; ++index) {
-    string str = absl::AsciiStrToUpper(ArgS(args, index));
-    if (str == "RAND") {
-      options.populate_random_values = true;
-    } else if (str == "TYPE") {
-      if (args.size() < index + 2) {
-        builder->SendError(kSyntaxErr);
-        return nullopt;
-      }
-      ++index;
-      options.type = absl::AsciiStrToUpper(ArgS(args, index));
-    } else if (str == "ELEMENTS") {
-      if (args.size() < index + 2) {
-        builder->SendError(kSyntaxErr);
-        return nullopt;
-      }
-      if (!absl::SimpleAtoi(ArgS(args, ++index), &options.elements)) {
-        builder->SendError(kSyntaxErr);
-        return nullopt;
-      }
-    } else if (str == "SLOTS") {
-      if (args.size() < index + 3) {
-        builder->SendError(kSyntaxErr);
-        return nullopt;
+  while (parser.HasNext()) {
+    PopulateFlag flag = parser.MapNext("RAND", FLAG_RAND, "TYPE", FLAG_TYPE, "ELEMENTS",
+                                       FLAG_ELEMENTS, "SLOTS", FLAG_SLOT, "EXPIRE", FLAG_EXPIRE);
+    switch (flag) {
+      case FLAG_RAND:
+        options.populate_random_values = true;
+        break;
+      case FLAG_TYPE:
+        options.type = absl::AsciiStrToUpper(parser.Next<string_view>());
+        break;
+      case FLAG_ELEMENTS:
+        options.elements = parser.Next<uint32_t>();
+        break;
+      case FLAG_SLOT: {
+        auto [start, end] = parser.Next<FInt<0, 16383>, FInt<0, 16383>>();
+        options.slot_range = cluster::SlotRange{SlotId(start), SlotId(end)};
+        break;
       }
-
-      auto parse_slot = [](string_view slot_str) -> OpResult<uint32_t> {
-        uint32_t slot_id;
-        if (!absl::SimpleAtoi(slot_str, &slot_id)) {
-          return facade::OpStatus::INVALID_INT;
-        }
-        if (slot_id > kMaxSlotNum) {
-          return facade::OpStatus::INVALID_VALUE;
+      case FLAG_EXPIRE: {
+        auto [min_ttl, max_ttl] = parser.Next<uint32_t, uint32_t>();
+        if (min_ttl >= max_ttl) {
+          builder->SendError(kExpiryOutOfRange);
+          (void)parser.Error();
+          return nullopt;
         }
-        return slot_id;
-      };
-
-      auto start = parse_slot(ArgS(args, ++index));
-      if (start.status() != facade::OpStatus::OK) {
-        builder->SendError(start.status());
-        return nullopt;
-      }
-      auto end = parse_slot(ArgS(args, ++index));
-      if (end.status() != facade::OpStatus::OK) {
-        builder->SendError(end.status());
-        return nullopt;
-      }
-      options.slot_range = cluster::SlotRange{.start = static_cast<SlotId>(start.value()),
-                                              .end = static_cast<SlotId>(end.value())};
-    } else if (str == "EXPIRE") {
-      if (args.size() < index + 3) {
-        builder->SendError(kSyntaxErr);
-        return nullopt;
-      }
-      uint32_t start, end;
-      if (!absl::SimpleAtoi(ArgS(args, ++index), &start)) {
-        builder->SendError(kSyntaxErr);
-        return nullopt;
-      }
-      if (!absl::SimpleAtoi(ArgS(args, ++index), &end)) {
-        builder->SendError(kSyntaxErr);
-        return nullopt;
-      }
-      if (start >= end) {
-        builder->SendError(kExpiryOutOfRange);
-        return nullopt;
+        options.expire_ttl_range = std::make_pair(min_ttl, max_ttl);
+        break;
       }
-      options.expire_ttl_range = std::make_pair(start, end);
-    } else {
-      builder->SendError(kSyntaxErr);
-      return nullopt;
+      default:
+        LOG(FATAL) << "Unexpected flag in PopulateArgs. Args: " << args;
+        break;
     }
   }
+  if (parser.HasError()) {
+    builder->SendError(parser.Error()->MakeReply());
+    return nullopt;
+  }
   return options;
 }
 

src/server/engine_shard.cc

@@ -261,6 +261,17 @@ __thread EngineShard* EngineShard::shard_ = nullptr;
 uint64_t TEST_current_time_ms = 0;
 
 ShardId Shard(string_view v, ShardId shard_num) {
+  // This cluster sharding is not necessary and may degrade keys distribution among shard threads.
+  // For example, if we have 3 shards, then no single-char keys will be assigned to shard 2 and
+  // 32 single char keys in range ['_' - '~'] will be assigned to shard 0.
+  // Yes, SlotId function does not have great distribution properties.
+  // On the other side, slot based sharding may help with pipeline squashing optimizations,
+  // because they rely on commands being single-sharded.
+  // TODO: once we improve our squashing logic, we can remove this.
+  if (IsClusterShardedBySlot()) {
+    return KeySlot(v) % shard_num;
+  }
+
   if (IsClusterShardedByTag()) {
     v = LockTagOptions::instance().Tag(v);
   }

src/server/hll_family_test.cc

@@ -194,6 +194,8 @@ TEST_F(HllFamilyTest, MergeOverlapping) {
 }
 
 TEST_F(HllFamilyTest, MergeInvalid) {
+  GTEST_SKIP() << "TBD: MergeInvalid test fails with multi-shard runs, see #5004";
+
   EXPECT_EQ(CheckedInt({"pfadd", "key1", "1", "2", "3"}), 1);
   EXPECT_EQ(Run({"set", "key2", "..."}), "OK");
   EXPECT_THAT(Run({"pfmerge", "key1", "key2"}), ErrArg(HllFamily::kInvalidHllErr));