Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
Original file line number Diff line number Diff line change
Expand Up @@ -202,8 +202,10 @@ waits for that gate.
produce a suffix that is dead from birth, which no removal will kill; the
splitting lane queues it with the prefix as its parent.
- **Background sweep.** `ThreadPoolIndexer` runs a `kv-maint` thread that
sweeps once 1,024 nodes have died since the last sweep or 30 s have passed,
checking every 50 ms. The walk takes no gates: it records holder-less
sweeps once the nodes that died since the last sweep reach the larger of
1,024 and an eighth of the nodes that sweep visited, or 30 s have passed,
checking every 50 ms. The walk visits the whole tree, so this keeps it to at
most eight node visits per death. The walk takes no gates: it records holder-less
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
children under the root and every anchor, descending only into nodes with
children. Each parent's candidates are re-checked and unlinked under one
try-locked gate, deepest parents first. The sweep catches what the drain
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,8 @@ pub struct ConcurrentRadixTreeCompressed {
sweep_schedule: SweepSchedule,
/// Nodes that lost their last holder since the last stale-leaf sweep began.
deaths_since_sweep: AtomicU64,
/// When the last stale-leaf sweep began.
last_sweep: parking_lot::Mutex<Instant>,
/// When the last stale-leaf sweep began, and how many nodes it visited.
last_sweep: parking_lot::Mutex<LastSweep>,
#[cfg(any(test, feature = "bench"))]
bench_metrics: CrtcBenchMetrics,
}
Expand All @@ -66,6 +66,10 @@ const SWEEP_TICK: Duration = Duration::from_millis(50);
const SWEEP_AFTER_DEATHS: u64 = 1024;
/// ...or once this long has passed, for nodes no death was counted for.
const SWEEP_MAX_INTERVAL: Duration = Duration::from_secs(30);
/// A sweep walks the whole tree, so it also waits for deaths in proportion to the nodes
/// the last sweep visited: one per this many. The walk then costs at most this many node
/// visits per death, and the dead leaves the lanes miss stay a small share of the tree.
const SWEEP_VISITS_PER_DEATH: u64 = 8;

/// Most queued dead leaves an idle event lane unlinks before it checks for new events.
const IDLE_UNLINK_LIMIT: usize = 256;
Expand All @@ -80,6 +84,12 @@ struct SweepSchedule {
max_interval: Duration,
}

struct LastSweep {
began: Instant,
/// Nodes the sweep's walk visited; zero before the first sweep.
nodes: u64,
}

#[cfg(any(test, feature = "bench"))]
struct CrtcBenchMetrics {
node_splits: AtomicU64,
Expand Down Expand Up @@ -152,7 +162,10 @@ impl ConcurrentRadixTreeCompressed {
max_interval: SWEEP_MAX_INTERVAL,
},
deaths_since_sweep: AtomicU64::new(0),
last_sweep: parking_lot::Mutex::new(Instant::now()),
last_sweep: parking_lot::Mutex::new(LastSweep {
began: Instant::now(),
nodes: 0,
}),
#[cfg(any(test, feature = "bench"))]
bench_metrics: CrtcBenchMetrics::new(),
}
Expand Down Expand Up @@ -181,19 +194,31 @@ impl ConcurrentRadixTreeCompressed {
self.deaths_since_sweep.fetch_add(1, Ordering::Relaxed);
}

