one to rule them all

Author: kostasrim

src/server/replica.cc

@@ -505,14 +505,23 @@ error_code Replica::InitiateDflySync(std::optional<LastMasterSyncData> last_mast
   // Initialize MultiShardExecution.
   multi_shard_exe_.reset(new MultiShardExecution());
 
-  // Initialize shard flows.
-  shard_flows_.resize(master_context_.num_flows);
-  DCHECK(!shard_flows_.empty());
-  for (unsigned i = 0; i < shard_flows_.size(); ++i) {
-    shard_flows_[i].reset(
-        new DflyShardReplica(server(), master_context_, i, &service_, multi_shard_exe_));
-  }
-  thread_flow_map_ = Partition(shard_flows_.size());
+  // Initialize shard flows. The update to the shard_flows_ should be done by this thread.
+  // Otherwise, there is a race condition between GetSummary() and the shard_flows_[i].reset()
+  // below.
+  decltype(shard_flows_) shard_flows_copy;
+  shard_flows_copy.resize(master_context_.num_flows);
+  DCHECK(!shard_flows_copy.empty());
+  thread_flow_map_ = Partition(shard_flows_copy.size());
+  const size_t pool_sz = shard_set->pool()->size();
+
+  shard_set->pool()->AwaitFiberOnAll([pool_sz, this, &shard_flows_copy](auto index, auto* ctx) {
+    for (unsigned i = index; i < shard_flows_copy.size(); i += pool_sz) {
+      shard_flows_copy[i].reset(
+          new DflyShardReplica(server(), master_context_, i, &service_, multi_shard_exe_));
+    }
+  });
+  // now update shard_flows on proactor thread
+  shard_flows_ = std::move(shard_flows_copy);
 
   // Blocked on until all flows got full sync cut.
   BlockingCounter sync_block{unsigned(shard_flows_.size())};
@@ -1171,6 +1180,7 @@ error_code Replica::ParseReplicationHeader(base::IoBuf* io_buf, PSyncResponse* d
 
 auto Replica::GetSummary() const -> Summary {
   auto f = [this]() {
+    DCHECK(Proactor() == ProactorBase::me());
     auto last_io_time = LastIoTime();
 
     // Note: we access LastIoTime from foreigh thread in unsafe manner. However, specifically here

src/server/server_family.cc

@@ -1970,11 +1970,11 @@ vector<facade::Listener*> ServerFamily::GetNonPriviligedListeners() const {
 }
 
 optional<Replica::Summary> ServerFamily::GetReplicaSummary() const {
-  util::fb2::LockGuard lk(replicaof_mu_);
-  if (replica_ == nullptr) {
+  if (tl_replica == nullptr) {
     return nullopt;
   } else {
-    return replica_->GetSummary();
+    auto replica = tl_replica;
+    return replica->GetSummary();
   }
 }
 
@@ -3095,15 +3095,12 @@ string ServerFamily::FormatInfoMetrics(const Metrics& m, std::string_view sectio
 
   auto add_repl_info = [&] {
     bool is_master = true;
-    // Thread local var is_master is updated under mutex replicaof_mu_ together with replica_,
-    // ensuring eventual consistency of is_master. When determining if the server is a replica and
-    // accessing the replica_ object, we must lock replicaof_mu_. Using is_master alone is
-    // insufficient in this scenario.
-    // Please note that we we do not use Metrics object here.
-    {
-      fb2::LockGuard lk(replicaof_mu_);
-      is_master = !replica_;
-    }
+
+    // Deep copy because tl_replica might be overwritten inbetween
+    auto replica = tl_replica;
+    auto cluster_replicas = tl_cluster_replicas;
+    is_master = !replica;
+
     if (is_master) {
       vector<ReplicaRoleInfo> replicas_info = dfly_cmd_->GetReplicasRoleInfo();
       append("role", "master");
@@ -3137,13 +3134,9 @@ string ServerFamily::FormatInfoMetrics(const Metrics& m, std::string_view sectio
         append("psync_attempts", rinfo.psync_attempts);
         append("psync_successes", rinfo.psync_successes);
       };
-      // Deep copy because tl_replica might be overwritten inbetween
-      auto replica = tl_replica;
 
       replication_info_cb(replica->GetSummary());
 
-      // Deep copy because tl_cluster_replicas might be overwritten inbetween
-      auto cluster_replicas = tl_cluster_replicas;
       // Special case, when multiple masters replicate to a single replica.
       for (const auto& replica : cluster_replicas) {
         replication_info_cb(replica->GetSummary());
@@ -3721,12 +3714,11 @@ void ServerFamily::ReplConf(CmdArgList args, const CommandContext& cmd_cntx) {
 
 void ServerFamily::Role(CmdArgList args, const CommandContext& cmd_cntx) {
   auto* rb = static_cast<RedisReplyBuilder*>(cmd_cntx.rb);
-  util::fb2::LockGuard lk(replicaof_mu_);
   // Thread local var is_master is updated under mutex replicaof_mu_ together with replica_,
   // ensuring eventual consistency of is_master. When determining if the server is a replica and
   // accessing the replica_ object, we must lock replicaof_mu_. Using is_master alone is
   // insufficient in this scenario.
-  if (!replica_) {
+  if (!tl_replica) {
     rb->StartArray(2);
     rb->SendBulkString("master");
     auto vec = dfly_cmd_->GetReplicasRoleInfo();
@@ -3739,6 +3731,10 @@ void ServerFamily::Role(CmdArgList args, const CommandContext& cmd_cntx) {
     }
 
   } else {
+    // Deep copy because tl_replica might be overwritten inbetween
+    auto replica = tl_replica;
+    auto cluster_replicas = tl_cluster_replicas;
+
     rb->StartArray(4 + cluster_replicas_.size() * 3);
     rb->SendBulkString(GetFlag(FLAGS_info_replication_valkey_compatible) ? "slave" : "replica");
 
@@ -3756,9 +3752,9 @@ void ServerFamily::Role(CmdArgList args, const CommandContext& cmd_cntx) {
         rb->SendBulkString("connecting");
       }
     };
-    send_replica_info(replica_->GetSummary());
-    for (const auto& replica : cluster_replicas_) {
-      send_replica_info(replica->GetSummary());
+    send_replica_info(replica->GetSummary());
+    for (const auto& repl : cluster_replicas) {
+      send_replica_info(repl->GetSummary());
     }
   }
 }