diff --git a/include/neug/transaction/README.md b/include/neug/transaction/README.md index 31eb2b033..800f8da4e 100644 --- a/include/neug/transaction/README.md +++ b/include/neug/transaction/README.md @@ -124,3 +124,28 @@ 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 directly contend the existing admission phase; acquisition order +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. + +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. diff --git a/include/neug/transaction/timestamp_lease.h b/include/neug/transaction/timestamp_lease.h index 8ceffa3a4..a87316e28 100644 --- a/include/neug/transaction/timestamp_lease.h +++ b/include/neug/transaction/timestamp_lease.h @@ -15,6 +15,7 @@ #pragma once #include +#include #include namespace neug { @@ -31,6 +32,8 @@ class IVersionManager; class UpdateTimestampLease { public: 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/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..32f822ac6 100644 --- a/include/neug/transaction/version_manager.h +++ b/include/neug/transaction/version_manager.h @@ -16,6 +16,7 @@ #include #include +#include #include #include "neug/transaction/runtime_wait.h" @@ -79,6 +80,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() = 0; virtual void begin_update_commit(uint32_t ts) = 0; // May invoke the runtime waiter. Checkpoint callers must enter commit and @@ -102,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. @@ -239,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; int thread_num_; @@ -249,6 +259,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_until( + std::chrono::steady_clock::time_point deadline); + uint32_t reserve_update_timestamp(); + void release_insert_admission(); void complete_write_timestamp(uint32_t ts); void advance_read_ts_locked(); RuntimeWaitFn runtime_wait_impl() const noexcept override; 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..5ffe78f7b 100644 --- a/src/transaction/timestamp_lease.cc +++ b/src/transaction/timestamp_lease.cc @@ -29,6 +29,14 @@ UpdateTimestampLease::UpdateTimestampLease(IVersionManager& version_manager) CHECK_NE(timestamp_, kInactiveTimestamp); } +UpdateTimestampLease::UpdateTimestampLease( + IVersionManager& version_manager, + std::chrono::steady_clock::time_point deadline) + : version_manager_(&version_manager), + timestamp_(version_manager.acquire_update_timestamp_until(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..c65d05d8b 100644 --- a/src/transaction/version_manager.cc +++ b/src/transaction/version_manager.cc @@ -16,6 +16,7 @@ #include "neug/transaction/version_manager.h" #include +#include #include #include #include @@ -26,7 +27,75 @@ namespace neug { -// VersionManager implementation +namespace { + +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)) { + 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; + } + 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) {} @@ -154,7 +223,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 = ReserveWriteTimestamp(write_ts_, read_ts_); + 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 +249,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)) { @@ -210,8 +301,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)}), @@ -277,15 +366,75 @@ void VersionManager::wait_for_inserters_to_drain() { } while (OperationGateWord::inserters(observed) != 0); } +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); + 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); + } while (OperationGateWord::inserters(observed) != 0); + return true; +} + RuntimeWaitFn VersionManager::runtime_wait_impl() const noexcept { return runtime_wait_.load(std::memory_order_acquire); } +uint32_t VersionManager::reserve_update_timestamp() { + const auto reservation = ReserveWriteTimestamp(write_ts_, read_ts_); + 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() { enter_admission_phase(AdmissionState::kInsertsBlocked); wait_for_inserters_to_drain(); + return reserve_update_timestamp(); +} - return write_ts_.fetch_add(1, std::memory_order_acq_rel); +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 (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); + } + + if (!wait_for_inserters_to_drain_until(deadline)) { + transition_admission_phase(AdmissionState::kInsertsBlocked, + AdmissionState::kOpen); + THROW_TRANSACTION_TIMEOUT("waiting for active inserts to finish"); + } + 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) { @@ -358,7 +507,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 = ReserveWriteTimestamp(write_ts_, read_ts_); + 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..080742359 100644 --- a/tests/transaction/test_read_view_publication.cc +++ b/tests/transaction/test_read_view_publication.cc @@ -99,6 +99,10 @@ class ScriptedVersionManager : public IVersionManager { 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_until( + std::chrono::steady_clock::time_point) 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..923541034 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,10 @@ 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::mutex g_blocking_wait_lock; +std::condition_variable g_blocking_wait_cv; +bool g_block_waiters = false; void RecordRuntimeWait(RuntimeWaitAction action) noexcept { g_runtime_wait_calls.fetch_add(1, std::memory_order_relaxed); @@ -60,6 +68,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 +127,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 +393,139 @@ TEST(VersionManagerWaitTest, AllContendedPathsUseBackoff) { } } +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 waiter_timed_out; + std::promise successor_timestamp; + auto timeout_result = waiter_timed_out.get_future(); + auto successor_result = successor_timestamp.get_future(); + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(50); + std::thread timed_waiter([&] { + try { + UpdateTimestampLease lease(manager, deadline); + waiter_timed_out.set_value(false); + } catch (const exception::TransactionTimeoutException&) { + waiter_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)); + + std::this_thread::sleep_until(deadline); + ReleaseRuntimeWaiters(); + 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); + timed_waiter.join(); + successor.join(); +} + +TEST(VersionManagerUpdateAdmissionTest, + InserterDrainTimeoutRestoresAdmissionWithoutTimestamp) { + VersionManager manager; + 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(); + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(50); + std::thread update([&] { + try { + UpdateTimestampLease lease(manager, deadline); + timed_out.set_value(false); + } catch (const exception::TransactionTimeoutException&) { + timed_out.set_value(true); + } + }); + ASSERT_TRUE(WaitForBlockedWaiters(1)); + + std::this_thread::sleep_until(deadline); + 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..d60a9f5be 100644 --- a/tests/transaction/test_transaction_release_order.cc +++ b/tests/transaction/test_transaction_release_order.cc @@ -61,6 +61,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_until( + std::chrono::steady_clock::time_point) override { + return 1; + } void begin_update_commit(uint32_t) override {} void drain_readers() override {} void finish_update_timestamp(uint32_t, @@ -214,6 +218,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);