/// One maintenance tick: sweeps stale leaves once enough nodes have died or the
/// schedule's longest interval has passed.
/// One maintenance tick: sweeps stale leaves once enough nodes have died, for the
/// schedule and the size of the tree the last sweep walked, or the schedule's longest
/// interval has passed.
fn sweep_if_due(&self) {
let schedule = &self.sweep_schedule;
let due = self.deaths_since_sweep.load(Ordering::Relaxed) >= schedule.after_deaths
|| self.last_sweep.lock().elapsed() >= schedule.max_interval;
let mut last = self.last_sweep.lock();
let after_deaths = schedule
.after_deaths
.max(last.nodes / SWEEP_VISITS_PER_DEATH);
let due = self.deaths_since_sweep.load(Ordering::Relaxed) >= after_deaths
|| last.began.elapsed() >= schedule.max_interval;
if !due {
return;
}
// Deaths during the sweep count toward the next one.
self.deaths_since_sweep.store(0, Ordering::Relaxed);
*self.last_sweep.lock() = Instant::now();
self.sweep_stale_children();
last.began = Instant::now();
// Only the maintenance thread sweeps on schedule, so holding the lock blocks no one.
let sweep = self.sweep_stale_children();
tracing::debug!(
visited = sweep.visited,
unlinked = sweep.unlinked,
"Swept stale index leaves"
);
last.nodes = sweep.visited;
}

#[cfg(test)]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,28 +3,37 @@

use super::*;

/// What one stale-leaf sweep did.
pub(super) struct Sweep {
pub(super) unlinked: usize,
/// Nodes the walk visited.
pub(super) visited: u64,
}

impl ConcurrentRadixTreeCompressed {
#[cfg(test)]
pub(crate) fn run_cleanup_for_test(&self) -> usize {
self.sweep_stale_children()
self.sweep_stale_children().unlinked
}

/// Unlinks holder-less, childless leaves that nothing else references, under the root
/// and every anchor, and returns how many it unlinked.
/// and every anchor.
///
/// The walk takes no gate: it records only holder-less children and descends only
/// into nodes with children, visiting children by reference. Each parent's candidates
/// are then re-checked and unlinked under one try-locked exclusive gate, deepest
/// parents first, so a dead chain goes in one pass. A parent whose gate stays taken
/// keeps its dead children until the next sweep.
pub(super) fn sweep_stale_children(&self) -> usize {
pub(super) fn sweep_stale_children(&self) -> Sweep {
const GATE_ATTEMPTS: usize = 4;
let mut queue = VecDeque::from([self.root.clone()]);
queue.extend(self.anchor_nodes.iter().map(|entry| entry.value().clone()));
let mut candidates = Vec::new();
let mut visited = 0u64;

while let Some(parent) = queue.pop_front() {
parent.for_each_child(|key, child| {
visited += 1;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
if child.is_holderless() {
candidates.push(CleanupEdge {
parent: Arc::downgrade(&parent),
Expand Down Expand Up @@ -76,7 +85,10 @@ impl ConcurrentRadixTreeCompressed {
}
}
}
removed
Sweep {
unlinked: removed,
visited,
}
}

/// Queues a leaf this lane's removal just killed for the lane's next idle drain. A
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1979,6 +1979,30 @@ mod reclamation_tests {
assert_eq!(dead_node_count(&index), 0);
}

#[test]
fn the_sweep_waits_for_deaths_in_proportion_to_the_tree_it_last_walked() {
// Sixteen root leaves; deaths are never drained, so only the sweep unlinks them.
let index = ConcurrentRadixTreeCompressed::new().with_sweep_schedule(None, 1, HOUR);
let mut lookup = direct_lookup();
for hash in 1..=16 {
apply(&index, &mut lookup, make_store_event(0, &[hash]));
}
let mut kill = |hash| apply(&index, &mut lookup, make_remove_event(0, &[hash]));

// Before any sweep, one death meets the threshold, and the walk visits 16 nodes.
kill(1);
index.sweep_if_due();
assert_eq!(dead_node_count(&index), 0);

// Now the sweep waits for 16 / SWEEP_VISITS_PER_DEATH = 2 deaths.
kill(2);
index.sweep_if_due();
assert_eq!(dead_node_count(&index), 1);
kill(3);
index.sweep_if_due();
assert_eq!(dead_node_count(&index), 0);
}

#[tokio::test]
async fn the_maintenance_thread_sweeps_a_leaf_the_lane_cannot_place() {
// One death meets a threshold of one, so a tick sweeps it.
Expand Down
Loading