From dfc5404b16c4ddd19b6661315596386e9a77061e Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Wed, 5 Aug 2026 12:32:19 +0800 Subject: [PATCH 1/7] fix: deadline-aware FIFO update admission and TimestampWindow capacity safety Update contenders previously CAS-ed the admission phase directly: later arrivals could overtake earlier ones, and waits had no deadline or same-class fairness. TimestampWindow keyed completion slots by ts % W and re-cleaned stale ranges, so a long-unresolved insert gap plus continued inserts spanning the window could corrupt slots and advance read_ts incorrectly. - acquire_update_timestamp() accepts an optional steady_clock absolute deadline (nullopt keeps the unbounded legacy wait without reading the clock); same-class update waiters are admitted FIFO; timeout paths remove the waiter and restore the admission phase with no leaked timestamp. Adds TransactionTimeoutException (ERR_TX_TIMEOUT). - TimestampWindow slots carry exact timestamp identity; write timestamps are reserved with a candidate - read_ts <= W capacity check; uint32 timeline exhaustion fails fast without wrap-around; slide_window() is removed. - Inserts hitting a full window drop their inserter admission count before backing off, so update/compact drains never wait on a timestamp-less inserter. Address review comments: consistent deadline alias and richer fail-fast diagnostics - UpdateTimestampLease takes MonotonicTimePoint (alias redeclared locally so the RAII header stays independent of the full version manager header). - The kWindowFull invariant-break message now carries read_ts/write_ts and window size for production triage. --- include/neug/transaction/timestamp_lease.h | 9 + include/neug/transaction/timestamp_window.h | 17 +- include/neug/transaction/version_manager.h | 34 ++- include/neug/utils/exception/exception.h | 11 + src/transaction/timestamp_lease.cc | 7 + src/transaction/timestamp_window.cc | 35 +-- src/transaction/version_manager.cc | 192 +++++++++++- src/utils/exception/exception.cc | 8 + .../transaction/test_read_view_publication.cc | 5 +- tests/transaction/test_runtime_wait.cc | 285 ++++++++++++++++++ .../test_transaction_release_order.cc | 20 +- 11 files changed, 570 insertions(+), 53 deletions(-) diff --git a/include/neug/transaction/timestamp_lease.h b/include/neug/transaction/timestamp_lease.h index 8ceffa3a4..29b18258d 100644 --- a/include/neug/transaction/timestamp_lease.h +++ b/include/neug/transaction/timestamp_lease.h @@ -15,12 +15,19 @@ #pragma once #include +#include #include namespace neug { class IVersionManager; +// Same deadline time type as IVersionManager::acquire_update_timestamp. +// Redeclared here so this RAII header does not need to include the full +// version manager header; diverging from the manager's alias is a compile +// error wherever both are visible. +using MonotonicTimePoint = std::chrono::steady_clock::time_point; + /** * @brief RAII owner of an update timestamp and its admission-state lifecycle. * @@ -31,6 +38,8 @@ class IVersionManager; class UpdateTimestampLease { public: explicit UpdateTimestampLease(IVersionManager& version_manager); + UpdateTimestampLease(IVersionManager& version_manager, + MonotonicTimePoint deadline); UpdateTimestampLease(UpdateTimestampLease&& other) noexcept; ~UpdateTimestampLease() noexcept; diff --git a/include/neug/transaction/timestamp_window.h b/include/neug/transaction/timestamp_window.h index 65f982867..14418c3f0 100644 --- a/include/neug/transaction/timestamp_window.h +++ b/include/neug/transaction/timestamp_window.h @@ -40,21 +40,16 @@ class TimestampWindow { // Clear a timestamp (called after read_ts advances past it) void clear(uint32_t ts); - // Advance the window base position (sliding window maintenance) - void slide_window(uint32_t current_ts); + static constexpr size_t kWindowSize = 65536; private: - static constexpr size_t kWindowSize = - 65536; // Window size for timestamp tracking - // Convert timestamp to array index inline size_t ts_index(uint32_t ts) const { return ts % kWindowSize; } - // Completed timestamp bitmap - std::unique_ptr[]> completed_ts_; - - // Base position of sliding window - uint32_t window_base_{0}; + // A slot contains the exact completed timestamp, or zero when empty. This + // prevents a timestamp that reuses the same ring index from being mistaken + // for an older completion. + std::unique_ptr[]> completed_ts_; }; -} // namespace neug \ No newline at end of file +} // namespace neug diff --git a/include/neug/transaction/version_manager.h b/include/neug/transaction/version_manager.h index 2d407d845..3b1dc5461 100644 --- a/include/neug/transaction/version_manager.h +++ b/include/neug/transaction/version_manager.h @@ -16,6 +16,9 @@ #include #include +#include +#include +#include #include #include "neug/transaction/runtime_wait.h" @@ -25,6 +28,8 @@ namespace neug { class UpdateTimestampLease; +using MonotonicTimePoint = std::chrono::steady_clock::time_point; +using MonotonicNowFn = MonotonicTimePoint (*)() noexcept; /** * @brief Atomically published reader-visible state. @@ -79,7 +84,8 @@ class IVersionManager { virtual void release_read_view() = 0; virtual uint32_t acquire_insert_timestamp() = 0; virtual void release_insert_timestamp(uint32_t ts) = 0; - virtual uint32_t acquire_update_timestamp() = 0; + virtual uint32_t acquire_update_timestamp( + std::optional deadline = std::nullopt) = 0; virtual void begin_update_commit(uint32_t ts) = 0; // May invoke the runtime waiter. Checkpoint callers must enter commit and // drain readers before acquiring checkpoint-manager or other @@ -216,6 +222,7 @@ static_assert((OperationGateWord::kPhaseMask | OperationGateWord::kReaderMask | class VersionManager : public IVersionManager { public: VersionManager(); + explicit VersionManager(MonotonicNowFn monotonic_now); ~VersionManager() override = default; void init_ts(PublishedReadView initial_read_view, int thread_num) override; @@ -226,7 +233,8 @@ class VersionManager : public IVersionManager { void release_read_view() override; uint32_t acquire_insert_timestamp() override; void release_insert_timestamp(uint32_t ts) override; - uint32_t acquire_update_timestamp() override; + uint32_t acquire_update_timestamp( + std::optional deadline = std::nullopt) override; void begin_update_commit(uint32_t ts) override; void drain_readers() override; void finish_update_timestamp( @@ -241,6 +249,15 @@ class VersionManager : public IVersionManager { using OperationGateWord = detail::OperationGateWord; void finish_update_and_reset_timeline(uint32_t ts) noexcept override; + struct UpdateWaiter {}; + + enum class TimestampReservationState { kReserved, kWindowFull, kExhausted }; + + struct TimestampReservation { + TimestampReservationState state; + uint32_t timestamp{0}; + }; + int thread_num_; // These helpers may suspend the logical task. Callers must not hold an // OS-thread-owned lock or retain an ordinary TLS pointer across the call. @@ -249,6 +266,15 @@ class VersionManager : public IVersionManager { AdmissionState desired_phase); void wait_for_readers_to_drain(); void wait_for_inserters_to_drain(); + bool wait_for_inserters_to_drain(std::optional deadline, + RuntimeBackoff& wait); + bool deadline_expired( + std::optional deadline) const noexcept; + void remove_update_waiter(UpdateWaiter* waiter); + TimestampReservation reserve_write_timestamp(); + void release_insert_admission(); + [[noreturn]] void throw_timestamp_reservation_failure( + TimestampReservationState state, uint32_t read_ts, uint32_t write_ts); void complete_write_timestamp(uint32_t ts); void advance_read_ts_locked(); RuntimeWaitFn runtime_wait_impl() const noexcept override; @@ -260,10 +286,14 @@ class VersionManager : public IVersionManager { std::atomic operation_gate_state_{0}; + std::mutex update_waiters_lock_; + std::deque update_waiters_; + TimestampWindow ts_window_; SpinLock lock_; std::atomic runtime_wait_; + MonotonicNowFn monotonic_now_; }; } // namespace neug diff --git a/include/neug/utils/exception/exception.h b/include/neug/utils/exception/exception.h index 16c4be378..7e0cef1c5 100644 --- a/include/neug/utils/exception/exception.h +++ b/include/neug/utils/exception/exception.h @@ -240,6 +240,14 @@ class NEUG_API TxStateConflictException : public Exception { const std::string& file_line); }; +class NEUG_API TransactionTimeoutException : public Exception { + public: + explicit TransactionTimeoutException(const std::string& msg); + + TransactionTimeoutException(const std::string& msg, + const std::string& file_line); +}; + } // namespace exception } // namespace neug @@ -344,6 +352,9 @@ class NEUG_API TxStateConflictException : public Exception { #define THROW_TX_STATE_CONFLICT(msg) \ THROW_EXCEPTION_WITH_FILE_LINE_AND_TYPE(TxStateConflictException, msg) +#define THROW_TRANSACTION_TIMEOUT(msg) \ + THROW_EXCEPTION_WITH_FILE_LINE_AND_TYPE(TransactionTimeoutException, msg) + #define THROW_IF_ARROW_NOT_OK(expr) \ do { \ auto status = (expr); \ diff --git a/src/transaction/timestamp_lease.cc b/src/transaction/timestamp_lease.cc index fe07d4a37..13600b6a2 100644 --- a/src/transaction/timestamp_lease.cc +++ b/src/transaction/timestamp_lease.cc @@ -29,6 +29,13 @@ UpdateTimestampLease::UpdateTimestampLease(IVersionManager& version_manager) CHECK_NE(timestamp_, kInactiveTimestamp); } +UpdateTimestampLease::UpdateTimestampLease(IVersionManager& version_manager, + MonotonicTimePoint deadline) + : version_manager_(&version_manager), + timestamp_(version_manager.acquire_update_timestamp(deadline)) { + CHECK_NE(timestamp_, kInactiveTimestamp); +} + UpdateTimestampLease::UpdateTimestampLease( UpdateTimestampLease&& other) noexcept : version_manager_(std::exchange(other.version_manager_, nullptr)), diff --git a/src/transaction/timestamp_window.cc b/src/transaction/timestamp_window.cc index 328ed3e4b..df8eb2f55 100644 --- a/src/transaction/timestamp_window.cc +++ b/src/transaction/timestamp_window.cc @@ -20,48 +20,33 @@ namespace neug { TimestampWindow::TimestampWindow() { - // Initialize completed timestamp bitmap - completed_ts_ = std::make_unique[]>(kWindowSize); + completed_ts_ = std::make_unique[]>(kWindowSize); for (size_t i = 0; i < kWindowSize; ++i) { - completed_ts_[i].store(false, std::memory_order_relaxed); + completed_ts_[i].store(0, std::memory_order_relaxed); } } TimestampWindow::~TimestampWindow() = default; void TimestampWindow::init() { - window_base_ = 0; for (size_t i = 0; i < kWindowSize; ++i) { - completed_ts_[i].store(false, std::memory_order_relaxed); + completed_ts_[i].store(0, std::memory_order_relaxed); } } void TimestampWindow::mark_completed(uint32_t ts) { - size_t idx = ts_index(ts); - // Correctness still holds even with the buffer overflow - completed_ts_[idx].store(true, std::memory_order_release); + DCHECK_NE(ts, 0U); + completed_ts_[ts_index(ts)].store(ts, std::memory_order_release); } bool TimestampWindow::is_completed(uint32_t ts) const { - size_t idx = ts_index(ts); - return completed_ts_[idx].load(std::memory_order_acquire); + return completed_ts_[ts_index(ts)].load(std::memory_order_acquire) == ts; } void TimestampWindow::clear(uint32_t ts) { - size_t idx = ts_index(ts); - completed_ts_[idx].store(false, std::memory_order_relaxed); + uint32_t expected = ts; + (void) completed_ts_[ts_index(ts)].compare_exchange_strong( + expected, 0, std::memory_order_relaxed, std::memory_order_relaxed); } -void TimestampWindow::slide_window(uint32_t current_ts) { - // Sliding window (if advanced significantly) - if (current_ts > window_base_ + kWindowSize / 2) { - // Clean up old window - uint32_t new_base = current_ts - kWindowSize / 4; - for (uint32_t ts = window_base_; ts < new_base; ++ts) { - clear(ts); - } - window_base_ = new_base; - } -} - -} // namespace neug \ No newline at end of file +} // namespace neug diff --git a/src/transaction/version_manager.cc b/src/transaction/version_manager.cc index b1b358c95..9874a00c8 100644 --- a/src/transaction/version_manager.cc +++ b/src/transaction/version_manager.cc @@ -16,6 +16,8 @@ #include "neug/transaction/version_manager.h" #include +#include +#include #include #include #include @@ -26,9 +28,20 @@ namespace neug { -// VersionManager implementation +namespace { -VersionManager::VersionManager() : runtime_wait_(&NativeRuntimeWait) {} +MonotonicTimePoint DefaultMonotonicNow() noexcept { + return std::chrono::steady_clock::now(); +} + +} // namespace + +VersionManager::VersionManager() : VersionManager(&DefaultMonotonicNow) {} + +VersionManager::VersionManager(MonotonicNowFn monotonic_now) + : runtime_wait_(&NativeRuntimeWait), monotonic_now_(monotonic_now) { + CHECK_NE(monotonic_now_, nullptr); +} void VersionManager::init_ts(PublishedReadView initial_read_view, int thread_num) { @@ -70,6 +83,13 @@ bool VersionManager::try_set_runtime_wait_if_quiescent( return false; } + std::unique_lock waiters_lock(update_waiters_lock_, std::try_to_lock); + if (!waiters_lock.owns_lock() || !update_waiters_.empty()) { + operation_gate_state_.store(OperationGateWord::empty(AdmissionState::kOpen), + std::memory_order_release); + return false; + } + runtime_wait_.store(runtime_wait, std::memory_order_release); operation_gate_state_.store(OperationGateWord::empty(AdmissionState::kOpen), std::memory_order_release); @@ -154,7 +174,25 @@ uint32_t VersionManager::acquire_insert_timestamp() { if (operation_gate_state_.compare_exchange_weak( observed, desired, std::memory_order_acquire, std::memory_order_relaxed)) { - return write_ts_.fetch_add(1, std::memory_order_acq_rel); + const auto reservation = reserve_write_timestamp(); + if (reservation.state == TimestampReservationState::kReserved) { + return reservation.timestamp; + } + + // A waiter that has not reserved a timestamp must not keep insert + // admission while waiting for window capacity: an update would otherwise + // close admission and wait for this count forever. + release_insert_admission(); + if (reservation.state == TimestampReservationState::kExhausted) { + throw_timestamp_reservation_failure( + reservation.state, read_ts_.load(std::memory_order_relaxed), + write_ts_.load(std::memory_order_relaxed)); + } + if (!wait) { + wait.emplace(make_runtime_backoff()); + } + (*wait)(); + observed = operation_gate_state_.load(std::memory_order_relaxed); } } } @@ -162,6 +200,10 @@ uint32_t VersionManager::acquire_insert_timestamp() { void VersionManager::release_insert_timestamp(uint32_t ts) { complete_write_timestamp(ts); + release_insert_admission(); +} + +void VersionManager::release_insert_admission() { const uint64_t observed = operation_gate_state_.load(std::memory_order_relaxed); if (NEUG_UNLIKELY(OperationGateWord::inserters(observed) == 0)) { @@ -174,6 +216,45 @@ void VersionManager::release_insert_timestamp(uint32_t ts) { DCHECK_GT(OperationGateWord::inserters(previous), 0U); } +VersionManager::TimestampReservation VersionManager::reserve_write_timestamp() { + uint32_t candidate = write_ts_.load(std::memory_order_relaxed); + while (true) { + if (candidate == std::numeric_limits::max()) { + return {TimestampReservationState::kExhausted}; + } + + const uint32_t current_read_ts = read_ts_.load(std::memory_order_acquire); + DCHECK_GT(candidate, current_read_ts); + const uint64_t outstanding = static_cast(candidate) - + static_cast(current_read_ts); + if (outstanding > TimestampWindow::kWindowSize) { + return {TimestampReservationState::kWindowFull}; + } + + if (write_ts_.compare_exchange_weak(candidate, candidate + 1, + std::memory_order_acq_rel, + std::memory_order_relaxed)) { + return {TimestampReservationState::kReserved, candidate}; + } + } +} + +[[noreturn]] void VersionManager::throw_timestamp_reservation_failure( + TimestampReservationState state, uint32_t read_ts, uint32_t write_ts) { + if (state == TimestampReservationState::kWindowFull) { + THROW_INTERNAL_EXCEPTION( + "TimestampWindow invariant broken: write timestamp reservation found " + "the window full despite exclusive write admission (read_ts=" + + std::to_string(read_ts) + ", write_ts=" + std::to_string(write_ts) + + ", window_size=" + std::to_string(TimestampWindow::kWindowSize) + + "); this indicates admission/window bookkeeping corruption, not " + "recoverable backpressure"); + } + THROW_RUNTIME_ERROR( + "Transaction timestamp space exhausted; checkpoint/reset the timeline " + "before reopening the database"); +} + void VersionManager::complete_write_timestamp(uint32_t ts) { // Mark completion (lock-free atomic operation) ts_window_.mark_completed(ts); @@ -210,8 +291,6 @@ void VersionManager::advance_read_ts_locked() { read_ts_.store(current, std::memory_order_release); } - // Sliding window maintenance - ts_window_.slide_window(current); published_read_view_.store( PackPublishedReadView({current, installed_snapshot_generation_.load( std::memory_order_relaxed)}), @@ -265,27 +344,106 @@ void VersionManager::wait_for_readers_to_drain() { } void VersionManager::wait_for_inserters_to_drain() { - uint64_t observed = operation_gate_state_.load(std::memory_order_acquire); + const uint64_t observed = + operation_gate_state_.load(std::memory_order_acquire); if (OperationGateWord::inserters(observed) == 0) { return; } RuntimeBackoff wait = make_runtime_backoff(); - do { + wait_for_inserters_to_drain(std::nullopt, wait); +} + +bool VersionManager::wait_for_inserters_to_drain( + std::optional deadline, RuntimeBackoff& wait) { + uint64_t observed = operation_gate_state_.load(std::memory_order_acquire); + while (OperationGateWord::inserters(observed) != 0) { + if (deadline_expired(deadline)) { + return false; + } wait(); observed = operation_gate_state_.load(std::memory_order_acquire); - } while (OperationGateWord::inserters(observed) != 0); + } + return !deadline_expired(deadline); } RuntimeWaitFn VersionManager::runtime_wait_impl() const noexcept { return runtime_wait_.load(std::memory_order_acquire); } -uint32_t VersionManager::acquire_update_timestamp() { - enter_admission_phase(AdmissionState::kInsertsBlocked); - wait_for_inserters_to_drain(); +bool VersionManager::deadline_expired( + std::optional deadline) const noexcept { + return deadline && monotonic_now_() >= *deadline; +} - return write_ts_.fetch_add(1, std::memory_order_acq_rel); +void VersionManager::remove_update_waiter(UpdateWaiter* waiter) { + std::lock_guard lock(update_waiters_lock_); + const auto it = + std::find(update_waiters_.begin(), update_waiters_.end(), waiter); + // A missing waiter means admission bookkeeping is already corrupt. Aborting + // is safer than throwing here, where the caller may hold an admission phase + // that only a successful removal would ever release. + CHECK(it != update_waiters_.end()) << "Update waiter is not queued"; + update_waiters_.erase(it); +} + +uint32_t VersionManager::acquire_update_timestamp( + std::optional deadline) { + UpdateWaiter waiter; + { + std::lock_guard lock(update_waiters_lock_); + update_waiters_.push_back(&waiter); + } + // Capture after enqueue. A successful runtime change requires this queue to + // be empty, so a queued waiter keeps one wait policy for its whole attempt. + RuntimeBackoff wait = make_runtime_backoff(); + + while (true) { + if (deadline_expired(deadline)) { + remove_update_waiter(&waiter); + THROW_TRANSACTION_TIMEOUT("waiting for update admission"); + } + + bool is_head = false; + { + std::lock_guard lock(update_waiters_lock_); + is_head = !update_waiters_.empty() && update_waiters_.front() == &waiter; + } + if (!is_head) { + wait(); + continue; + } + + uint64_t observed = operation_gate_state_.load(std::memory_order_relaxed); + if (OperationGateWord::phase(observed) == AdmissionState::kOpen && + OperationGateWord::try_change_phase(operation_gate_state_, observed, + AdmissionState::kInsertsBlocked)) { + break; + } + wait(); + } + + remove_update_waiter(&waiter); + if (!wait_for_inserters_to_drain(deadline, wait)) { + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + THROW_TRANSACTION_TIMEOUT("waiting for active inserts to finish"); + } + if (deadline_expired(deadline)) { + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + THROW_TRANSACTION_TIMEOUT("reserving update timestamp"); + } + + const auto reservation = reserve_write_timestamp(); + if (reservation.state == TimestampReservationState::kReserved) { + return reservation.timestamp; + } + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + throw_timestamp_reservation_failure( + reservation.state, read_ts_.load(std::memory_order_relaxed), + write_ts_.load(std::memory_order_relaxed)); } void VersionManager::begin_update_commit(uint32_t ts) { @@ -358,7 +516,15 @@ uint32_t VersionManager::acquire_compact_timestamp() { wait_for_readers_to_drain(); wait_for_inserters_to_drain(); - return write_ts_.fetch_add(1, std::memory_order_acq_rel); + const auto reservation = reserve_write_timestamp(); + if (reservation.state == TimestampReservationState::kReserved) { + return reservation.timestamp; + } + transition_admission_phase(AdmissionState::kAllBlocked, + AdmissionState::kOpen); + throw_timestamp_reservation_failure( + reservation.state, read_ts_.load(std::memory_order_relaxed), + write_ts_.load(std::memory_order_relaxed)); } void VersionManager::release_compact_timestamp(uint32_t ts) { diff --git a/src/utils/exception/exception.cc b/src/utils/exception/exception.cc index 84583da22..47626a276 100644 --- a/src/utils/exception/exception.cc +++ b/src/utils/exception/exception.cc @@ -252,5 +252,13 @@ TxStateConflictException::TxStateConflictException(const std::string& msg, : Exception("Transaction state conflict: " + msg, file_line, neug::StatusCode::ERR_TX_STATE_CONFLICT) {} +TransactionTimeoutException::TransactionTimeoutException(const std::string& msg) + : Exception("Transaction timeout: " + msg, + neug::StatusCode::ERR_TX_TIMEOUT) {} +TransactionTimeoutException::TransactionTimeoutException( + const std::string& msg, const std::string& file_line) + : Exception("Transaction timeout: " + msg, file_line, + neug::StatusCode::ERR_TX_TIMEOUT) {} + } // namespace exception } // namespace neug diff --git a/tests/transaction/test_read_view_publication.cc b/tests/transaction/test_read_view_publication.cc index ac9411e01..366ed92af 100644 --- a/tests/transaction/test_read_view_publication.cc +++ b/tests/transaction/test_read_view_publication.cc @@ -98,7 +98,10 @@ class ScriptedVersionManager : public IVersionManager { void release_read_view() override { release_count_.fetch_add(1); } uint32_t acquire_insert_timestamp() override { return 1; } void release_insert_timestamp(uint32_t) override {} - uint32_t acquire_update_timestamp() override { return 1; } + uint32_t acquire_update_timestamp( + std::optional) override { + return 1; + } void begin_update_commit(uint32_t) override {} void drain_readers() override {} void finish_update_timestamp(uint32_t, diff --git a/tests/transaction/test_runtime_wait.cc b/tests/transaction/test_runtime_wait.cc index 2327a6553..f4e76db35 100644 --- a/tests/transaction/test_runtime_wait.cc +++ b/tests/transaction/test_runtime_wait.cc @@ -18,8 +18,11 @@ #include #include #include +#include #include +#include #include +#include #include #include #include @@ -31,6 +34,7 @@ #endif #include "neug/transaction/timestamp_lease.h" #include "neug/transaction/version_manager.h" +#include "neug/utils/exception/exception.h" namespace neug { @@ -40,6 +44,18 @@ constexpr auto kWaitTimeout = std::chrono::seconds(10); std::atomic g_runtime_wait_calls{0}; std::atomic g_yield_calls{0}; std::atomic g_sleep_calls{0}; +std::atomic g_blocked_waiters{0}; +std::atomic g_fake_now_ticks{0}; +std::atomic g_fake_now_calls{0}; +std::mutex g_blocking_wait_lock; +std::condition_variable g_blocking_wait_cv; +bool g_block_waiters = false; + +MonotonicTimePoint FakeMonotonicNow() noexcept { + g_fake_now_calls.fetch_add(1, std::memory_order_relaxed); + return MonotonicTimePoint(std::chrono::milliseconds( + g_fake_now_ticks.load(std::memory_order_acquire))); +} void RecordRuntimeWait(RuntimeWaitAction action) noexcept { g_runtime_wait_calls.fetch_add(1, std::memory_order_relaxed); @@ -60,6 +76,26 @@ void CountingNativeRuntimeWait(RuntimeWaitAction action) noexcept { NativeRuntimeWait(action); } +void BlockingRuntimeWait(RuntimeWaitAction) noexcept { + g_blocked_waiters.fetch_add(1, std::memory_order_release); + std::unique_lock lock(g_blocking_wait_lock); + g_blocking_wait_cv.wait(lock, [] { return !g_block_waiters; }); +} + +void BlockRuntimeWaiters() { + g_blocked_waiters.store(0, std::memory_order_relaxed); + std::lock_guard lock(g_blocking_wait_lock); + g_block_waiters = true; +} + +void ReleaseRuntimeWaiters() { + { + std::lock_guard lock(g_blocking_wait_lock); + g_block_waiters = false; + } + g_blocking_wait_cv.notify_all(); +} + void InitManager(VersionManager& manager) { manager.init_ts({1, 0}, 4); EXPECT_TRUE(manager.try_set_runtime_wait_if_quiescent(&CountingRuntimeWait)); @@ -99,6 +135,12 @@ bool WaitForSleep() { []() { return g_sleep_calls.load(std::memory_order_relaxed) != 0; }); } +bool WaitForBlockedWaiters(uint32_t expected) { + return WaitUntil([&] { + return g_blocked_waiters.load(std::memory_order_acquire) >= expected; + }); +} + void ResetRuntimeWaitCalls() { g_runtime_wait_calls.store(0, std::memory_order_relaxed); g_yield_calls.store(0, std::memory_order_relaxed); @@ -359,6 +401,249 @@ TEST(VersionManagerWaitTest, AllContendedPathsUseBackoff) { } } +TEST(VersionManagerUpdateAdmissionTest, UpdateWaitersAcquireInFifoOrder) { + VersionManager manager(&FakeMonotonicNow); + InitManager(manager); + ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); + const auto holder = manager.acquire_update_timestamp(); + + BlockRuntimeWaiters(); + std::mutex order_lock; + std::vector order; + auto acquire_and_finish = [&](int id) { + const auto ts = manager.acquire_update_timestamp(); + { + std::lock_guard lock(order_lock); + order.push_back(id); + } + FinishUpdate(manager, ts); + }; + + std::thread first(acquire_and_finish, 1); + ASSERT_TRUE(WaitForBlockedWaiters(1)); + std::thread second(acquire_and_finish, 2); + ASSERT_TRUE(WaitForBlockedWaiters(2)); + std::thread third(acquire_and_finish, 3); + ASSERT_TRUE(WaitForBlockedWaiters(3)); + EXPECT_FALSE(manager.try_set_runtime_wait_if_quiescent(&NativeRuntimeWait)); + + ReleaseRuntimeWaiters(); + FinishUpdate(manager, holder); + first.join(); + second.join(); + third.join(); + EXPECT_EQ(order, (std::vector{1, 2, 3})); +} + +TEST(VersionManagerUpdateAdmissionTest, NoDeadlinePathsSkipMonotonicClock) { + g_fake_now_calls.store(0, std::memory_order_relaxed); + VersionManager manager(&FakeMonotonicNow); + InitManager(manager); + ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); + + // Contended no-deadline update: a queued waiter must not read the clock. + const auto holder = manager.acquire_update_timestamp(); + BlockRuntimeWaiters(); + std::thread waiter([&manager] { + const auto ts = manager.acquire_update_timestamp(); + FinishUpdate(manager, ts); + }); + ASSERT_TRUE(WaitForBlockedWaiters(1)); + FinishUpdate(manager, holder); + ReleaseRuntimeWaiters(); + waiter.join(); + + // Uncontended no-deadline update lease and no-deadline insert path. + { UpdateTimestampLease lease(manager); } + const auto insert_ts = manager.acquire_insert_timestamp(); + manager.release_insert_timestamp(insert_ts); + + EXPECT_EQ(g_fake_now_calls.load(std::memory_order_relaxed), 0U); + + // Sanity: a deadline path reads the clock, proving the counter is live. + EXPECT_THROW(UpdateTimestampLease(manager, MonotonicTimePoint::min()), + exception::TransactionTimeoutException); + EXPECT_GT(g_fake_now_calls.load(std::memory_order_relaxed), 0U); +} + +TEST(VersionManagerUpdateAdmissionTest, HeadTimeoutDoesNotBlockSuccessor) { + g_fake_now_ticks.store(10, std::memory_order_release); + VersionManager manager(&FakeMonotonicNow); + InitManager(manager); + ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); + const auto holder = manager.acquire_update_timestamp(); + + BlockRuntimeWaiters(); + std::promise head_timed_out; + std::promise successor_timestamp; + auto head_result = head_timed_out.get_future(); + auto successor_result = successor_timestamp.get_future(); + std::thread head([&] { + try { + (void) manager.acquire_update_timestamp( + MonotonicTimePoint(std::chrono::milliseconds(11))); + head_timed_out.set_value(false); + } catch (const exception::TransactionTimeoutException&) { + head_timed_out.set_value(true); + } + }); + ASSERT_TRUE(WaitForBlockedWaiters(1)); + std::thread successor([&] { + const auto ts = manager.acquire_update_timestamp(); + successor_timestamp.set_value(ts); + FinishUpdate(manager, ts); + }); + ASSERT_TRUE(WaitForBlockedWaiters(2)); + + g_fake_now_ticks.store(11, std::memory_order_release); + ReleaseRuntimeWaiters(); + EXPECT_EQ(head_result.wait_for(kWaitTimeout), std::future_status::ready); + EXPECT_TRUE(head_result.get()); + FinishUpdate(manager, holder); + EXPECT_EQ(successor_result.wait_for(kWaitTimeout), std::future_status::ready); + EXPECT_EQ(successor_result.get(), 3U); + head.join(); + successor.join(); +} + +TEST(VersionManagerUpdateAdmissionTest, NonHeadTimeoutPreservesFifoOrder) { + g_fake_now_ticks.store(20, std::memory_order_release); + VersionManager manager(&FakeMonotonicNow); + InitManager(manager); + ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); + const auto holder = manager.acquire_update_timestamp(); + + BlockRuntimeWaiters(); + std::mutex order_lock; + std::vector order; + std::promise middle_timed_out; + auto middle_result = middle_timed_out.get_future(); + auto acquire_and_finish = [&](int id) { + const auto ts = manager.acquire_update_timestamp(); + { + std::lock_guard lock(order_lock); + order.push_back(id); + } + FinishUpdate(manager, ts); + }; + std::thread first(acquire_and_finish, 1); + ASSERT_TRUE(WaitForBlockedWaiters(1)); + std::thread middle([&] { + try { + (void) manager.acquire_update_timestamp( + MonotonicTimePoint(std::chrono::milliseconds(21))); + middle_timed_out.set_value(false); + } catch (const exception::TransactionTimeoutException&) { + middle_timed_out.set_value(true); + } + }); + ASSERT_TRUE(WaitForBlockedWaiters(2)); + std::thread third(acquire_and_finish, 3); + ASSERT_TRUE(WaitForBlockedWaiters(3)); + + g_fake_now_ticks.store(21, std::memory_order_release); + ReleaseRuntimeWaiters(); + EXPECT_EQ(middle_result.wait_for(kWaitTimeout), std::future_status::ready); + EXPECT_TRUE(middle_result.get()); + FinishUpdate(manager, holder); + first.join(); + middle.join(); + third.join(); + EXPECT_EQ(order, (std::vector{1, 3})); +} + +TEST(VersionManagerUpdateAdmissionTest, + InserterDrainTimeoutRestoresAdmissionWithoutTimestamp) { + g_fake_now_ticks.store(30, std::memory_order_release); + VersionManager manager(&FakeMonotonicNow); + InitManager(manager); + ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); + const auto insert_ts = manager.acquire_insert_timestamp(); + + BlockRuntimeWaiters(); + std::promise timed_out; + auto timeout_result = timed_out.get_future(); + std::thread update([&] { + try { + (void) manager.acquire_update_timestamp( + MonotonicTimePoint(std::chrono::milliseconds(31))); + timed_out.set_value(false); + } catch (const exception::TransactionTimeoutException&) { + timed_out.set_value(true); + } + }); + ASSERT_TRUE(WaitForBlockedWaiters(1)); + + g_fake_now_ticks.store(31, std::memory_order_release); + ReleaseRuntimeWaiters(); + EXPECT_EQ(timeout_result.wait_for(kWaitTimeout), std::future_status::ready); + EXPECT_TRUE(timeout_result.get()); + update.join(); + + manager.release_insert_timestamp(insert_ts); + const auto next_update = manager.acquire_update_timestamp(); + EXPECT_EQ(next_update, 3U); + FinishUpdate(manager, next_update); +} + +TEST(VersionManagerTimestampWindowTest, + FullWindowBackpressuresWithoutBlockingUpdateDrain) { + VersionManager manager; + InitManager(manager); + ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); + const auto oldest = manager.acquire_insert_timestamp(); + for (size_t i = 0; i < TimestampWindow::kWindowSize - 1; ++i) { + const auto ts = manager.acquire_insert_timestamp(); + manager.release_insert_timestamp(ts); + } + EXPECT_EQ(manager.acquire_read_view().visibility_ts, 1U); + manager.release_read_view(); + + BlockRuntimeWaiters(); + std::promise waiting_insert; + auto waiting_result = waiting_insert.get_future(); + std::thread waiter([&] { + const auto ts = manager.acquire_insert_timestamp(); + waiting_insert.set_value(ts); + manager.release_insert_timestamp(ts); + }); + ASSERT_TRUE(WaitForBlockedWaiters(1)); + + std::promise update_timestamp; + auto update_result = update_timestamp.get_future(); + std::thread update([&] { + const auto ts = manager.acquire_update_timestamp(); + FinishUpdate(manager, ts); + update_timestamp.set_value(ts); + }); + // The full-window insert has released admission. The update can therefore + // close insert admission and waits only for the still-active oldest insert. + ASSERT_TRUE(WaitForBlockedWaiters(2)); + manager.release_insert_timestamp(oldest); + ReleaseRuntimeWaiters(); + + EXPECT_EQ(update_result.wait_for(kWaitTimeout), std::future_status::ready); + EXPECT_EQ(update_result.get(), TimestampWindow::kWindowSize + 2); + EXPECT_EQ(waiting_result.wait_for(kWaitTimeout), std::future_status::ready); + EXPECT_EQ(waiting_result.get(), TimestampWindow::kWindowSize + 3); + update.join(); + waiter.join(); +} + +TEST(VersionManagerTimestampWindowTest, TimestampExhaustionRestoresAdmission) { + VersionManager manager; + manager.init_ts({std::numeric_limits::max() - 1, 0}, 1); + + EXPECT_THROW(manager.acquire_insert_timestamp(), exception::RuntimeError); + EXPECT_THROW(manager.acquire_update_timestamp(), exception::RuntimeError); + EXPECT_THROW(manager.acquire_compact_timestamp(), exception::RuntimeError); + + const auto read = manager.acquire_read_view(); + EXPECT_EQ(read.visibility_ts, std::numeric_limits::max() - 1); + manager.release_read_view(); + EXPECT_TRUE(manager.try_set_runtime_wait_if_quiescent(&NativeRuntimeWait)); +} + TEST(VersionManagerAdmissionTest, CompactDoesNotOverlapAdmittedReadersOrInserters) { VersionManager manager; diff --git a/tests/transaction/test_transaction_release_order.cc b/tests/transaction/test_transaction_release_order.cc index 67e493385..0aaa60c6e 100644 --- a/tests/transaction/test_transaction_release_order.cc +++ b/tests/transaction/test_transaction_release_order.cc @@ -60,7 +60,10 @@ class ReleaseOrderVersionManager : public IVersionManager { } PublishedReadView acquire_read_view() override { return {1, 0}; } uint32_t acquire_insert_timestamp() override { return 1; } - uint32_t acquire_update_timestamp() override { return 1; } + uint32_t acquire_update_timestamp( + std::optional) override { + return 1; + } void begin_update_commit(uint32_t) override {} void drain_readers() override {} void finish_update_timestamp(uint32_t, @@ -214,6 +217,21 @@ TEST(UpdateTimestampLeaseTest, MoveTransfersTimestampOwnership) { version_manager.finish_update_timestamp(next_timestamp, std::nullopt); } +TEST(UpdateTimestampLeaseTest, DeadlineFailureDoesNotCreateLeaseOwnership) { + VersionManager version_manager; + version_manager.init_ts({0, 0}, 2); + + const auto holder = version_manager.acquire_update_timestamp(); + EXPECT_THROW( + UpdateTimestampLease(version_manager, std::chrono::steady_clock::now()), + exception::TransactionTimeoutException); + + version_manager.finish_update_timestamp(holder, std::nullopt); + const auto next_timestamp = version_manager.acquire_update_timestamp(); + EXPECT_EQ(next_timestamp, 2U); + version_manager.finish_update_timestamp(next_timestamp, std::nullopt); +} + TEST(APInPlaceConcurrencyTest, ExistingReaderBlocksWriterMutationPhase) { VersionManager version_manager; version_manager.init_ts({0, 0}, 2); From eba10f26da7a0a5ab5e83138224d49d963c7e678 Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Wed, 5 Aug 2026 15:01:00 +0800 Subject: [PATCH 2/7] update doc --- include/neug/transaction/README.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/include/neug/transaction/README.md b/include/neug/transaction/README.md index 31eb2b033..ef3cebd4d 100644 --- a/include/neug/transaction/README.md +++ b/include/neug/transaction/README.md @@ -124,3 +124,23 @@ cannot advance `read_ts_` past an earlier unfinished transaction. Insert commit appends WAL before replaying into the live graph. Update commit appends WAL before publishing its COW snapshot. Both complete their timestamps only after the graph change is visible. + +Update waiters are FIFO within their own class. A caller may provide an absolute +`steady_clock` deadline when acquiring an update timestamp; expiry before a +timestamp is reserved returns `ERR_TX_TIMEOUT` and restores admission. Legacy +callers provide no deadline and retain infinite-wait behavior. + +When `VersionManager::begin_update_commit` is called, the admission state changes from `kInsertsBlocked` to `kAllBlocked`. New reads and new inserts are blocked until the `UpdateTransaction` is committed or aborted. Already-acquired reads continue unaffected on their pinned snapshot. + +Timestamp completion uses a fixed ring whose slots contain the exact completed +timestamp, not a boolean bit. Before assigning a new write timestamp, +`VersionManager` limits unresolved timestamps to the ring capacity. An insert +that encounters this intentional backpressure first releases its inserter +admission, so it cannot prevent an update or compact operation from draining +existing inserts. + +## Serializability + +For a `ReadTransaction`, it will be assigned a graph timestamp. All insert or update transactions with timestamp less than or equal to that timestamp have been committed and are visible through timestamp filtering and the pinned snapshot. + +For each `InsertTransaction` or `UpdateTransaction`, a unique timestamp will be assigned. When committing, a write-ahead log will be written to the disk and all modifications will be applied to the graph atomically. From e93904b6b4fd2800d45140df86a28b5b8f81e5ba Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Wed, 5 Aug 2026 18:17:43 +0800 Subject: [PATCH 3/7] refine --- include/neug/transaction/README.md | 9 +- include/neug/transaction/timestamp_lease.h | 6 +- include/neug/transaction/version_manager.h | 13 +- src/transaction/timestamp_lease.cc | 11 +- src/transaction/version_manager.cc | 104 ++++---------- tests/transaction/test_runtime_wait.cc | 158 +++------------------ 6 files changed, 63 insertions(+), 238 deletions(-) diff --git a/include/neug/transaction/README.md b/include/neug/transaction/README.md index ef3cebd4d..8224c385b 100644 --- a/include/neug/transaction/README.md +++ b/include/neug/transaction/README.md @@ -125,10 +125,11 @@ Insert commit appends WAL before replaying into the live graph. Update commit appends WAL before publishing its COW snapshot. Both complete their timestamps only after the graph change is visible. -Update waiters are FIFO within their own class. A caller may provide an absolute -`steady_clock` deadline when acquiring an update timestamp; expiry before a -timestamp is reserved returns `ERR_TX_TIMEOUT` and restores admission. Legacy -callers provide no deadline and retain infinite-wait behavior. +Update waiters directly contend the existing admission phase; acquisition order +is unspecified. A caller may provide an absolute `steady_clock` deadline when +acquiring an update timestamp; expiry before a timestamp is reserved returns +`ERR_TX_TIMEOUT` and restores any phase acquired by that attempt. Legacy callers +provide no deadline and retain infinite-wait behavior. When `VersionManager::begin_update_commit` is called, the admission state changes from `kInsertsBlocked` to `kAllBlocked`. New reads and new inserts are blocked until the `UpdateTransaction` is committed or aborted. Already-acquired reads continue unaffected on their pinned snapshot. diff --git a/include/neug/transaction/timestamp_lease.h b/include/neug/transaction/timestamp_lease.h index 29b18258d..7902130d2 100644 --- a/include/neug/transaction/timestamp_lease.h +++ b/include/neug/transaction/timestamp_lease.h @@ -37,9 +37,9 @@ using MonotonicTimePoint = std::chrono::steady_clock::time_point; */ class UpdateTimestampLease { public: - explicit UpdateTimestampLease(IVersionManager& version_manager); - UpdateTimestampLease(IVersionManager& version_manager, - MonotonicTimePoint deadline); + explicit UpdateTimestampLease( + IVersionManager& version_manager, + std::optional deadline = std::nullopt); UpdateTimestampLease(UpdateTimestampLease&& other) noexcept; ~UpdateTimestampLease() noexcept; diff --git a/include/neug/transaction/version_manager.h b/include/neug/transaction/version_manager.h index 3b1dc5461..87995b611 100644 --- a/include/neug/transaction/version_manager.h +++ b/include/neug/transaction/version_manager.h @@ -17,8 +17,6 @@ #include #include #include -#include -#include #include #include "neug/transaction/runtime_wait.h" @@ -29,7 +27,6 @@ namespace neug { class UpdateTimestampLease; using MonotonicTimePoint = std::chrono::steady_clock::time_point; -using MonotonicNowFn = MonotonicTimePoint (*)() noexcept; /** * @brief Atomically published reader-visible state. @@ -84,6 +81,8 @@ class IVersionManager { virtual void release_read_view() = 0; virtual uint32_t acquire_insert_timestamp() = 0; virtual void release_insert_timestamp(uint32_t ts) = 0; + // Waiters directly contend the admission phase. Acquisition order is + // intentionally unspecified; the successful phase CAS linearizes ownership. virtual uint32_t acquire_update_timestamp( std::optional deadline = std::nullopt) = 0; virtual void begin_update_commit(uint32_t ts) = 0; @@ -222,7 +221,6 @@ static_assert((OperationGateWord::kPhaseMask | OperationGateWord::kReaderMask | class VersionManager : public IVersionManager { public: VersionManager(); - explicit VersionManager(MonotonicNowFn monotonic_now); ~VersionManager() override = default; void init_ts(PublishedReadView initial_read_view, int thread_num) override; @@ -249,8 +247,6 @@ class VersionManager : public IVersionManager { using OperationGateWord = detail::OperationGateWord; void finish_update_and_reset_timeline(uint32_t ts) noexcept override; - struct UpdateWaiter {}; - enum class TimestampReservationState { kReserved, kWindowFull, kExhausted }; struct TimestampReservation { @@ -270,7 +266,6 @@ class VersionManager : public IVersionManager { RuntimeBackoff& wait); bool deadline_expired( std::optional deadline) const noexcept; - void remove_update_waiter(UpdateWaiter* waiter); TimestampReservation reserve_write_timestamp(); void release_insert_admission(); [[noreturn]] void throw_timestamp_reservation_failure( @@ -286,14 +281,10 @@ class VersionManager : public IVersionManager { std::atomic operation_gate_state_{0}; - std::mutex update_waiters_lock_; - std::deque update_waiters_; - TimestampWindow ts_window_; SpinLock lock_; std::atomic runtime_wait_; - MonotonicNowFn monotonic_now_; }; } // namespace neug diff --git a/src/transaction/timestamp_lease.cc b/src/transaction/timestamp_lease.cc index 13600b6a2..6dd595756 100644 --- a/src/transaction/timestamp_lease.cc +++ b/src/transaction/timestamp_lease.cc @@ -23,14 +23,9 @@ namespace neug { -UpdateTimestampLease::UpdateTimestampLease(IVersionManager& version_manager) - : version_manager_(&version_manager), - timestamp_(version_manager.acquire_update_timestamp()) { - CHECK_NE(timestamp_, kInactiveTimestamp); -} - -UpdateTimestampLease::UpdateTimestampLease(IVersionManager& version_manager, - MonotonicTimePoint deadline) +UpdateTimestampLease::UpdateTimestampLease( + IVersionManager& version_manager, + std::optional deadline) : version_manager_(&version_manager), timestamp_(version_manager.acquire_update_timestamp(deadline)) { CHECK_NE(timestamp_, kInactiveTimestamp); diff --git a/src/transaction/version_manager.cc b/src/transaction/version_manager.cc index 9874a00c8..36296306c 100644 --- a/src/transaction/version_manager.cc +++ b/src/transaction/version_manager.cc @@ -16,7 +16,6 @@ #include "neug/transaction/version_manager.h" #include -#include #include #include #include @@ -28,20 +27,7 @@ namespace neug { -namespace { - -MonotonicTimePoint DefaultMonotonicNow() noexcept { - return std::chrono::steady_clock::now(); -} - -} // namespace - -VersionManager::VersionManager() : VersionManager(&DefaultMonotonicNow) {} - -VersionManager::VersionManager(MonotonicNowFn monotonic_now) - : runtime_wait_(&NativeRuntimeWait), monotonic_now_(monotonic_now) { - CHECK_NE(monotonic_now_, nullptr); -} +VersionManager::VersionManager() : runtime_wait_(&NativeRuntimeWait) {} void VersionManager::init_ts(PublishedReadView initial_read_view, int thread_num) { @@ -83,13 +69,6 @@ bool VersionManager::try_set_runtime_wait_if_quiescent( return false; } - std::unique_lock waiters_lock(update_waiters_lock_, std::try_to_lock); - if (!waiters_lock.owns_lock() || !update_waiters_.empty()) { - operation_gate_state_.store(OperationGateWord::empty(AdmissionState::kOpen), - std::memory_order_release); - return false; - } - runtime_wait_.store(runtime_wait, std::memory_order_release); operation_gate_state_.store(OperationGateWord::empty(AdmissionState::kOpen), std::memory_order_release); @@ -373,66 +352,43 @@ RuntimeWaitFn VersionManager::runtime_wait_impl() const noexcept { bool VersionManager::deadline_expired( std::optional deadline) const noexcept { - return deadline && monotonic_now_() >= *deadline; -} - -void VersionManager::remove_update_waiter(UpdateWaiter* waiter) { - std::lock_guard lock(update_waiters_lock_); - const auto it = - std::find(update_waiters_.begin(), update_waiters_.end(), waiter); - // A missing waiter means admission bookkeeping is already corrupt. Aborting - // is safer than throwing here, where the caller may hold an admission phase - // that only a successful removal would ever release. - CHECK(it != update_waiters_.end()) << "Update waiter is not queued"; - update_waiters_.erase(it); + return deadline && std::chrono::steady_clock::now() >= *deadline; } uint32_t VersionManager::acquire_update_timestamp( std::optional deadline) { - UpdateWaiter waiter; - { - std::lock_guard lock(update_waiters_lock_); - update_waiters_.push_back(&waiter); - } - // Capture after enqueue. A successful runtime change requires this queue to - // be empty, so a queued waiter keeps one wait policy for its whole attempt. - RuntimeBackoff wait = make_runtime_backoff(); - - while (true) { - if (deadline_expired(deadline)) { - remove_update_waiter(&waiter); - THROW_TRANSACTION_TIMEOUT("waiting for update admission"); - } - - bool is_head = false; - { - std::lock_guard lock(update_waiters_lock_); - is_head = !update_waiters_.empty() && update_waiters_.front() == &waiter; - } - if (!is_head) { + if (!deadline) { + // Preserve the legacy auto-commit fast path: no clock read and no backoff + // object unless admission or inserter drain is actually contended. + enter_admission_phase(AdmissionState::kInsertsBlocked); + wait_for_inserters_to_drain(); + } else { + RuntimeBackoff wait = make_runtime_backoff(); + uint64_t observed = operation_gate_state_.load(std::memory_order_relaxed); + while (true) { + if (deadline_expired(deadline)) { + THROW_TRANSACTION_TIMEOUT("waiting for update admission"); + } + if (OperationGateWord::phase(observed) == AdmissionState::kOpen && + OperationGateWord::try_change_phase( + operation_gate_state_, observed, + AdmissionState::kInsertsBlocked)) { + break; + } wait(); - continue; + observed = operation_gate_state_.load(std::memory_order_relaxed); } - uint64_t observed = operation_gate_state_.load(std::memory_order_relaxed); - if (OperationGateWord::phase(observed) == AdmissionState::kOpen && - OperationGateWord::try_change_phase(operation_gate_state_, observed, - AdmissionState::kInsertsBlocked)) { - break; + if (!wait_for_inserters_to_drain(deadline, wait)) { + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + THROW_TRANSACTION_TIMEOUT("waiting for active inserts to finish"); + } + if (deadline_expired(deadline)) { + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + THROW_TRANSACTION_TIMEOUT("reserving update timestamp"); } - wait(); - } - - remove_update_waiter(&waiter); - if (!wait_for_inserters_to_drain(deadline, wait)) { - transition_admission_phase(AdmissionState::kInsertsBlocked, - AdmissionState::kOpen); - THROW_TRANSACTION_TIMEOUT("waiting for active inserts to finish"); - } - if (deadline_expired(deadline)) { - transition_admission_phase(AdmissionState::kInsertsBlocked, - AdmissionState::kOpen); - THROW_TRANSACTION_TIMEOUT("reserving update timestamp"); } const auto reservation = reserve_write_timestamp(); diff --git a/tests/transaction/test_runtime_wait.cc b/tests/transaction/test_runtime_wait.cc index f4e76db35..9a8d62991 100644 --- a/tests/transaction/test_runtime_wait.cc +++ b/tests/transaction/test_runtime_wait.cc @@ -45,18 +45,10 @@ std::atomic g_runtime_wait_calls{0}; std::atomic g_yield_calls{0}; std::atomic g_sleep_calls{0}; std::atomic g_blocked_waiters{0}; -std::atomic g_fake_now_ticks{0}; -std::atomic g_fake_now_calls{0}; std::mutex g_blocking_wait_lock; std::condition_variable g_blocking_wait_cv; bool g_block_waiters = false; -MonotonicTimePoint FakeMonotonicNow() noexcept { - g_fake_now_calls.fetch_add(1, std::memory_order_relaxed); - return MonotonicTimePoint(std::chrono::milliseconds( - g_fake_now_ticks.load(std::memory_order_acquire))); -} - void RecordRuntimeWait(RuntimeWaitAction action) noexcept { g_runtime_wait_calls.fetch_add(1, std::memory_order_relaxed); if (action == RuntimeWaitAction::kYield) { @@ -401,90 +393,26 @@ TEST(VersionManagerWaitTest, AllContendedPathsUseBackoff) { } } -TEST(VersionManagerUpdateAdmissionTest, UpdateWaitersAcquireInFifoOrder) { - VersionManager manager(&FakeMonotonicNow); - InitManager(manager); - ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); - const auto holder = manager.acquire_update_timestamp(); - - BlockRuntimeWaiters(); - std::mutex order_lock; - std::vector order; - auto acquire_and_finish = [&](int id) { - const auto ts = manager.acquire_update_timestamp(); - { - std::lock_guard lock(order_lock); - order.push_back(id); - } - FinishUpdate(manager, ts); - }; - - std::thread first(acquire_and_finish, 1); - ASSERT_TRUE(WaitForBlockedWaiters(1)); - std::thread second(acquire_and_finish, 2); - ASSERT_TRUE(WaitForBlockedWaiters(2)); - std::thread third(acquire_and_finish, 3); - ASSERT_TRUE(WaitForBlockedWaiters(3)); - EXPECT_FALSE(manager.try_set_runtime_wait_if_quiescent(&NativeRuntimeWait)); - - ReleaseRuntimeWaiters(); - FinishUpdate(manager, holder); - first.join(); - second.join(); - third.join(); - EXPECT_EQ(order, (std::vector{1, 2, 3})); -} - -TEST(VersionManagerUpdateAdmissionTest, NoDeadlinePathsSkipMonotonicClock) { - g_fake_now_calls.store(0, std::memory_order_relaxed); - VersionManager manager(&FakeMonotonicNow); - InitManager(manager); - ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); - - // Contended no-deadline update: a queued waiter must not read the clock. - const auto holder = manager.acquire_update_timestamp(); - BlockRuntimeWaiters(); - std::thread waiter([&manager] { - const auto ts = manager.acquire_update_timestamp(); - FinishUpdate(manager, ts); - }); - ASSERT_TRUE(WaitForBlockedWaiters(1)); - FinishUpdate(manager, holder); - ReleaseRuntimeWaiters(); - waiter.join(); - - // Uncontended no-deadline update lease and no-deadline insert path. - { UpdateTimestampLease lease(manager); } - const auto insert_ts = manager.acquire_insert_timestamp(); - manager.release_insert_timestamp(insert_ts); - - EXPECT_EQ(g_fake_now_calls.load(std::memory_order_relaxed), 0U); - - // Sanity: a deadline path reads the clock, proving the counter is live. - EXPECT_THROW(UpdateTimestampLease(manager, MonotonicTimePoint::min()), - exception::TransactionTimeoutException); - EXPECT_GT(g_fake_now_calls.load(std::memory_order_relaxed), 0U); -} - -TEST(VersionManagerUpdateAdmissionTest, HeadTimeoutDoesNotBlockSuccessor) { - g_fake_now_ticks.store(10, std::memory_order_release); - VersionManager manager(&FakeMonotonicNow); +TEST(VersionManagerUpdateAdmissionTest, + ContendedTimeoutDoesNotBlockOtherWaiters) { + VersionManager manager; InitManager(manager); ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); const auto holder = manager.acquire_update_timestamp(); BlockRuntimeWaiters(); - std::promise head_timed_out; + std::promise waiter_timed_out; std::promise successor_timestamp; - auto head_result = head_timed_out.get_future(); + auto timeout_result = waiter_timed_out.get_future(); auto successor_result = successor_timestamp.get_future(); - std::thread head([&] { + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(50); + std::thread timed_waiter([&] { try { - (void) manager.acquire_update_timestamp( - MonotonicTimePoint(std::chrono::milliseconds(11))); - head_timed_out.set_value(false); + (void) manager.acquire_update_timestamp(deadline); + waiter_timed_out.set_value(false); } catch (const exception::TransactionTimeoutException&) { - head_timed_out.set_value(true); + waiter_timed_out.set_value(true); } }); ASSERT_TRUE(WaitForBlockedWaiters(1)); @@ -495,67 +423,20 @@ TEST(VersionManagerUpdateAdmissionTest, HeadTimeoutDoesNotBlockSuccessor) { }); ASSERT_TRUE(WaitForBlockedWaiters(2)); - g_fake_now_ticks.store(11, std::memory_order_release); + std::this_thread::sleep_until(deadline); ReleaseRuntimeWaiters(); - EXPECT_EQ(head_result.wait_for(kWaitTimeout), std::future_status::ready); - EXPECT_TRUE(head_result.get()); + EXPECT_EQ(timeout_result.wait_for(kWaitTimeout), std::future_status::ready); + EXPECT_TRUE(timeout_result.get()); FinishUpdate(manager, holder); EXPECT_EQ(successor_result.wait_for(kWaitTimeout), std::future_status::ready); EXPECT_EQ(successor_result.get(), 3U); - head.join(); + timed_waiter.join(); successor.join(); } -TEST(VersionManagerUpdateAdmissionTest, NonHeadTimeoutPreservesFifoOrder) { - g_fake_now_ticks.store(20, std::memory_order_release); - VersionManager manager(&FakeMonotonicNow); - InitManager(manager); - ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); - const auto holder = manager.acquire_update_timestamp(); - - BlockRuntimeWaiters(); - std::mutex order_lock; - std::vector order; - std::promise middle_timed_out; - auto middle_result = middle_timed_out.get_future(); - auto acquire_and_finish = [&](int id) { - const auto ts = manager.acquire_update_timestamp(); - { - std::lock_guard lock(order_lock); - order.push_back(id); - } - FinishUpdate(manager, ts); - }; - std::thread first(acquire_and_finish, 1); - ASSERT_TRUE(WaitForBlockedWaiters(1)); - std::thread middle([&] { - try { - (void) manager.acquire_update_timestamp( - MonotonicTimePoint(std::chrono::milliseconds(21))); - middle_timed_out.set_value(false); - } catch (const exception::TransactionTimeoutException&) { - middle_timed_out.set_value(true); - } - }); - ASSERT_TRUE(WaitForBlockedWaiters(2)); - std::thread third(acquire_and_finish, 3); - ASSERT_TRUE(WaitForBlockedWaiters(3)); - - g_fake_now_ticks.store(21, std::memory_order_release); - ReleaseRuntimeWaiters(); - EXPECT_EQ(middle_result.wait_for(kWaitTimeout), std::future_status::ready); - EXPECT_TRUE(middle_result.get()); - FinishUpdate(manager, holder); - first.join(); - middle.join(); - third.join(); - EXPECT_EQ(order, (std::vector{1, 3})); -} - TEST(VersionManagerUpdateAdmissionTest, InserterDrainTimeoutRestoresAdmissionWithoutTimestamp) { - g_fake_now_ticks.store(30, std::memory_order_release); - VersionManager manager(&FakeMonotonicNow); + VersionManager manager; InitManager(manager); ASSERT_TRUE(manager.try_set_runtime_wait_if_quiescent(&BlockingRuntimeWait)); const auto insert_ts = manager.acquire_insert_timestamp(); @@ -563,10 +444,11 @@ TEST(VersionManagerUpdateAdmissionTest, BlockRuntimeWaiters(); std::promise timed_out; auto timeout_result = timed_out.get_future(); + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(50); std::thread update([&] { try { - (void) manager.acquire_update_timestamp( - MonotonicTimePoint(std::chrono::milliseconds(31))); + (void) manager.acquire_update_timestamp(deadline); timed_out.set_value(false); } catch (const exception::TransactionTimeoutException&) { timed_out.set_value(true); @@ -574,7 +456,7 @@ TEST(VersionManagerUpdateAdmissionTest, }); ASSERT_TRUE(WaitForBlockedWaiters(1)); - g_fake_now_ticks.store(31, std::memory_order_release); + std::this_thread::sleep_until(deadline); ReleaseRuntimeWaiters(); EXPECT_EQ(timeout_result.wait_for(kWaitTimeout), std::future_status::ready); EXPECT_TRUE(timeout_result.get()); From 47cc7837f96f4fd0bdc01f0ad107039355e9551d Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Wed, 5 Aug 2026 20:46:08 +0800 Subject: [PATCH 4/7] fix: refine update timestamp admission --- include/neug/transaction/timestamp_lease.h | 12 +- include/neug/transaction/version_manager.h | 21 +-- src/transaction/timestamp_lease.cc | 10 +- src/transaction/version_manager.cc | 131 +++++++++++------- .../transaction/test_read_view_publication.cc | 5 +- tests/transaction/test_runtime_wait.cc | 4 +- .../test_transaction_release_order.cc | 5 +- 7 files changed, 109 insertions(+), 79 deletions(-) diff --git a/include/neug/transaction/timestamp_lease.h b/include/neug/transaction/timestamp_lease.h index 7902130d2..a87316e28 100644 --- a/include/neug/transaction/timestamp_lease.h +++ b/include/neug/transaction/timestamp_lease.h @@ -22,12 +22,6 @@ namespace neug { class IVersionManager; -// Same deadline time type as IVersionManager::acquire_update_timestamp. -// Redeclared here so this RAII header does not need to include the full -// version manager header; diverging from the manager's alias is a compile -// error wherever both are visible. -using MonotonicTimePoint = std::chrono::steady_clock::time_point; - /** * @brief RAII owner of an update timestamp and its admission-state lifecycle. * @@ -37,9 +31,9 @@ using MonotonicTimePoint = std::chrono::steady_clock::time_point; */ class UpdateTimestampLease { public: - explicit UpdateTimestampLease( - IVersionManager& version_manager, - std::optional deadline = std::nullopt); + explicit UpdateTimestampLease(IVersionManager& version_manager); + UpdateTimestampLease(IVersionManager& version_manager, + std::chrono::steady_clock::time_point deadline); UpdateTimestampLease(UpdateTimestampLease&& other) noexcept; ~UpdateTimestampLease() noexcept; diff --git a/include/neug/transaction/version_manager.h b/include/neug/transaction/version_manager.h index 87995b611..cca398c59 100644 --- a/include/neug/transaction/version_manager.h +++ b/include/neug/transaction/version_manager.h @@ -26,7 +26,6 @@ namespace neug { class UpdateTimestampLease; -using MonotonicTimePoint = std::chrono::steady_clock::time_point; /** * @brief Atomically published reader-visible state. @@ -83,8 +82,7 @@ class IVersionManager { virtual void release_insert_timestamp(uint32_t ts) = 0; // Waiters directly contend the admission phase. Acquisition order is // intentionally unspecified; the successful phase CAS linearizes ownership. - virtual uint32_t acquire_update_timestamp( - std::optional deadline = std::nullopt) = 0; + virtual uint32_t acquire_update_timestamp() = 0; virtual void begin_update_commit(uint32_t ts) = 0; // May invoke the runtime waiter. Checkpoint callers must enter commit and // drain readers before acquiring checkpoint-manager or other @@ -107,6 +105,11 @@ class IVersionManager { private: friend class UpdateTimestampLease; + // Timed acquisition is intentionally lease-only: callers must not receive a + // raw timestamp without immediately establishing RAII ownership. + virtual uint32_t acquire_update_timestamp_until( + std::chrono::steady_clock::time_point deadline) = 0; + /// Complete an exclusive update after external state has moved to a new /// timeline. Preserve the current snapshot generation and publish visibility /// timestamp zero before reopening admission. @@ -231,8 +234,7 @@ class VersionManager : public IVersionManager { void release_read_view() override; uint32_t acquire_insert_timestamp() override; void release_insert_timestamp(uint32_t ts) override; - uint32_t acquire_update_timestamp( - std::optional deadline = std::nullopt) override; + uint32_t acquire_update_timestamp() override; void begin_update_commit(uint32_t ts) override; void drain_readers() override; void finish_update_timestamp( @@ -245,6 +247,8 @@ class VersionManager : public IVersionManager { private: using AdmissionState = detail::AdmissionState; using OperationGateWord = detail::OperationGateWord; + uint32_t acquire_update_timestamp_until( + std::chrono::steady_clock::time_point deadline) override; void finish_update_and_reset_timeline(uint32_t ts) noexcept override; enum class TimestampReservationState { kReserved, kWindowFull, kExhausted }; @@ -262,11 +266,10 @@ class VersionManager : public IVersionManager { AdmissionState desired_phase); void wait_for_readers_to_drain(); void wait_for_inserters_to_drain(); - bool wait_for_inserters_to_drain(std::optional deadline, - RuntimeBackoff& wait); - bool deadline_expired( - std::optional deadline) const noexcept; + bool wait_for_inserters_to_drain_until( + std::chrono::steady_clock::time_point deadline); TimestampReservation reserve_write_timestamp(); + uint32_t reserve_update_timestamp(); void release_insert_admission(); [[noreturn]] void throw_timestamp_reservation_failure( TimestampReservationState state, uint32_t read_ts, uint32_t write_ts); diff --git a/src/transaction/timestamp_lease.cc b/src/transaction/timestamp_lease.cc index 6dd595756..5ffe78f7b 100644 --- a/src/transaction/timestamp_lease.cc +++ b/src/transaction/timestamp_lease.cc @@ -23,11 +23,17 @@ namespace neug { +UpdateTimestampLease::UpdateTimestampLease(IVersionManager& version_manager) + : version_manager_(&version_manager), + timestamp_(version_manager.acquire_update_timestamp()) { + CHECK_NE(timestamp_, kInactiveTimestamp); +} + UpdateTimestampLease::UpdateTimestampLease( IVersionManager& version_manager, - std::optional deadline) + std::chrono::steady_clock::time_point deadline) : version_manager_(&version_manager), - timestamp_(version_manager.acquire_update_timestamp(deadline)) { + timestamp_(version_manager.acquire_update_timestamp_until(deadline)) { CHECK_NE(timestamp_, kInactiveTimestamp); } diff --git a/src/transaction/version_manager.cc b/src/transaction/version_manager.cc index 36296306c..e868b69a8 100644 --- a/src/transaction/version_manager.cc +++ b/src/transaction/version_manager.cc @@ -27,6 +27,14 @@ namespace neug { +namespace { + +bool DeadlineExpired(std::chrono::steady_clock::time_point deadline) noexcept { + return std::chrono::steady_clock::now() >= deadline; +} + +} // namespace + VersionManager::VersionManager() : runtime_wait_(&NativeRuntimeWait) {} void VersionManager::init_ts(PublishedReadView initial_read_view, @@ -203,10 +211,23 @@ VersionManager::TimestampReservation VersionManager::reserve_write_timestamp() { } const uint32_t current_read_ts = read_ts_.load(std::memory_order_acquire); - DCHECK_GT(candidate, current_read_ts); const uint64_t outstanding = static_cast(candidate) - static_cast(current_read_ts); - if (outstanding > TimestampWindow::kWindowSize) { + if (NEUG_UNLIKELY(outstanding > TimestampWindow::kWindowSize)) { + if (candidate <= current_read_ts) { + // A failed CAS leaves candidate at the then-current write_ts, but + // another inserter may allocate and complete that timestamp before + // this load of read_ts. The acquire load above orders this refresh + // after frontier publication. + candidate = write_ts_.load(std::memory_order_relaxed); + if (NEUG_UNLIKELY(candidate <= current_read_ts)) { + // Keep release builds fail-fast without pulling glog formatting and + // a stack frame into this reservation hot path. + DCHECK_GT(candidate, current_read_ts); + __builtin_trap(); + } + continue; + } return {TimestampReservationState::kWindowFull}; } @@ -323,83 +344,87 @@ void VersionManager::wait_for_readers_to_drain() { } void VersionManager::wait_for_inserters_to_drain() { - const uint64_t observed = - operation_gate_state_.load(std::memory_order_acquire); + uint64_t observed = operation_gate_state_.load(std::memory_order_acquire); if (OperationGateWord::inserters(observed) == 0) { return; } RuntimeBackoff wait = make_runtime_backoff(); - wait_for_inserters_to_drain(std::nullopt, wait); + do { + wait(); + observed = operation_gate_state_.load(std::memory_order_acquire); + } while (OperationGateWord::inserters(observed) != 0); } -bool VersionManager::wait_for_inserters_to_drain( - std::optional deadline, RuntimeBackoff& wait) { +bool VersionManager::wait_for_inserters_to_drain_until( + std::chrono::steady_clock::time_point deadline) { uint64_t observed = operation_gate_state_.load(std::memory_order_acquire); - while (OperationGateWord::inserters(observed) != 0) { - if (deadline_expired(deadline)) { + if (OperationGateWord::inserters(observed) == 0) { + return true; + } + + RuntimeBackoff wait = make_runtime_backoff(); + do { + if (DeadlineExpired(deadline)) { return false; } wait(); observed = operation_gate_state_.load(std::memory_order_acquire); - } - return !deadline_expired(deadline); + } while (OperationGateWord::inserters(observed) != 0); + return true; } RuntimeWaitFn VersionManager::runtime_wait_impl() const noexcept { return runtime_wait_.load(std::memory_order_acquire); } -bool VersionManager::deadline_expired( - std::optional deadline) const noexcept { - return deadline && std::chrono::steady_clock::now() >= *deadline; +uint32_t VersionManager::reserve_update_timestamp() { + const auto reservation = reserve_write_timestamp(); + if (reservation.state == TimestampReservationState::kReserved) { + return reservation.timestamp; + } + const uint32_t current_read_ts = read_ts_.load(std::memory_order_relaxed); + const uint32_t current_write_ts = write_ts_.load(std::memory_order_relaxed); + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + throw_timestamp_reservation_failure(reservation.state, current_read_ts, + current_write_ts); } -uint32_t VersionManager::acquire_update_timestamp( - std::optional deadline) { - if (!deadline) { - // Preserve the legacy auto-commit fast path: no clock read and no backoff - // object unless admission or inserter drain is actually contended. - enter_admission_phase(AdmissionState::kInsertsBlocked); - wait_for_inserters_to_drain(); - } else { - RuntimeBackoff wait = make_runtime_backoff(); - uint64_t observed = operation_gate_state_.load(std::memory_order_relaxed); - while (true) { - if (deadline_expired(deadline)) { - THROW_TRANSACTION_TIMEOUT("waiting for update admission"); - } - if (OperationGateWord::phase(observed) == AdmissionState::kOpen && - OperationGateWord::try_change_phase( - operation_gate_state_, observed, - AdmissionState::kInsertsBlocked)) { - break; - } - wait(); - observed = operation_gate_state_.load(std::memory_order_relaxed); - } +uint32_t VersionManager::acquire_update_timestamp() { + enter_admission_phase(AdmissionState::kInsertsBlocked); + wait_for_inserters_to_drain(); + return reserve_update_timestamp(); +} - if (!wait_for_inserters_to_drain(deadline, wait)) { - transition_admission_phase(AdmissionState::kInsertsBlocked, - AdmissionState::kOpen); - THROW_TRANSACTION_TIMEOUT("waiting for active inserts to finish"); +uint32_t VersionManager::acquire_update_timestamp_until( + std::chrono::steady_clock::time_point deadline) { + RuntimeBackoff admission_wait = make_runtime_backoff(); + uint64_t observed = operation_gate_state_.load(std::memory_order_relaxed); + while (true) { + if (DeadlineExpired(deadline)) { + THROW_TRANSACTION_TIMEOUT("waiting for update admission"); } - if (deadline_expired(deadline)) { - transition_admission_phase(AdmissionState::kInsertsBlocked, - AdmissionState::kOpen); - THROW_TRANSACTION_TIMEOUT("reserving update timestamp"); + if (OperationGateWord::phase(observed) == AdmissionState::kOpen && + OperationGateWord::try_change_phase(operation_gate_state_, observed, + AdmissionState::kInsertsBlocked)) { + break; } + admission_wait(); + observed = operation_gate_state_.load(std::memory_order_relaxed); } - const auto reservation = reserve_write_timestamp(); - if (reservation.state == TimestampReservationState::kReserved) { - return reservation.timestamp; + if (!wait_for_inserters_to_drain_until(deadline)) { + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + THROW_TRANSACTION_TIMEOUT("waiting for active inserts to finish"); } - transition_admission_phase(AdmissionState::kInsertsBlocked, - AdmissionState::kOpen); - throw_timestamp_reservation_failure( - reservation.state, read_ts_.load(std::memory_order_relaxed), - write_ts_.load(std::memory_order_relaxed)); + if (DeadlineExpired(deadline)) { + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + THROW_TRANSACTION_TIMEOUT("reserving update timestamp"); + } + return reserve_update_timestamp(); } void VersionManager::begin_update_commit(uint32_t ts) { diff --git a/tests/transaction/test_read_view_publication.cc b/tests/transaction/test_read_view_publication.cc index 366ed92af..080742359 100644 --- a/tests/transaction/test_read_view_publication.cc +++ b/tests/transaction/test_read_view_publication.cc @@ -98,8 +98,9 @@ class ScriptedVersionManager : public IVersionManager { void release_read_view() override { release_count_.fetch_add(1); } uint32_t acquire_insert_timestamp() override { return 1; } void release_insert_timestamp(uint32_t) override {} - uint32_t acquire_update_timestamp( - std::optional) override { + uint32_t acquire_update_timestamp() override { return 1; } + uint32_t acquire_update_timestamp_until( + std::chrono::steady_clock::time_point) override { return 1; } void begin_update_commit(uint32_t) override {} diff --git a/tests/transaction/test_runtime_wait.cc b/tests/transaction/test_runtime_wait.cc index 9a8d62991..923541034 100644 --- a/tests/transaction/test_runtime_wait.cc +++ b/tests/transaction/test_runtime_wait.cc @@ -409,7 +409,7 @@ TEST(VersionManagerUpdateAdmissionTest, std::chrono::steady_clock::now() + std::chrono::milliseconds(50); std::thread timed_waiter([&] { try { - (void) manager.acquire_update_timestamp(deadline); + UpdateTimestampLease lease(manager, deadline); waiter_timed_out.set_value(false); } catch (const exception::TransactionTimeoutException&) { waiter_timed_out.set_value(true); @@ -448,7 +448,7 @@ TEST(VersionManagerUpdateAdmissionTest, std::chrono::steady_clock::now() + std::chrono::milliseconds(50); std::thread update([&] { try { - (void) manager.acquire_update_timestamp(deadline); + UpdateTimestampLease lease(manager, deadline); timed_out.set_value(false); } catch (const exception::TransactionTimeoutException&) { timed_out.set_value(true); diff --git a/tests/transaction/test_transaction_release_order.cc b/tests/transaction/test_transaction_release_order.cc index 0aaa60c6e..d60a9f5be 100644 --- a/tests/transaction/test_transaction_release_order.cc +++ b/tests/transaction/test_transaction_release_order.cc @@ -60,8 +60,9 @@ class ReleaseOrderVersionManager : public IVersionManager { } PublishedReadView acquire_read_view() override { return {1, 0}; } uint32_t acquire_insert_timestamp() override { return 1; } - uint32_t acquire_update_timestamp( - std::optional) override { + uint32_t acquire_update_timestamp() override { return 1; } + uint32_t acquire_update_timestamp_until( + std::chrono::steady_clock::time_point) override { return 1; } void begin_update_commit(uint32_t) override {} From 7c071121e894498695d9a550727794e6a8658727 Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Thu, 6 Aug 2026 14:08:41 +0800 Subject: [PATCH 5/7] refine --- include/neug/transaction/version_manager.h | 10 -- src/transaction/version_manager.cc | 119 +++++++++++---------- 2 files changed, 64 insertions(+), 65 deletions(-) diff --git a/include/neug/transaction/version_manager.h b/include/neug/transaction/version_manager.h index cca398c59..32f822ac6 100644 --- a/include/neug/transaction/version_manager.h +++ b/include/neug/transaction/version_manager.h @@ -251,13 +251,6 @@ class VersionManager : public IVersionManager { std::chrono::steady_clock::time_point deadline) override; void finish_update_and_reset_timeline(uint32_t ts) noexcept override; - enum class TimestampReservationState { kReserved, kWindowFull, kExhausted }; - - struct TimestampReservation { - TimestampReservationState state; - uint32_t timestamp{0}; - }; - int thread_num_; // These helpers may suspend the logical task. Callers must not hold an // OS-thread-owned lock or retain an ordinary TLS pointer across the call. @@ -268,11 +261,8 @@ class VersionManager : public IVersionManager { void wait_for_inserters_to_drain(); bool wait_for_inserters_to_drain_until( std::chrono::steady_clock::time_point deadline); - TimestampReservation reserve_write_timestamp(); uint32_t reserve_update_timestamp(); void release_insert_admission(); - [[noreturn]] void throw_timestamp_reservation_failure( - TimestampReservationState state, uint32_t read_ts, uint32_t write_ts); void complete_write_timestamp(uint32_t ts); void advance_read_ts_locked(); RuntimeWaitFn runtime_wait_impl() const noexcept override; diff --git a/src/transaction/version_manager.cc b/src/transaction/version_manager.cc index e868b69a8..1bc483baf 100644 --- a/src/transaction/version_manager.cc +++ b/src/transaction/version_manager.cc @@ -33,6 +33,67 @@ bool DeadlineExpired(std::chrono::steady_clock::time_point deadline) noexcept { return std::chrono::steady_clock::now() >= deadline; } +enum class TimestampReservationState { kReserved, kWindowFull, kExhausted }; + +struct TimestampReservation { + TimestampReservationState state; + uint32_t timestamp{0}; +}; + +TimestampReservation ReserveWriteTimestamp( + std::atomic& write_ts, const std::atomic& read_ts) { + uint32_t candidate = write_ts.load(std::memory_order_relaxed); + while (true) { + if (candidate == std::numeric_limits::max()) { + return {TimestampReservationState::kExhausted}; + } + + const uint32_t current_read_ts = read_ts.load(std::memory_order_acquire); + const uint64_t outstanding = static_cast(candidate) - + static_cast(current_read_ts); + if (NEUG_UNLIKELY(outstanding > TimestampWindow::kWindowSize)) { + if (candidate <= current_read_ts) { + // A failed CAS leaves candidate at the then-current write_ts, but + // another inserter may allocate and complete that timestamp before + // this load of read_ts. The acquire load above orders this refresh + // after frontier publication. + candidate = write_ts.load(std::memory_order_relaxed); + if (NEUG_UNLIKELY(candidate <= current_read_ts)) { + // Keep release builds fail-fast without pulling glog formatting and + // a stack frame into this reservation hot path. + DCHECK_GT(candidate, current_read_ts); + __builtin_trap(); + } + continue; + } + return {TimestampReservationState::kWindowFull}; + } + + if (write_ts.compare_exchange_weak(candidate, candidate + 1, + std::memory_order_acq_rel, + std::memory_order_relaxed)) { + return {TimestampReservationState::kReserved, candidate}; + } + RuntimeCpuRelax(); + } +} + +[[noreturn]] void throw_timestamp_reservation_failure( + TimestampReservationState state, uint32_t read_ts, uint32_t write_ts) { + if (state == TimestampReservationState::kWindowFull) { + THROW_INTERNAL_EXCEPTION( + "TimestampWindow invariant broken: write timestamp reservation found " + "the window full despite exclusive write admission (read_ts=" + + std::to_string(read_ts) + ", write_ts=" + std::to_string(write_ts) + + ", window_size=" + std::to_string(TimestampWindow::kWindowSize) + + "); this indicates admission/window bookkeeping corruption, not " + "recoverable backpressure"); + } + THROW_RUNTIME_ERROR( + "Transaction timestamp space exhausted; checkpoint/reset the timeline " + "before reopening the database"); +} + } // namespace VersionManager::VersionManager() : runtime_wait_(&NativeRuntimeWait) {} @@ -161,7 +222,7 @@ uint32_t VersionManager::acquire_insert_timestamp() { if (operation_gate_state_.compare_exchange_weak( observed, desired, std::memory_order_acquire, std::memory_order_relaxed)) { - const auto reservation = reserve_write_timestamp(); + const auto reservation = ReserveWriteTimestamp(write_ts_, read_ts_); if (reservation.state == TimestampReservationState::kReserved) { return reservation.timestamp; } @@ -203,58 +264,6 @@ void VersionManager::release_insert_admission() { DCHECK_GT(OperationGateWord::inserters(previous), 0U); } -VersionManager::TimestampReservation VersionManager::reserve_write_timestamp() { - uint32_t candidate = write_ts_.load(std::memory_order_relaxed); - while (true) { - if (candidate == std::numeric_limits::max()) { - return {TimestampReservationState::kExhausted}; - } - - const uint32_t current_read_ts = read_ts_.load(std::memory_order_acquire); - const uint64_t outstanding = static_cast(candidate) - - static_cast(current_read_ts); - if (NEUG_UNLIKELY(outstanding > TimestampWindow::kWindowSize)) { - if (candidate <= current_read_ts) { - // A failed CAS leaves candidate at the then-current write_ts, but - // another inserter may allocate and complete that timestamp before - // this load of read_ts. The acquire load above orders this refresh - // after frontier publication. - candidate = write_ts_.load(std::memory_order_relaxed); - if (NEUG_UNLIKELY(candidate <= current_read_ts)) { - // Keep release builds fail-fast without pulling glog formatting and - // a stack frame into this reservation hot path. - DCHECK_GT(candidate, current_read_ts); - __builtin_trap(); - } - continue; - } - return {TimestampReservationState::kWindowFull}; - } - - if (write_ts_.compare_exchange_weak(candidate, candidate + 1, - std::memory_order_acq_rel, - std::memory_order_relaxed)) { - return {TimestampReservationState::kReserved, candidate}; - } - } -} - -[[noreturn]] void VersionManager::throw_timestamp_reservation_failure( - TimestampReservationState state, uint32_t read_ts, uint32_t write_ts) { - if (state == TimestampReservationState::kWindowFull) { - THROW_INTERNAL_EXCEPTION( - "TimestampWindow invariant broken: write timestamp reservation found " - "the window full despite exclusive write admission (read_ts=" + - std::to_string(read_ts) + ", write_ts=" + std::to_string(write_ts) + - ", window_size=" + std::to_string(TimestampWindow::kWindowSize) + - "); this indicates admission/window bookkeeping corruption, not " - "recoverable backpressure"); - } - THROW_RUNTIME_ERROR( - "Transaction timestamp space exhausted; checkpoint/reset the timeline " - "before reopening the database"); -} - void VersionManager::complete_write_timestamp(uint32_t ts) { // Mark completion (lock-free atomic operation) ts_window_.mark_completed(ts); @@ -379,7 +388,7 @@ RuntimeWaitFn VersionManager::runtime_wait_impl() const noexcept { } uint32_t VersionManager::reserve_update_timestamp() { - const auto reservation = reserve_write_timestamp(); + const auto reservation = ReserveWriteTimestamp(write_ts_, read_ts_); if (reservation.state == TimestampReservationState::kReserved) { return reservation.timestamp; } @@ -497,7 +506,7 @@ uint32_t VersionManager::acquire_compact_timestamp() { wait_for_readers_to_drain(); wait_for_inserters_to_drain(); - const auto reservation = reserve_write_timestamp(); + const auto reservation = ReserveWriteTimestamp(write_ts_, read_ts_); if (reservation.state == TimestampReservationState::kReserved) { return reservation.timestamp; } From 2b189904c968d84b351bc18fb89fc5677d131692 Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Thu, 6 Aug 2026 14:14:30 +0800 Subject: [PATCH 6/7] refine doc --- include/neug/transaction/README.md | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/include/neug/transaction/README.md b/include/neug/transaction/README.md index 8224c385b..800f8da4e 100644 --- a/include/neug/transaction/README.md +++ b/include/neug/transaction/README.md @@ -126,10 +126,14 @@ appends WAL before publishing its COW snapshot. Both complete their timestamps only after the graph change is visible. Update waiters directly contend the existing admission phase; acquisition order -is unspecified. A caller may provide an absolute `steady_clock` deadline when -acquiring an update timestamp; expiry before a timestamp is reserved returns -`ERR_TX_TIMEOUT` and restores any phase acquired by that attempt. Legacy callers -provide no deadline and retain infinite-wait behavior. +is unspecified. The public manager API retains its no-deadline fast path. +The deadline overload of `UpdateTimestampLease` invokes a private lease-only +manager hook. If its absolute `steady_clock` deadline expires before timestamp +reservation, lease construction reports `ERR_TX_TIMEOUT` and restores any phase +acquired by that attempt. Admission contention and inserter draining use separate +backoff cursors. Existing production callers retain infinite-wait behavior and +do not read the clock; future explicit-transaction integration will pass its +write-wait deadline through this overload. When `VersionManager::begin_update_commit` is called, the admission state changes from `kInsertsBlocked` to `kAllBlocked`. New reads and new inserts are blocked until the `UpdateTransaction` is committed or aborted. Already-acquired reads continue unaffected on their pinned snapshot. From 33e9133cdcdf1f5d9b90164fcdc5afa00f470697 Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Thu, 6 Aug 2026 14:36:00 +0800 Subject: [PATCH 7/7] minor fix --- src/transaction/version_manager.cc | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/transaction/version_manager.cc b/src/transaction/version_manager.cc index 1bc483baf..c65d05d8b 100644 --- a/src/transaction/version_manager.cc +++ b/src/transaction/version_manager.cc @@ -59,10 +59,11 @@ TimestampReservation ReserveWriteTimestamp( // after frontier publication. candidate = write_ts.load(std::memory_order_relaxed); if (NEUG_UNLIKELY(candidate <= current_read_ts)) { - // Keep release builds fail-fast without pulling glog formatting and - // a stack frame into this reservation hot path. - DCHECK_GT(candidate, current_read_ts); - __builtin_trap(); + THROW_INTERNAL_EXCEPTION( + "Write timestamp reservation invariant broken after refresh: " + "write_ts=" + + std::to_string(candidate) + " must be greater than read_ts=" + + std::to_string(current_read_ts)); } continue; }