chore: clear logging and tx manager API

Author: zz-jason

include/leanstore/buffer-manager/guarded_buffer_frame.hpp

@@ -2,7 +2,6 @@
 
 #include "leanstore/buffer-manager/buffer_frame.hpp"
 #include "leanstore/buffer-manager/buffer_manager.hpp"
-#include "leanstore/concurrency/logging_impl.hpp"
 #include "leanstore/concurrency/tx_manager.hpp"
 #include "leanstore/concurrency/wal_payload_handler.hpp"
 #include "leanstore/sync/hybrid_guard.hpp"
@@ -180,13 +179,12 @@ class GuardedBufferFrame {
 
   template <typename WT, typename... Args>
   cr::WalPayloadHandler<WT> ReserveWALPayload(uint64_t wal_size, Args&&... args) {
-    LEAN_DCHECK(CoroEnv::CurTxMgr().ActiveTx().is_durable_);
     LEAN_DCHECK(guard_.state_ == GuardState::kExclusivePessimistic);
 
     const auto page_id = bf_->header_.page_id_;
     const auto tree_id = bf_->page_.btree_id_;
     wal_size = ((wal_size - 1) / 8 + 1) * 8;
-    auto handler = CoroEnv::CurLogging().ReserveWALEntryComplex<WT, Args...>(
+    auto handler = CoroEnv::CurTxMgr().ReserveWALEntryComplex<WT, Args...>(
         sizeof(WT) + wal_size, page_id, bf_->page_.psn_, tree_id, std::forward<Args>(args)...);
 
     return handler;

include/leanstore/concurrency/logging.hpp

@@ -2,6 +2,7 @@
 
 #include "leanstore-c/perf_counters.h"
 #include "leanstore/concurrency/transaction.hpp"
+#include "leanstore/concurrency/wal_entry.hpp"
 #include "leanstore/sync/optimistic_guarded.hpp"
 #include "leanstore/units.hpp"
 #include "leanstore/utils/counter_util.hpp"
@@ -54,6 +55,10 @@ class WalPayloadHandler;
 /// Helps to transaction concurrenct control and write-ahead logging.
 class Logging {
 public:
+  /// Logical sequence number, i.e., the unique ID of each WAL.
+  LID lsn_clock_ = 0;
+
+  /// The previous LSN of the current transaction, used to link WAL entries.
   LID prev_lsn_;
 
   /// The active complex WalEntry for the current transaction, usually used for insert, update,
@@ -85,18 +90,17 @@ class Logging {
 
   storage::OptimisticGuarded<WalFlushReq> wal_flush_req_;
 
-  /// The ring buffer of the current worker thread. All the wal entries of the current worker are
-  /// writtern to this ring buffer firstly, then flushed to disk by the group commit thread.
-  ALIGNAS(512) uint8_t* wal_buffer_;
+  /// The maximum writtern system transaction ID in the worker.
+  TXID sys_tx_writtern_ = 0;
 
-  /// The size of the wal ring buffer.
-  uint64_t wal_buffer_bytes_;
+  /// File descriptor for the write-ahead log.
+  int32_t wal_fd_ = -1;
 
-  /// Used to track the write order of wal entries.
-  LID lsn_clock_ = 0;
+  /// Start offset of the next WalEntry.
+  uint64_t wal_size_ = 0;
 
-  /// The maximum writtern system transaction ID in the worker.
-  TXID sys_tx_writtern_ = 0;
+  /// The size of the wal ring buffer.
+  uint64_t wal_buffer_bytes_;
 
   /// Written offset of the wal ring buffer.
   uint64_t wal_buffered_ = 0;
@@ -106,19 +110,14 @@ class Logging {
   /// thread.
   std::atomic<uint64_t> wal_flushed_ = 0;
 
-  /// The first WAL record of the current active transaction.
-  uint64_t tx_wal_begin_;
-
-  /// File descriptor for the write-ahead log.
-  int32_t wal_fd_ = -1;
-
-  /// Start offset of the next WalEntry.
-  uint64_t wal_size_ = 0;
+  /// The ring buffer of the current worker thread. All the wal entries of the current worker are
+  /// writtern to this ring buffer firstly, then flushed to disk by the group commit thread.
+  ALIGNAS(512) uint8_t* wal_buffer_;
 
 public:
   Logging(uint64_t wal_buffer_bytes)
-      : wal_buffer_((uint8_t*)(std::aligned_alloc(512, wal_buffer_bytes))),
-        wal_buffer_bytes_(wal_buffer_bytes) {
+      : wal_buffer_bytes_(wal_buffer_bytes),
+        wal_buffer_((uint8_t*)(std::aligned_alloc(512, wal_buffer_bytes))) {
     std::memset(wal_buffer_, 0, wal_buffer_bytes);
   }
 
@@ -129,6 +128,12 @@ class Logging {
     }
   }
 
+  LID ReserveLsn() {
+    return lsn_clock_++;
+  }
+
+  void ReserveWalBuffer(uint32_t requested_size);
+
   void UpdateSignaledCommitTs(const LID signaled_commit_ts) {
     signaled_commit_ts_.store(signaled_commit_ts, std::memory_order_release);
   }
@@ -139,19 +144,14 @@ class Logging {
     }
   }
 
-  void ReserveContiguousBuffer(uint32_t requested_size);
-
   /// Iterate over current TX entries
-  void IterateCurrentTxWALs(std::function<void(const WalEntry& entry)> callback);
+  void IterateCurrentTxWALs(uint64_t first_wal,
+                            std::function<void(const WalEntry& entry)> callback);
 
   void WriteWalTxAbort();
   void WriteWalTxFinish();
   void WriteWalCarriageReturn();
 
-  template <typename T, typename... Args>
-  WalPayloadHandler<T> ReserveWALEntryComplex(uint64_t payload_size, PID page_id, LID psn,
-                                              TREEID tree_id, Args&&... args);
-
   /// Submits wal record to group committer when it is ready to flush to disk.
   /// @param totalSize size of the wal record to be flush.
   void SubmitWALEntryComplex(uint64_t total_size);

include/leanstore/concurrency/logging_impl.hpp

@@ -5,9 +5,7 @@
 #include "leanstore/units.hpp"
 #include "leanstore/utils/managed_thread.hpp"
 
-namespace leanstore {
-namespace cr {
-
+namespace leanstore ::cr {
 enum class TxState { kIdle, kStarted, kCommitted, kAborted };
 
 struct TxStatUtil {
@@ -59,6 +57,13 @@ class Transaction {
   /// tx_isolation_level_ is the isolation level for the current transaction.
   IsolationLevel tx_isolation_level_ = IsolationLevel::kSnapshotIsolation;
 
+  /// The first WAL record of the current active transaction.
+  uint64_t first_wal_ = 0;
+
+  /// The previous WAL record of the current active transaction, 0 for the first
+  /// WAL record. Used when writing compensation log records for tx abort.
+  uint64_t prev_wal_lsn_ = 0;
+
   /// Whether the transaction has any data writes. Transaction writes can be
   /// detected once it generates a WAL entry.
   bool has_wrote_ = false;
@@ -80,16 +85,13 @@ class Transaction {
 
   // Start a new transaction, initialize all fields
   void Start(TxMode mode, IsolationLevel level) {
+    Transaction tx_inited;
+    *this = tx_inited;
+
     state_ = TxState::kStarted;
-    start_ts_ = 0;
-    commit_ts_ = 0;
-    max_observed_sys_tx_id_ = 0;
-    has_remote_dependency_ = false;
     tx_mode_ = mode;
     tx_isolation_level_ = level;
-    has_wrote_ = false;
     is_durable_ = CoroEnv::CurStore()->store_option_->enable_wal_;
-    wal_exceed_buffer_ = false;
   }
 
   /// Check whether a user transaction with remote dependencies can be committed.
@@ -98,5 +100,4 @@ class Transaction {
   }
 };
 
-} // namespace cr
-} // namespace leanstore
+} // namespace leanstore::cr

include/leanstore/concurrency/transaction.hpp

@@ -98,6 +98,29 @@ class TxManager {
     last_committed_usr_tx_.store(usr_tx_id, std::memory_order_release);
   }
 
+  template <typename T, typename... Args>
+  WalPayloadHandler<T> ReserveWALEntryComplex(uint64_t payload_size, PID page_id, LID psn,
+                                              TREEID tree_id, Args&&... args) {
+    auto& logging = CoroEnv::CurLogging();
+
+    // write transaction start on demand
+    auto prev_lsn = active_tx_.prev_wal_lsn_;
+    active_tx_.has_wrote_ = true;
+    SCOPED_DEFER(active_tx_.prev_wal_lsn_ = logging.active_walentry_complex_->lsn_);
+
+    auto entry_lsn = logging.lsn_clock_++;
+    auto* entry_ptr = logging.wal_buffer_ + logging.wal_buffered_;
+    auto entry_size = sizeof(WalEntryComplex) + payload_size;
+    logging.ReserveWalBuffer(entry_size);
+
+    logging.active_walentry_complex_ = new (entry_ptr) WalEntryComplex(
+        entry_lsn, prev_lsn, entry_size, worker_id_, active_tx_.start_ts_, psn, page_id, tree_id);
+
+    auto* payload_ptr = logging.active_walentry_complex_->payload_;
+    auto wal_payload = new (payload_ptr) T(std::forward<Args>(args)...);
+    return {wal_payload, entry_size};
+  }
+
   static constexpr uint64_t kRcBit = (1ull << 63);
   static constexpr uint64_t kLongRunningBit = (1ull << 62);
   static constexpr uint64_t kCleanBitsMask = ~(kRcBit | kLongRunningBit);

include/leanstore/concurrency/tx_manager.hpp

@@ -132,8 +132,8 @@ class PACKED WalEntryComplex : public WalEntry {
 
   uint32_t ComputeCRC32() const {
     auto type_field_size = sizeof(Type);
-    auto crc32FieldSize = sizeof(uint32_t);
-    auto crc_skip_size = type_field_size + crc32FieldSize;
+    auto crc_field_size = sizeof(uint32_t);
+    auto crc_skip_size = type_field_size + crc_field_size;
     const auto* src = reinterpret_cast<const uint8_t*>(this) + crc_skip_size;
     auto src_size = size_ - crc_skip_size;
     return utils::CRC(src, src_size);

include/leanstore/concurrency/wal_entry.hpp

@@ -1,6 +1,5 @@
 #pragma once
 
-#include "leanstore/concurrency/group_committer.hpp"
 #include "leanstore/concurrency/tx_manager.hpp"
 
 namespace leanstore::cr {

include/leanstore/concurrency/wal_payload_handler.hpp

@@ -5,7 +5,6 @@
 #include "leanstore/exceptions.hpp"
 #include "leanstore/utils/log.hpp"
 #include "utils/coroutine/coro_env.hpp"
-#include "utils/coroutine/coroutine.hpp"
 #include "utils/to_json.hpp"
 
 #include <cstring>
@@ -20,7 +19,7 @@ uint32_t Logging::WalContiguousFreeSpace() {
   return flushed - wal_buffered_;
 }
 
-void Logging::ReserveContiguousBuffer(uint32_t bytes_required) {
+void Logging::ReserveWalBuffer(uint32_t bytes_required) {
   // Spin until there is enough space. The wal ring buffer space is reclaimed
   // when the group commit thread commits the written wal entries.
   while (true) {
@@ -52,7 +51,7 @@ void Logging::ReserveContiguousBuffer(uint32_t bytes_required) {
 void Logging::WriteWalTxAbort() {
   // Reserve space
   auto size = sizeof(WalTxAbort);
-  ReserveContiguousBuffer(size);
+  ReserveWalBuffer(size);
 
   // Initialize a WalTxAbort
   auto* data = wal_buffer_ + wal_buffered_;
@@ -70,7 +69,7 @@ void Logging::WriteWalTxAbort() {
 void Logging::WriteWalTxFinish() {
   // Reserve space
   auto size = sizeof(WalTxFinish);
-  ReserveContiguousBuffer(size);
+  ReserveWalBuffer(size);
 
   // Initialize a WalTxFinish
   auto* data = wal_buffer_ + wal_buffered_;
@@ -116,8 +115,9 @@ void Logging::PublishWalFlushReq() {
 }
 
 // Called by worker, so concurrent writes on the buffer
-void Logging::IterateCurrentTxWALs(std::function<void(const WalEntry& entry)> callback) {
-  uint64_t cursor = tx_wal_begin_;
+void Logging::IterateCurrentTxWALs(uint64_t first_wal,
+                                   std::function<void(const WalEntry& entry)> callback) {
+  uint64_t cursor = first_wal;
   while (cursor != wal_buffered_) {
     const WalEntry& entry = *reinterpret_cast<WalEntry*>(wal_buffer_ + cursor);
     DEBUG_BLOCK() {

src/concurrency/logging.cpp

@@ -53,7 +53,7 @@ void TxManager::StartTx(TxMode mode, IsolationLevel level, bool is_read_only) {
   active_tx_.max_observed_sys_tx_id_ = store_->MvccManager()->GetMinCommittedSysTx();
 
   // Init wal and group commit related transaction information
-  CoroEnv::CurLogging().tx_wal_begin_ = CoroEnv::CurLogging().wal_buffered_;
+  active_tx_.first_wal_ = CoroEnv::CurLogging().wal_buffered_;
 
   // For now, we only support SI and SSI
   if (level < IsolationLevel::kSnapshotIsolation) {
@@ -172,7 +172,7 @@ void TxManager::AbortTx() {
   // TODO(jian.z): support reading from WAL file once
   LEAN_DCHECK(!active_tx_.wal_exceed_buffer_, "Aborting from WAL file is not supported yet");
   std::vector<const WalEntry*> entries;
-  CoroEnv::CurLogging().IterateCurrentTxWALs([&](const WalEntry& entry) {
+  CoroEnv::CurLogging().IterateCurrentTxWALs(active_tx_.first_wal_, [&](const WalEntry& entry) {
     if (entry.type_ == WalEntry::Type::kComplex) {
       entries.push_back(&entry);
     }

src/concurrency/tx_manager.cpp

@@ -25,13 +25,11 @@ namespace leanstore::test {
 class CoroLeanStoreTest : public LeanTestSuite {};
 
 TEST_F(CoroLeanStoreTest, BasicKv) {
-  // GTEST_SKIP() << "Skipping test BasicKv, as logging is not correctly implemented yet.";
-
   static constexpr auto kBtreeName = "test_btree";
   static constexpr auto kNumKeys = 100;
   static constexpr auto kKeyPattern = "key_btree_LL_xxxxxxxxxxxx_{}";
   static constexpr auto kValPattern = "VAL_BTREE_LL_YYYYYYYYYYYY_{}";
-  static constexpr auto kEnableWal = false;
+  static constexpr auto kEnableWal = true;
   static constexpr auto kBtreeConfig = BTreeConfig{
       .enable_wal_ = kEnableWal,
       .use_bulk_insert_ = false,