From 5b7129c7baf5696a10a8cecd3600debf275d0268 Mon Sep 17 00:00:00 2001 From: thawk105 Date: Mon, 5 Oct 2026 19:23:16 +0900 Subject: [PATCH] Fix Silo TPC-C progress and tuple lifetime under contention Reset absent per operation and release the current lock on an absent target to prevent Delivery from leaving permanently locked Order records. Register published INSERT ownership before node validation and retire removed tuples with epochs sampled after index removal. Pin epochs at begin, abort reads of pending INSERTs, and observe quit during lock waits. Preserve staged updates and stable scan results so repeated Stock updates and read-own-writes behave consistently. Add 14 deterministic regression cases and run them in CI. All cases fail against the baseline and pass in normal and ASan builds; the standard mix with eight workers and one warehouse now completes the 30-second check. --- .github/workflows/build.yml | 10 +- CMakeLists.txt | 4 + cc/silo/CMakeLists.txt | 14 + cc/silo/include/transaction.hh | 10 +- cc/silo/test/transaction_test.cc | 389 ++++++++++++++++++++++++++++ cc/silo/transaction.cc | 144 +++++++--- cc/silo/util.cc | 3 +- include/tpcc/tpcc_tx_orderstatus.hh | 2 +- 8 files changed, 528 insertions(+), 48 deletions(-) create mode 100644 cc/silo/test/transaction_test.cc diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index ebab4584..481d8c55 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -66,13 +66,14 @@ jobs: ccache --zero-stats - name: Configure (top-level cmake) - # Sanitizer adds significant link-time memory/cpu cost and CI does - # not run binaries, so it adds no value here. Use Release for the - # build verification. Wire ccache via compiler launchers. + # Keep the full-protocol build and deterministic regressions in + # Release mode. Debug+ASan workload checks run separately. + # Wire ccache via compiler launchers. run: | cmake -S . -B build \ -DCMAKE_BUILD_TYPE=Release \ -DENABLE_SANITIZER=OFF \ + -DCCBENCH_BUILD_SILO_TESTS=ON \ -DCMAKE_C_COMPILER_LAUNCHER=ccache \ -DCMAKE_CXX_COMPILER_LAUNCHER=ccache @@ -82,6 +83,9 @@ jobs: # longer build that here.) run: cmake --build build -j $(nproc) + - name: Test Silo transaction regressions + run: ctest --test-dir build -R '^silo_regression_test$' --output-on-failure + - name: ccache stats if: always() run: ccache --show-stats diff --git a/CMakeLists.txt b/CMakeLists.txt index 6305d3cc..b9e99b9a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -16,6 +16,10 @@ include(Options) option(ENABLE_SANITIZER "enable sanitizer on debug build" ON) option(ENABLE_UB_SANITIZER "enable undefined behavior sanitizer on debug build" OFF) option(ENABLE_COVERAGE "enable coverage on debug build" OFF) +option(CCBENCH_BUILD_SILO_TESTS "build deterministic Silo regression tests" OFF) +if(CCBENCH_BUILD_SILO_TESTS) + enable_testing() +endif() # Auto-enable ccache when available — local builds also benefit from # the cross-target dedup we get in CI. Override with -DCCBENCH_CCACHE=OFF. diff --git a/cc/silo/CMakeLists.txt b/cc/silo/CMakeLists.txt index c17dedb2..f28db130 100644 --- a/cc/silo/CMakeLists.txt +++ b/cc/silo/CMakeLists.txt @@ -22,3 +22,17 @@ target_link_libraries(replay_test.exe PRIVATE target_compile_definitions(replay_test.exe PRIVATE ${_universal_defs}) # Same -Wall -Wextra -Werror as the ccbench_add_protocol() binaries (#43). set_compile_options(replay_test.exe) + +if(CCBENCH_BUILD_SILO_TESTS) + add_executable(silo_regression_test + test/transaction_test.cc transaction.cc util.cc) + target_link_libraries(silo_regression_test PRIVATE + ccbench_common ccbench::masstree ccbench::mimalloc GTest::gtest_main) + # Keep protocol switches and Result's layout identical to the workload target. + get_target_property(_silo_test_defs tpcc_silo.exe COMPILE_DEFINITIONS) + target_compile_definitions(silo_regression_test PRIVATE ${_silo_test_defs}) + set_compile_options(silo_regression_test) + add_test(NAME silo_regression_test COMMAND silo_regression_test) + # The old read protocol spins forever on these deliberately pending INSERTs. + set_tests_properties(silo_regression_test PROPERTIES TIMEOUT 20) +endif() diff --git a/cc/silo/include/transaction.hh b/cc/silo/include/transaction.hh index f0c8976d..816f89e9 100644 --- a/cc/silo/include/transaction.hh +++ b/cc/silo/include/transaction.hh @@ -32,10 +32,15 @@ class TxScanCallback; class TxExecutor { public: + struct RetiredRecord { + Tuple* record; + uint64_t epoch; + }; + std::vector> read_set_; std::vector> write_set_; std::vector pro_set_; - std::deque gc_records_; + std::deque gc_records_; std::unordered_map node_map_; std::vector log_set_; @@ -161,6 +166,9 @@ public: void leaderWork(); void gc_records(); + +private: + void retire_insert(WriteElement& element); }; static_assert(TxExecutorLike); diff --git a/cc/silo/test/transaction_test.cc b/cc/silo/test/transaction_test.cc new file mode 100644 index 00000000..88de225c --- /dev/null +++ b/cc/silo/test/transaction_test.cc @@ -0,0 +1,389 @@ +#include +#include +#include +#include +#include +#include +#include + +#define GLOBAL_VALUE_DEFINE + +#include "../include/atomic_tool.hh" +#include "../include/transaction.hh" +#include "../../../include/tpcc/tpcc_tx_neworder.hh" + +#include + +namespace { + +template +TupleBody MakeBody(std::string_view key, const T& value) { + HeapObject object; + object.allocate(); + object.cast_to() = value; + return TupleBody(key, std::move(object)); +} + +class SiloTransactionTest : public ::testing::Test { +protected: + bool quit_ = false; + Result results_[2]; + uint64_t_64byte epochs_[2]; + uint64_t_64byte commit_tids_[2]; + TxExecutor owner_{0, &results_[0], quit_}; + TxExecutor observer_{1, &results_[1], quit_}; + std::vector> keys_; + + void SetUp() override { + TotalThreadNum = 2; + ThLocalEpoch = epochs_; + CTIDW = commit_tids_; + storeRelease(GlobalEpoch.obj_, uint64_t{1}); + storeRelease(ReclamationEpoch, uint32_t{0}); + FLAGS_tpcc_interactive_ms = 0; + Backoff::Backoff_.store(0); + MasstreeWrapper::thread_init(0); + owner_.begin(); + observer_.begin(); + } + + void TearDown() override { + // No transaction or Tuple pointer remains in use after the test body. + owner_.abort(); + observer_.abort(); + storeRelease(ReclamationEpoch, std::numeric_limits::max()); + owner_.gc_records(); + observer_.gc_records(); + for (const auto& [storage, key] : keys_) { + auto& tree = Masstrees[get_storage(storage)]; + Tuple* tuple = tree.get_value(key); + if (tuple != nullptr) { + EXPECT_EQ(Status::OK, tree.remove_value(key)); + delete tuple; + } + } + ThLocalEpoch = nullptr; + CTIDW = nullptr; + } + + static std::string Key(uint64_t value) { + const auto encoded = __builtin_bswap64(value); + return std::string(reinterpret_cast(&encoded), + sizeof(encoded)); + } + + template + Tuple* Seed(Storage storage, std::string_view key, const T& value) { + auto tuple = std::make_unique(); + tuple->init(0, MakeBody(key, value), nullptr); + const Status status = + Masstrees[get_storage(storage)].insert_value(key, tuple.get()); + EXPECT_EQ(Status::OK, status); + if (status != Status::OK) return nullptr; + keys_.emplace_back(storage, key); + return tuple.release(); + } + + Status Insert(TxExecutor& tx, Storage storage, std::string_view key, + uint64_t value) { + keys_.emplace_back(storage, key); + return tx.insert(storage, key, MakeBody(key, value)); + } +}; + +TEST_F(SiloTransactionTest, PendingInsertPointReadAbortsWithoutWaiting) { + const auto key = Key(10); + ASSERT_EQ(Status::OK, Insert(owner_, Storage::Order, key, 42)); + TupleBody sentinel; + TupleBody* body = &sentinel; + + EXPECT_EQ(Status::ERROR_CONCURRENT_WRITE_OR_DELETE, + observer_.read(Storage::Order, key, &body)); + EXPECT_EQ(TransactionStatus::aborted, observer_.status_); + EXPECT_EQ(&sentinel, body); + EXPECT_TRUE(observer_.read_set_.empty()); +} + +TEST_F(SiloTransactionTest, LimitedScanOfPendingInsertAbortsWithoutWaiting) { + const auto first = Key(10); + const auto committed = Key(20); + ASSERT_EQ(Status::OK, Insert(owner_, Storage::NewOrder, first, 10)); + ASSERT_NE(nullptr, Seed(Storage::NewOrder, committed, uint64_t{20})); + std::vector result; + + // A pending first row must not be reported as an empty district or allow + // Delivery to choose the later committed row when LIMIT is one. + EXPECT_EQ(Status::ERROR_CONCURRENT_WRITE_OR_DELETE, + observer_.scan(Storage::NewOrder, first, false, Key(30), true, + result, 1)); + EXPECT_EQ(TransactionStatus::aborted, observer_.status_); + EXPECT_TRUE(result.empty()); +} + +TEST_F(SiloTransactionTest, AbortedInsertRemainsAliveUntilReclamationEpoch) { + const auto key = Key(10); + ASSERT_EQ(Status::OK, Insert(owner_, Storage::Order, key, 42)); + Tuple* retained = Masstrees[get_storage(Storage::Order)].get_value(key); + ASSERT_NE(nullptr, retained); + storeRelease(GlobalEpoch.obj_, uint64_t{4}); + + owner_.abort(); + + EXPECT_EQ(nullptr, Masstrees[get_storage(Storage::Order)].get_value(key)); + ASSERT_EQ(1U, owner_.gc_records_.size()); + Tidword retired; + retired.obj_ = loadAcquire(retained->tidword_.obj_); + EXPECT_TRUE(retired.absent); + EXPECT_FALSE(retired.lock); + EXPECT_EQ(4U, retired.epoch); + EXPECT_EQ(42U, retained->body_.get_value().cast_to()); + EXPECT_EQ(Status::WARN_NOT_FOUND, + observer_.read_internal(Storage::Order, key, retained)); + + storeRelease(ReclamationEpoch, uint32_t{3}); + owner_.gc_records(); + ASSERT_EQ(1U, owner_.gc_records_.size()); + storeRelease(ReclamationEpoch, uint32_t{4}); + owner_.gc_records(); + EXPECT_TRUE(owner_.gc_records_.empty()); +} + +TEST_F(SiloTransactionTest, IndexVersionConflictInsertIsOwnedAndRolledBack) { + const auto anchor = Key(10); + const auto concurrent = Key(20); + const auto rejected = Key(30); + ASSERT_NE(nullptr, Seed(Storage::Order, anchor, uint64_t{10})); + std::vector result; + ASSERT_EQ(Status::OK, + owner_.scan(Storage::Order, anchor, false, Key(40), true, result)); + ASSERT_FALSE(owner_.node_map_.empty()); + + // Another transaction changes the scanned leaf before owner's INSERT. + ASSERT_EQ(Status::OK, Insert(observer_, Storage::Order, concurrent, 20)); + ASSERT_TRUE(observer_.commit()); + EXPECT_EQ(Status::ERROR_CONCURRENT_WRITE_OR_DELETE, + Insert(owner_, Storage::Order, rejected, 30)); + EXPECT_EQ(TransactionStatus::aborted, owner_.status_); + ASSERT_NE(nullptr, owner_.searchWriteSet(Storage::Order, rejected)); + Tuple* retained = Masstrees[get_storage(Storage::Order)].get_value(rejected); + ASSERT_NE(nullptr, retained); + + owner_.abort(); + + EXPECT_EQ(nullptr, + Masstrees[get_storage(Storage::Order)].get_value(rejected)); + ASSERT_EQ(1U, owner_.gc_records_.size()); + EXPECT_FALSE(retained->tidword_.lock); + EXPECT_TRUE(retained->tidword_.absent); +} + +TEST_F(SiloTransactionTest, BeginPinsEpochUntilTransactionCompletes) { + storeRelease(GlobalEpoch.obj_, uint64_t{3}); + owner_.begin(); + EXPECT_EQ(3U, loadAcquire(ThLocalEpoch[0].obj_)); + const auto key = Key(10); + ASSERT_NE(nullptr, Seed(Storage::Order, key, uint64_t{10})); + TupleBody* body = nullptr; + ASSERT_EQ(Status::OK, owner_.read(Storage::Order, key, &body)); + + storeRelease(GlobalEpoch.obj_, uint64_t{5}); + ASSERT_TRUE(owner_.validationPhase()); + EXPECT_EQ(3U, loadAcquire(ThLocalEpoch[0].obj_)); + owner_.writePhase(); + owner_.begin(); + EXPECT_EQ(5U, loadAcquire(ThLocalEpoch[0].obj_)); +} + +TEST_F(SiloTransactionTest, DeleteRetirementWaitsForRemovalEpoch) { + const auto key = Key(10); + Tuple* retained = Seed(Storage::NewOrder, key, uint64_t{10}); + ASSERT_NE(nullptr, retained); + ASSERT_EQ(Status::OK, owner_.delete_record(Storage::NewOrder, key)); + ASSERT_TRUE(owner_.validationPhase()); + + // The pinned transaction epoch precedes removal from the index. A reader + // may still have acquired this pointer in the newer epoch before removal. + storeRelease(GlobalEpoch.obj_, uint64_t{3}); + owner_.writePhase(); + EXPECT_TRUE(retained->tidword_.absent); + ASSERT_EQ(1U, owner_.gc_records_.size()); + storeRelease(ReclamationEpoch, uint32_t{1}); + owner_.gc_records(); + ASSERT_EQ(1U, owner_.gc_records_.size()); + storeRelease(ReclamationEpoch, uint32_t{3}); + owner_.gc_records(); + EXPECT_TRUE(owner_.gc_records_.empty()); +} + +TEST_F(SiloTransactionTest, ReadsAndScansSeeOwnInsertBody) { + const auto key = Key(10); + ASSERT_EQ(Status::OK, Insert(owner_, Storage::Order, key, 42)); + TupleBody* body = nullptr; + ASSERT_EQ(Status::OK, owner_.read(Storage::Order, key, &body)); + ASSERT_NE(nullptr, body); + ASSERT_TRUE(body->get_value().is_compatible()); + EXPECT_EQ(42U, body->get_value().cast_to()); + + std::vector result; + ASSERT_EQ(Status::OK, + owner_.scan(Storage::Order, key, false, Key(20), true, result)); + ASSERT_EQ(1U, result.size()); + EXPECT_EQ(body, result.front()); +} + +TEST_F(SiloTransactionTest, OwnUpdatedReadsOverrideSnapshotWithinStorage) { + const auto key = Key(10); + ASSERT_NE(nullptr, Seed(Storage::Warehouse, key, uint64_t{1})); + ASSERT_NE(nullptr, Seed(Storage::District, key, uint64_t{2})); + TupleBody* body = nullptr; + ASSERT_EQ(Status::OK, owner_.read(Storage::Warehouse, key, &body)); + ASSERT_EQ(Status::OK, owner_.read(Storage::District, key, &body)); + ASSERT_EQ(Status::OK, + owner_.update(Storage::District, key, MakeBody(key, uint64_t{3}))); + ASSERT_EQ(Status::OK, owner_.read(Storage::District, key, &body)); + EXPECT_EQ(3U, body->get_value().cast_to()); + ASSERT_EQ(Status::OK, owner_.read(Storage::Warehouse, key, &body)); + EXPECT_EQ(1U, body->get_value().cast_to()); + + std::vector result; + ASSERT_EQ(Status::OK, + owner_.scan(Storage::District, key, false, Key(20), true, result)); + ASSERT_EQ(1U, result.size()); + EXPECT_EQ(3U, result.front()->get_value().cast_to()); +} + +TEST_F(SiloTransactionTest, ScanReallocationKeepsResultsInKeyOrder) { + const auto first = Key(10); + const auto second = Key(20); + ASSERT_NE(nullptr, Seed(Storage::Order, first, uint64_t{10})); + ASSERT_NE(nullptr, Seed(Storage::Order, second, uint64_t{20})); + TupleBody* body = nullptr; + ASSERT_EQ(Status::OK, owner_.read(Storage::Order, first, &body)); + owner_.read_set_.shrink_to_fit(); + ASSERT_EQ(owner_.read_set_.size(), owner_.read_set_.capacity()); + + std::vector result; + ASSERT_EQ(Status::OK, + owner_.scan(Storage::Order, first, false, Key(30), true, result)); + ASSERT_EQ(2U, result.size()); + auto* first_read = owner_.searchReadSet(Storage::Order, first); + auto* second_read = owner_.searchReadSet(Storage::Order, second); + ASSERT_NE(nullptr, first_read); + ASSERT_NE(nullptr, second_read); + // Check ownership before dereferencing: the old scan returns a freed + // first-row pointer after appending the second row reallocates read_set_. + ASSERT_EQ(&first_read->body_, result[0]); + ASSERT_EQ(&second_read->body_, result[1]); + EXPECT_EQ(first, result[0]->get_key()); + EXPECT_EQ(second, result[1]->get_key()); + EXPECT_EQ(10U, result[0]->get_value().cast_to()); + EXPECT_EQ(20U, result[1]->get_value().cast_to()); +} + +TEST_F(SiloTransactionTest, RepeatedStockUpdatesAccumulateQuantitiesAndCounts) { + Stock initial{}; + initial.S_W_ID = 1; + initial.S_I_ID = 1; + initial.S_QUANTITY = 16; + initial.S_YTD = 100; + initial.S_ORDER_CNT = 10; + initial.S_REMOTE_CNT = 1; + std::strcpy(initial.S_DIST_01, "district-data"); + SimpleKey<8> stock_key; + initial.createKey(stock_key.ptr()); + ASSERT_NE(nullptr, Seed(Storage::Stock, stock_key.view(), initial)); + + Stock first{}; + Stock second{}; + ASSERT_TRUE((get_and_update_stock( + owner_, 1, 1, 7, false, first))); + ASSERT_TRUE((get_and_update_stock( + owner_, 1, 1, 9, true, second))); + EXPECT_EQ(100, first.S_QUANTITY); + EXPECT_EQ(91, second.S_QUANTITY); + EXPECT_EQ(116U, second.S_YTD); + EXPECT_EQ(12U, second.S_ORDER_CNT); + EXPECT_EQ(2U, second.S_REMOTE_CNT); + EXPECT_STREQ("district-data", first.S_DIST_01); + EXPECT_STREQ("district-data", second.S_DIST_01); + ASSERT_EQ(1U, owner_.write_set_.size()); + ASSERT_TRUE(owner_.commit()); + + TupleBody* body = nullptr; + ASSERT_EQ(Status::OK, + observer_.read(Storage::Stock, stock_key.view(), &body)); + const auto& stored = body->get_value().cast_to(); + EXPECT_EQ(91, stored.S_QUANTITY); + EXPECT_EQ(116U, stored.S_YTD); + EXPECT_EQ(12U, stored.S_ORDER_CNT); + EXPECT_EQ(2U, stored.S_REMOTE_CNT); + EXPECT_STREQ("district-data", stored.S_DIST_01); +} + +TEST_F(SiloTransactionTest, DeleteBeforeUpdateDoesNotMarkUpdatedRowAbsent) { + const auto key = Key(10); + ASSERT_NE(nullptr, Seed(Storage::NewOrder, key, uint64_t{10})); + Tuple* order = Seed(Storage::Order, key, uint64_t{20}); + ASSERT_NE(nullptr, order); + ASSERT_EQ(Status::OK, owner_.delete_record(Storage::NewOrder, key)); + ASSERT_EQ(Status::OK, + owner_.update(Storage::Order, key, MakeBody(key, uint64_t{30}))); + ASSERT_TRUE(owner_.commit()); + + EXPECT_FALSE(order->tidword_.absent); + TupleBody* body = nullptr; + ASSERT_EQ(Status::OK, observer_.read(Storage::Order, key, &body)); + EXPECT_EQ(30U, body->get_value().cast_to()); + EXPECT_EQ(Status::WARN_NOT_FOUND, + observer_.read(Storage::NewOrder, key, &body)); +} + +TEST_F(SiloTransactionTest, ConcurrentDeleteRetiresRecordOnlyOnce) { + const auto key = Key(10); + ASSERT_NE(nullptr, Seed(Storage::NewOrder, key, uint64_t{10})); + ASSERT_EQ(Status::OK, owner_.delete_record(Storage::NewOrder, key)); + ASSERT_EQ(Status::OK, observer_.delete_record(Storage::NewOrder, key)); + ASSERT_TRUE(owner_.commit()); + + EXPECT_FALSE(observer_.commit()); + EXPECT_EQ(TransactionStatus::aborted, observer_.status_); + EXPECT_EQ(1U, owner_.gc_records_.size()); + EXPECT_TRUE(observer_.gc_records_.empty()); +} + +TEST_F(SiloTransactionTest, UpdatingRemovedRecordReleasesItsAcquiredLock) { + const auto key = Key(10); + Tuple* retained = Seed(Storage::Order, key, uint64_t{10}); + ASSERT_NE(nullptr, retained); + ASSERT_EQ(Status::OK, + observer_.update(Storage::Order, key, MakeBody(key, uint64_t{20}))); + ASSERT_EQ(Status::OK, owner_.delete_record(Storage::Order, key)); + ASSERT_TRUE(owner_.commit()); + + EXPECT_FALSE(observer_.commit()); + EXPECT_EQ(TransactionStatus::aborted, observer_.status_); + Tidword removed; + removed.obj_ = loadAcquire(retained->tidword_.obj_); + EXPECT_TRUE(removed.absent); + EXPECT_FALSE(removed.lock); +} + +TEST_F(SiloTransactionTest, QuitStopsReadWaitingOnCommittedRecordLock) { + const auto key = Key(10); + Tuple* tuple = Seed(Storage::Order, key, uint64_t{10}); + ASSERT_NE(nullptr, tuple); + Tidword locked = tuple->tidword_; + locked.lock = true; + storeRelease(tuple->tidword_.obj_, locked.obj_); + storeRelease(quit_, true); + TupleBody* body = nullptr; + + EXPECT_EQ(Status::ERROR_PREEMPTIVE_ABORT, + observer_.read(Storage::Order, key, &body)); + EXPECT_EQ(TransactionStatus::aborted, observer_.status_); + EXPECT_EQ(nullptr, body); + locked.lock = false; + storeRelease(tuple->tidword_.obj_, locked.obj_); +} + +} // namespace diff --git a/cc/silo/transaction.cc b/cc/silo/transaction.cc index 66e0b786..801cb45c 100644 --- a/cc/silo/transaction.cc +++ b/cc/silo/transaction.cc @@ -12,24 +12,35 @@ extern void siloLeaderWork(uint64_t& epoch_timer_start, uint64_t& epoch_timer_stop); void TxExecutor::gc_records() { - const auto r_epoch = ReclamationEpoch; + const auto r_epoch = __atomic_load_n(&ReclamationEpoch, __ATOMIC_ACQUIRE); // for records while (!gc_records_.empty()) { - Tuple* rec = gc_records_.front(); - if (rec->tidword_.epoch > r_epoch) break; - delete rec; + const auto& retired = gc_records_.front(); + if (retired.epoch > r_epoch) break; + delete retired.record; gc_records_.pop_front(); } } +void TxExecutor::retire_insert(WriteElement& element) { + Masstrees[get_storage(element.storage_)].remove_value_if_present( + element.key_); + // Readers may already hold this pointer. Publish a tombstone before + // reclaiming it after every transaction has left its protected epoch. + Tidword removed; + removed.obj_ = loadAcquire(element.rcdptr_->tidword_.obj_); + removed.epoch = atomicLoadGE(); + removed.absent = true; + removed.lock = false; + storeRelease(element.rcdptr_->tidword_.obj_, removed.obj_); + gc_records_.push_back({element.rcdptr_, removed.epoch}); +} + void TxExecutor::abort() { // remove inserted records for (auto& we : write_set_) { - if (we.op_ == OpType::INSERT) { - Masstrees[get_storage(we.storage_)].remove_value_if_present(we.key_); - delete we.rcdptr_; - } + if (we.op_ == OpType::INSERT) { retire_insert(we); } } gc_records(); @@ -52,6 +63,10 @@ void TxExecutor::abort() { } void TxExecutor::begin() { + // Announce the epoch before obtaining index pointers, and keep it pinned + // until this transaction has finished using its read/write sets. + atomicStoreThLocalEpoch(thid_, atomicLoadGE()); + __atomic_thread_fence(__ATOMIC_SEQ_CST); status_ = TransactionStatus::inflight; max_wset_.obj_ = 0; max_rset_.obj_ = 0; @@ -89,6 +104,9 @@ Status TxExecutor::insert(Storage s, std::string_view key, TupleBody&& body) { delete tuple; return stat; } + // The index now exposes this tuple. Abort must own it even if the node + // version check below fails; otherwise its INSERT lock is never released. + write_set_.emplace_back(s, key, tuple, OpType::INSERT); if (insert_info.node) { if (!node_map_.empty()) { auto it = node_map_.find((void*) insert_info.node); @@ -105,8 +123,6 @@ Status TxExecutor::insert(Storage s, std::string_view key, TupleBody&& body) { ERR; } - write_set_.emplace_back(s, key, tuple, OpType::INSERT); - #if ADD_ANALYSIS result_->local_write_latency_ += rdtscp() - start; #endif @@ -119,10 +135,18 @@ Status TxExecutor::delete_record(Storage s, std::string_view key) { #endif Tidword tidw; - // cancel previous write + // Coalesce a previous write without invalidating an iterator or losing + // ownership of an uncommitted INSERT. for (auto itr = write_set_.begin(); itr != write_set_.end(); ++itr) { if ((*itr).storage_ != s) continue; - if ((*itr).key_ == key) { write_set_.erase(itr); } + if ((*itr).key_ != key) continue; + if (itr->op_ == OpType::INSERT) { + retire_insert(*itr); + write_set_.erase(itr); + } else { + itr->op_ = OpType::DELETE; + } + return Status::OK; } Tuple* tuple = Masstrees[get_storage(s)].get_value(key); @@ -149,6 +173,11 @@ void TxExecutor::lockWriteSet() { if (itr->op_ == OpType::INSERT) continue; expected.obj_ = loadAcquire((*itr).rcdptr_->tidword_.obj_); for (;;) { + if (loadAcquire(quit_)) { + this->status_ = TransactionStatus::aborted; + unlockWriteSet(itr); + return; + } if (expected.lock) { #if NO_WAIT_LOCKING_IN_VALIDATION this->status_ = TransactionStatus::aborted; @@ -158,6 +187,8 @@ void TxExecutor::lockWriteSet() { if (itr != write_set_.begin()) unlockWriteSet(itr); goto retry; #endif + _mm_pause(); + expected.obj_ = loadAcquire((*itr).rcdptr_->tidword_.obj_); } else { desired = expected; desired.lock = 1; @@ -166,8 +197,8 @@ void TxExecutor::lockWriteSet() { break; } } - if (itr->op_ == OpType::UPDATE && itr->rcdptr_->tidword_.absent) { - unlockWriteSet(itr); + if (expected.absent) { + unlockWriteSet(itr + 1); this->status_ = TransactionStatus::aborted; return; } @@ -192,16 +223,17 @@ Status TxExecutor::read(Storage s, std::string_view key, TupleBody** body) { /** * read-own-writes or re-read from local read set. */ + we = searchWriteSet(s, key); + if (we) { + if (we->op_ == OpType::DELETE) return Status::WARN_NOT_FOUND; + *body = we->op_ == OpType::INSERT ? &(we->rcdptr_->body_) : &(we->body_); + goto FINISH_READ; + } re = searchReadSet(s, key); if (re) { *body = &(re->body_); goto FINISH_READ; } - we = searchWriteSet(s, key); - if (we) { - *body = &(we->body_); - goto FINISH_READ; - } /** * Search tuple from data structure. @@ -236,7 +268,24 @@ Status TxExecutor::read_internal(Storage s, std::string_view key, // spinning until the lock is clear for (;;) { - while (expected.lock) { expected.obj_ = loadAcquire(tuple->tidword_.obj_); } + if (loadAcquire(quit_)) { + status_ = TransactionStatus::aborted; + return Status::ERROR_PREEMPTIVE_ABORT; + } + while (expected.lock) { + // Uncommitted INSERTs are owned before validation, so waiting here + // can prevent their owning transactions from making progress. + if (expected.absent) { + status_ = TransactionStatus::aborted; + return Status::ERROR_CONCURRENT_WRITE_OR_DELETE; + } + if (loadAcquire(quit_)) { + status_ = TransactionStatus::aborted; + return Status::ERROR_PREEMPTIVE_ABORT; + } + _mm_pause(); + expected.obj_ = loadAcquire(tuple->tidword_.obj_); + } //(b) checks whether the record is the latest version // omit. because this is implemented by single version @@ -277,7 +326,6 @@ Status TxExecutor::scan(const Storage s, std::string_view left_key, bool r_exclusive, std::vector& result, int64_t limit) { result.clear(); - auto rset_init_size = read_set_.size(); std::vector scan_res; Masstrees[get_storage(s)].scan( @@ -285,27 +333,30 @@ Status TxExecutor::scan(const Storage s, std::string_view left_key, l_exclusive, right_key.empty() ? nullptr : right_key.data(), right_key.size(), r_exclusive, &scan_res, limit, callback_); - for (auto&& itr : scan_res) { - ReadElement* re = searchReadSet(s, itr->body_.get_key()); - if (re) { - result.emplace_back(&(re->body_)); - continue; - } + if (status_ == TransactionStatus::aborted) { + return Status::ERROR_CONCURRENT_WRITE_OR_DELETE; + } + for (auto&& itr : scan_res) { WriteElement* we = searchWriteSet(s, itr->body_.get_key()); - if (we) { - result.emplace_back(&(we->body_)); - continue; - } + if (we) continue; + if (searchReadSet(s, itr->body_.get_key())) continue; Status stat = read_internal(s, itr->body_.get_key(), itr); if (stat != Status::OK && stat != Status::WARN_NOT_FOUND) { return stat; } } - if (rset_init_size != read_set_.size()) { - for (auto itr = read_set_.begin() + rset_init_size; itr != read_set_.end(); - ++itr) { - result.emplace_back(&((*itr).body_)); + // Adding reads can relocate read_set_ elements. Materialize pointers only + // after all reads are collected, preserving the index scan's key order. + for (auto* tuple : scan_res) { + if (auto* we = searchWriteSet(s, tuple->body_.get_key())) { + if (we->op_ == OpType::INSERT) { + result.emplace_back(&we->rcdptr_->body_); + } else if (we->op_ == OpType::UPDATE) { + result.emplace_back(&we->body_); + } + } else if (auto* re = searchReadSet(s, tuple->body_.get_key())) { + result.emplace_back(&re->body_); } } @@ -369,10 +420,6 @@ bool TxExecutor::validationPhase() { // Validation Phase lockWriteSet(); if (this->status_ == TransactionStatus::aborted) return false; - asm volatile("" ::: "memory"); - atomicStoreThLocalEpoch(thid_, atomicLoadGE()); - asm volatile("" ::: "memory"); - /* Phase 2 abort if any condition of below is satisfied. * 1. tid of read_set_ changed from it that was got in Read Phase. * 2. not latest version @@ -458,7 +505,17 @@ Status TxExecutor::update(Storage s, std::string_view key, TupleBody&& body) { std::uint64_t start = rdtscp(); #endif - if (searchWriteSet(s, key)) goto FINISH_WRITE; + if (auto* we = searchWriteSet(s, key)) { + if (we->op_ == OpType::DELETE) return Status::WARN_NOT_FOUND; + if (we->op_ == OpType::INSERT) { + // The index exposes this key even while the INSERT is locked. + // Keep it immutable for concurrent scans. + we->rcdptr_->body_.swap_value(body); + } else { + we->body_ = std::move(body); + } + goto FINISH_WRITE; + } /** * Search tuple from data structure. @@ -505,7 +562,7 @@ void TxExecutor::writePhase() { tid_b.tid++; // calculates (c) - tid_c.epoch = ThLocalEpoch[thid_].obj_; + tid_c.epoch = atomicLoadGE(); // compare a, b, c Tidword maxtid = std::max({tid_a, tid_b, tid_c}); @@ -522,6 +579,7 @@ void TxExecutor::writePhase() { // update and unlock switch ((*itr).op_) { case OpType::UPDATE: { + maxtid.absent = false; memcpy((*itr).rcdptr_->body_.get_val_ptr(), (*itr).body_.get_val_ptr(), (*itr).body_.get_val_size()); storeRelease((*itr).rcdptr_->tidword_.obj_, maxtid.obj_); @@ -540,7 +598,9 @@ void TxExecutor::writePhase() { (*itr).key_); storeRelease((*itr).rcdptr_->tidword_.obj_, maxtid.obj_); // create information for garbage collection - gc_records_.push_back((*itr).rcdptr_); + // A commit TID can have been chosen before the epoch advanced. + // Tag retirement after removal, independently of the commit TID. + gc_records_.push_back({(*itr).rcdptr_, atomicLoadGE()}); break; } default: diff --git a/cc/silo/util.cc b/cc/silo/util.cc index 38d2c3b1..26bf60fd 100644 --- a/cc/silo/util.cc +++ b/cc/silo/util.cc @@ -154,7 +154,8 @@ void siloLeaderWork(uint64_t& epoch_timer_start, uint64_t& epoch_timer_stop) { chkEpochLoaded()) { atomicAddGE(); uint32_t cur_epoch = atomicLoadGE(); - ReclamationEpoch = cur_epoch > 2 ? cur_epoch - 2 : 0; + __atomic_store_n(&ReclamationEpoch, cur_epoch > 2 ? cur_epoch - 2 : 0, + __ATOMIC_RELEASE); epoch_timer_start = epoch_timer_stop; } } diff --git a/include/tpcc/tpcc_tx_orderstatus.hh b/include/tpcc/tpcc_tx_orderstatus.hh index 75dfeada..3af625c1 100644 --- a/include/tpcc/tpcc_tx_orderstatus.hh +++ b/include/tpcc/tpcc_tx_orderstatus.hh @@ -34,7 +34,7 @@ bool get_order_key_by_customer_id(TxExecutor& tx, uint16_t w_id, uint8_t d_id, "cannot get order ID by scanning order-secondary with customer ID"); return false; } - if (result.size() != 1) ERR; + if (result.size() != 1) { return false; } TupleBody* body = *result.begin(); o_key = body->get_value().cast_to>(); return true;