fix CR

Signed-off-by: adi_holden <[email protected]>

Author: adiholden

src/server/cluster/cluster_config.cc

@@ -70,6 +70,24 @@ SlotId ClusterConfig::KeySlot(string_view key) {
   return crc16(tag.data(), tag.length()) & kMaxSlotNum;
 }
 
+OpResult<SlotRange> ClusterConfig::SlotRangeFromStr(std::string_view start, std::string_view end) {
+  uint32_t slot_id_start, slot_id_end;
+  if (!absl::SimpleAtoi(start, &slot_id_start)) {
+    return facade::OpStatus::INVALID_INT;
+  }
+  if (slot_id_start > ClusterConfig::kMaxSlotNum) {
+    return facade::OpStatus::INVALID_VALUE;
+  }
+  if (!absl::SimpleAtoi(end, &slot_id_end)) {
+    return facade::OpStatus::INVALID_INT;
+  }
+  if ((slot_id_end > ClusterConfig::kMaxSlotNum) || (slot_id_end < slot_id_start)) {
+    return facade::OpStatus::INVALID_VALUE;
+  }
+  return SlotRange{.start = static_cast<uint16_t>(slot_id_start),
+                   .end = static_cast<uint16_t>(slot_id_end)};
+}
+
 namespace {
 bool HasValidNodeIds(const ClusterConfig::ClusterShards& new_config) {
   absl::flat_hash_set<string_view> nodes;

src/server/cluster/cluster_config.h

@@ -10,11 +10,14 @@
 #include <vector>
 
 #include "core/json/json_object.h"
+#include "src/facade/op_status.h"
 #include "src/server/cluster/slot_set.h"
 #include "src/server/common.h"
 
 namespace dfly {
 
+using facade::OpResult;
+
 // MigrationState constants are ordered in state changing order
 enum class MigrationState : uint8_t {
   C_NO_STATE,
@@ -57,6 +60,7 @@ class ClusterConfig {
   using ClusterShards = std::vector<ClusterShard>;
 
   static SlotId KeySlot(std::string_view key);
+  static OpResult<SlotRange> SlotRangeFromStr(std::string_view start, std::string_view end);
 
   static void Initialize();
   static bool IsEnabled();

src/server/cluster/slot_set.h

@@ -99,7 +99,7 @@ class SlotSet {
     std::string slots_str;
     for (SlotId i = 0; i < kSlotsNumber; ++i) {
       if (slots_->test(i)) {
-        absl::StrAppend(&slots_str, absl::StrCat(i));
+        absl::StrAppend(&slots_str, absl::StrCat(i, " "));
       }
     }
     return slots_str;

src/server/common.cc

@@ -395,4 +395,17 @@ std::string AbslUnparseFlag(const dfly::MemoryBytesFlag& flag) {
   return strings::HumanReadableNumBytes(flag.value);
 }
 
+std::ostream& operator<<(std::ostream& os, const GlobalState& state) {
+  return os << GlobalStateName(state);
+}
+
+std::ostream& operator<<(std::ostream& os, ArgSlice list) {
+  os << "[";
+  if (!list.empty()) {
+    std::for_each(list.begin(), list.end() - 1, [&os](const auto& val) { os << val << ", "; });
+    os << (*(list.end() - 1));
+  }
+  return os << "]";
+}
+
 }  // namespace dfly

src/server/common.h

@@ -160,6 +160,12 @@ enum class GlobalState : uint8_t {
   TAKEN_OVER,
 };
 
+const char* GlobalStateName(GlobalState gs);
+
+std::ostream& operator<<(std::ostream& os, const GlobalState& state);
+
+std::ostream& operator<<(std::ostream& os, ArgSlice list);
+
 enum class TimeUnit : uint8_t { SEC, MSEC };
 
 inline void ToUpper(const MutableSlice* val) {
@@ -195,8 +201,6 @@ int64_t GetMallocCurrentCommitted();
 // set upon server start.
 extern unsigned kernel_version;
 
-const char* GlobalStateName(GlobalState gs);
-
 template <typename RandGen> std::string GetRandomHex(RandGen& gen, size_t len) {
   static_assert(std::is_same<uint64_t, decltype(gen())>::value);
   std::string res(len, '\0');

src/server/main_service.cc

@@ -2405,35 +2405,38 @@ VarzValue::Map Service::GetVarzStats() {
 
 GlobalState Service::SwitchState(GlobalState from, GlobalState to) {
   lock_guard lk(mu_);
-  // Counting switchs to loading state enables us to do several loading processes (replications)
-  // at the same time. Only when all loading processes finish we exit loading state.
-  // When server is in loading state, calling SwitchState to loading will increase
-  // loading_state_counter_. When server is in loading state, calling SwitchState to active will
-  // decrease loading_state_counter_, if loading_state_counter_ is 0 switch to active.
-  if (global_state_ == to && global_state_ == GlobalState::LOADING) {
-    ++loading_state_counter_;
-    return global_state_;
-  }
-  if (global_state_ == from && global_state_ == GlobalState::LOADING &&
-      loading_state_counter_ > 0) {
-    --loading_state_counter_;
-    if (loading_state_counter_ > 0) {
-      return global_state_;
-    }
-  }
-
   if (global_state_ != from) {
     return global_state_;
   }
 
   VLOG(1) << "Switching state from " << GlobalStateName(from) << " to " << GlobalStateName(to);
-
   global_state_ = to;
 
   pp_.Await([&](ProactorBase*) { ServerState::tlocal()->set_gstate(to); });
   return to;
 }
 
+void Service::RequestLoadingState() {
+  unique_lock lk(mu_);
+  ++loading_state_counter_;
+  if (global_state_ != GlobalState::LOADING) {
+    DCHECK_EQ(global_state_, GlobalState::ACTIVE);
+    lk.unlock();
+    SwitchState(GlobalState::ACTIVE, GlobalState::LOADING);
+  }
+}
+
+void Service::RemoveLoadingState() {
+  unique_lock lk(mu_);
+  DCHECK_EQ(global_state_, GlobalState::LOADING);
+  DCHECK_GT(loading_state_counter_, 0u);
+  --loading_state_counter_;
+  if (loading_state_counter_ == 0) {
+    lk.unlock();
+    SwitchState(GlobalState::LOADING, GlobalState::ACTIVE);
+  }
+}
+
 GlobalState Service::GetGlobalState() const {
   lock_guard lk(mu_);
   return global_state_;

src/server/main_service.h

@@ -87,6 +87,9 @@ class Service : public facade::ServiceInterface {
   // Upon switch, updates cached global state in threadlocal ServerState struct.
   GlobalState SwitchState(GlobalState from, GlobalState to);
 
+  void RequestLoadingState();
+  void RemoveLoadingState();
+
   GlobalState GetGlobalState() const;
 
   void ConfigureHttpHandlers(util::HttpListenerBase* base, bool is_privileged) final;

src/server/replica.cc

@@ -376,10 +376,8 @@ error_code Replica::InitiatePSync() {
     io::PrefixSource ps{io_buf.InputBuffer(), Sock()};
 
     // Set LOADING state.
-    CHECK(service_.SwitchState(GlobalState::ACTIVE, GlobalState::LOADING) == GlobalState::LOADING);
-    absl::Cleanup cleanup = [this]() {
-      service_.SwitchState(GlobalState::LOADING, GlobalState::ACTIVE);
-    };
+    service_.RequestLoadingState();
+    absl::Cleanup cleanup = [this]() { service_.RemoveLoadingState(); };
 
     if (slot_range_.has_value()) {
       JournalExecutor{&service_}.FlushSlots(slot_range_.value());
@@ -434,14 +432,6 @@ error_code Replica::InitiatePSync() {
 error_code Replica::InitiateDflySync() {
   auto start_time = absl::Now();
 
-  absl::Cleanup cleanup = [this]() {
-    // We do the following operations regardless of outcome.
-    JoinDflyFlows();
-    service_.SwitchState(GlobalState::LOADING, GlobalState::ACTIVE);
-    state_mask_.fetch_and(~R_SYNCING);
-    last_journal_LSNs_.reset();
-  };
-
   // Initialize MultiShardExecution.
   multi_shard_exe_.reset(new MultiShardExecution());
 
@@ -471,10 +461,19 @@ error_code Replica::InitiateDflySync() {
   RETURN_ON_ERR(cntx_.SwitchErrorHandler(std::move(err_handler)));
 
   // Make sure we're in LOADING state.
-  CHECK(service_.SwitchState(GlobalState::ACTIVE, GlobalState::LOADING) == GlobalState::LOADING);
+  service_.RequestLoadingState();
 
   // Start full sync flows.
   state_mask_.fetch_or(R_SYNCING);
+
+  absl::Cleanup cleanup = [this]() {
+    // We do the following operations regardless of outcome.
+    JoinDflyFlows();
+    service_.RemoveLoadingState();
+    state_mask_.fetch_and(~R_SYNCING);
+    last_journal_LSNs_.reset();
+  };
+
   std::string_view sync_type = "full";
   {
     // Going out of the way to avoid using std::vector<bool>...

src/server/server_family.cc

@@ -551,53 +551,49 @@ string_view GetRedisMode() {
 }
 
 struct ReplicaOfArgs {
-  bool no_one = false;
   string_view host;
   string_view port_sv;
-  uint32_t port;
+  uint16_t port;
   std::optional<SlotRange> slot_range;
   static OpResult<ReplicaOfArgs> FromParams(string_view host, string_view port,
                                             std::optional<string_view> slot_start,
                                             std::optional<string_view> slot_end);
+  bool IsReplicaOfNoOne();
 };
 
-OpResult<ReplicaOfArgs> ReplicaOfArgs::FromParams(string_view host, string_view port,
+bool ReplicaOfArgs::IsReplicaOfNoOne() {
+  if (absl::EqualsIgnoreCase(host, "no") && absl::EqualsIgnoreCase(port_sv, "one")) {
+    return true;
+  }
+  return false;
+}
+
+OpResult<ReplicaOfArgs> ReplicaOfArgs::FromParams(string_view host, string_view port_sv,
                                                   std::optional<string_view> slot_start,
                                                   std::optional<string_view> slot_end) {
   ReplicaOfArgs replicaof_args;
   replicaof_args.host = host;
-  replicaof_args.port_sv = port;
-  if (absl::EqualsIgnoreCase(replicaof_args.host, "no") &&
-      absl::EqualsIgnoreCase(replicaof_args.port_sv, "one")) {
-    replicaof_args.no_one = true;
+  replicaof_args.port_sv = port_sv;
+  if (replicaof_args.IsReplicaOfNoOne()) {
     return replicaof_args;
   }
 
-  if (!absl::SimpleAtoi(replicaof_args.port_sv, &replicaof_args.port) || replicaof_args.port < 1 ||
-      replicaof_args.port > 65535) {
+  uint32_t port;
+  if (!absl::SimpleAtoi(port_sv, &port) || port < 1 || port > 65535) {
     return facade::OpStatus::INVALID_INT;
   }
+  replicaof_args.port = static_cast<uint16_t>(port);
 
   if (slot_start.has_value()) {
     if (!slot_end.has_value()) {
       return facade::OpStatus::SYNTAX_ERR;
     }
 
-    uint32_t slot_id_start, slot_id_end;
-    if (!absl::SimpleAtoi(*slot_start, &slot_id_start)) {
-      return facade::OpStatus::INVALID_INT;
-    }
-    if (slot_id_start > ClusterConfig::kMaxSlotNum) {
-      return facade::OpStatus::INVALID_VALUE;
+    OpResult<SlotRange> slot_range = ClusterConfig::SlotRangeFromStr(*slot_start, *slot_end);
+    if (!slot_range) {
+      return slot_range.status();
     }
-    if (!absl::SimpleAtoi(*slot_end, &slot_id_end)) {
-      return facade::OpStatus::INVALID_INT;
-    }
-    if (slot_id_end > ClusterConfig::kMaxSlotNum) {
-      return facade::OpStatus::INVALID_VALUE;
-    }
-    replicaof_args.slot_range = SlotRange{.start = static_cast<uint16_t>(slot_id_start),
-                                          .end = static_cast<uint16_t>(slot_id_end)};
+    replicaof_args.slot_range = slot_range.value();
   }
   return replicaof_args;
 }
@@ -846,9 +842,9 @@ void ServerFamily::Shutdown() {
     unique_lock lk(replicaof_mu_);
     if (replica_) {
       replica_->Stop();
-      for (auto& replica : cluster_replicas_) {
-        replica->Stop();
-      }
+    }
+    for (auto& replica : cluster_replicas_) {
+      replica->Stop();
     }
 
     dfly_cmd_->Shutdown();
@@ -2344,7 +2340,7 @@ void ServerFamily::AddReplicaOf(CmdArgList args, ConnectionContext* cntx) {
   if (!replicaof_args) {
     return cntx->SendError(replicaof_args.status());
   }
-  if (replicaof_args->no_one) {
+  if (replicaof_args->IsReplicaOfNoOne()) {
     return cntx->SendError("ADDREPLICAOF does not supprot no one");
   }
   LOG(INFO) << "Add Replica " << replicaof_args->host << ":" << replicaof_args->port_sv;
@@ -2377,7 +2373,7 @@ void ServerFamily::ReplicaOfInternal(std::string_view host, std::string_view por
   }
 
   // If NO ONE was supplied, just stop the current replica (if it exists)
-  if (replicaof_args->no_one) {
+  if (replicaof_args->IsReplicaOfNoOne()) {
     if (!ServerState::tlocal()->is_master) {
       CHECK(replica_);
 
@@ -2618,9 +2614,7 @@ void ServerFamily::Role(CmdArgList args, ConnectionContext* cntx) {
 
   } else {
     unique_lock lk{replicaof_mu_};
-
-    size_t additional_replication = cluster_replicas_.size();
-    rb->StartArray(4 + additional_replication * 3);
+    rb->StartArray(4 + cluster_replicas_.size() * 3);
     rb->SendBulkString("replica");
 
     auto send_replica_info = [rb](Replica::Info rinfo) {

src/server/test_utils.cc

@@ -31,15 +31,6 @@ ABSL_FLAG(bool, force_epoll, false, "If true, uses epoll api instead iouring to
 
 namespace dfly {
 
-std::ostream& operator<<(std::ostream& os, ArgSlice list) {
-  os << "[";
-  if (!list.empty()) {
-    std::for_each(list.begin(), list.end() - 1, [&os](const auto& val) { os << val << ", "; });
-    os << (*(list.end() - 1));
-  }
-  return os << "]";
-}
-
 std::ostream& operator<<(std::ostream& os, const DbStats& stats) {
   os << "keycount: " << stats.key_count << ", tiered_size: " << stats.tiered_size
      << ", tiered_entries: " << stats.tiered_entries << "\n";