perf: support scalable logging

benchmarks/ycsb/YcsbWiredTiger.hpp

@@ -198,7 +198,7 @@ class YcsbWiredTiger : public YcsbExecutor {
 
     std::string configString(
         "create, direct_io=[data, log, checkpoint], "
-        "log=(enabled=true,archive=true), statistics_log=(wait=1), "
+        "log=(enabled),transaction_sync=(enabled=true,method=none), statistics_log=(wait=1), "
         "statistics=(all, clear), session_max=2000, eviction=(threads_max=4), cache_size=" +
         std::to_string(FLAGS_ycsb_mem_kb / 1024) + "M");
     int ret = wiredtiger_open(dataDir.c_str(), nullptr, configString.c_str(), &mConn);

include/leanstore/LeanStore.hpp

@@ -8,6 +8,7 @@
 #include <atomic>
 #include <cstdint>
 #include <expected>
+#include <format>
 #include <functional>
 #include <memory>
 #include <string>
@@ -52,9 +53,6 @@ class LeanStore {
   //! The file descriptor for pages
   int32_t mPageFd;
 
-  //! The file descriptor for write-ahead log
-  int32_t mWalFd;
-
   //! The tree registry
   std::unique_ptr<storage::TreeRegistry> mTreeRegistry;
 
@@ -144,8 +142,8 @@ class LeanStore {
     return std::string(mStoreOption->mStoreDir) + "/db.pages";
   }
 
-  std::string GetWalFilePath() const {
-    return std::string(mStoreOption->mStoreDir) + "/db.wal";
+  std::string GetDbWalPath(uint64_t workerId) const {
+    return std::format("{}/worker_{}.wal", getDbWalDir(), workerId);
   }
 
 private:
@@ -163,6 +161,10 @@ class LeanStore {
   void deserializeFlags();
 
   void initPageAndWalFd();
+
+  std::string getDbWalDir() const {
+    return std::format("{}/wal", mStoreOption->mStoreDir);
+  }
 };
 
 } // namespace leanstore

include/leanstore/btree/core/PessimisticExclusiveIterator.hpp

@@ -2,7 +2,9 @@
 
 #include "leanstore/KVInterface.hpp"
 #include "leanstore/btree/core/PessimisticIterator.hpp"
+#include "leanstore/concurrency/WorkerContext.hpp"
 #include "leanstore/utils/CounterUtil.hpp"
+#include "leanstore/utils/Defer.hpp"
 #include "leanstore/utils/Log.hpp"
 #include "leanstore/utils/RandomGenerator.hpp"
 #include "leanstore/utils/UserThread.hpp"
@@ -63,7 +65,9 @@ class PessimisticExclusiveIterator : public PessimisticIterator {
   }
 
   void SplitForKey(Slice key) {
-    auto sysTxId = mBTree.mStore->AllocSysTxTs();
+    TXID sysTxId = cr::WorkerContext::My().StartSysTx();
+    SCOPED_DEFER(cr::WorkerContext::My().CommitSysTx());
+
     while (true) {
       JUMPMU_TRY() {
         if (!Valid() || !KeyInCurrentNode(key)) {
@@ -163,18 +167,18 @@ class PessimisticExclusiveIterator : public PessimisticIterator {
 
       mSlotId = -1;
       JUMPMU_TRY() {
-        TXID sysTxId = mBTree.mStore->AllocSysTxTs();
+        TXID sysTxId = cr::WorkerContext::My().StartSysTx();
         mBTree.TrySplitMayJump(sysTxId, *mGuardedLeaf.mBf, splitSlot);
+        cr::WorkerContext::My().CommitSysTx();
 
         COUNTER_INC(&leanstore::cr::tlsPerfCounters.mContentionSplitSucceed);
-        LS_DLOG("[Contention Split] succeed, pageId={}, contention pct={}, split "
-                "slot={}",
+        LS_DLOG("[Contention Split] succeed, pageId={}, contentionPct={}, splitSlot={}",
                 mGuardedLeaf.mBf->mHeader.mPageId, contentionPct, splitSlot);
       }
       JUMPMU_CATCH() {
         COUNTER_INC(&leanstore::cr::tlsPerfCounters.mContentionSplitFailed);
-        Log::Info("[Contention Split] contention split failed, pageId={}, contention "
-                  "pct={}, split slot={}",
+        Log::Info("[Contention Split] contention split failed, pageId={}, "
+                  "contentionPct={}, splitSlot={}",
                   mGuardedLeaf.mBf->mHeader.mPageId, contentionPct, splitSlot);
       }
     }
@@ -196,8 +200,9 @@ class PessimisticExclusiveIterator : public PessimisticIterator {
       mGuardedLeaf.unlock();
       mSlotId = -1;
       JUMPMU_TRY() {
-        TXID sysTxId = mBTree.mStore->AllocSysTxTs();
+        TXID sysTxId = cr::WorkerContext::My().StartSysTx();
         mBTree.TryMergeMayJump(sysTxId, *mGuardedLeaf.mBf);
+        cr::WorkerContext::My().CommitSysTx();
       }
       JUMPMU_CATCH() {
         LS_DLOG("TryMergeIfNeeded failed, pageId={}", mGuardedLeaf.mBf->mHeader.mPageId);

include/leanstore/btree/core/PessimisticIterator.hpp

@@ -6,6 +6,7 @@
 #include "leanstore/Slice.hpp"
 #include "leanstore/btree/core/BTreeNode.hpp"
 #include "leanstore/buffer-manager/GuardedBufferFrame.hpp"
+#include "leanstore/concurrency/WorkerContext.hpp"
 #include "leanstore/sync/HybridLatch.hpp"
 #include "leanstore/utils/Log.hpp"
 #include "leanstore/utils/UserThread.hpp"
@@ -397,8 +398,9 @@ inline void PessimisticIterator::Next() {
     if (mGuardedLeaf->mNumSlots == 0) {
       SetCleanUpCallback([&, toMerge = mGuardedLeaf.mBf]() {
         JUMPMU_TRY() {
-          TXID sysTxId = mBTree.mStore->AllocSysTxTs();
+          TXID sysTxId = cr::WorkerContext::My().StartSysTx();
           mBTree.TryMergeMayJump(sysTxId, *toMerge, true);
+          cr::WorkerContext::My().CommitSysTx();
         }
         JUMPMU_CATCH() {
         }

include/leanstore/buffer-manager/GuardedBufferFrame.hpp

@@ -158,9 +158,6 @@ class GuardedBufferFrame {
 
     // update system transaction id
     mBf->mPage.mSysTxId = sysTxId;
-
-    // update the maximum system transaction id written by the worker
-    cr::WorkerContext::My().mLogging.UpdateSysTxWrittern(sysTxId);
   }
 
   //! Check remote dependency
@@ -172,8 +169,8 @@ class GuardedBufferFrame {
     }
 
     if (mBf->mHeader.mLastWriterWorker != cr::WorkerContext::My().mWorkerId &&
-        mBf->mPage.mSysTxId > cr::ActiveTx().mMaxObservedSysTxId) {
-      cr::ActiveTx().mMaxObservedSysTxId = mBf->mPage.mSysTxId;
+        mBf->mPage.mSysTxId > cr::ActiveTx().mDependentSysTx) {
+      cr::ActiveTx().mDependentSysTx = mBf->mPage.mSysTxId;
       cr::ActiveTx().mHasRemoteDependency = true;
     }
   }

include/leanstore/concurrency/CRManager.hpp

@@ -14,7 +14,6 @@ class LeanStore;
 namespace cr {
 
 struct WaterMarkInfo;
-class GroupCommitter;
 
 //! Manages a fixed number of worker threads and group committer threads.
 class CRManager {
@@ -30,19 +29,13 @@ class CRManager {
 
   WaterMarkInfo mGlobalWmkInfo;
 
-  //! The group committer thread, created and started if WAL is enabled when the
-  //! CRManager instance is created.
-  ///
-  //! NOTE: It should be created after all the worker threads are created and
-  //! started.
-  std::unique_ptr<GroupCommitter> mGroupCommitter;
-
 public:
+  //! Construct a CRManager.
   CRManager(leanstore::LeanStore* store);
 
+  //! Destruct a CRManager.
   ~CRManager();
 
-public:
   // State Serialization
   StringMap Serialize();