fix: remove coordinator_index_ from tx & fix short circuit (#1640)

fix: remove coordinator_index_ from tx

Signed-off-by: Vladislav Oleshko <vlad@dragonflydb.io>

src/server/blocking_controller_test.cc

@@ -45,7 +45,7 @@ void BlockingControllerTest::SetUp() {
   shard_set = new EngineShardSet(pp_.get());
   shard_set->Init(kNumThreads, false);
 
-  trans_.reset(new Transaction{&cid_, 0});
+  trans_.reset(new Transaction{&cid_});
 
   str_vec_.assign({"blpop", "x", "z", "0"});
   for (auto& s : str_vec_) {

src/server/debugcmd.cc

@@ -224,8 +224,7 @@ void DebugCmd::Load(string_view filename) {
   };
 
   const CommandId* cid = sf_.service().FindCmd("FLUSHALL");
-  intrusive_ptr<Transaction> flush_trans(
+  intrusive_ptr<Transaction> flush_trans(new Transaction{cid});
-      new Transaction{cid, ServerState::tlocal()->thread_index()});
   flush_trans->InitByArgs(0, {});
   VLOG(1) << "Performing flush";
   error_code ec = sf_.Drakarys(flush_trans.get(), DbSlice::kDbAll);

src/server/main_service.cc

@@ -943,7 +943,7 @@ void Service::DispatchCommand(CmdArgList args, facade::ConnectionContext* cntx)
     DCHECK(dfly_cntx->transaction == nullptr);
 
     if (cid->IsTransactional()) {
-      dist_trans.reset(new Transaction{cid, etl.thread_index()});
+      dist_trans.reset(new Transaction{cid});
 
       if (!dist_trans->IsMulti()) {  // Multi command initialize themself based on their mode.
         if (auto st = dist_trans->InitByArgs(dfly_cntx->conn_state.db_index, args_no_cmd);

src/server/multi_command_squasher.cc

@@ -41,7 +41,7 @@ MultiCommandSquasher::ShardExecInfo& MultiCommandSquasher::PrepareShardInfo(Shar
     if (IsAtomic()) {
       sinfo.local_tx = new Transaction{cntx_->transaction};
     } else {
-      sinfo.local_tx = new Transaction{cntx_->transaction->GetCId(), sid};
+      sinfo.local_tx = new Transaction{cntx_->transaction->GetCId()};
       sinfo.local_tx->StartMultiNonAtomic();
     }
   }

src/server/server_family.cc

@@ -1110,8 +1110,7 @@ GenericError DoPartialSave(fs::path full_filename, const dfly::StringVec& script
 GenericError ServerFamily::DoSave() {
   const CommandId* cid = service().FindCmd("SAVE");
   CHECK_NOTNULL(cid);
-  boost::intrusive_ptr<Transaction> trans(
+  boost::intrusive_ptr<Transaction> trans(new Transaction{cid});
-      new Transaction{cid, ServerState::tlocal()->thread_index()});
   trans->InitByArgs(0, {});
   return DoSave(absl::GetFlag(FLAGS_df_snapshot_format), {}, trans.get());
 }

src/server/transaction.cc

@@ -41,8 +41,7 @@ IntentLock::Mode Transaction::Mode() const {
  * @param ess
  * @param cs
  */
-Transaction::Transaction(const CommandId* cid, uint32_t thread_index)
+Transaction::Transaction(const CommandId* cid) : cid_{cid} {
-    : cid_{cid}, coordinator_index_(thread_index) {
   string_view cmd_name(cid_->name());
   if (cmd_name == "EXEC" || cmd_name == "EVAL" || cmd_name == "EVALSHA") {
     multi_.reset(new MultiData);
@@ -695,7 +694,8 @@ OpStatus Transaction::ScheduleSingleHop(RunnableType cb) {
       }
     };
 
-    if (coordinator_index_ == unique_shard_id_ && ServerState::tlocal()->AllowInlineScheduling()) {
+    if (auto* ss = ServerState::tlocal();
+        ss->thread_index() == unique_shard_id_ && ss->AllowInlineScheduling()) {
       DVLOG(2) << "Inline scheduling a transaction";
       schedule_cb();
     } else {
@@ -837,9 +837,9 @@ void Transaction::ExecuteAsync() {
   // IsArmedInShard in other threads.
   run_count_.store(unique_shard_cnt_, memory_order_release);
 
-  // Execute inline when we can. We can't use coordinator_index_ because we may offload this
+  auto* ss = ServerState::tlocal();
-  // function to run in a different thread.
+  if (unique_shard_cnt_ == 1 && ss->thread_index() == unique_shard_id_ &&
-  if (unique_shard_cnt_ == 1 && ServerState::tlocal()->thread_index() == unique_shard_id_) {
+      ss->AllowInlineScheduling()) {
     DVLOG(1) << "Short-circuit ExecuteAsync " << DebugId();
     EngineShard::tlocal()->PollExecution("exec_cb", this);
     intrusive_ptr_release(this);  // against use_count_.fetch_add above.

src/server/transaction.h

@@ -138,7 +138,7 @@ class Transaction {
   };
 
  public:
-  explicit Transaction(const CommandId* cid, uint32_t thread_index);
+  explicit Transaction(const CommandId* cid);
 
   explicit Transaction(const Transaction* parent);
 
@@ -560,7 +560,6 @@ class Transaction {
   // they read this variable the coordinator thread is stalled and can not cause data races.
   // If COORDINATOR_XXX has been set, it means we passed or crossed stage XXX.
   uint8_t coordinator_state_ = 0;
-  uint32_t coordinator_index_;  // thread_index of the coordinator thread.
 
   // Used for single-hop transactions with unique_shards_ == 1, hence no data-race.
   OpStatus local_result_ = OpStatus::OK;