lace_tokio/
compaction.rs

1//! Offline compaction of a native root (Lace-100 "Compacting a root").
2
3use std::path::Path;
4use std::sync::Arc;
5use std::sync::atomic::{AtomicBool, Ordering};
6
7use lace_sans_io::core::{MachineError, RetentionSweep};
8use lace_sans_io::runtime_support::compaction::DEFAULT_COMPACTION_BATCH_BYTES;
9pub use lace_sans_io::runtime_support::compaction::{ArenaRelease, CompactionReport};
10use lace_sans_io::runtime_support::native_filesystem_store::compact_filesystem_root;
11
12use crate::{Error, IntoPolicy, Result};
13
14/// Remove every record of the filesystem root at `root` that none of
15/// `policies` selects, and release its bytes in place, on a blocking thread.
16///
17/// The root must have no open Lace and no live arena-backed record in any
18/// process. A root holding an interrupted compaction resumes it without
19/// evaluating `policies`. Dropping the future stops the work at its next
20/// store operation or batch boundary; the root then resumes on the next call.
21pub async fn compact_filesystem<P: IntoPolicy>(
22    root: impl AsRef<Path>,
23    policies: impl IntoIterator<Item = P>,
24) -> Result<CompactionReport> {
25    compact_on_blocking_thread(root.as_ref(), policies, compact_filesystem_root).await
26}
27
28/// Remove every record of the LMDB root at `root` that none of `policies`
29/// selects, and release its arena bytes in place, on a blocking thread.
30///
31/// No other process may have the root open and no arena-backed record of it
32/// may be alive anywhere; other processes' opens wait until compaction ends. A
33/// root holding an interrupted compaction resumes it without evaluating
34/// `policies`. Dropping the future stops the work at its next store operation
35/// or batch boundary; the root then resumes on the next call.
36#[cfg(feature = "lmdb")]
37pub async fn compact_lmdb<P: IntoPolicy>(
38    root: impl AsRef<Path>,
39    policies: impl IntoIterator<Item = P>,
40) -> Result<CompactionReport> {
41    compact_on_blocking_thread(root.as_ref(), policies, |root, sweep, batch, stop| {
42        lmdb_lace::NativeLmdbStore::compact(root, sweep, batch, stop)
43    })
44    .await
45}
46
47/// One backend's synchronous compaction driver.
48type CompactRoot = fn(
49    &Path,
50    RetentionSweep,
51    u64,
52    &dyn Fn() -> bool,
53) -> std::result::Result<CompactionReport, MachineError>;
54
55/// Prepare `policies`, then run `compact` on a blocking thread with a stop
56/// flag that the returned future's drop sets.
57async fn compact_on_blocking_thread<P: IntoPolicy>(
58    root: &Path,
59    policies: impl IntoIterator<Item = P>,
60    compact: CompactRoot,
61) -> Result<CompactionReport> {
62    let policies = policies
63        .into_iter()
64        .map(IntoPolicy::into_policy)
65        .collect::<Result<Vec<_>>>()?;
66    let sweep = RetentionSweep::new(&policies)?;
67    let root = root.to_path_buf();
68    let stop = StopOnDrop(Arc::new(AtomicBool::new(false)));
69    let requested = Arc::clone(&stop.0);
70    let work = tokio::task::spawn_blocking(move || {
71        compact(&root, sweep, DEFAULT_COMPACTION_BATCH_BYTES, &|| {
72            requested.load(Ordering::Relaxed)
73        })
74    });
75    match work.await {
76        Ok(result) => result.map_err(|error| Error::from_machine(error, None)),
77        Err(error) => match error.try_into_panic() {
78            Ok(panic) => std::panic::resume_unwind(panic),
79            Err(error) => Err(Error::integration(format!(
80                "compaction task failed: {error}"
81            ))),
82        },
83    }
84}
85
86/// Requests a stop when the compaction future drops before completion.
87struct StopOnDrop(Arc<AtomicBool>);
88
89impl Drop for StopOnDrop {
90    fn drop(&mut self) {
91        self.0.store(true, Ordering::Relaxed);
92    }
93}
94
95#[cfg(test)]
96#[path = "compaction_tests.rs"]
97mod tests;
lace_tokio/compaction.rs

lace_tokio/
compaction.rs
