Skip to content
Open
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion conanfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ def requirements(self):
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.48.0 <1]", **VISIBLE)
self.requires("llfs/0.48.1#c96bb3c69c41c786cccab4a6e4f6e3ca", **VISIBLE)
Comment thread
gabrielbornstein marked this conversation as resolved.
Resolved
self.requires("pcg-cpp/cci.20220409", **VISIBLE)
self.requires("yaml-cpp/[>=0.9.0 <1]")
self.requires("zlib/1.3.1", **OVERRIDE)
Expand Down
78 changes: 54 additions & 24 deletions src/turtle_kv/checkpoint_generator.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -205,37 +205,42 @@ StatusOr<std::unique_ptr<CheckpointJob>> CheckpointGenerator::finalize_checkpoin
checkpoint_job->edit_offset_upper_bound = edit_offset_upper_bound;
checkpoint_job->batch_count = batch_count;

this->active_checkpoints_.push_back(PackedCheckpoint{
.edit_offset_upper_bound = this->base_checkpoint_.edit_offset_upper_bound().value(),
.new_tree_root = llfs::PackedPageId::from(this->base_checkpoint_.root_id()),
});
{
std::lock_guard<std::mutex> lock{this->checkpoint_mutex_};

checkpoint_job->active_checkpoints.emplace(
llfs::PackAsVariant<CheckpointLogEvent, PackedActiveCheckpoints>{
this->active_checkpoints_,
});
this->active_checkpoints_.push_back(PackedCheckpoint{
.edit_offset_upper_bound = this->base_checkpoint_.edit_offset_upper_bound().value(),
.new_tree_root = llfs::PackedPageId::from(this->base_checkpoint_.root_id()),
});

// Package the job up with a PackedActiveCheckpoints event record so we can append it to the Volume.
//
StatusOr<llfs::AppendableJob> appendable_job =
llfs::make_appendable_job(std::move(this->job_),
llfs::PackableRef{*checkpoint_job->active_checkpoints});
checkpoint_job->active_checkpoints.emplace(
llfs::PackAsVariant<CheckpointLogEvent, PackedActiveCheckpoints>{
this->active_checkpoints_,
});

BATT_REQUIRE_OK(appendable_job);
// Package the job up with a PackedActiveCheckpoints event record so we can append it to the
// Volume.
//
StatusOr<llfs::AppendableJob> appendable_job =
llfs::make_appendable_job(std::move(this->job_),
llfs::PackableRef{*checkpoint_job->active_checkpoints});

// Reserve slot grant for the current checkpoint in checkpoint-log.
//
auto grant_size = appendable_job->calculate_grant_size();
StatusOr<batt::Grant> checkpoint_grant = this->reserve_slot_grant_for_checkpoints(grant_size);
BATT_REQUIRE_OK(appendable_job);

BATT_REQUIRE_OK(checkpoint_grant);
// Reserve slot grant for the current checkpoint in checkpoint-log.
//
auto grant_size = appendable_job->calculate_grant_size();
StatusOr<batt::Grant> checkpoint_grant = this->reserve_slot_grant_for_checkpoints(grant_size);

checkpoint_job->append_job_grant.emplace(std::move(*checkpoint_grant));
checkpoint_job->appendable_job.emplace(std::move(*appendable_job));
checkpoint_job->prepare_slot_sequencer.emplace(this->slot_sequencer_);
BATT_REQUIRE_OK(checkpoint_grant);

this->base_job_ = checkpoint_job->appendable_job->job.finalized_job();
this->slot_sequencer_ = this->slot_sequencer_.get_next();
checkpoint_job->append_job_grant.emplace(std::move(*checkpoint_grant));
checkpoint_job->appendable_job.emplace(std::move(*appendable_job));
checkpoint_job->prepare_slot_sequencer.emplace(this->slot_sequencer_);

this->base_job_ = checkpoint_job->appendable_job->job.finalized_job();
this->slot_sequencer_ = this->slot_sequencer_.get_next();
}

BATT_CHECK_EQ(this->job_, nullptr);

Expand All @@ -244,4 +249,29 @@ StatusOr<std::unique_ptr<CheckpointJob>> CheckpointGenerator::finalize_checkpoin
return {std::move(checkpoint_job)};
}

//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - -
//
Status CheckpointGenerator::delete_checkpoint(
EditOffset edit_offset,
PackedActiveCheckpoints& active_checkpoints_out) noexcept
{
std::lock_guard<std::mutex> lock{this->checkpoint_mutex_};

if (this->active_checkpoints_.num_active_checkpoints == 0) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Re: your comment in PackedActiveCheckpoints::erase: these checks should move into that function, making this one as simple as 1. lock the mutex, 2. call erase.

return {batt::StatusCode::kNotFound};
}

if (edit_offset.value() == this->active_checkpoints_.newest().edit_offset_upper_bound) {
return {batt::StatusCode::kFailedPrecondition};
}

if (!this->active_checkpoints_.erase(edit_offset.value())) {
return {batt::StatusCode::kNotFound};
}

active_checkpoints_out = this->active_checkpoints_;

return OkStatus();
}

} // namespace turtle_kv
14 changes: 14 additions & 0 deletions src/turtle_kv/checkpoint_generator.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
#include <batteries/metrics/metric_collectors.hpp>
#include <batteries/small_vec.hpp>

#include <mutex>

