#![forbid(unsafe_code)]
use std::collections::{HashMap, HashSet};
use crate::core::extent::ChunkId;
use crate::format::codec::CodecError;
use crate::store::Store;
use crate::store::StoreError;
use crate::store::inode::{Inode, InodeData};
use crate::store::object::Location;
use crate::store::root::Root;
use crate::store::segment::{self, SegmentWriter};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MarkKind {
Root,
Inode,
TreeInodeIndex,
TreeDirectory,
TreeExtent,
TreeChunkIndex,
TreeSnapshot,
TreeXattr,
Object,
}
pub struct LiveMark {
pub live: HashSet<ChunkId>,
pub referenced: HashSet<ChunkId>,
pub live_descriptors: HashSet<Vec<u8>>,
}
pub fn mark_live(store: &Store) -> Result<HashSet<ChunkId>, StoreError> {
Ok(mark_live_full(store)?.live)
}
pub fn mark_live_full(store: &Store) -> Result<LiveMark, StoreError> {
let mut live: HashSet<ChunkId> = HashSet::new();
let mut worklist: Vec<(ChunkId, MarkKind)> = Vec::new();
let mut referenced: HashSet<ChunkId> = HashSet::new();
let mut live_descriptors: HashSet<Vec<u8>> = HashSet::new();
worklist.push((store.current_root().id(), MarkKind::Root));
let snapshots = crate::store::snapshot::list(
store.current_root().snapshot_tree_root,
crate::store::BTREE_ORDER,
store.config().limits.max_fanout,
store,
)?;
for (_, entry) in snapshots {
worklist.push((entry.root_id, MarkKind::Root));
}
while let Some((id, kind)) = worklist.pop() {
if !live.insert(id) {
continue;
}
match kind {
MarkKind::Root => {
let root = decode_root(store, &id)?;
worklist.push((root.inode_index_root, MarkKind::TreeInodeIndex));
worklist.push((root.chunk_index_root, MarkKind::TreeChunkIndex));
if !root.snapshot_tree_root.is_zero() {
worklist.push((root.snapshot_tree_root, MarkKind::TreeSnapshot));
}
if !root.model_index_root.is_zero() {
worklist.push((root.model_index_root, MarkKind::TreeChunkIndex));
}
}
MarkKind::Inode => {
let inode = decode_inode(store, &id)?;
if !inode.xattr_root.is_zero() {
worklist.push((inode.xattr_root, MarkKind::TreeXattr));
}
match &inode.data {
InodeData::Directory { dir_root } if !dir_root.is_zero() => {
worklist.push((*dir_root, MarkKind::TreeDirectory));
}
InodeData::File { extent_root } if !extent_root.is_zero() => {
worklist.push((*extent_root, MarkKind::TreeExtent));
}
_ => {}
}
}
MarkKind::TreeInodeIndex => walk_tree(
store,
&id,
TreeValue::InodeId,
&mut live,
&mut worklist,
&mut referenced,
&mut live_descriptors,
)?,
MarkKind::TreeDirectory => walk_tree(
store,
&id,
TreeValue::Directory,
&mut live,
&mut worklist,
&mut referenced,
&mut live_descriptors,
)?,
MarkKind::TreeExtent => walk_tree(
store,
&id,
TreeValue::ExtentDescriptor,
&mut live,
&mut worklist,
&mut referenced,
&mut live_descriptors,
)?,
MarkKind::TreeChunkIndex => walk_tree(
store,
&id,
TreeValue::ChunkIndexEntry,
&mut live,
&mut worklist,
&mut referenced,
&mut live_descriptors,
)?,
MarkKind::TreeSnapshot => walk_tree(
store,
&id,
TreeValue::Snapshot,
&mut live,
&mut worklist,
&mut referenced,
&mut live_descriptors,
)?,
MarkKind::TreeXattr => walk_tree(
store,
&id,
TreeValue::Xattr,
&mut live,
&mut worklist,
&mut referenced,
&mut live_descriptors,
)?,
MarkKind::Object => {}
}
}
let limits = store.config().limits;
let mut queue: Vec<ChunkId> = referenced.iter().copied().collect();
let mut seen: HashSet<ChunkId> = HashSet::new();
while let Some(cid) = queue.pop() {
if !seen.insert(cid) {
continue;
}
let Some(bytes) = store.chunk_descriptor(&cid)? else {
continue;
};
let desc = match crate::format::descriptor::decode(
&bytes,
limits.max_descriptor_bytes,
limits.max_inline_bytes,
limits.max_palette,
limits.max_period,
limits.max_chunk_size,
) {
Ok(d) => d,
Err(_) => continue,
};
mark_descriptor_refs(&bytes, store, &mut live, &mut worklist)?;
use crate::core::representation::Representation;
let next = match &desc {
Representation::ExactRef { target, .. } => Some(*target),
Representation::BaseResidual { base, .. } => Some(*base),
_ => None,
};
if let Some(n) = next {
if !seen.contains(&n) {
queue.push(n);
}
}
}
Ok(LiveMark {
live,
referenced: seen,
live_descriptors,
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TreeValue {
InodeId,
Directory,
ExtentDescriptor,
ChunkIndexEntry,
Snapshot,
Xattr,
}
fn walk_tree(
store: &Store,
node_id: &ChunkId,
value_kind: TreeValue,
live: &mut HashSet<ChunkId>,
worklist: &mut Vec<(ChunkId, MarkKind)>,
referenced: &mut HashSet<ChunkId>,
live_descriptors: &mut HashSet<Vec<u8>>,
) -> Result<(), StoreError> {
if node_id.is_zero() {
return Ok(());
}
let payload = store
.fetch_object(node_id)?
.ok_or_else(|| StoreError::Invariant(format!("missing tree node {node_id}")))?;
let node = crate::store::index::Node::decode(
&payload,
crate::store::BTREE_ORDER,
store.config().limits.max_fanout,
)
.map_err(|e| StoreError::Index(e.to_string()))?;
match node {
crate::store::index::Node::Internal {
first_child,
entries,
} => {
let kind = match value_kind {
TreeValue::InodeId => MarkKind::TreeInodeIndex,
TreeValue::Directory => MarkKind::TreeDirectory,
TreeValue::ExtentDescriptor => MarkKind::TreeExtent,
TreeValue::ChunkIndexEntry => MarkKind::TreeChunkIndex,
TreeValue::Snapshot => MarkKind::TreeSnapshot,
TreeValue::Xattr => MarkKind::TreeXattr,
};
worklist.push((first_child, kind));
for e in entries {
let child = ChunkId::new(e.value.as_slice().try_into().expect("32-byte id"));
worklist.push((child, kind));
}
}
crate::store::index::Node::Leaf { entries } => {
for e in entries {
match value_kind {
TreeValue::InodeId => {
let inode_id =
ChunkId::new(e.value.as_slice().try_into().map_err(|_| {
StoreError::Invariant("inode value not 32 bytes".into())
})?);
worklist.push((inode_id, MarkKind::Inode));
}
TreeValue::Directory | TreeValue::Xattr | TreeValue::ChunkIndexEntry => {}
TreeValue::ExtentDescriptor => {
mark_descriptor_refs(&e.value, store, live, worklist)?;
collect_descriptor_refs(&e.value, store, referenced)?;
live_descriptors.insert(e.value.clone());
}
TreeValue::Snapshot => {
let entry = crate::store::snapshot::SnapshotEntry::decode(&e.value)
.map_err(|e| StoreError::Descriptor(e.to_string()))?;
worklist.push((entry.root_id, MarkKind::Root));
}
}
}
}
}
Ok(())
}
fn collect_descriptor_refs(
bytes: &[u8],
store: &Store,
referenced: &mut HashSet<ChunkId>,
) -> Result<(), StoreError> {
let l = store.config().limits;
let desc = match crate::format::descriptor::decode(
bytes,
l.max_descriptor_bytes,
l.max_inline_bytes,
l.max_palette,
l.max_period,
l.max_chunk_size,
) {
Ok(d) => d,
Err(_) => return Ok(()),
};
use crate::core::representation::Representation;
match &desc {
Representation::ExactRef { target, .. } => {
referenced.insert(*target);
}
Representation::BaseResidual { base, .. } => {
referenced.insert(*base);
}
Representation::SequenceDict { dictionary, .. } => {
referenced.insert(*dictionary);
}
Representation::SequenceSharedDict {
dictionary, shared, ..
} => {
if !dictionary.is_zero() {
referenced.insert(*dictionary);
}
referenced.insert(*shared);
}
_ => {}
}
Ok(())
}
fn mark_descriptor_refs(
bytes: &[u8],
store: &Store,
live: &mut HashSet<ChunkId>,
worklist: &mut Vec<(ChunkId, MarkKind)>,
) -> Result<(), StoreError> {
let l = store.config().limits;
let desc = match crate::format::descriptor::decode(
bytes,
l.max_descriptor_bytes,
l.max_inline_bytes,
l.max_palette,
l.max_period,
l.max_chunk_size,
) {
Ok(d) => d,
Err(_) => return Ok(()), };
use crate::core::representation::{Representation, Residual};
let mut refs = Vec::new();
match &desc {
Representation::Raw { obj, .. } => refs.push(*obj),
Representation::Rans { model, enc_obj, .. } => {
refs.push(*model);
refs.push(*enc_obj);
}
Representation::SequenceRans { model, enc_obj, .. } => {
refs.push(*model);
refs.push(*enc_obj);
}
Representation::SparseBlock64 { model, enc_obj, .. } => {
refs.push(*model);
refs.push(*enc_obj);
}
Representation::SequenceDict { model, enc_obj, .. } => {
refs.push(*model);
refs.push(*enc_obj);
}
Representation::SequenceSharedDict { model, enc_obj, .. } => {
refs.push(*model);
refs.push(*enc_obj);
}
Representation::BaseResidual {
residual: Residual::RansCoded { enc_obj, model, .. },
..
}
| Representation::EntropyRef {
residual: Residual::RansCoded { enc_obj, model, .. },
..
} => {
refs.push(*enc_obj);
refs.push(*model);
}
Representation::BaseResidual {
residual: Residual::BaseSequence { enc_obj, model, .. },
..
} => {
refs.push(*enc_obj);
refs.push(*model);
}
_ => {}
}
for r in refs {
if live.insert(r) {
worklist.push((r, MarkKind::Object));
}
}
Ok(())
}
fn decode_root(store: &Store, id: &ChunkId) -> Result<Root, StoreError> {
let payload = store
.fetch_object(id)?
.ok_or_else(|| StoreError::Invariant(format!("missing root object {id}")))?;
Root::decode(&payload).map_err(|e| StoreError::Superblock(e.to_string()))
}
fn decode_inode(store: &Store, id: &ChunkId) -> Result<Inode, StoreError> {
let payload = store
.fetch_object(id)?
.ok_or_else(|| StoreError::Invariant(format!("missing inode object {id}")))?;
Inode::decode(&payload).map_err(|e| StoreError::Descriptor(e.to_string()))
}
pub fn live_ratios(
store: &Store,
live: &HashSet<ChunkId>,
) -> Result<HashMap<u64, (u64, u64)>, StoreError> {
let mut map: HashMap<u64, (u64, u64)> = HashMap::new(); for (id, loc) in store.object_index().iter() {
let entry = map.entry(loc.segment_seq).or_insert((0, 0));
entry.1 += loc.total_size();
if live.contains(&id) {
entry.0 += loc.total_size();
}
}
Ok(map)
}
fn collect_tree_node_ids(
store: &Store,
root: &ChunkId,
out: &mut HashSet<ChunkId>,
) -> Result<(), StoreError> {
if root.is_zero() {
return Ok(());
}
let mut stack = vec![*root];
while let Some(id) = stack.pop() {
if !out.insert(id) {
continue;
}
let payload = store
.fetch_object(&id)?
.ok_or_else(|| StoreError::Invariant(format!("missing tree node {id}")))?;
let node = crate::store::index::Node::decode(
&payload,
crate::store::BTREE_ORDER,
store.config().limits.max_fanout,
)
.map_err(|e| StoreError::Index(e.to_string()))?;
match node {
crate::store::index::Node::Internal {
first_child,
entries,
} => {
stack.push(first_child);
for e in entries {
let child = ChunkId::new(e.value.as_slice().try_into().expect("32-byte id"));
stack.push(child);
}
}
crate::store::index::Node::Leaf { .. } => {}
}
}
Ok(())
}
struct RebuildProvider<'a> {
writer: &'a mut SegmentWriter,
new_seq: u64,
pending: HashMap<ChunkId, Vec<u8>>,
new_locations: &'a mut Vec<(ChunkId, Location)>,
}
impl crate::store::index::ObjectProvider for RebuildProvider<'_> {
fn get(&self, id: &ChunkId) -> Result<Option<Vec<u8>>, crate::store::index::BTreeError> {
Ok(self.pending.get(id).cloned())
}
fn put(&mut self, id: ChunkId, bytes: Vec<u8>) {
if self.pending.contains_key(&id) {
return;
}
let encoded = crate::format::record::encode(
crate::format::version::RecordTag::BtreeNode,
0,
None,
&bytes,
);
let offset = self.writer.durable_end() + self.writer.buffered_len();
self.writer.append(encoded);
self.new_locations.push((
id,
Location {
segment_seq: self.new_seq,
offset,
stored_len: bytes.len() as u64,
materialized_len: None,
tag: crate::format::version::RecordTag::BtreeNode,
},
));
self.pending.insert(id, bytes);
}
}
struct RebuiltIndex {
root: ChunkId,
old_only: HashSet<ChunkId>,
old_nodes: HashSet<ChunkId>,
}
fn rebuild_chunk_index(
store: &Store,
writer: &mut SegmentWriter,
new_seq: u64,
mark: &LiveMark,
new_locations: &mut Vec<(ChunkId, Location)>,
) -> Result<RebuiltIndex, StoreError> {
let limits = store.config().limits;
let old_root = store.current_root().chunk_index_root;
let mut old_nodes: HashSet<ChunkId> = HashSet::new();
if !old_root.is_zero() {
collect_tree_node_ids(store, &old_root, &mut old_nodes)?;
}
let mut kept: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
if !old_root.is_zero() {
let entries = crate::store::index::scan_all(
old_root,
crate::store::BTREE_ORDER,
limits.max_fanout,
store,
)?;
for (key, value) in entries {
let cid = ChunkId::new(
key.as_slice()
.try_into()
.map_err(|_| StoreError::Invariant("chunk index key not 32 bytes".into()))?,
);
if mark.referenced.contains(&cid) || mark.live_descriptors.contains(&value) {
kept.push((key, value));
}
}
}
let mut provider = RebuildProvider {
writer,
new_seq,
pending: HashMap::new(),
new_locations,
};
let mut new_root = ChunkId::ZERO;
for (key, value) in &kept {
new_root = crate::store::index::insert(
new_root,
key,
value,
crate::store::BTREE_ORDER,
limits.max_fanout,
&mut provider,
)?;
}
let new_nodes: HashSet<ChunkId> = provider.pending.keys().copied().collect();
let old_only: HashSet<ChunkId> = old_nodes.difference(&new_nodes).copied().collect();
Ok(RebuiltIndex {
root: new_root,
old_only,
old_nodes,
})
}
pub fn collect(
store: &Store,
hooks: &crate::store::transaction::CrashHooks,
) -> Result<u64, StoreError> {
let mark = mark_live_full(store)?;
let live = &mark.live;
let ratios = live_ratios(store, live)?;
let target = store.config().gc_target_ratio;
let victims: Vec<u64> = ratios
.iter()
.filter(|(_, (live_b, total))| total > &0 && (*live_b as f64 / *total as f64) < target)
.map(|(seq, _)| *seq)
.collect();
if victims.is_empty() {
return Ok(0);
}
let new_seq = store.current_segment_seq() + 1;
let mut writer = SegmentWriter::open(store.dir(), new_seq)?;
let mut new_locations: Vec<(ChunkId, Location)> = Vec::new();
let rebuilt = rebuild_chunk_index(store, &mut writer, new_seq, &mark, &mut new_locations)?;
let mut reclaimable = 0u64;
for (id, loc) in store.object_index().iter() {
if victims.contains(&loc.segment_seq)
&& (!live.contains(&id) || rebuilt.old_only.contains(&id))
{
reclaimable += loc.total_size();
}
}
let mut copy_candidates: Vec<(ChunkId, Location)> = store
.object_index()
.iter()
.into_iter()
.filter(|(id, loc)| {
victims.contains(&loc.segment_seq)
&& live.contains(id)
&& !rebuilt.old_nodes.contains(id)
})
.collect();
copy_candidates.sort_by_key(|(_, loc)| (loc.segment_seq, loc.offset));
for (id, loc) in copy_candidates {
let payload = store.read_payload_at(&loc)?;
let flags = if loc.materialized_len.is_some() {
crate::format::record::FLAG_HAS_MATERIALIZED_LEN
} else {
0
};
let encoded = crate::format::record::encode(loc.tag, flags, loc.materialized_len, &payload);
let offset = writer.durable_end() + writer.buffered_len();
writer.append(encoded);
new_locations.push((
id,
Location {
segment_seq: new_seq,
offset,
stored_len: payload.len() as u64,
materialized_len: loc.materialized_len,
tag: loc.tag,
},
));
}
writer.flush()?;
writer.fdatasync()?;
SegmentWriter::sync_dir(store.dir())?;
let mut root = store.current_root();
root.chunk_index_root = rebuilt.root;
root.segment_seq = new_seq;
root.index_epoch = root.index_epoch.saturating_add(1);
root.generation = store.generation() + 1;
let root_bytes = root.encode();
let root_id = ChunkId::of(&root_bytes);
let encoded = crate::format::record::encode(
crate::format::version::RecordTag::Root,
0,
None,
&root_bytes,
);
let offset = writer.durable_end();
writer.append(encoded);
writer.flush()?;
writer.fdatasync()?;
hooks.hit(crate::store::transaction::CrashPoint::AfterRootWrite)?;
store.write_superblock(root_id, &root)?;
hooks.hit(crate::store::transaction::CrashPoint::AfterSuperblockWrite)?;
store.fsync_superblock()?;
hooks.hit(crate::store::transaction::CrashPoint::AfterSuperblockFsync)?;
for (id, loc) in new_locations {
store.object_index().insert(id, loc);
}
store.publish_commit(&root, root_id)?;
let root_loc = Location {
segment_seq: new_seq,
offset,
stored_len: root_bytes.len() as u64,
materialized_len: None,
tag: crate::format::version::RecordTag::Root,
};
store.object_index().insert(root_id, root_loc);
store.install_segment(writer);
hooks.hit(crate::store::transaction::CrashPoint::BeforeOldSegmentDelete)?;
for seq in &victims {
segment::delete_segment(store.dir(), *seq)?;
}
let dead: Vec<ChunkId> = store
.object_index()
.iter()
.into_iter()
.filter(|(id, loc)| {
victims.contains(&loc.segment_seq)
&& (!live.contains(id) || rebuilt.old_only.contains(id))
})
.map(|(id, _)| id)
.collect();
for id in dead {
store.object_index().remove(&id);
}
Ok(reclaimable)
}
pub fn unreachable_bytes(store: &Store) -> Result<u64, StoreError> {
let live = mark_live(store)?;
let mut unreachable = 0u64;
for (id, loc) in store.object_index().iter() {
if !live.contains(&id) {
unreachable += loc.total_size();
}
}
Ok(unreachable)
}
pub fn unreachable_bytes_by_record_tag(
store: &Store,
) -> Result<std::collections::BTreeMap<String, u64>, StoreError> {
let live = mark_live(store)?;
let mut by_tag: std::collections::BTreeMap<String, u64> = std::collections::BTreeMap::new();
for (id, loc) in store.object_index().iter() {
if !live.contains(&id) {
*by_tag.entry(format!("{:?}", loc.tag)).or_insert(0) += loc.total_size();
}
}
Ok(by_tag)
}
#[allow(unused_imports)]
use CodecError as _CodecError;