diff --git a/README.md b/README.md index e57af1d..2298511 100644 --- a/README.md +++ b/README.md @@ -147,13 +147,8 @@ We currently support: Currently missing/under development: -- Removing deletion tombstones from checkpoint trees -- Recovery of existing KVStore instances after shutdown - Time-consistent isolated multi-key reads -- Key-only queries (i.e., `has_key`/`contains`; note, key-only scans - *are* currently supported) - Saving/loading full point-in-time snapshots -- `fsync` like blocking call to ensure updates are durable (in the WAL) Limitations (out-of-scope): diff --git a/conan.lock b/conan.lock index 13b562b..64fcade 100644 --- a/conan.lock +++ b/conan.lock @@ -2,7 +2,7 @@ "version": "0.5", "requires": [ "abseil/20250127.0", - "batteries/0.70.2", + "batteries/0.71.1", "boost/1.88.0", "bzip2/1.0.8", "cli11/2.5.0", @@ -15,7 +15,7 @@ "libpfm4/4.13.0", "libunwind/1.8.1", "liburing/2.11", - "llfs/0.44.0", + "llfs/0.46.0", "openssl/3.6.0", "pcg-cpp/cci.20220409", "protobuf/3.21.12", diff --git a/conanfile.py b/conanfile.py index 31cc052..74989f2 100644 --- a/conanfile.py +++ b/conanfile.py @@ -60,6 +60,10 @@ class TurtleKvRecipe(ConanFile): "profile_queries": True, } + package_id_embed_mode = "full_mode" + package_id_non_embed_mode = "full_mode" + package_id_unknown_mode = "full_mode" + #+++++++++++-+-+--+----- --- -- - - - - # Optional metadata # @@ -85,10 +89,10 @@ def requirements(self): } self.requires("abseil/20250127.0", **VISIBLE, **OVERRIDE) - self.requires("batteries/[>=0.70.2 <1]", **VISIBLE, **OVERRIDE) + self.requires("batteries/[>=0.71.1 <1]", **VISIBLE, **OVERRIDE) self.requires("boost/1.88.0", **VISIBLE, **OVERRIDE) self.requires("glog/0.7.1", **VISIBLE) - self.requires("llfs/[>=0.44.0 <1]", **VISIBLE) + self.requires("llfs/[>=0.46.0 <1]", **VISIBLE) self.requires("pcg-cpp/cci.20220409", **VISIBLE) self.requires("yaml-cpp/[>=0.9.0 <1]") self.requires("zlib/1.3.1", **OVERRIDE) diff --git a/src/turtle_kv/change_log/change_log.test.cpp b/src/turtle_kv/change_log/change_log.test.cpp index 2ff54d3..ca42822 100644 --- a/src/turtle_kv/change_log/change_log.test.cpp +++ b/src/turtle_kv/change_log/change_log.test.cpp @@ -18,9 +18,11 @@ #include #include +#include #include #include +#include #include #include #include @@ -28,13 +30,29 @@ namespace turtle_kv { +// TODO [tastolfi 2026-06-18] Move this to batteries. +// +#define ASSERT_OK(convertible_to_status) \ + ASSERT_NO_FATAL_FAILURE([&](Status status) { \ + ASSERT_TRUE(status.ok()) << BATT_INSPECT(status); \ + }(batt::to_status((convertible_to_status)))) + class ChangeLogTest : public ::testing::Test { protected: + using AppendCallback = std::function; + + //+++++++++++-+-+--+----- --- -- - - - - + void SetUp() override { + this->test_main_thread_ = std::this_thread::get_id(); + batt::StatusOr root = turtle_kv::data_root(); - ASSERT_TRUE(root.ok()); + ASSERT_OK(root); this->test_dir_ = *root / "turtle_kv_Test"; this->test_file_ = this->test_dir_ / "test_change_log.log"; @@ -47,151 +65,237 @@ class ChangeLogTest : public ::testing::Test return; } + /** \brief Creates the configured change log file. + */ + Status create_log_file(RemoveExisting remove_existing) + { + return ChangeLogFile::create(this->test_file_, this->config_, remove_existing); + } + + /** \brief Opens an already-created log file. + */ + Status open_log_file() + { + BATT_ASSIGN_OK_RESULT(this->log_file_, ChangeLogFile::open(this->test_file_)); + return OkStatus(); + } + + /** \brief Creates and returns a `ChangeLogWriter` instance. + */ + Status create_writer(RemoveExisting remove_existing) + { + BATT_ASSIGN_OK_RESULT(this->writer_, + ChangeLogWriter::open_or_create(this->test_file_, + this->config_, + this->writer_options_, + remove_existing)); + + this->writer_->start(batt::Runtime::instance().default_scheduler().schedule_task()); + + return OkStatus(); + } + + /** \brief Halts the specified `ChangeLogWriter` instance. If specified, this function first waits + * for the writer to process appends before halting. + */ + void shutdown_writer(bool flush) + { + BATT_CHECK_NOT_NULLPTR(this->writer_); + + if (this->context_) { + this->context_ = None; + } + + if (flush) { + // Wait for writer to process appends before halting. + // + ASSERT_TRUE(this->writer_->wait_for_flush()); + } + + this->writer_->halt(); + this->writer_->join(); + } + + /** \brief Appends the payload in `data` to a new slot within some `BlockBuffer` owned by the + * specified `context`. Optionally takes in a callback function that would execute after the + * data is copied into the slot when specified. + */ + Status append_slot(ChangeLogWriter::Context& context, + const std::string_view& data, + batt::WaitForResource wait_for_resource = batt::WaitForResource::kTrue, + Optional callback_fn = None) + { + return context.append_slot( + this->min_edit_offset_lower_bound_, + data.size(), + wait_for_resource, + [&data, &callback_fn](FirstVisitToBlock first_visit, + ChangeLogBlock* block, + MutableBuffer buffer, + EditOffset offset) { + VLOG(1) << "Appending block with lower_bound: " << block->edit_offset_lower_bound() + << ", on slot: " << offset << "\n" + << BATT_INSPECT(first_visit) << BATT_INSPECT(block->slot_count()) + << BATT_INSPECT(block->edit_offset_range()); + + std::memcpy(buffer.data(), data.data(), data.size()); + + if (callback_fn) { + (*callback_fn)(first_visit, block, buffer, offset); + } + }); + } + + /** \brief Appends a slot from the main test thread, lazily creating a context the first time if + * necessary. + */ + Status append_slot(const std::string_view& data, + batt::WaitForResource wait_for_resource = batt::WaitForResource::kTrue, + Optional callback_fn = None) + { + BATT_CHECK_EQ(this->test_main_thread_, std::this_thread::get_id()) + << "The no-Context overload of append_slot may *only* be called from the main test thread!"; + + if (!this->context_) { + BATT_CHECK_NOT_NULLPTR(this->writer_); + this->context_.emplace(*this->writer_); + } + return this->append_slot(*this->context_, data, wait_for_resource, callback_fn); + } + + /** \brief Opens a `ChangeLogReader` instance and visits slots with the `visitor_fn` function + * specified. + * + * \return the number of slots recovered and visited. + */ + template + requires std::invocable && + std::same_as, + Status> + usize open_reader_and_visit(VisitorFn&& visitor_fn) + { + StatusOr> reader = ChangeLogReader::open(this->test_file_); + BATT_CHECK_OK(reader); + + usize slots_read = 0; + auto counting_visitor = [&](FirstVisitToBlock first_visit, + ChangeLogBlock* block, + EditOffset edit_offset, + ConstBuffer payload) -> Status { + ++slots_read; + return visitor_fn(slots_read, first_visit, block, edit_offset, payload); + }; + + batt::Status visit_status = (*reader)->visit_slots(counting_visitor).status(); + BATT_CHECK_OK(visit_status); + + return slots_read; + } + //+++++++++++-+-+--+----- --- -- - - - - + std::thread::id test_main_thread_; std::filesystem::path test_dir_; std::filesystem::path test_file_; + ChangeLogFile::Config config_ = ChangeLogFile::Config::with_default_values(); + std::unique_ptr log_file_; + ChangeLogWriter::Options writer_options_ = ChangeLogWriter::Options::with_default_values(); + std::unique_ptr writer_; + Optional context_; + EditOffset min_edit_offset_lower_bound_{0}; }; //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // TEST_F(ChangeLogTest, CreateAndOpenFile) { - ChangeLogFile::Config config = ChangeLogFile::Config::with_default_values(); - Status status = ChangeLogFile::create(this->test_file_, config, RemoveExisting{true}); - ASSERT_TRUE(status.ok()) << BATT_INSPECT(status) << BATT_INSPECT(this->test_file_); - - StatusOr> log_file = ChangeLogFile::open(this->test_file_); - ASSERT_TRUE(log_file.ok()); - ASSERT_NE(log_file->get(), nullptr); + ASSERT_OK(this->create_log_file(RemoveExisting{true})); + ASSERT_OK(this->open_log_file()); - EXPECT_EQ((*log_file)->config().block_size, config.block_size); - EXPECT_EQ((*log_file)->config().block_count, config.block_count); - EXPECT_EQ((*log_file)->config().block0_offset, config.block0_offset); + EXPECT_EQ(this->log_file_->config().block_size, this->config_.block_size); + EXPECT_EQ(this->log_file_->config().block_count, this->config_.block_count); + EXPECT_EQ(this->log_file_->config().block0_offset, this->config_.block0_offset); } //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // TEST_F(ChangeLogTest, WriterBasicOperations) { - ChangeLogFile::Config config = ChangeLogFile::Config::with_default_values(); - ChangeLogWriter::Options options = ChangeLogWriter::Options::with_default_values(); - - StatusOr> writer = - ChangeLogWriter::open_or_create(this->test_file_, config, options, RemoveExisting{true}); - ASSERT_TRUE(writer.ok()); - - (*writer)->start(batt::Runtime::instance().default_scheduler().schedule_task()); - - ChangeLogWriter::Context context(**writer); + ASSERT_OK(this->create_writer(RemoveExisting{true})); // Write some test data // std::string test_data = "Hello, ChangeLog!"; - Status write_status = context.append_slot( - /*min_edit_offset_lower_bound=*/EditOffset{0}, - test_data.size(), - [&test_data, this](FirstVisitToBlock first_visit, - ChangeLogBlock* block, - MutableBuffer buffer, - EditOffset offset) { - VLOG(1) << "Appending block with lower_bound: " << block->edit_offset_lower_bound() - << ", on slot: " << offset; - VLOG(1) << BATT_INSPECT(first_visit) << BATT_INSPECT(block->slot_count()) - << BATT_INSPECT(block->edit_offset_range()); - std::memcpy(buffer.data(), test_data.data(), test_data.size()); - }); - ASSERT_TRUE(write_status.ok()); - - // Wait for writer to process appends before halting. - // - ASSERT_TRUE((*writer)->wait_for_flush()); + ASSERT_OK(this->append_slot(test_data)); - (*writer)->halt(); - (*writer)->join(); + this->shutdown_writer(/*flush=*/true); - EXPECT_GT((*writer)->metrics().received_user_byte_count.load(), 0); + EXPECT_GT(this->writer_->metrics().received_user_byte_count.load(), 0); } //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // TEST_F(ChangeLogTest, WriteAndReadMultipleSlots) { - ChangeLogFile::Config config = ChangeLogFile::Config::with_default_values(); - config.block_size = BlockSize{4096}; - config.block_count = BlockCount{10}; - std::vector test_data = {"First slot data", - "Second slot data with more content", - "Third slot", - "Fourth slot with even more data to test", - "Fifth and final slot"}; + this->config_.block_size = BlockSize{4096}; + this->config_.block_count = BlockCount{10}; + + std::vector test_data = { + "First slot data", + "Second slot data with more content", + "Third slot", + "Fourth slot with even more data to test", + "Fifth and final slot", + }; // Write phase // { - StatusOr> writer = - ChangeLogWriter::open_or_create(this->test_file_, - config, - ChangeLogWriter::Options::with_default_values(), - RemoveExisting{true}); - ASSERT_TRUE(writer.ok()); - - (*writer)->start(batt::Runtime::instance().default_scheduler().schedule_task()); - - ChangeLogWriter::Context context(**writer); + ASSERT_OK(this->create_writer(RemoveExisting{true})); // Write multiple slots // for (size_t i = 0; i < test_data.size(); ++i) { - Status write_status = context.append_slot( - /*min_edit_offset_lower_bound=*/EditOffset{0}, - test_data[i].size(), - [&data = test_data[i], this](FirstVisitToBlock first_visit, - ChangeLogBlock* block, - MutableBuffer buffer, - EditOffset offset) { - VLOG(1) << "Appending block with lower_bound: " << block->edit_offset_lower_bound() - << ", on slot: " << offset; - VLOG(1) << BATT_INSPECT(first_visit) << BATT_INSPECT(block->slot_count()) - << BATT_INSPECT(block->edit_offset_range()); - std::memcpy(buffer.data(), data.data(), data.size()); - }); - ASSERT_TRUE(write_status.ok()) << "Failed to write slot " << i; + Status write_status = this->append_slot(test_data[i]); + ASSERT_TRUE(write_status.ok()) << "Failed to write slot " << i << BATT_INSPECT(write_status); } - // Wait for writer to process appends before halting. - // - ASSERT_TRUE((*writer)->wait_for_flush()); - - (*writer)->halt(); - (*writer)->join(); + this->shutdown_writer(/*flush=*/true); } // Read phase // { - StatusOr> reader = ChangeLogReader::open(this->test_file_); - ASSERT_TRUE(reader.ok()); - std::vector read_data; std::vector edit_offsets; - auto visitor_fn = [&](FirstVisitToBlock first_visit, - ChangeLogBlock* block, - EditOffset edit_offset, - ConstBuffer payload) -> Status { + usize slots_read = this->open_reader_and_visit([&](usize, + FirstVisitToBlock first_visit, + ChangeLogBlock* block, + EditOffset edit_offset, + ConstBuffer payload) -> Status { VLOG(1) << "Reading block with lower_bound: " << block->edit_offset_lower_bound() - << ", on slot: " << edit_offset; - VLOG(1) << BATT_INSPECT(first_visit) << BATT_INSPECT(block->slot_count()) + << ", on slot: " << edit_offset << "\n" + << BATT_INSPECT(first_visit) << BATT_INSPECT(block->slot_count()) << BATT_INSPECT(block->edit_offset_range()); - std::string data(reinterpret_cast(payload.data()), payload.size()); - read_data.push_back(data); + + read_data.emplace_back(static_cast(payload.data()), payload.size()); edit_offsets.push_back(edit_offset); + return OkStatus(); - }; + }); - batt::Status visit_status = (*reader)->visit_slots(visitor_fn).status(); - ASSERT_TRUE(visit_status.ok()) << BATT_INSPECT(visit_status); + EXPECT_EQ(slots_read, 5); // Verify we read all slots. // @@ -215,50 +319,42 @@ TEST_F(ChangeLogTest, WriteAndReadMultipleSlots) // TEST_F(ChangeLogTest, ConcurrentWritesMultipleContexts) { - ChangeLogFile::Config config = ChangeLogFile::Config::with_default_values(); - config.block_count = BlockCount{20}; const int num_threads = 4; const int slots_per_thread = 10; + const int total_slots_expected = num_threads * slots_per_thread; + batt::Mutex> offsets; // Write Phase // { - StatusOr> writer = - ChangeLogWriter::open_or_create(this->test_file_, - config, - ChangeLogWriter::Options::with_default_values(), - RemoveExisting{true}); - ASSERT_TRUE(writer.ok()); + // Size the log for worst-case scenario (every slot is written to its own block). + // + this->config_.block_count = BlockCount{total_slots_expected}; - (*writer)->start(batt::Runtime::instance().default_scheduler().schedule_task()); + ASSERT_OK(this->create_writer(RemoveExisting{true})); std::vector threads; - std::atomic total_writes{0}; + std::atomic append_ok_count{0}; for (int t = 0; t < num_threads; ++t) { threads.emplace_back([&, thread_id = t]() { - ChangeLogWriter::Context context(**writer); + ChangeLogWriter::Context context{*this->writer_}; for (int i = 0; i < slots_per_thread; ++i) { - std::string data = "Thread " + std::to_string(thread_id) + " Slot " + std::to_string(i); + std::string data = batt::to_string("Thread ", thread_id, " Slot ", i); - Status write_status = context.append_slot( - /*min_edit_offset_lower_bound=*/EditOffset{0}, - data.size(), - [&data, &offsets](FirstVisitToBlock, - ChangeLogBlock* block, - MutableBuffer buffer, - EditOffset offset) { - VLOG(1) << "Appending block with lower_bound: " << block->edit_offset_lower_bound() - << ", on slot: " << offset << BATT_INSPECT(data.size()); + Status write_status = this->append_slot( + context, + data, + batt::WaitForResource::kTrue, + [&offsets](FirstVisitToBlock, ChangeLogBlock*, MutableBuffer, EditOffset offset) { batt::ScopedLock> locked_offsets{offsets}; locked_offsets->insert(offset.value()); - std::memcpy(buffer.data(), data.data(), data.size()); }); if (write_status.ok()) { - total_writes.fetch_add(1); + append_ok_count.fetch_add(1); } } }); @@ -268,32 +364,22 @@ TEST_F(ChangeLogTest, ConcurrentWritesMultipleContexts) t.join(); } - // Wait for writer to process appends before halting. - // - ASSERT_TRUE((*writer)->wait_for_flush()); - - (*writer)->halt(); - (*writer)->join(); + this->shutdown_writer(/*flush=*/true); - EXPECT_EQ(total_writes.load(), num_threads * slots_per_thread); + EXPECT_EQ(append_ok_count.load(), total_slots_expected); } // Read Phase // { - StatusOr> reader = ChangeLogReader::open(this->test_file_); - ASSERT_TRUE(reader.ok()); - - int slots_read = 0; - auto visitor_fn = [&](FirstVisitToBlock, - ChangeLogBlock* block, - EditOffset edit_offset, - ConstBuffer payload) -> Status { + usize slots_read = this->open_reader_and_visit([&](usize, + FirstVisitToBlock, + ChangeLogBlock* block, + EditOffset edit_offset, + ConstBuffer payload) -> Status { VLOG(1) << "Reading block with lower_bound: " << block->edit_offset_lower_bound() << ", on slot: " << edit_offset << ", payload size: " << payload.size(); - slots_read++; - batt::ScopedLock> locked_offsets{offsets}; // Check that edit_offset was in the set of offsets we wrote @@ -301,10 +387,7 @@ TEST_F(ChangeLogTest, ConcurrentWritesMultipleContexts) BATT_REQUIRE_NE(locked_offsets->find(edit_offset.value()), locked_offsets->end()); locked_offsets->erase(edit_offset.value()); return OkStatus(); - }; - - batt::Status visit_status = (*reader)->visit_slots(visitor_fn).status(); - ASSERT_TRUE(visit_status.ok()) << BATT_INSPECT(visit_status); + }); EXPECT_EQ(slots_read, num_threads * slots_per_thread); @@ -319,89 +402,53 @@ TEST_F(ChangeLogTest, ConcurrentWritesMultipleContexts) // TEST_F(ChangeLogTest, BlockBoundaryConditions) { - ChangeLogFile::Config config = ChangeLogFile::Config::with_default_values(); - config.block_size = BlockSize{1024}; // Small blocks to test boundaries - config.block_count = BlockCount{5}; - int num_appends = 6; - - StatusOr> writer = - ChangeLogWriter::open_or_create(this->test_file_, - config, - ChangeLogWriter::Options::with_default_values(), - RemoveExisting{true}); - ASSERT_TRUE(writer.ok()); - - (*writer)->start(batt::Runtime::instance().default_scheduler().schedule_task()); + const int num_appends = 6; - ChangeLogWriter::Context context(**writer); + this->config_.block_size = BlockSize{1024}; // Small blocks to test boundaries + this->config_.block_count = BlockCount{5}; + ASSERT_OK(this->create_writer(RemoveExisting{true})); // Write data that will span multiple blocks // std::string large_data(900, 'X'); // Almost fills a block for (int i = 0; i < num_appends; ++i) { - Status write_status = context.append_slot( - /*min_edit_offset_lower_bound=*/EditOffset{0}, - large_data.size(), - [&large_data, &writer, i, this](FirstVisitToBlock first_visit, - ChangeLogBlock* block, - MutableBuffer buffer, - EditOffset offset) { - VLOG(1) << "Appending block with lower_bound: " << block->edit_offset_lower_bound() - << ", on slot: " << offset; - VLOG(1) << BATT_INSPECT(first_visit) << BATT_INSPECT(block->slot_count()) - << BATT_INSPECT(block->edit_offset_range()); - std::memcpy(buffer.data(), large_data.data(), large_data.size()); - - (*writer)->trim(offset + EditOffsetDelta{(i64)large_data.size()}).IgnoreError(); + Status write_status = this->append_slot( + large_data, + batt::WaitForResource::kTrue, + [&, i](FirstVisitToBlock, ChangeLogBlock*, MutableBuffer, EditOffset offset) { + this->writer_->trim(offset + EditOffsetDelta{(i64)large_data.size()}).IgnoreError(); }); ASSERT_TRUE(write_status.ok()) << BATT_INSPECT(write_status); } - // Wait for writer to process appends before halting. - // - ASSERT_TRUE((*writer)->wait_for_flush()); - - (*writer)->halt(); - (*writer)->join(); + this->shutdown_writer(/*flush=*/true); } //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // TEST_F(ChangeLogTest, ReadEmptyLog) { - ChangeLogFile::Config config = ChangeLogFile::Config::with_default_values(); - - StatusOr> writer = - ChangeLogWriter::open_or_create(this->test_file_, - config, - ChangeLogWriter::Options::with_default_values(), - RemoveExisting{true}); - ASSERT_TRUE(writer.ok()); + ASSERT_OK(this->create_writer(RemoveExisting{true})); // Don't write anything, just close // - writer->reset(); + ASSERT_NE(this->writer_, nullptr); + this->writer_ = nullptr; // Try to read // - StatusOr> reader = ChangeLogReader::open(this->test_file_); - ASSERT_TRUE(reader.ok()); - - int slots_read = 0; - auto visitor_fn = [&](FirstVisitToBlock, - ChangeLogBlock* block, - EditOffset edit_offset, - ConstBuffer payload) -> Status { + + usize slots_read = this->open_reader_and_visit([&](usize, + FirstVisitToBlock, + ChangeLogBlock* block, + EditOffset edit_offset, + ConstBuffer payload) -> Status { VLOG(1) << "Reading block with lower_bound: " << block->edit_offset_lower_bound() << ", on slot: " << edit_offset << ", payload size: " << payload.size(); - slots_read++; return OkStatus(); - }; - - batt::Status visit_status = (*reader)->visit_slots(visitor_fn).status(); - ASSERT_TRUE(visit_status.ok()) << BATT_INSPECT(visit_status); + }); EXPECT_EQ(slots_read, 0); } @@ -410,11 +457,10 @@ TEST_F(ChangeLogTest, ReadEmptyLog) // TEST_F(ChangeLogTest, ExceedCapacityWrapAround) { - ChangeLogFile::Config config = ChangeLogFile::Config::with_default_values(); - config.block_size = BlockSize{4096}; - config.block_count = BlockCount{8}; // Small capacity to test wrap-around + this->config_.block_size = BlockSize{4096}; + this->config_.block_count = BlockCount{8}; // Small capacity to test wrap-around - const i64 total_capacity = config.block_size * config.block_count; + const i64 total_capacity = this->config_.block_size * this->config_.block_count; const i64 target_data_size = total_capacity * 2.5; // Write 2.5x the capacity // Generate varying sizes of data @@ -430,16 +476,7 @@ TEST_F(ChangeLogTest, ExceedCapacityWrapAround) // Write Phase // { - StatusOr> writer = - ChangeLogWriter::open_or_create(this->test_file_, - config, - ChangeLogWriter::Options::with_default_values(), - RemoveExisting{true}); - ASSERT_TRUE(writer.ok()); - - (*writer)->start(batt::Runtime::instance().default_scheduler().schedule_task()); - - ChangeLogWriter::Context context(**writer); + ASSERT_OK(this->create_writer(RemoveExisting{true})); i64 expected_next_offset = 0; @@ -452,28 +489,13 @@ TEST_F(ChangeLogTest, ExceedCapacityWrapAround) Status write_status = batt::StatusCode::kUnknown; for (;;) { - write_status = context.append_slot( - /*min_edit_offset_lower_bound=*/EditOffset{0}, - slot_data.size(), + write_status = this->append_slot( + slot_data, offsets.empty() ? batt::WaitForResource::kTrue : batt::WaitForResource::kFalse, - [&slot_data, &offsets, &expected_next_offset](FirstVisitToBlock, - ChangeLogBlock* block, - MutableBuffer buffer, - EditOffset offset) { - VLOG(1) << "Appending block with lower_bound: " << block->edit_offset_lower_bound() - << ", on slot: " << offset << " size=" << slot_data.size(); - + [&](FirstVisitToBlock, ChangeLogBlock*, MutableBuffer, EditOffset offset) { BATT_CHECK_EQ(offset.value(), expected_next_offset); expected_next_offset += slot_data.size(); - - VLOG(1) << " (updated)" << BATT_INSPECT(expected_next_offset); - - offsets.emplace(/*lower_bound=*/offset.value(), - /*upper_bound=*/offset.value() + (i64)slot_data.size()); - - BATT_CHECK_GE(buffer.size(), slot_data.size()); - - std::memcpy(buffer.data(), slot_data.data(), slot_data.size()); + offsets.emplace(offset.value(), offset.value() + (i64)slot_data.size()); }); if (write_status.ok() || write_status != batt::StatusCode::kGrantUnavailable || @@ -485,7 +507,7 @@ TEST_F(ChangeLogTest, ExceedCapacityWrapAround) // retrying. // VLOG(1) << "trimming to " << offsets.begin()->second; - Status trim_status = (*writer)->trim(EditOffset{offsets.begin()->second}); + Status trim_status = this->writer_->trim(EditOffset{offsets.begin()->second}); ASSERT_TRUE(trim_status.ok()) << BATT_INSPECT(trim_status); offsets.erase(offsets.begin()); ++slots_trimmed; @@ -501,10 +523,7 @@ TEST_F(ChangeLogTest, ExceedCapacityWrapAround) // Give writer time to flush remaining data // - ASSERT_TRUE((*writer)->wait_for_flush()); - - (*writer)->halt(); - (*writer)->join(); + this->shutdown_writer(/*flush=*/true); LOG(INFO) << "Wrap-around test stats:" << " total_written=" << total_written << " capacity=" << total_capacity; @@ -514,42 +533,36 @@ TEST_F(ChangeLogTest, ExceedCapacityWrapAround) EXPECT_GT(total_written, total_capacity * 2); EXPECT_GT(successful_writes, 0); - auto& metrics = (*writer)->metrics(); + auto& metrics = this->writer_->metrics(); + EXPECT_GT(metrics.written_user_byte_count.load(), 0); EXPECT_GT(metrics.write_count.load(), 0); - EXPECT_GT(ChangeLogBlock::metrics().block_alloc_count.get(), config.block_count.value()); + EXPECT_GT(ChangeLogBlock::metrics().block_alloc_count.get(), this->config_.block_count.value()); } // Read Phase // { - StatusOr> reader = ChangeLogReader::open(this->test_file_); - ASSERT_TRUE(reader.ok()); - - int slots_read = 0; std::unordered_set unique_blocks; - auto visitor_fn = [&](FirstVisitToBlock first_visit, - ChangeLogBlock* block, - EditOffset edit_offset, - ConstBuffer payload) -> Status { + + usize slots_read = this->open_reader_and_visit([&](usize, + FirstVisitToBlock first_visit, + ChangeLogBlock* block, + EditOffset edit_offset, + ConstBuffer payload) -> Status { VLOG(1) << "Reading block with lower_bound: " << block->edit_offset_lower_bound() << ", on slot: " << edit_offset << ", payload size: " << payload.size() << BATT_INSPECT(first_visit) << BATT_INSPECT(block->get_block_index()); - slots_read++; - BATT_REQUIRE_NE(offsets.find(edit_offset.value()), offsets.end()); offsets.erase(edit_offset.value()); unique_blocks.insert(block->edit_offset_lower_bound().value()); return OkStatus(); - }; - - batt::Status visit_status = (*reader)->visit_slots(visitor_fn).status(); - ASSERT_TRUE(visit_status.ok()) << BATT_INSPECT(visit_status); + }); EXPECT_GT(slots_read, 0); - EXPECT_LE(unique_blocks.size(), config.block_count.value()); + EXPECT_LE(unique_blocks.size(), this->config_.block_count.value()); EXPECT_EQ(slots_read, successful_writes - slots_trimmed); EXPECT_TRUE(offsets.empty()); } @@ -559,9 +572,8 @@ TEST_F(ChangeLogTest, ExceedCapacityWrapAround) // TEST_F(ChangeLogTest, CorruptBlockInMiddle) { - ChangeLogFile::Config config = ChangeLogFile::Config::with_default_values(); - config.block_size = BlockSize{4096}; - config.block_count = BlockCount{10}; + this->config_.block_size = BlockSize{4096}; + this->config_.block_count = BlockCount{10}; const usize data_size_per_slot = 4000; const int num_blocks_to_write = 5; @@ -574,41 +586,21 @@ TEST_F(ChangeLogTest, CorruptBlockInMiddle) // Write phase // { - StatusOr> writer = - ChangeLogWriter::open_or_create(this->test_file_, - config, - ChangeLogWriter::Options::with_default_values(), - RemoveExisting{true}); - ASSERT_TRUE(writer.ok()); - - (*writer)->start(batt::Runtime::instance().default_scheduler().schedule_task()); - - ChangeLogWriter::Context context(**writer); + ASSERT_OK(this->create_writer(RemoveExisting{true})); // Write slots s.t. each block has one slot // for (int i = 0; i < num_blocks_to_write; ++i) { - Status write_status = context.append_slot( - /*min_edit_offset_lower_bound=*/EditOffset{0}, - test_data[i].size(), - [&data = test_data[i], - i](FirstVisitToBlock, ChangeLogBlock* block, MutableBuffer buffer, EditOffset offset) { - VLOG(1) << "Writing slot " << i - << " to block with lower_bound: " << block->edit_offset_lower_bound() - << ", at edit_offset: " << offset; - - std::memcpy(buffer.data(), data.data(), data.size()); - }); - - ASSERT_TRUE(write_status.ok()) << "Failed to write slot " << i; - } + Status write_status = this->append_slot(test_data[i]); + ASSERT_TRUE(write_status.ok()) << "Failed to write slot " << i << BATT_INSPECT(write_status); - // Wait for writer to flush all data - // - ASSERT_TRUE((*writer)->wait_for_flush()); + // Sync the blocks as we go to make sure they are recovered in the same order they were + // appended. + // + ASSERT_OK(this->writer_->sync_latest()); + } - (*writer)->halt(); - (*writer)->join(); + this->shutdown_writer(/*flush=*/true); } // Corrupt a block by overwriting its magic number @@ -625,7 +617,7 @@ TEST_F(ChangeLogTest, CorruptBlockInMiddle) }); const i64 corrupt_block_offset = - config.block0_offset + (corrupt_block_index * config.block_size); + this->config_.block0_offset + (corrupt_block_index * this->config_.block_size); LOG(INFO) << "Corrupting block at index " << corrupt_block_index << ", file offset: " << corrupt_block_offset; @@ -637,6 +629,7 @@ TEST_F(ChangeLogTest, CorruptBlockInMiddle) Status write_status = llfs::write_fd(*fd, ConstBuffer{&invalid_magic, sizeof(invalid_magic)}, corrupt_block_offset); + ASSERT_TRUE(write_status.ok()) << "Failed to corrupt block: " << write_status; // TODO: [Gabe Bornstein 4/3/26] Consider updating the corrupt block @@ -648,30 +641,21 @@ TEST_F(ChangeLogTest, CorruptBlockInMiddle) // Read phase - ChangeLogReader should handle the corrupt block gracefully // { - StatusOr> reader = ChangeLogReader::open(this->test_file_); - ASSERT_TRUE(reader.ok()); - - int slots_read = 0; std::vector recovered_offsets; - auto visitor_fn = [&](FirstVisitToBlock, - ChangeLogBlock* block, - EditOffset edit_offset, - ConstBuffer payload) -> Status { - slots_read++; + usize slots_read = open_reader_and_visit([&](usize slot_index, + FirstVisitToBlock, + ChangeLogBlock* block, + EditOffset edit_offset, + ConstBuffer payload) -> Status { recovered_offsets.push_back(edit_offset); - LOG(INFO) << "Post-corruption read: slot " << slots_read << " at edit_offset: " << edit_offset + LOG(INFO) << "Post-corruption read: slot " << slot_index << " at edit_offset: " << edit_offset << ", block lower_bound: " << block->edit_offset_lower_bound() << ", payload size: " << payload.size(); return OkStatus(); - }; - - Status visit_status = (*reader)->visit_slots(visitor_fn).status(); - - // The visit should succeed but only read blocks before the corruption - ASSERT_TRUE(visit_status.ok()) << "Visit failed with: " << visit_status; + }); EXPECT_EQ(slots_read, corrupt_block_index) << "Expected to read " << corrupt_block_index @@ -681,4 +665,166 @@ TEST_F(ChangeLogTest, CorruptBlockInMiddle) } } +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +TEST_F(ChangeLogTest, Sync) +{ + ASSERT_OK(this->create_writer(RemoveExisting{true})); + + // Append a slot and capture the edit offset after it. + // + std::string test_data = "Slot data for sync test"; + ASSERT_OK(this->append_slot(test_data)); + + const EditOffset target = this->writer_->next_edit_offset(); + + Status sync_status = this->writer_->sync(target); + EXPECT_TRUE(sync_status.ok()) << BATT_INSPECT(sync_status); + + // After sync returns, durable_upper_bound must return target. + // + EXPECT_EQ(this->writer_->durable_upper_bound().value(), target.value()); + + this->shutdown_writer(/*flush=*/false); + + LOG(INFO) << BATT_INSPECT(this->writer_->metrics().advance_sync_upper_bound_latency); +} + +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +TEST_F(ChangeLogTest, MultipleSync) +{ + this->config_.block_count = BlockCount{20}; + ASSERT_OK(this->create_writer(RemoveExisting{true})); + + // Append a slot so we have a target offset to sync to. + // + std::string test_data = "Multiple sync test slot"; + ASSERT_OK(this->append_slot(test_data)); + + const EditOffset target = this->writer_->next_edit_offset(); + const usize num_waiters = std::thread::hardware_concurrency(); + + std::atomic start{false}; + std::atomic completed{0}; + + std::vector threads; + threads.reserve(num_waiters); + + for (usize i = 0; i < num_waiters; ++i) { + threads.emplace_back([&]() { + while (!start.load()) { + continue; + } + + Status s = this->writer_->sync(target); + EXPECT_TRUE(s.ok()) << BATT_INSPECT(s); + + completed.fetch_add(1); + }); + } + + start.store(true); + + for (auto& t : threads) { + t.join(); + } + + // All threads must have completed successfully. + // + EXPECT_EQ(completed.load(), num_waiters); + + this->shutdown_writer(/*flush=*/false); + + LOG(INFO) << BATT_INSPECT(this->writer_->metrics().advance_sync_upper_bound_latency); +} + +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +TEST_F(ChangeLogTest, SyncStaggeredOffsets) +{ + // Use small blocks with large payloads to force slots into separate blocks and + // separate write batches. + // + this->config_.block_size = BlockSize{512}; + this->config_.block_count = BlockCount{200}; + ASSERT_OK(this->create_writer(RemoveExisting{true})); + + const usize num_slots = 200; + const usize slot_size = 400; + + // Pre-compute the target offsets. + // + std::vector targets; + targets.reserve(num_slots); + for (usize i = 0; i < num_slots; ++i) { + targets.push_back(EditOffset{(i64)(i + 1) * (i64)slot_size}); + } + + // Launch sync threads before appending data, so they block on await_true. + // + std::vector> completion_order(num_slots); + std::atomic completion_counter{0}; + + // Use a barrier to block the append thread from continuing until one of the sync threads has + // "consumed" the corresponding flush event. + // + std::barrier sync_barrier{2, batt::DoNothing{}}; + + std::vector sync_threads; + sync_threads.reserve(num_slots); + + for (usize i = 0; i < num_slots; ++i) { + sync_threads.emplace_back([&, i]() { + Status s = this->writer_->sync(targets[i]); + EXPECT_TRUE(s.ok()) << BATT_INSPECT(s); + + completion_order[i].store(completion_counter.fetch_add(1)); + + // Signal to the appender thread that a sync thread has completed. + // + sync_barrier.arrive_and_wait(); + }); + } + + // Append slots one at a time, waiting for each to flush before appending the next. + // + std::thread appender([&]() { + ChangeLogWriter::Context context(*this->writer_); + + for (usize i = 0; i < num_slots; ++i) { + std::string data(slot_size, 'A' + ((char)i % 26)); + Status write_status = this->append_slot(context, data); + + ASSERT_TRUE(write_status.ok()); + ASSERT_TRUE(this->writer_->wait_for_flush()); + + // Wait for a sync thread to consume the slot. + // + sync_barrier.arrive_and_wait(); + } + }); + + appender.join(); + + for (auto& t : sync_threads) { + t.join(); + } + + // Verify that a thread waiting on a smaller offset must complete no later than + // a thread waiting on a larger offset. + // + for (usize i = 0; i < num_slots; ++i) { + for (usize j = i + 1; j < num_slots; ++j) { + EXPECT_LE(completion_order[i].load(), completion_order[j].load()) + << "Thread waiting on offset " << targets[i] << " completed after thread waiting on " + << targets[j]; + } + } + + this->shutdown_writer(/*flush=*/false); + + LOG(INFO) << BATT_INSPECT(this->writer_->metrics().advance_sync_upper_bound_latency); +} + } // namespace turtle_kv diff --git a/src/turtle_kv/change_log/change_log_block.hpp b/src/turtle_kv/change_log/change_log_block.hpp index b9391c4..ee6675d 100644 --- a/src/turtle_kv/change_log/change_log_block.hpp +++ b/src/turtle_kv/change_log/change_log_block.hpp @@ -30,6 +30,7 @@ #include #include +#include namespace turtle_kv { @@ -203,6 +204,39 @@ class ChangeLogBlock BATT_OK_RESULT_OR_PANIC(Self::read_slot_edit_offset_delta(this->get_slot(i))); } + EditOffsetDelta next_edit_offset_of_slot(usize slot_index) const noexcept + { + return EditOffsetDelta{ + static_cast(this->slot_size(slot_index) - sizeof(PackedEditOffsetDelta))}; + } + + /** \brief Returns the index of the first slot with EditOffset >= `target`, or None if no such + * slot exists. + */ + Optional lower_bound_slot(EditOffset target) const noexcept + { + if (this->slot_count() == 0) { + return None; + } + + // SlotInfo pointers are in descending EditOffset order. + // + auto slot_index_range = boost::irange(0, this->slot_count()); + auto first = std::begin(slot_index_range); + auto last = std::end(slot_index_range); + + auto it = std::lower_bound(first, last, target, [this](usize slot_index, EditOffset value) { + return this->slot_edit_offset(slot_index) < value; + }); + + // If all edit offsets are strictly less than `target`, return None. + // + if (it == last) { + return None; + } + return *it; + } + /** \brief Adds `count` references to this buffer. */ void add_ref(i32 count) noexcept; diff --git a/src/turtle_kv/change_log/change_log_blocks_visitor.hpp b/src/turtle_kv/change_log/change_log_blocks_visitor.hpp new file mode 100644 index 0000000..e767864 --- /dev/null +++ b/src/turtle_kv/change_log/change_log_blocks_visitor.hpp @@ -0,0 +1,165 @@ +//=##=##=#==#=#==#===#+==#+==========+==+=+=+=+=+=++=+++=+++++=-++++=-+++++++++++ +// +// Part of the TurtleKV Project, under Apache License v2.0. +// See https://www.apache.org/licenses/LICENSE-2.0 for license information. +// SPDX short identifier: Apache-2.0 +// +//+++++++++++-+-+--+----- --- -- - - - - + +#pragma once +#define TURTLE_KV_CHANGE_LOG_BLOCKS_VISITOR_HPP + +#include +#include +#include + +#include + +#include +#include + +#include + +#include + +#include +#include + +namespace turtle_kv { + +//=#=#==#==#===============+=+=+=+=++=++++++++++++++-++-+--+-+----+--------------- + +/** \brief Used to read slots from a block. Tracks which slot to read next with `next_slot_i`. + */ +struct BlockIterator { + boost::intrusive_ptr block; + usize next_slot_i = 0; + bool visited = false; + + explicit BlockIterator(boost::intrusive_ptr&& block_arg, usize slot_i) noexcept + : block{std::move(block_arg)} + , next_slot_i{slot_i} + , visited{false} + { + } + + BlockIterator() = default; + + bool has_more() const noexcept + { + return this->next_slot_i < this->block->slot_count(); + } + + EditOffset current_edit_offset() const noexcept + { + BATT_CHECK(this->has_more()); + return this->block->slot_edit_offset(this->next_slot_i); + } +}; + +using BlockIteratorMap = absl::flat_hash_map; + +//=#=#==#==#===============+=+=+=+=++=++++++++++++++-++-+--+-+----+--------------- + +template +concept BlockSlotVisitorFn = + std::invocable; + +//=#=#==#==#===============+=+=+=+=++=++++++++++++++-++-+--+-+----+--------------- + +class ChangeLogBlocksVisitor +{ + public: + explicit ChangeLogBlocksVisitor(EditOffset upper_bound) noexcept + : visited_upper_bound_{upper_bound} + { + } + + ChangeLogBlocksVisitor() noexcept : visited_upper_bound_{0} + { + } + + EditOffset visited_upper_bound() const noexcept + { + return this->visited_upper_bound_; + } + + void set_visited_upper_bound(EditOffset value) noexcept + { + this->visited_upper_bound_ = value; + } + + const BlockIteratorMap& pending_blocks() const noexcept + { + return this->pending_blocks_; + } + + void add_block(boost::intrusive_ptr&& block) + { + // Find the edit offset lower bound in the block w.r.t. the visited upper bound, to account + // for cases where a trim has happened mid-block. + // + Optional slot_i = block->lower_bound_slot(this->visited_upper_bound_); + if (!slot_i) { + return; + } + const EditOffset slot_offset = block->slot_edit_offset(*slot_i); + this->pending_blocks_[slot_offset] = BlockIterator{std::move(block), *slot_i}; + } + + /** \brief Walks from current_offset_start_ forward through a set of pending blocks, + * consuming contiguous slots in order. Calls `slot_fn` for each slot consumed. Stops at the first + * gap. + * + * \param slot_fn Called for each consumed slot with + * (FirstVisitToBlock, ChangeLogBlock*, slot_index, EditOffset). + * + * \return The new visited_upper_bound_ value after walking. + */ + template + EditOffset visit_change_log_blocks(SlotFn&& slot_fn) + { + for (;;) { + auto it = this->pending_blocks_.find(this->visited_upper_bound_); + if (it == this->pending_blocks_.end()) { + break; + } + + BlockIterator entry = std::move(it->second); + this->pending_blocks_.erase(it); + + do { + auto first_visit = FirstVisitToBlock{!entry.visited}; + entry.visited = true; + + BATT_INVOKE_LOOP_FN((slot_fn, + first_visit, + entry.block.get(), + entry.next_slot_i, + this->visited_upper_bound_)); + + this->visited_upper_bound_ += entry.block->next_edit_offset_of_slot(entry.next_slot_i); + + ++entry.next_slot_i; + } while (entry.has_more() && entry.current_edit_offset() == this->visited_upper_bound_); + + if (entry.has_more()) { + const EditOffset block_next_edit_offset = entry.current_edit_offset(); + this->pending_blocks_[block_next_edit_offset] = std::move(entry); + } + } + return this->visited_upper_bound_; + } + + private: + /** \brief The upper bound of the contiguous range of slots that have been visited. + */ + EditOffset visited_upper_bound_; + + /** \brief Map from slot EditOffset to block entry; entries are consumed as the + * visited_upper_bound_ advances. Blocks with remaining non-contiguous slots are re-inserted. + */ + BlockIteratorMap pending_blocks_; +}; + +} // namespace turtle_kv diff --git a/src/turtle_kv/change_log/change_log_reader.cpp b/src/turtle_kv/change_log/change_log_reader.cpp index f74123d..bb32c19 100644 --- a/src/turtle_kv/change_log/change_log_reader.cpp +++ b/src/turtle_kv/change_log/change_log_reader.cpp @@ -9,74 +9,16 @@ #include // +#include + #include #include +#include namespace turtle_kv { namespace { -// Used to read slots from a block. Tracks which slot to read next with `next_slot_i`. -// -struct BlockIterator { - boost::intrusive_ptr block; - usize next_slot_i = 0; - bool visited = false; - - //+++++++++++-+-+--+----- --- -- - - - - - - explicit BlockIterator(boost::intrusive_ptr&& block_arg) noexcept - : block{std::move(block_arg)} - { - } - - BlockIterator(BlockIterator&& other) noexcept - : block{std::exchange(other.block, nullptr)} - , next_slot_i{std::exchange(other.next_slot_i, 0)} - , visited{std::exchange(other.visited, false)} - { - } - - BlockIterator(const BlockIterator&) = delete; - BlockIterator& operator=(const BlockIterator&) = delete; - - // Get the EditOffset of the current slot. - // - EditOffset current_edit_offset() const - { - BATT_CHECK(this->has_more()); - return this->block->slot_edit_offset(this->next_slot_i); - } - - // Check if there are more slots to process. - // - bool has_more() const - { - return next_slot_i < block->slot_count(); - } - - bool operator<(const BlockIterator& other) const - { - // If this has no more, it cannot be before anything else. - // - if (!this->has_more()) { - return false; - } - - // If the other has no more, this comes before; otherwise we know both have more so compare the - // current edit offsets. - // - return !other.has_more() || this->current_edit_offset() < other.current_edit_offset(); - } -}; - -struct BlockIteratorCompare { - bool operator()(BlockIterator* left, BlockIterator* right) const - { - return *left < *right; - } -}; - //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // /** \brief Changes the passed `recovered_state` against `read_meta_state`; if it is different, @@ -130,9 +72,9 @@ StatusOr ChangeLogReader::visit_slots( // std::unordered_set visited_block_set; - // Create block iterators, filtering out empty blocks. + // Build the visitor, filtering out empty/trimmed blocks. // - std::vector block_iterators; + ChangeLogBlocksVisitor blocks_visitor{target_trim_edit_offset}; for (auto& block : blocks_vec) { if (block->edit_offset_upper_bound() <= target_trim_edit_offset) { @@ -142,13 +84,13 @@ StatusOr ChangeLogReader::visit_slots( continue; } if (block->slot_count() > 0) { - block_iterators.emplace_back(batt::make_copy(block)); + blocks_visitor.add_block(batt::make_copy(block)); } } // If there's no slots to process, return early. // - if (block_iterators.empty()) { + if (blocks_visitor.pending_blocks().empty()) { RecoveredChangeLogState recovered_state; recovered_state.block_range = Interval{BlockIndex{0}, BlockIndex{0}}; recovered_state.trim_edit_offset = target_trim_edit_offset; @@ -161,71 +103,34 @@ StatusOr ChangeLogReader::visit_slots( return {recovered_state}; } - StackMerger heap{ - Slice{as_slice(block_iterators)}}; - - // The first expected slot edit offset is the trim edit offset. - // - EditOffset expected_next_edit_offset = target_trim_edit_offset; - //+++++++++++-+-+--+----- --- -- - - - - // Process slots in EditOffset order. // - while (!heap.empty()) { - BlockIterator* current = heap.first(); - ConstBuffer slot_buffer = current->block->get_slot(current->next_slot_i); - EditOffset edit_offset = current->current_edit_offset(); + Status visit_status = OkStatus(); - // Advance to the next slot when we finish each iteration of the loop. - // - auto on_loop_iter_exit = batt::finally([&] { - ++current->next_slot_i; - if (current->has_more()) { - heap.update_first(); - } else { - heap.remove_first(); - } - }); - - // Skip slots below the trim bound. - // - if (edit_offset < target_trim_edit_offset) { - continue; - } + EditOffset expected_next_edit_offset{blocks_visitor.visit_change_log_blocks( + [&](FirstVisitToBlock first_visit, ChangeLogBlock* block, usize slot_i, + EditOffset edit_offset) -> Optional { + if (first_visit) { + auto [_, inserted] = + visited_block_set.insert(block->get_block_index().value_or_panic()); + BATT_CHECK(inserted); + } else { + BATT_CHECK(visited_block_set.count(block->get_block_index().value_or_panic()) > 0); + } - // If there's a gap in our slots, we're missing data and can't continue. - // - if (expected_next_edit_offset != edit_offset) { - VLOG(1) << "Gap found;" << BATT_INSPECT(expected_next_edit_offset) - << BATT_INSPECT(edit_offset) << BATT_INSPECT(current->block->get_block_index()); - break; - } + ConstBuffer slot_buffer = block->get_slot(slot_i); + ConstBuffer payload = slot_buffer + sizeof(PackedEditOffsetDelta); - expected_next_edit_offset = - edit_offset + - EditOffsetDelta{static_cast(slot_buffer.size() - sizeof(PackedEditOffsetDelta))}; + visit_status = visitor(first_visit, block, edit_offset, payload); - auto first_visit_to_block = FirstVisitToBlock{!current->visited}; - ChangeLogBlock* block = current->block.get(); + if (!visit_status.ok()) { + return batt::seq::LoopControl::kBreak; + } + return None; + })}; - // If recovering state, add each block which contains recovered slots to the - // `block_edit_ranges` hash table. - // - if (first_visit_to_block) { - visited_block_set.insert(block->get_block_index().value_or_panic()); - BATT_CHECK_EQ(false, current->visited); - current->visited = true; - } else { - BATT_CHECK(current->visited); - } - - // Move the payload past the EditOffset. - // - ConstBuffer payload = slot_buffer + sizeof(PackedEditOffsetDelta); - - Status visit_status = visitor(first_visit_to_block, block, edit_offset, payload); - BATT_REQUIRE_OK(visit_status); - } + BATT_REQUIRE_OK(visit_status); //+++++++++++-+-+--+----- --- -- - - - - // Initialize the `recovered_state` object. diff --git a/src/turtle_kv/change_log/change_log_reader.hpp b/src/turtle_kv/change_log/change_log_reader.hpp index 5b6ffe3..827653e 100644 --- a/src/turtle_kv/change_log/change_log_reader.hpp +++ b/src/turtle_kv/change_log/change_log_reader.hpp @@ -15,8 +15,6 @@ #include #include -#include - #include #include diff --git a/src/turtle_kv/change_log/change_log_writer.cpp b/src/turtle_kv/change_log/change_log_writer.cpp index 97eb10e..db196d3 100644 --- a/src/turtle_kv/change_log/change_log_writer.cpp +++ b/src/turtle_kv/change_log/change_log_writer.cpp @@ -13,6 +13,8 @@ #include +#include + #include #include #include @@ -136,6 +138,24 @@ struct ChangeLogWriter::WrittenBlocksState { } }; +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +struct ChangeLogWriter::AdvanceSyncState { + /** \brief Visits all slots in all flushed blocks, to track the current durable 'sync' upper + * bound EditOffset. + */ + ChangeLogBlocksVisitor visitor; + + //----- --- -- - - - - + + AdvanceSyncState() = delete; + + explicit AdvanceSyncState(EditOffset recovered_upper_bound) noexcept + : visitor{recovered_upper_bound} + { + } +}; + //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // struct ChangeLogWriter::ActiveBlocksState : ChangeLogFile::MetaState { @@ -358,6 +378,7 @@ void ChangeLogWriter::Context::push_buffer(BlockBuffer*& buffer, , free_block_tokens_{BATT_CHECKED_CAST(u64, this->change_log_->config().block_count.value())} , metrics_{} , next_edit_offset_{recovered_state.next_edit_offset.value()} + , sync_upper_bound_{recovered_state.next_edit_offset.value()} { // Initialize the meta-block buffer to reflect the on-disk state. // @@ -410,6 +431,7 @@ void ChangeLogWriter::start(batt::Task::executor_type&& executor) noexcept // void ChangeLogWriter::halt() noexcept { + this->sync_upper_bound_.close(); this->halt_requested_.store(true); this->free_block_tokens_.close(); } @@ -506,6 +528,7 @@ void ChangeLogWriter::writer_task_main() noexcept CollectedBlocksState collected; PreparedBlocksState prepared{this->config(), *this->state_.lock()->active_blocks_state_}; WrittenBlocksState written; + AdvanceSyncState synced{EditOffset{this->sync_upper_bound_.get_value()}}; for (;;) { // Collect BlockBuffers from writer contexts. @@ -517,9 +540,11 @@ void ChangeLogWriter::writer_task_main() noexcept BATT_ASSIGN_OK_RESULT(BlockBufferStats prepare_stats, this->prepare_blocks(collected, prepared)); - // If there are no updates, then sleep before polling again (unless halt requested). + // If there are no updates, then sleep before polling again (unless halt requested or + // there is pending urgent work). // - if ((force_sleep || prepared.empty()) && this->halt_requested_.load() == false) { + if ((force_sleep || prepared.empty()) && this->halt_requested_.load() == false && + !this->has_pending_urgent_sync_work()) { force_sleep = false; inactive_count += 1; this->metrics_.sleep_count.add(1); @@ -528,7 +553,7 @@ void ChangeLogWriter::writer_task_main() noexcept // appending; in this case, enter our timed polling loop. // const i64 delay_usec = pick_delay_usec(rng); - batt::Task::sleep(std::chrono::microseconds(delay_usec)); + [[maybe_unused]] auto ec = batt::Task::sleep(std::chrono::microseconds(delay_usec)); // After allowing other tasks to run, we should immediately poll updates again to see if we // have more data. @@ -543,17 +568,25 @@ void ChangeLogWriter::writer_task_main() noexcept const BlockBufferStats block_stats = prepare_stats + write_stats; - // Force a sleep if the collected buffers weren't full enough to hit the target density. - // - force_sleep = block_stats.is_under_target(); - // "Activate" the written blocks by adding them to the active blocks state; this allows // accurate trimming and reclamation of storage resources. + // + batt::SmallVec, kStaticQueueSize> newly_activated; { batt::ScopedLock locked_state{this->state_}; - BATT_REQUIRE_OK(this->activate_blocks(written, *locked_state->active_blocks_state_)); + BATT_REQUIRE_OK( + this->activate_blocks(written, *locked_state->active_blocks_state_, newly_activated)); } + // Advance the durable upper bound with the newly activated blocks. + // + this->advance_sync_upper_bound(newly_activated, synced); + + // Force a sleep if the collected buffers weren't full enough to hit the target density, + // unless there is pending urgent work. + // + force_sleep = block_stats.is_under_target() && !this->has_pending_urgent_sync_work(); + // If halt is requested and we don't appear to be making any progress, then return. // if (this->halt_requested_.load() // @@ -767,8 +800,10 @@ auto ChangeLogWriter::write_blocks(PreparedBlocksState& input, WrittenBlocksStat //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // -Status ChangeLogWriter::activate_blocks(WrittenBlocksState& input, - ActiveBlocksState& output) noexcept +Status ChangeLogWriter::activate_blocks( + WrittenBlocksState& input, + ActiveBlocksState& output, + batt::SmallVecBase>& newly_activated) noexcept { if (input.blocks.empty()) { return OkStatus(); @@ -835,6 +870,12 @@ Status ChangeLogWriter::activate_blocks(WrittenBlocksState& input, continue; } + // Collect blocks with slots for advancing the durable upper bound. + // + if (next_block->slot_count() > 0) { + newly_activated.emplace_back(next_block); + } + // Update active blocks edit offset upper bound. // output.block_upper_bounds[*input.block_index] = next_block->edit_offset_upper_bound().value(); @@ -940,6 +981,80 @@ Optional ChangeLogWriter::ActiveBlocksState::apply_trim( return released_grant; } +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +Status ChangeLogWriter::sync(EditOffset upper_bound, bool urgent) noexcept +{ + // Return early if upper bound is already synced. + // + if (this->sync_upper_bound_.get_value() >= upper_bound.value()) { + return OkStatus(); + } + + if (urgent) { + this->urgent_sync_counter_.fetch_add(1); + } + + auto on_exit = batt::finally([&] { + if (urgent) { + this->urgent_sync_counter_.fetch_sub(1); + } + }); + + if (this->task_) { + this->task_->wake(); + } + + // Block until the writer main task advances sync_upper_bound_ past our target. + // + BATT_ASSIGN_OK_RESULT([[maybe_unused]] const i64 observed, + this->sync_upper_bound_.await_true([upper_bound](i64 observed) { + return observed >= upper_bound.value(); + })); + + return OkStatus(); +} + +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +i64 ChangeLogWriter::get_unflushed_byte_count() const noexcept +{ + const i64 count = + std::max(0, this->next_edit_offset_.load() - this->sync_upper_bound_.get_value()); + return count; +} + +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +bool ChangeLogWriter::has_pending_urgent_sync_work() const noexcept +{ + return this->urgent_sync_counter_.load() > 0; +} + +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +void ChangeLogWriter::advance_sync_upper_bound( + batt::SmallVecBase>& newly_activated, + AdvanceSyncState& sync_state) noexcept +{ + LatencyTimer timer{Every2ToTheConst<0>{}, this->metrics_.advance_sync_upper_bound_latency}; + + BATT_CHECK_EQ(sync_state.visitor.visited_upper_bound(), + EditOffset{this->sync_upper_bound_.get_value()}) + << "The ChangeLogWriter::write_task_main thread must be the only modifier of " + "sync_upper_bound_!"; + + for (auto& block_ptr : newly_activated) { + sync_state.visitor.add_block(std::move(block_ptr)); + } + + sync_state.visitor.visit_change_log_blocks(batt::DoNothing{}); + // + // Nothing to do with the visited blocks; we just care about the new visited upper bound. + + this->sync_upper_bound_.set_value(sync_state.visitor.visited_upper_bound().value()); +} + //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // Status ChangeLogWriter::refresh_meta_block(ActiveBlocksState& active_blocks) noexcept diff --git a/src/turtle_kv/change_log/change_log_writer.hpp b/src/turtle_kv/change_log/change_log_writer.hpp index 7204cac..01ce618 100644 --- a/src/turtle_kv/change_log/change_log_writer.hpp +++ b/src/turtle_kv/change_log/change_log_writer.hpp @@ -11,6 +11,7 @@ #include #include +#include #include #include @@ -110,6 +111,7 @@ class ChangeLogWriter return (double)this->received_user_byte_count.load() / ((double)this->received_block_byte_count.load() + 1e-6); }}; + LatencyMetric advance_sync_upper_bound_latency; }; //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - @@ -347,6 +349,23 @@ class ChangeLogWriter return false; } + Status sync(EditOffset upper_bound, bool urgent = false) noexcept; + + Status sync_latest(bool urgent = false) noexcept + { + return this->sync(this->next_edit_offset(), urgent); + } + + EditOffset durable_upper_bound() const noexcept + { + return EditOffset{this->sync_upper_bound_.get_value()}; + } + + /** \brief Returns the number of bytes between the sync upper bound and the next edit offset. + * Updates the unflushed_byte_count metric. + */ + i64 get_unflushed_byte_count() const noexcept; + //+++++++++++-+-+--+----- --- -- - - - - private: //+++++++++++-+-+--+----- --- -- - - - - @@ -371,6 +390,10 @@ class ChangeLogWriter struct ActiveBlocksState; + // activate_blocks() -> AdvanceSyncState -> advance_sync_upper_bound() + // + struct AdvanceSyncState; + //+++++++++++-+-+--+----- --- -- - - - - struct State { @@ -491,13 +514,32 @@ class ChangeLogWriter * * ChangeLogBlock (BlockBuffer) objects removed from `input.blocks` are released by decrementing * their ref count via `remove_ref`. + * + * Blocks with slots are appended to `newly_activated` for post-activation advancing of the + * durable upper bound. */ - Status activate_blocks(WrittenBlocksState& input, ActiveBlocksState& output) noexcept; + Status activate_blocks( + WrittenBlocksState& input, + ActiveBlocksState& output, + batt::SmallVecBase>& newly_activated) noexcept; /** \brief Refreshes the meta-block in the change log file. */ Status refresh_meta_block(ActiveBlocksState& active_blocks) noexcept; + /** \brief Inserts newly activated blocks into the pending map and advances sync_upper_bound_ by + * walking slots from the current upper bound. Called after activate_blocks, outside the + * state mutex. + */ + void advance_sync_upper_bound( + batt::SmallVecBase>& newly_activated, + AdvanceSyncState& sync_state) noexcept; + + /** \brief Returns true when the writer task should stay awake: there are pending urgent syncs and + * unflushed bytes. + */ + bool has_pending_urgent_sync_work() const noexcept; + //+++++++++++-+-+--+----- --- -- - - - - /** \brief The state of the log file. @@ -536,6 +578,14 @@ class ChangeLogWriter /** \brief The background writer task. */ Optional task_; + + /** \brief The confirmed durable EditOffset upper bound. + */ + batt::Watch sync_upper_bound_; + + /** \brief The number of pending sync callers that have an urgent priority. + */ + std::atomic urgent_sync_counter_{0}; }; // #=##=##=#==#=#==#===#+==#+==========+==+=+=+=+=+=++=+++=+++++=-++++=-+++++++++++ diff --git a/src/turtle_kv/change_log/edit_offset.hpp b/src/turtle_kv/change_log/edit_offset.hpp index 5691632..7e0b34d 100644 --- a/src/turtle_kv/change_log/edit_offset.hpp +++ b/src/turtle_kv/change_log/edit_offset.hpp @@ -23,6 +23,13 @@ class WrappedInt using Self = WrappedInt; using IntT = Int; + struct Hash { + decltype(auto) operator()(const Self& wrapped_int) const + { + return std::hash{}(wrapped_int.value()); + } + }; + //+++++++++++-+-+--+----- --- -- - - - - constexpr explicit WrappedInt(IntT value) noexcept : value_{value} @@ -131,9 +138,15 @@ inline EditOffsetDelta operator-(EditOffset left, EditOffset right) return EditOffsetDelta{left.value() - right.value()}; } +inline EditOffset& operator+=(EditOffset& left, EditOffsetDelta right) +{ + left = EditOffset{left.value() + right.value()}; + return left; +} + inline EditOffset operator+(EditOffset left, EditOffsetDelta right) { - return EditOffset{left.value() + right.value()}; + return left += right; } inline EditOffset operator+(EditOffset left, SlotEditOffsetDelta right) diff --git a/src/turtle_kv/kv_store.cpp b/src/turtle_kv/kv_store.cpp index 773ab82..6e07922 100644 --- a/src/turtle_kv/kv_store.cpp +++ b/src/turtle_kv/kv_store.cpp @@ -639,7 +639,9 @@ boost::intrusive_ptr KVStore::create_mem_table(EditOffset edit_offset_ //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // -Status KVStore::put(const KeyView& key, const ValueView& value) noexcept /*override*/ +StatusOr KVStore::put(const KeyView& key, + const ValueView& value, + Optional write_options) noexcept { for (usize retry_i = 0; retry_i != KVStore::kMaxUpdateRetries; ++retry_i) { #if TURTLE_KV_PROFILE_UPDATES @@ -667,18 +669,25 @@ Status KVStore::put(const KeyView& key, const ValueView& value) noexcept /*overr // Insert the key/value pair into the active MemTable; this will also append a change log // buffer. // - Status status = + StatusOr result = observed_mem_table->put(thread_context.change_log_writer_context_, key, value); #if TURTLE_KV_PROFILE_UPDATES put_mem_table_timer.stop(); #endif + if (result.ok()) { + if (write_options && write_options->sync) { + BATT_REQUIRE_OK(this->change_log_writer_->sync(*result, write_options->urgent_sync)); + } + return result; + } + // On success and unrecoverable errors, just return immediately. // - if (status != batt::StatusCode::kResourceExhausted && - status != batt::StatusCode::kGrantUnavailable) { - return status; + if (result.status() != batt::StatusCode::kResourceExhausted && + result.status() != batt::StatusCode::kGrantUnavailable) { + return result.status(); } // Grab a (owning) reference to the MemTable. @@ -716,7 +725,15 @@ Status KVStore::put(const KeyView& key, const ValueView& value) noexcept /*overr // // Out of retries. - return batt::StatusCode::kUnavailable; + return Status{batt::StatusCode::kUnavailable}; +} + + +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +Status KVStore::put(const KeyView& key, const ValueView& value) noexcept /*override*/ +{ + return this->put(key, value, /*write_options=*/None).status(); } //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - @@ -1039,6 +1056,14 @@ Status KVStore::remove(const KeyView& key) noexcept /*override*/ return this->put(key, ValueView::deleted()); } +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +StatusOr KVStore::remove(const KeyView& key, + Optional write_options) noexcept +{ + return this->put(key, ValueView::deleted(), write_options); +} + //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // Status KVStore::finalize_mem_table(boost::intrusive_ptr&& old_mem_table) @@ -1324,6 +1349,18 @@ using CheckpointEvent = llfs::PackedVariant; prev_checkpoint.second); } +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +Status KVStore::sync(Optional upper_bound, Optional write_options) noexcept +{ + EditOffset target = + upper_bound ? *upper_bound : this->change_log_writer_->next_edit_offset(); + + bool urgent = write_options && write_options->urgent_sync ? true : false; + + return this->change_log_writer_->sync(target, urgent); +} + //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // void KVStore::info_task_main() noexcept diff --git a/src/turtle_kv/kv_store.hpp b/src/turtle_kv/kv_store.hpp index b33a54d..8d4045d 100644 --- a/src/turtle_kv/kv_store.hpp +++ b/src/turtle_kv/kv_store.hpp @@ -61,6 +61,7 @@ class KVStore : public Table using Config = KVStoreConfig; using RuntimeOptions = KVStoreRuntimeOptions; + using WriteOptions = KVStoreWriteOptions; struct ThreadContext { llfs::PageCache& page_cache; @@ -204,6 +205,10 @@ class KVStore : public Table Status put(const KeyView& key, const ValueView& value) noexcept override; + StatusOr put(const KeyView& key, + const ValueView& value, + Optional write_options) noexcept; + StatusOr get(const KeyView& key) noexcept override; StatusOr scan(const KeyView& min_key, @@ -213,6 +218,8 @@ class KVStore : public Table Status remove(const KeyView& key) noexcept override; + StatusOr remove(const KeyView& key, Optional write_options) noexcept; + //+++++++++++-+-+--+----- --- -- - - - - const TreeOptions& tree_options() const @@ -260,6 +267,14 @@ class KVStore : public Table */ void release_thread_context() noexcept; + /** \brief Allows for the explicit syncing of data up till a specified upper bound `EditOffset`. + * Callers of this function are guaranteed that the data up till the upper bound is hardened to + * disk after this function returns successfully. If no upper bound `EditOffset` is provided to + * this function, the upper bound of the last inserted edit is used. + */ + Status sync(Optional upper_bound = None, + Optional write_options = None) noexcept; + //+++++++++++-+-+--+----- --- -- - - - - private: enum struct RecoveryStatus : i32 { diff --git a/src/turtle_kv/kv_store.test.cpp b/src/turtle_kv/kv_store.test.cpp index 70f9171..7c61eda 100644 --- a/src/turtle_kv/kv_store.test.cpp +++ b/src/turtle_kv/kv_store.test.cpp @@ -13,18 +13,20 @@ #include #include -#include - #include "data_root.test.hpp" +#include +#include #include -#include - #include +#include -#include +#include +#include -#include +#include +#include +#include namespace { @@ -33,11 +35,15 @@ using namespace turtle_kv::constants; using llfs::PageSize; +using turtle_kv::EditOffset; using turtle_kv::KeyView; using turtle_kv::KVStore; using turtle_kv::LatencyMetric; using turtle_kv::LatencyTimer; +using turtle_kv::None; +using turtle_kv::ObjectThreadStorage; using turtle_kv::OkStatus; +using turtle_kv::Optional; using turtle_kv::RemoveExisting; using turtle_kv::Slice; using turtle_kv::Status; @@ -129,22 +135,56 @@ class KVStoreTest : public ::testing::Test this->kv_store_config.tree_options, this->runtime_options); } + void PopulateKVStore(KVStore& kv_store, u64 num_puts, - std::map* out_data = nullptr) + std::map* out_data = nullptr, + double delete_proportion = 0.0, + std::set* out_deleted = nullptr, + Optional write_options = None) { for (u64 i = 0; i < num_puts; ++i) { std::string key = this->generate_key(this->rng); std::string value = this->generate_value(); - Status put_status = kv_store.put(KeyView{key}, ValueView::from_str(value)); - ASSERT_TRUE(put_status.ok()) << BATT_INSPECT(put_status); + if (write_options) { + StatusOr result = + kv_store.put(KeyView{key}, ValueView::from_str(value), *write_options); + ASSERT_TRUE(result.ok()) << BATT_INSPECT(result.status()); + } else { + Status put_status = kv_store.put(KeyView{key}, ValueView::from_str(value)); + ASSERT_TRUE(put_status.ok()) << BATT_INSPECT(put_status); + } if (out_data) { (*out_data)[key] = value; } VLOG(3) << "Put key==" << key << ", value==" << value; } + + if (delete_proportion > 0.0 && out_data) { + const u64 num_to_delete = static_cast(num_puts * delete_proportion); + u64 deleted = 0; + + // TODO [tastolfi 2026-06-16] Add an option to pick keys at random rather than in-order. + // + for (const auto& [key, value] : *out_data) { + if (deleted >= num_to_delete) { + break; + } + if (write_options) { + StatusOr result = kv_store.remove(KeyView{key}, *write_options); + ASSERT_TRUE(result.ok()) << BATT_INSPECT(result.status()); + } else { + Status result = kv_store.remove(KeyView{key}); + ASSERT_TRUE(result.ok()) << BATT_INSPECT(result); + } + if (out_deleted) { + out_deleted->insert(key); + } + ++deleted; + } + } } void ShutdownKVStore(std::unique_ptr& kv_store) @@ -445,6 +485,7 @@ TEST_P(CheckpointTest, CheckpointRecovery) std::map expected_keys_values; u64 num_checkpoints_created = 0; + EditOffset last_checkpoint_bound{0}; u64 keys_per_checkpoint; if (this->num_checkpoints_to_create == 0) { @@ -473,7 +514,9 @@ TEST_P(CheckpointTest, CheckpointRecovery) if (keys_since_checkpoint >= keys_per_checkpoint && this->num_checkpoints_to_create != 0) { keys_since_checkpoint = 0; ++num_checkpoints_created; - BATT_CHECK_OK(kv_store->force_checkpoint()); + StatusOr checkpoint_bound = kv_store->force_checkpoint(); + BATT_CHECK_OK(checkpoint_bound); + last_checkpoint_bound = *checkpoint_bound; VLOG(2) << "Created " << num_checkpoints_created << " checkpoints"; if (num_checkpoints_created == this->num_checkpoints_to_create) { break; @@ -484,7 +527,9 @@ TEST_P(CheckpointTest, CheckpointRecovery) // Handle off by one error where we create one less checkpoint than expected // if (num_checkpoints_created < this->num_checkpoints_to_create) { - BATT_CHECK_OK(kv_store->force_checkpoint()); + StatusOr checkpoint_bound = kv_store->force_checkpoint(); + BATT_CHECK_OK(checkpoint_bound); + last_checkpoint_bound = *checkpoint_bound; ++num_checkpoints_created; VLOG(1) << "Created " << num_checkpoints_created << " checkpoints after rounding error"; } @@ -492,9 +537,7 @@ TEST_P(CheckpointTest, CheckpointRecovery) BATT_CHECK_EQ(num_checkpoints_created, this->num_checkpoints_to_create) << "Did not take the correct number of checkpoints. There is a bug in this test."; - // TODO: [Gabe Bornstein 3/17/26] Replace with fsync once it's implemented. - // - std::this_thread::sleep_for(std::chrono::seconds(1)); + BATT_CHECK_OK(kv_store->wait_for_checkpoint(last_checkpoint_bound)); this->ShutdownKVStore(kv_store); batt::StatusOr> checkpoint_log_volume = @@ -571,9 +614,7 @@ TEST_P(KVStoreRecoveryTest, KVStoreRecovery) this->PopulateKVStore(*kv_store, this->num_puts, &expected_keys_values); - // TODO: [Gabe Bornstein 3/17/26] Replace with fsync once it's implemented. - // - std::this_thread::sleep_for(std::chrono::seconds(1)); + BATT_CHECK_OK(kv_store->sync()); this->ShutdownKVStore(kv_store); } @@ -585,10 +626,6 @@ TEST_P(KVStoreRecoveryTest, KVStoreRecovery) ASSERT_TRUE(recovered_kv_store.ok()) << BATT_INSPECT(recovered_kv_store.status()); - // TODO: [Gabe Bornstein 4/14/26] Replace with fsync once it's implemented. - // - std::this_thread::sleep_for(std::chrono::seconds(1)); - for (const auto& [key, expected_value] : expected_keys_values) { turtle_kv::KeyView key_view{key}; batt::StatusOr actual_value = (*recovered_kv_store)->get(key_view); @@ -604,6 +641,289 @@ TEST_P(KVStoreRecoveryTest, KVStoreRecovery) // deletes, not just inserts. // +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +TEST_F(KVStoreTest, SyncWriteOptions) +{ + std::filesystem::path test_kv_store_dir = this->data_root / "turtle_kv_Test" / "sync_write_opts"; + + std::map expected_keys_values; + std::set deleted_keys; + + { + StatusOr> open_result = this->CreateAndOpenKVStore(test_kv_store_dir); + ASSERT_TRUE(open_result.ok()) << BATT_INSPECT(open_result.status()); + + std::unique_ptr& kv_store = *open_result; + + KVStore::WriteOptions opts{.sync = true}; + this->PopulateKVStore(*kv_store, 50, &expected_keys_values, 0.25, &deleted_keys, opts); + + this->ShutdownKVStore(kv_store); + } + + // Recover and verify state. + // + { + StatusOr> recovered_kv_store = + turtle_kv::KVStore::open(test_kv_store_dir, + this->kv_store_config.tree_options, + this->runtime_options); + + ASSERT_TRUE(recovered_kv_store.ok()) << BATT_INSPECT(recovered_kv_store.status()); + + for (const auto& [key, expected_value] : expected_keys_values) { + StatusOr actual_value = (*recovered_kv_store)->get(KeyView{key}); + + if (deleted_keys.count(key)) { + ASSERT_EQ(actual_value.status(), batt::StatusCode::kNotFound); + } else { + ASSERT_TRUE(actual_value.ok()) << "Didn't find key after recovery: " << key; + EXPECT_EQ(actual_value->as_str(), expected_value) + << "Wrong value for key after recovery: " << key; + } + } + + (*recovered_kv_store)->halt(); + (*recovered_kv_store)->join(); + } +} + +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +TEST_F(KVStoreTest, SyncExplicit) +{ + std::filesystem::path test_kv_store_dir = + this->data_root / "turtle_kv_Test" / "sync_explicit_recovery"; + + std::map expected_keys_values; + std::set deleted_keys; + + { + StatusOr> open_result = this->CreateAndOpenKVStore(test_kv_store_dir); + ASSERT_TRUE(open_result.ok()) << BATT_INSPECT(open_result.status()); + + std::unique_ptr& kv_store = *open_result; + + this->PopulateKVStore(*kv_store, 200, &expected_keys_values, 0.25, &deleted_keys); + + BATT_CHECK_OK(kv_store->sync()); + + this->ShutdownKVStore(kv_store); + } + + // Recover and verify state. + // + { + StatusOr> recovered_kv_store = + turtle_kv::KVStore::open(test_kv_store_dir, + this->kv_store_config.tree_options, + this->runtime_options); + + ASSERT_TRUE(recovered_kv_store.ok()) << BATT_INSPECT(recovered_kv_store.status()); + + for (const auto& [key, expected_value] : expected_keys_values) { + StatusOr actual_value = (*recovered_kv_store)->get(KeyView{key}); + + if (deleted_keys.count(key)) { + ASSERT_EQ(actual_value.status(), batt::StatusCode::kNotFound); + } else { + ASSERT_TRUE(actual_value.ok()) << "Didn't find key after recovery: " << key; + EXPECT_EQ(actual_value->as_str(), expected_value) + << "Wrong value for key after recovery: " << key; + } + } + + (*recovered_kv_store)->halt(); + (*recovered_kv_store)->join(); + } +} + +//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - +// +TEST_F(KVStoreTest, SyncMultithreadedStress) +{ + std::filesystem::path test_kv_store_dir = + this->data_root / "turtle_kv_Test" / "sync_multithread_stress"; + + StatusOr> open_result = this->CreateAndOpenKVStore(test_kv_store_dir); + ASSERT_TRUE(open_result.ok()) << BATT_INSPECT(open_result.status()); + + std::unique_ptr& kv_store = *open_result; + + const usize num_threads = std::thread::hardware_concurrency(); + const usize ops_per_thread = 5000; + + struct PerThreadState { + std::unordered_map live_keys; + std::unordered_set removed_keys; + usize sync_ok_count = 0; + usize sync_error_count = 0; + usize put_ok_count = 0; + usize put_error_count = 0; + usize remove_ok_count = 0; + usize remove_error_count = 0; + }; + + ObjectThreadStorage::ScopedSlot per_thread_state; + + std::vector threads; + + std::barrier workload_done{isize(num_threads + 1), batt::DoNothing{}}; + std::barrier ok_to_exit{isize(num_threads + 1), batt::DoNothing{}}; + + for (usize t = 0; t < num_threads; ++t) { + threads.emplace_back( + [&kv_store, &per_thread_state, &workload_done, &ok_to_exit, this, thread_id = t]() { + std::default_random_engine thread_rng{(usize)(42 + thread_id)}; + RandomStringGenerator gen_key{}; + + PerThreadState& state = per_thread_state.get(); + + for (usize i = 0; i < ops_per_thread; ++i) { + std::string key = gen_key(thread_rng); + std::string value = this->generate_value(); + + // Alternate between sync put, non-sync put + explicit sync, non-sync put, and remove. + // + const usize op = i % 4; + if (op == 0) { + // Sync put. + // + KVStore::WriteOptions opts{.sync = true}; + StatusOr result = + kv_store->put(KeyView{key}, ValueView::from_str(value), opts); + + if (result.ok()) { + state.put_ok_count += 1; + state.sync_ok_count += 1; + state.live_keys[key] = value; + state.removed_keys.erase(key); + } else { + state.put_error_count += 1; + state.sync_error_count += 1; + } + + } else if (op == 1) { + // Non-sync put followed by explicit sync. + // + Status put_result = kv_store->put(KeyView{key}, ValueView::from_str(value)); + + if (!put_result.ok()) { + state.put_error_count += 1; + } else { + state.put_ok_count += 1; + state.live_keys[key] = value; + state.removed_keys.erase(key); + + Status sync_result = kv_store->sync(); + if (sync_result.ok()) { + state.sync_ok_count += 1; + } else { + state.sync_error_count += 1; + } + } + } else if (op == 2) { + // Non-sync put (no sync at all). + // + Status put_result = kv_store->put(KeyView{key}, ValueView::from_str(value)); + + if (put_result.ok()) { + state.put_ok_count += 1; + state.live_keys[key] = value; + state.removed_keys.erase(key); + } else { + state.put_error_count += 1; + } + + } else { + // Remove a key that this thread previously inserted. + // + if (state.live_keys.empty()) { + continue; + } + std::string remove_key = state.live_keys.begin()->first; + + KVStore::WriteOptions opts{.sync = true}; + StatusOr result = kv_store->remove(KeyView{remove_key}, opts); + if (result.ok()) { + state.remove_ok_count += 1; + state.sync_ok_count += 1; + state.live_keys.erase(remove_key); + state.removed_keys.insert(remove_key); + } else { + state.remove_error_count += 1; + state.sync_error_count += 1; + } + } + } + + // Signal to the main test thread that we are done. + // + workload_done.arrive_and_wait(); + + // Wait for the test thread to finish inspecting per-thread state before exiting. + // + ok_to_exit.arrive_and_wait(); + }); + } + + workload_done.arrive_and_wait(); + + // Final sync to ensure everything is flushed. + // + Status final_sync = kv_store->sync(); + ASSERT_TRUE(final_sync.ok()) << BATT_INSPECT(final_sync); + + // Verify all live keys are readable with correct values, and removed keys are gone. + // + usize visit_count = 0; + per_thread_state.visit_each([&](PerThreadState& state) -> bool { + ++visit_count; + + // Each time through the loop does exactly one put or remove. + // + EXPECT_EQ(state.put_ok_count + state.remove_ok_count, ops_per_thread); + + // Syncs happen on 3 of 4 ops; but removes don't always happen. + // + EXPECT_GE(state.sync_ok_count, ops_per_thread / 2); + EXPECT_LE(state.sync_ok_count, ops_per_thread * 3 / 4); + + // No errors, please! + // + EXPECT_EQ(state.put_error_count, 0); + EXPECT_EQ(state.sync_error_count, 0); + EXPECT_EQ(state.remove_error_count, 0); + + for (const auto& [key, expected_value] : state.live_keys) { + StatusOr actual_value = kv_store->get(KeyView{key}); + EXPECT_TRUE(actual_value.ok()) << "Missing key: " << key; + if (actual_value.ok()) { + EXPECT_EQ(actual_value->as_str(), expected_value) << "Wrong value for key: " << key; + } + } + + for (const std::string& key : state.removed_keys) { + StatusOr actual_value = kv_store->get(KeyView{key}); + EXPECT_EQ(actual_value.status(), batt::StatusCode::kNotFound) + << "Key should have been removed: " << key; + } + + return false; + }); + EXPECT_EQ(visit_count, num_threads); + + // Allow the threads to continue past the second barrier, then join all. + // + ok_to_exit.arrive_and_wait(); + for (auto& t : threads) { + t.join(); + } + + this->ShutdownKVStore(kv_store); +} + } // namespace //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - diff --git a/src/turtle_kv/kv_store_config.hpp b/src/turtle_kv/kv_store_config.hpp index c53210c..6024a03 100644 --- a/src/turtle_kv/kv_store_config.hpp +++ b/src/turtle_kv/kv_store_config.hpp @@ -72,6 +72,23 @@ struct KVStoreRuntimeOptions { static Self with_default_values() noexcept; }; +//=#=#==#==#===============+=+=+=+=++=++++++++++++++-++-+--+-+----+--------------- +// +/** \brief Options for KVStore write operations. + */ +struct KVStoreWriteOptions { + /** \brief If true, the write operation blocks until the data is guaranteed to be durable. + */ + bool sync = false; + + /** \brief Controls the polling behavior of the background task that writes blocks of edit data + * to disk. If true, sync must also be true and the background task will more aggressively poll + * for blocks to write. Setting to `true` may reduce the latency of a sync operation, but it may + * cause the change log storage utilization efficiency to decrease. + */ + bool urgent_sync = false; +}; + BATT_OBJECT_PRINT_IMPL((inline), KVStoreRuntimeOptions, (initial_checkpoint_distance, diff --git a/src/turtle_kv/mem_table/mem_table.hpp b/src/turtle_kv/mem_table/mem_table.hpp index 780c7b4..c135c2b 100644 --- a/src/turtle_kv/mem_table/mem_table.hpp +++ b/src/turtle_kv/mem_table/mem_table.hpp @@ -173,7 +173,9 @@ class BasicMemTable : public MemTableBase /** \brief Applies a single key/value update to the MemTable, recording the update in the * change log via the passed context. */ - Status put(StorageWriterContext& context, const KeyView& key, const ValueView& value) noexcept; + StatusOr put(StorageWriterContext& context, + const KeyView& key, + const ValueView& value) noexcept; /** \brief Returns the value currently bound to the passed key, if present; otherwise, returns * None. diff --git a/src/turtle_kv/mem_table/mem_table.ipp b/src/turtle_kv/mem_table/mem_table.ipp index 4f5930b..0ec2d8e 100644 --- a/src/turtle_kv/mem_table/mem_table.ipp +++ b/src/turtle_kv/mem_table/mem_table.ipp @@ -72,7 +72,7 @@ BasicMemTable::~BasicMemTable() noexcept //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - - // template -Status BasicMemTable::put( +StatusOr BasicMemTable::put( StorageWriterContext& storage_writer_context, const KeyView& key, const ValueView& value) noexcept @@ -101,9 +101,9 @@ Status BasicMemTable::put( BATT_REQUIRE_OK(this->art_index_.insert(key, inserter)); BATT_CHECK_NOT_NULLPTR(inserter.entry_out); - } - return OkStatus(); + return inserter.edit_range_out.upper_bound; + } // // ~on_scope_exit calls commit_edit. } diff --git a/src/turtle_kv/mem_table/mem_table.test.cpp b/src/turtle_kv/mem_table/mem_table.test.cpp index 4ad2647..68a684b 100644 --- a/src/turtle_kv/mem_table/mem_table.test.cpp +++ b/src/turtle_kv/mem_table/mem_table.test.cpp @@ -195,16 +195,17 @@ class MemTableTest : public ::testing::Test })); } - Status status = this->mem_table->put(this->storage_writer_context, key, value); - EXPECT_EQ((status == batt::StatusCode::kResourceExhausted), expect_overflow); + StatusOr edit_status = + this->mem_table->put(this->storage_writer_context, key, value); + EXPECT_EQ((edit_status.status() == batt::StatusCode::kResourceExhausted), expect_overflow); - if (status.ok()) { + if (edit_status.ok()) { this->total_inserted_items_size += item_size; } Mock::VerifyAndClearExpectations(&this->storage_writer_context); - return status; + return edit_status.status(); } /** \brief Returns a worst-case estimate of the number of batches which will be produced by the diff --git a/src/turtle_kv/mem_table/mem_table_entry.hpp b/src/turtle_kv/mem_table/mem_table_entry.hpp index 6224ad9..741dc2b 100644 --- a/src/turtle_kv/mem_table/mem_table_entry.hpp +++ b/src/turtle_kv/mem_table/mem_table_entry.hpp @@ -21,6 +21,7 @@ #include #include +#include #include #include @@ -142,6 +143,7 @@ struct MemTableValueEntryInserter { // Outputs MemTableValueEntry* entry_out = nullptr; + Interval edit_range_out; //+++++++++++-+-+--+----- --- -- - - - - @@ -157,6 +159,7 @@ struct MemTableValueEntryInserter { pack_key_value_slot(this->key, this->value, buffer); this->entry_out = new (entry_memory) MemTableValueEntry{packed_pair, edit_offset}; + this->set_edit_range_out(buffer, edit_offset); })); return OkStatus(); @@ -175,12 +178,19 @@ struct MemTableValueEntryInserter { p_entry->update_value(packed_pair, edit_offset); this->entry_out = p_entry; + this->set_edit_range_out(buffer, edit_offset); })); return OkStatus(); } //+++++++++++-+-+--+----- --- -- - - - - + private: + void set_edit_range_out(const MutableBuffer& buffer, EditOffset edit_offset) noexcept + { + this->edit_range_out.lower_bound = edit_offset.value(); + this->edit_range_out.upper_bound = this->edit_range_out.lower_bound + buffer.size(); + } static_assert(MemTableEntryInserter); };