Fix the another pre-emption bug with inline scheduling

Author: royjacobson

src/server/journal/journal.cc

@@ -86,6 +86,10 @@ void Journal::UnregisterOnChange(uint32_t id) {
   journal_slice.UnregisterOnChange(id);
 }
 
+bool Journal::HasRegisteredCallbacks() const {
+  return journal_slice.HasRegisteredCallbacks();
+}
+
 LSN Journal::GetLsn() const {
   return journal_slice.cur_lsn();
 }

src/server/journal/journal.h

@@ -33,6 +33,7 @@ class Journal {
 
   uint32_t RegisterOnChange(ChangeCallback cb);
   void UnregisterOnChange(uint32_t id);
+  bool HasRegisteredCallbacks() const;
 
   /*
   void AddCmd(TxId txid, Op opcode, Span args) {

src/server/journal/journal_slice.h

@@ -44,6 +44,9 @@ class JournalSlice {
 
   uint32_t RegisterOnChange(ChangeCallback cb);
   void UnregisterOnChange(uint32_t);
+  bool HasRegisteredCallbacks() const {
+    return !change_cb_arr_.empty();
+  }
 
  private:
   struct RingItem;

src/server/server_state.cc

@@ -13,6 +13,7 @@ extern "C" {
 #include "base/flags.h"
 #include "base/logging.h"
 #include "facade/conn_context.h"
+#include "server/journal/journal.h"
 
 ABSL_FLAG(uint32_t, interpreter_per_thread, 10, "Lua interpreters per thread");
 
@@ -70,6 +71,24 @@ void ServerState::Destroy() {
   state_ = nullptr;
 }
 
+bool ServerState::AllowInlineScheduling() const {
+  // We can't allow inline scheduling during a full sync, because then journaling transactions
+  // will be scheduled before RdbLoader::LoadItemsBuffer is finished. We can't use the regular
+  // locking mechanism because RdbLoader is not using transactions.
+
+  // Journal callbacks can preempt; This means we have to disallow inline scheduling
+  // because then we might interleave the callbacks loop from an inlined-scheduled command
+  // and a normally-scheduled command.
+  // The problematic loop is in JournalSlice::AddLogRecord, going over all the callbacks.
+  if (gstate_ == GlobalState::LOADING)
+    return false;
+
+  if (journal_ && journal_->HasRegisteredCallbacks())
+    return false;
+
+  return true;
+}
+
 Interpreter* ServerState::BorrowInterpreter() {
   return interpreter_mgr_.Get();
 }

src/server/server_state.h

@@ -130,6 +130,8 @@ class ServerState {  // public struct - to allow initialization.
     gstate_ = s;
   }
 
+  bool AllowInlineScheduling() const;
+
   // Borrow interpreter from internal manager. Return int with ReturnInterpreter.
   Interpreter* BorrowInterpreter();
 

src/server/transaction.cc

@@ -689,11 +689,7 @@ OpStatus Transaction::ScheduleSingleHop(RunnableType cb) {
       }
     };
 
-    // We can't allow inline scheduling during a full sync, because then journaling transactions
-    // will be scheduled before RdbLoader::LoadItemsBuffer is finished. We can't use the regular
-    // locking mechanism because RdbLoader is not using transactions.
-    if (coordinator_index_ == unique_shard_id_ &&
-        ServerState::tlocal()->gstate() != GlobalState::LOADING) {
+    if (coordinator_index_ == unique_shard_id_ && ServerState::tlocal()->AllowInlineScheduling()) {
       DVLOG(2) << "Inline scheduling a transaction";
       schedule_cb();
     } else {