chore: support mvcc in coroutine mode, part 1

Author: zz-jason

.clang-format

@@ -21,6 +21,8 @@ Cpp11BracedListStyle: true
 PackConstructorInitializers: CurrentLine
 
 # arguments
+BinPackArguments: true
+AllowAllArgumentsOnNextLine: true
 AlignAfterOpenBracket: Align
 QualifierAlignment: Custom
 QualifierOrder: ['inline', 'static', 'const', 'constexpr', 'volatile', 'type']

CMakePresets.json

@@ -74,4 +74,4 @@
             }
         }
     ]
-}
+}

benchmarks/micro-benchmarks/insert_update_bench.cpp

@@ -39,7 +39,7 @@ static void BenchUpdateInsert(benchmark::State& state) {
   std::unordered_set<std::string> dedup;
   for (auto _ : state) {
     leanstore->ExecSync(0, [&]() {
-      cr::WorkerContext::My().StartTx();
+      cr::TxManager::My().StartTx();
       std::string key;
       std::string val;
       for (size_t i = 0; i < 16; i++) {
@@ -48,13 +48,13 @@ static void BenchUpdateInsert(benchmark::State& state) {
         btree->Insert(Slice((const uint8_t*)key.data(), key.size()),
                       Slice((const uint8_t*)val.data(), val.size()));
       }
-      cr::WorkerContext::My().CommitTx();
+      cr::TxManager::My().CommitTx();
     });
   }
 
   leanstore->ExecSync(0, [&]() {
-    cr::WorkerContext::My().StartTx();
-    SCOPED_DEFER(cr::WorkerContext::My().CommitTx());
+    cr::TxManager::My().StartTx();
+    SCOPED_DEFER(cr::TxManager::My().CommitTx());
     leanstore->DropTransactionKV(btree_name);
   });
 }

benchmarks/ycsb/ycsb_leanstore.hpp

@@ -6,7 +6,7 @@
 #include "leanstore/btree/basic_kv.hpp"
 #include "leanstore/btree/transaction_kv.hpp"
 #include "leanstore/concurrency/cr_manager.hpp"
-#include "leanstore/concurrency/worker_context.hpp"
+#include "leanstore/concurrency/tx_manager.hpp"
 #include "leanstore/kv_interface.hpp"
 #include "leanstore/lean_store.hpp"
 #include "leanstore/utils/jump_mu.hpp"
@@ -162,7 +162,7 @@ class YcsbLeanStore : public YcsbExecutor {
           utils::RandomGenerator::RandString(val, FLAGS_ycsb_val_size);
 
           if (bench_transaction_kv_) {
-            cr::WorkerContext::My().StartTx();
+            cr::TxManager::My().StartTx();
           }
 
           auto op_code =
@@ -172,7 +172,7 @@ class YcsbLeanStore : public YcsbExecutor {
           }
 
           if (bench_transaction_kv_) {
-            cr::WorkerContext::My().CommitTx();
+            cr::TxManager::My().CommitTx();
           }
         }
       };
