feat(server): Use new journal format (dragonflydb#563)

Author: dranikpg

src/facade/conn_context.h

@@ -50,7 +50,8 @@ class ConnectionContext {
   bool req_auth : 1;
   bool replica_conn : 1;
   bool authenticated : 1;
-  bool force_dispatch : 1;  // whether we should route all requests to the dispatch fiber.
+  bool force_dispatch : 1;    // whether we should route all requests to the dispatch fiber.
+  bool journal_emulated : 1;  // whether it is used to dispatch journal commands.
 
  private:
   Connection* owner_;

src/facade/facade.cc

@@ -118,6 +118,7 @@ ConnectionContext::ConnectionContext(::io::Sink* stream, Connection* owner) : ow
   replica_conn = false;
   authenticated = false;
   force_dispatch = false;
+  journal_emulated = false;
 }
 
 RedisReplyBuilder* ConnectionContext::operator->() {

src/server/CMakeLists.txt

@@ -19,7 +19,7 @@ add_library(dragonfly_lib  channel_slice.cc command_registry.cc
             snapshot.cc script_mgr.cc server_family.cc malloc_stats.cc
             set_family.cc stream_family.cc string_family.cc
             zset_family.cc version.cc bitops_family.cc container_utils.cc
-            serializer_commons.cc journal/serializer.cc)
+            serializer_commons.cc journal/serializer.cc journal/executor.cc)
 
 cxx_link(dragonfly_lib dfly_transaction dfly_facade redis_lib strings_lib html_lib
          absl::random_random TRDP::jsoncons zstd TRDP::lz4)

src/server/common.h

@@ -268,7 +268,10 @@ class Context : protected Cancellation {
   //
   // Note: this function blocks when called from inside an error handler.
   template <typename... T> GenericError Error(T... ts) {
-    std::lock_guard lk{mu_};
+    if (!mu_.try_lock())  // TODO: Maybe use two separate locks.
+      return GenericError{std::forward<T>(ts)...};
+
+    std::lock_guard lk{mu_, std::adopt_lock};
     if (err_)
       return err_;
 

src/server/dflycmd.cc

@@ -13,6 +13,7 @@
 #include "server/engine_shard_set.h"
 #include "server/error.h"
 #include "server/journal/journal.h"
+#include "server/journal/serializer.h"
 #include "server/rdb_save.h"
 #include "server/script_mgr.h"
 #include "server/server_family.h"
@@ -353,16 +354,16 @@ OpStatus DflyCmd::StartStableSyncInThread(FlowInfo* flow, EngineShard* shard) {
   // Register journal listener and cleanup.
   uint32_t cb_id = 0;
   if (shard != nullptr) {
-    cb_id = sf_->journal()->RegisterOnChange([flow](const journal::Entry& je) {
-      // TODO: Serialize event.
-      ReqSerializer serializer{flow->conn->socket()};
-      serializer.SendCommand(absl::StrCat("SET ", je.key, " ", je.pval_ptr->ToString()));
-    });
+    JournalWriter writer{flow->conn->socket()};
+    auto journal_cb = [flow, writer = std::move(writer)](const journal::Entry& je) mutable {
+      writer.Write(je);
+    };
+    cb_id = sf_->journal()->RegisterOnChange(std::move(journal_cb));
   }
 
   flow->cleanup = [flow, this, cb_id]() {
     if (cb_id)
-      sf_->journal()->Unregister(cb_id);
+      sf_->journal()->UnregisterOnChange(cb_id);
     flow->TryShutdownSocket();
   };
 

src/server/journal/executor.cc

@@ -0,0 +1,29 @@
+// Copyright 2022, DragonflyDB authors.  All rights reserved.
+// See LICENSE for licensing terms.
+//
+
+#include "server/journal/executor.h"
+
+#include "base/logging.h"
+#include "server/main_service.h"
+
+namespace dfly {
+
+JournalExecutor::JournalExecutor(Service* service) : service_{service} {
+}
+
+void JournalExecutor::Execute(journal::ParsedEntry&& entry) {
+  if (entry.payload) {
+    io::NullSink null_sink;
+    ConnectionContext conn_context{&null_sink, nullptr};
+    conn_context.is_replicating = true;
+    conn_context.journal_emulated = true;
+    conn_context.conn_state.db_index = entry.dbid;
+
+    auto span = CmdArgList{entry.payload->data(), entry.payload->size()};
+
+    service_->DispatchCommand(span, &conn_context);
+  }
+}
+
+}  // namespace dfly

src/server/journal/executor.h

@@ -0,0 +1,23 @@
+// Copyright 2022, DragonflyDB authors.  All rights reserved.
+// See LICENSE for licensing terms.
+//
+
+#pragma once
+
+#include "server/journal/types.h"
+
+namespace dfly {
+
+class Service;
+
+// JournalExecutor allows executing journal entries.
+class JournalExecutor {
+ public:
+  JournalExecutor(Service* service);
+  void Execute(journal::ParsedEntry&& entry);
+
+ private:
+  Service* service_;
+};
+
+}  // namespace dfly

src/server/journal/journal.cc

@@ -41,11 +41,11 @@ error_code Journal::OpenInThread(bool persistent, string_view dir) {
     }
   }
 
-      ServerState::tlocal()->set_journal(this);
-      EngineShard* shard = EngineShard::tlocal();
-      if (shard) {
-        shard->set_journal(this);
-      }
+  ServerState::tlocal()->set_journal(this);
+  EngineShard* shard = EngineShard::tlocal();
+  if (shard) {
+    shard->set_journal(this);
+  }
 
   return ec;
 }
@@ -83,16 +83,16 @@ uint32_t Journal::RegisterOnChange(ChangeCallback cb) {
   return journal_slice.RegisterOnChange(cb);
 }
 
-void Journal::Unregister(uint32_t id) {
-  journal_slice.Unregister(id);
+void Journal::UnregisterOnChange(uint32_t id) {
+  journal_slice.UnregisterOnChange(id);
 }
 
 bool Journal::SchedStartTx(TxId txid, unsigned num_keys, unsigned num_shards) {
   if (!journal_slice.IsOpen() || lameduck_.load(memory_order_relaxed))
     return false;
 
-  // TODO: to complete the metadata.
-  journal_slice.AddLogRecord(Entry::Sched(txid));
+  // TODO: Handle tx entries.
+  // journal_slice.AddLogRecord(Entry::Sched(txid));
 
   return true;
 }

src/server/journal/journal.h

@@ -13,7 +13,6 @@ class Transaction;
 
 namespace journal {
 
-
 class Journal {
  public:
   using Span = absl::Span<const std::string_view>;
@@ -32,9 +31,8 @@ class Journal {
 
   //******* The following functions must be called in the context of the owning shard *********//
 
-
   uint32_t RegisterOnChange(ChangeCallback cb);
-  void Unregister(uint32_t id);
+  void UnregisterOnChange(uint32_t id);
 
   // Returns true if transaction was scheduled, false if journal is inactive
   // or in lameduck mode and does not log new transactions.
@@ -58,7 +56,6 @@ class Journal {
   void RecordEntry(const Entry& entry);
 
  private:
-
   mutable boost::fibers::mutex state_mu_;
 
   std::atomic_bool lameduck_{false};

src/server/journal/journal_slice.cc

@@ -147,7 +147,7 @@ uint32_t JournalSlice::RegisterOnChange(ChangeCallback cb) {
   return id;
 }
 
-void JournalSlice::Unregister(uint32_t id) {
+void JournalSlice::UnregisterOnChange(uint32_t id) {
   CHECK(!iterating_cb_arr_);
 
   auto it = find_if(change_cb_arr_.begin(), change_cb_arr_.end(),