namespace turtle_kv {

//=#=#==#==#===============+=+=+=+=++=++++++++++++++-++-+--+-+----+---------------
Expand Down Expand Up @@ -100,6 +102,13 @@ class CheckpointGenerator
std::shared_ptr<batt::Grant::Issuer>&& token_issuer,
llfs::PageCacheOvercommit& overcommit) noexcept;

/** \brief Removes the checkpoint identified by `edit_offset` from the active set, durably writes
* the updated PackedActiveCheckpoints to the checkpoint volume, and returns the new
* PackedActiveCheckpoints via `active_checkpoints_out`.
*/
Status delete_checkpoint(EditOffset edit_offset,
PackedActiveCheckpoints& active_checkpoints_out) noexcept;

llfs::PageCacheJob& page_cache_job() const
{
return *this->job_;
Expand Down Expand Up @@ -194,6 +203,11 @@ class CheckpointGenerator
//
llfs::Volume& checkpoint_volume_;

// Guards active_checkpoints_ and checkpoint volume writes, ensuring mutual exclusion between
// finalize_checkpoint (update thread) and delete_checkpoint (caller thread).
//
std::mutex checkpoint_mutex_;

// The set of currently active (retained) checkpoints, in sorted order by
// edit_offset_upper_bound.
//
Expand Down
46 changes: 43 additions & 3 deletions src/turtle_kv/kv_store.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1055,6 +1055,20 @@ StatusOr<Snapshot> KVStore::get_snapshot(EditOffset checkpoint_edit_offset) noex
};
}

//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - -
//
StatusOr<Snapshot> KVStore::create_snapshot() noexcept
{
BATT_ASSIGN_OK_RESULT(EditOffset checkpoint_edit_offset, this->force_checkpoint());

// force_checkpoint only finalizes the active MemTable; the checkpoint itself is committed
// asynchronously, so wait for it to become active before looking it up.
//
BATT_REQUIRE_OK(this->wait_for_checkpoint(checkpoint_edit_offset));
Comment thread
gabrielbornstein marked this conversation as resolved.
Resolved

return this->get_snapshot(checkpoint_edit_offset);
}

//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - -
//
StatusOr<std::vector<Snapshot>> KVStore::get_active_snapshots() noexcept
Expand Down Expand Up @@ -1092,6 +1106,32 @@ StatusOr<std::vector<Snapshot>> KVStore::get_active_snapshots() noexcept
return snapshots;
}

//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - -
//
Status KVStore::delete_snapshot(EditOffset checkpoint_edit_offset) noexcept
{
// TODO: [Gabe Bornstein 10/12/26] commit_checkpoint does this. Do we need delete_snapshot to do

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we could replace this with BATT_CHECK_OK(this->wait_for_recovery()) -- At the point when its possible for application code to call delete_snapshot, we know recovery is done because KVStore::open has returned.

// it too?
//
BATT_REQUIRE_OK(this->wait_for_recovery());

PackedActiveCheckpoints updated_active_checkpoints;
BATT_REQUIRE_OK(this->checkpoint_generator_->delete_checkpoint(checkpoint_edit_offset,
updated_active_checkpoints));
{
batt::Toggle<State>::Writer state_writer{this->state_};
const State& old_state = state_writer.old_value();
State& new_state = state_writer.new_value();

new_state.mem_table_ = old_state.mem_table_;
new_state.deltas_ = old_state.deltas_;
new_state.base_checkpoint_ = old_state.base_checkpoint_;
new_state.active_checkpoints_ = updated_active_checkpoints;
}

return OkStatus();
}

//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - -
//
StatusOr<usize> KVStore::scan(
Expand Down Expand Up @@ -1484,9 +1524,9 @@ void KVStore::mem_table_batch_scanner_thread_main()
//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- - - - -
//
template <typename Fn>
requires std::invocable<Fn, std::unique_ptr<DeltaBatch>> Status
KVStore::scan_mem_table_to_build_batches(boost::intrusive_ptr<MemTable>&& mem_table,
Fn&& consume_fn)
requires std::invocable<Fn, std::unique_ptr<DeltaBatch>>
Status KVStore::scan_mem_table_to_build_batches(boost::intrusive_ptr<MemTable>&& mem_table,
Fn&& consume_fn)
{
MemTable::BatchCompactor batch_compactor{*mem_table,
/*byte_size_limit=*/this->tree_options_.flush_size()};
Expand Down
10 changes: 10 additions & 0 deletions src/turtle_kv/kv_store.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -273,10 +273,20 @@ class KVStore : public Table
*/
StatusOr<Snapshot> get_snapshot(EditOffset checkpoint_edit_offset) noexcept;

/** \brief Forces a checkpoint of all edits up to this point, waits for it to be durably
* committed, and returns a read-only Snapshot of that checkpoint.
*/
StatusOr<Snapshot> create_snapshot() noexcept;
Comment thread
gabrielbornstein marked this conversation as resolved.
Resolved

/** \brief Returns Snapshots for all currently active checkpoints, ordered oldest to newest.
*/
StatusOr<std::vector<Snapshot>> get_active_snapshots() noexcept;

/** \brief Deletes the snapshot/checkpoint identified by the given EditOffset.
* Returns kNotFound if no checkpoint exists at that offset.
*/
Status delete_snapshot(EditOffset checkpoint_edit_offset) noexcept;

std::function<void(std::ostream&)> debug_info() const noexcept;

void collect_stats(
Expand Down
Loading