@@ -252,10 +252,10 @@ class YcsbLeanStore : public YcsbExecutor {
                 // generate key for read
                 GenYcsbKey(zipf_random, key);
                 if (bench_transaction_kv_) {
-                  cr::WorkerContext::My().StartTx(TxMode::kShortRunning,
-                                                  IsolationLevel::kSnapshotIsolation, true);
+                  cr::TxManager::My().StartTx(TxMode::kShortRunning,
+                                              IsolationLevel::kSnapshotIsolation, true);
                   table->Lookup(Slice(key, FLAGS_ycsb_key_size), copy_value);
-                  cr::WorkerContext::My().CommitTx();
+                  cr::TxManager::My().CommitTx();
                 } else {
                   table->Lookup(Slice(key, FLAGS_ycsb_key_size), copy_value);
                 }
@@ -264,10 +264,10 @@ class YcsbLeanStore : public YcsbExecutor {
                 GenYcsbKey(zipf_random, key);
                 // generate val for update
                 if (bench_transaction_kv_) {
-                  cr::WorkerContext::My().StartTx();
+                  cr::TxManager::My().StartTx();
                   table->UpdatePartial(Slice(key, FLAGS_ycsb_key_size), update_call_back,
                                        *update_desc);
-                  cr::WorkerContext::My().CommitTx();
+                  cr::TxManager::My().CommitTx();
                 } else {
                   table->UpdatePartial(Slice(key, FLAGS_ycsb_key_size), update_call_back,
                                        *update_desc);

benchmarks/ycsb/ycsb_leanstore_client.hpp

@@ -4,7 +4,7 @@
 #include "leanstore/btree/basic_kv.hpp"
 #include "leanstore/btree/transaction_kv.hpp"
 #include "leanstore/concurrency/cr_manager.hpp"
-#include "leanstore/concurrency/worker_context.hpp"
+#include "leanstore/concurrency/tx_manager.hpp"
 #include "leanstore/kv_interface.hpp"
 #include "leanstore/lean_store.hpp"
 #include "leanstore/utils/log.hpp"

docker/Dockerfile

@@ -38,7 +38,7 @@ ENV VCPKG_DEFAULT_BINARY_CACHE=/opt/vcpkg_cache
 # install vcpkg packages
 RUN ./opt/vcpkg/bootstrap-vcpkg.sh \
  && vcpkg install benchmark gflags spdlog gtest rapidjson gperftools \
-          cpp-httplib crc32c libunwind cpptrace concurrentqueue boost-context
+          cpp-httplib crc32c libunwind cpptrace concurrentqueue boost-context rocksdb
 
 
 # install gcov

examples/cpp/basic_kv_example.cpp

@@ -1,7 +1,7 @@
 #include <leanstore-c/store_option.h>
 #include <leanstore/LeanStore.hpp>
 #include <leanstore/btree/BasicKV.hpp>
-#include <leanstore/concurrency/WorkerContext.hpp>
+#include <leanstore/concurrency/TxManager.hpp>
 
 #include <iostream>
 #include <memory>

include/leanstore/btree/chained_tuple.hpp

@@ -3,10 +3,11 @@
 #include "leanstore/btree/basic_kv.hpp"
 #include "leanstore/btree/core/btree_iter_mut.hpp"
 #include "leanstore/concurrency/cr_manager.hpp"
-#include "leanstore/concurrency/worker_context.hpp"
+#include "leanstore/concurrency/tx_manager.hpp"
 #include "leanstore/units.hpp"
 #include "leanstore/utils/portable.hpp"
 #include "tuple.hpp"
+#include "utils/coroutine/mvcc_manager.hpp"
 
 namespace leanstore::storage::btree {
 
@@ -62,10 +63,10 @@ class PACKED ChainedTuple : public Tuple {
   std::tuple<OpCode, uint16_t> GetVisibleTuple(Slice payload, ValCallback callback) const;
 
   void UpdateStats() {
-    if (cr::WorkerContext::My().cc_.VisibleForAll(tx_id_) ||
+    if (cr::TxManager::My().cc_.VisibleForAll(tx_id_) ||
         oldest_tx_ !=
             static_cast<uint16_t>(
-                cr::WorkerContext::My().store_->crmanager_->global_wmk_info_.oldest_active_tx_ &
+                cr::TxManager::My().store_->MvccManager()->GlobalWmkInfo().oldest_active_tx_ &
                 0xFFFF)) {
       oldest_tx_ = 0;
       total_updates_ = 0;
@@ -77,11 +78,11 @@ class PACKED ChainedTuple : public Tuple {
   bool ShouldConvertToFatTuple() {
     bool command_valid = command_id_ != kInvalidCommandid;
     bool has_long_running_olap =
-        cr::WorkerContext::My().store_->crmanager_->global_wmk_info_.HasActiveLongRunningTx();
+        cr::TxManager::My().store_->MvccManager()->GlobalWmkInfo().HasActiveLongRunningTx();
     bool frequently_updated =
-        total_updates_ > cr::WorkerContext::My().store_->store_option_->worker_threads_;
+        total_updates_ > cr::TxManager::My().store_->store_option_->worker_threads_;
     bool recent_updated_by_others =
-        worker_id_ != cr::WorkerContext::My().worker_id_ || tx_id_ != cr::ActiveTx().start_ts_;
+        worker_id_ != cr::TxManager::My().worker_id_ || tx_id_ != cr::ActiveTx().start_ts_;
     return command_valid && has_long_running_olap && recent_updated_by_others && frequently_updated;
   }
 

include/leanstore/btree/core/b_tree_generic.hpp

@@ -316,7 +316,7 @@ inline void BTreeGeneric::FindLeafCanJump(Slice key, GuardedBufferFrame<BTreeNod
   volatile uint16_t level = 0;
   while (!guarded_target->is_leaf_) {
     auto& child_swip = guarded_target->LookupInner(key);
-    LS_DCHECK(!child_swip.IsEmpty());
+    LEAN_DCHECK(!child_swip.IsEmpty());
     guarded_parent = std::move(guarded_target);
     if (level == height_ - 1) {
       guarded_target =
@@ -352,7 +352,7 @@ inline void BTreeGeneric::FindLeafCanJump(Slice key, GuardedBufferFrame<BTreeNod
 //   auto level = 0u;
 //   while (!child_node->is_leaf_) {
 //     auto& child_swip = child_node->LookupInner(key);
-//     LS_DCHECK(!child_swip.IsEmpty());
+//     LEAN_DCHECK(!child_swip.IsEmpty());
 //
 //     // TODO: yield and retry from the begining
 //     if (locked_parent.IsConflicted()) {
@@ -441,8 +441,8 @@ inline ParentSwipHandler BTreeGeneric::FindParent(BTreeGeneric& btree, BufferFra
     leanstore::JumpContext::Jump();
   }
 
-  LS_DCHECK(pos_in_parent != std::numeric_limits<uint32_t>::max(), "Invalid posInParent={}",
-            pos_in_parent);
+  LEAN_DCHECK(pos_in_parent != std::numeric_limits<uint32_t>::max(), "Invalid posInParent={}",
+              pos_in_parent);
   ParentSwipHandler parent_handler = {.parent_guard_ = std::move(guarded_child.guard_),
                                       .parent_bf_ = guarded_child.bf_,
                                       .child_swip_ = *child_swip,

include/leanstore/btree/core/b_tree_node.hpp

@@ -237,7 +237,7 @@ class BTreeNode : public BTreeNodeHeader {
   }
 
   Swip* ChildSwip(uint16_t slot_id) {
-    LS_DCHECK(slot_id < num_slots_);
+    LEAN_DCHECK(slot_id < num_slots_);
     return reinterpret_cast<Swip*>(ValData(slot_id));
   }
 
@@ -247,28 +247,28 @@ class BTreeNode : public BTreeNodeHeader {
 
   // Attention: the caller has to hold a copy of the existing payload
   void ShortenPayload(uint16_t slot_id, uint16_t target_size) {
-    LS_DCHECK(target_size <= slot_[slot_id].val_size_);
+    LEAN_DCHECK(target_size <= slot_[slot_id].val_size_);
     const uint16_t free_space = slot_[slot_id].val_size_ - target_size;
     space_used_ -= free_space;
     slot_[slot_id].val_size_ = target_size;
   }
 
   bool CanExtendPayload(uint16_t slot_id, uint16_t target_size) {
-    LS_DCHECK(target_size > ValSize(slot_id),
-              "Target size must be larger than current size, "
-              "targetSize={}, currentSize={}",
-              target_size, ValSize(slot_id));
+    LEAN_DCHECK(target_size > ValSize(slot_id),
+                "Target size must be larger than current size, "
+                "targetSize={}, currentSize={}",
+                target_size, ValSize(slot_id));
 
     const uint16_t extra_space_needed = target_size - ValSize(slot_id);
     return FreeSpaceAfterCompaction() >= extra_space_needed;
   }
 
   /// Move key-value pair to a new location
   void ExtendPayload(uint16_t slot_id, uint16_t target_size) {
-    LS_DCHECK(CanExtendPayload(slot_id, target_size),
-              "ExtendPayload failed, not enough space in the current node, "
-              "slotId={}, targetSize={}, FreeSpace={}, currentSize={}",
-              slot_id, target_size, FreeSpaceAfterCompaction(), ValSize(slot_id));
+    LEAN_DCHECK(CanExtendPayload(slot_id, target_size),
+                "ExtendPayload failed, not enough space in the current node, "
+                "slotId={}, targetSize={}, FreeSpace={}, currentSize={}",
+                slot_id, target_size, FreeSpaceAfterCompaction(), ValSize(slot_id));
     auto key_size_without_prefix = KeySizeWithoutPrefix(slot_id);
     const uint16_t old_total_size = key_size_without_prefix + ValSize(slot_id);
     const uint16_t new_total_size = key_size_without_prefix + target_size;
@@ -285,7 +285,7 @@ class BTreeNode : public BTreeNodeHeader {
     if (FreeSpace() < new_total_size) {
       Compactify();
     }
-    LS_DCHECK(FreeSpace() >= new_total_size);
+    LEAN_DCHECK(FreeSpace() >= new_total_size);
     advance_data_offset(new_total_size);
     slot_[slot_id].offset_ = data_offset_;
     slot_[slot_id].key_size_without_prefix_ = key_size_without_prefix;
@@ -498,7 +498,7 @@ inline int16_t BTreeNode::LinearSearchWithBias(Slice key, uint16_t start_pos, bo
     return -1;
   }
 
-  LS_DCHECK(key.size() >= prefix_size_ && bcmp(key.data(), LowerFenceAddr(), prefix_size_) == 0);
+  LEAN_DCHECK(key.size() >= prefix_size_ && bcmp(key.data(), LowerFenceAddr(), prefix_size_) == 0);
 
   // the compared key has the same prefix
   key.remove_prefix(prefix_size_);
@@ -575,8 +575,8 @@ inline int16_t BTreeNode::LowerBound(Slice key, bool* is_equal) {
 inline void BTreeNode::set_fences(Slice lower_key, Slice upper_key) {
   InsertFence(lower_fence_, lower_key);
   InsertFence(upper_fence_, upper_key);
-  LS_DCHECK(LowerFenceAddr() == nullptr || UpperFenceAddr() == nullptr ||
-            *LowerFenceAddr() <= *UpperFenceAddr());
+  LEAN_DCHECK(LowerFenceAddr() == nullptr || UpperFenceAddr() == nullptr ||
+              *LowerFenceAddr() <= *UpperFenceAddr());
 
   // prefix compression
   for (prefix_size_ = 0; (prefix_size_ < std::min(lower_key.size(), upper_key.size())) &&