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;