hyperfile_cleaner/compact.rs
1//! Offline compaction: rewriting the live blocks of mostly-dead objects so the
2//! objects can be reclaimed, and so a read of one file range costs fewer
3//! requests.
4//!
5//! ## What `prune` cannot do
6//!
7//! [`Cleaner::prune`](crate::Cleaner::prune) reclaims an object nothing points
8//! into. An object that is 5% live is not that object: those few blocks hold the
9//! file's only copy of their contents, so the other 95% cannot be given back
10//! while they stay there. Rewriting them into a new checkpoint is what turns the
11//! old object into one `prune` can take.
12//!
13//! The same rewrite is what makes a read cheaper. A request breaks where the
14//! object changes, so blocks of one file range scattered across many objects
15//! cost one request each; written out together they cost one. Both goals use the
16//! same mechanism and differ only in what they select.
17//!
18//! ## How the rewrite happens
19//!
20//! Not by a format this crate writes. Nothing here builds a segment, assigns a
21//! pointer or publishes an inode — that is `hyperfile`'s write path, and a second
22//! implementation of it would drift from the first. What this does instead is ask
23//! `hyperfile` to rewrite blocks it already knows how to rewrite:
24//!
25//! - [`Hyper::fs_relocate`] marks a run of blocks to be rewritten at the next flush
26//! with their current bytes. It reads them with the read path's own coalescing,
27//! so a contiguous run costs a few large requests rather than one per block; it
28//! skips holes, explicit zero blocks and anything already dirty, so a sparse
29//! file stays sparse; and it touches neither `mtime`, `ctime` nor the read cache.
30//! - One `fs_flush` at the end of the pass writes the dirty set as one checkpoint.
31//! The dirty set is keyed by block index, so the blocks land in file order and
32//! packed — which is both halves of what compaction is for.
33//!
34//! ## A compaction is not a change of content
35//!
36//! The pass has to publish a new checkpoint — blocks moved, so pointers, the map
37//! and the inode changed — and so `last_cno` advances. A consumer that reads a new
38//! `last_cno` as "the content changed" would take every compaction for a change.
39//! `hyperfile` therefore marks a publish whose every dirty block came from
40//! relocation with `InodeRaw::FLAG_COMPACTED`, and
41//! [`Hyper::fs_content_cno`] stays where it was across it. That is the number a
42//! consumer should compare.
43//!
44//! The mark is only set when the publish's number directly follows the previous
45//! one, so it holds for exactly one checkpoint. Two passes back to back would
46//! therefore move `content_cno` on the second, and [`S3Cleaner::compact`] refuses
47//! to start on a checkpoint that is already marked unless
48//! [`CompactOptions::allow_consecutive`] says otherwise.
49//!
50//! ## Beside the writer
51//!
52//! With [`CompactOptions::beside_writer`] the pass is not a second writer:
53//! `Hyper::fs_begin_compaction` takes up the stored version, `fs_relocate`
54//! copies the blocks as below, and `fs_publish_compaction` writes the copies'
55//! map and *attaches* it to the stored inode -- its object's metadata, with one
56//! conditional write that changes nothing else. The writer takes the
57//! compaction in before its next publish of data and keeps every block it
58//! changed, so `FailFast` on the writer's side still stops only a second writer.
59//! Until then readers read through the objects as they were, and the emptied
60//! objects become reclaimable only once the writer has published. What follows
61//! is about the default.
62//!
63//! ## Why offline
64//!
65//! A compactor is a writer, and it writes *old* bytes — the content it read
66//! earlier. `hyperfile`'s default conflict policy, `RetryLastWriterWins`,
67//! refreshes from storage on a conflict and re-publishes the dirty set, which for
68//! a compactor means republishing a stale copy of a block another writer has
69//! since changed. So compaction runs with
70//! [`FlushConflictPolicy::FailFast`](hyperfile::config::FlushConflictPolicy::FailFast)
71//! and the caller must be the container's only writer. A concurrent publish is
72//! then detected — the inode carries an `If-Match` on its ETag — and the pass
73//! fails rather than losing the other writer's data.
74//!
75//! A log is the other half of that contract. This refuses to run when the
76//! container's prefix holds keys that are neither the inode nor segment objects,
77//! because a WAL is the usual reason for them and a compactor that does not read
78//! the log would publish over entries recovery has not replayed yet.
79
80use std::io::{Error, ErrorKind, Result};
81use std::collections::{BTreeMap, BTreeSet};
82use log::{debug, info, warn};
83use hyperfile::{BlockIndex, Cno, SegmentId};
84use hyperfile::config::{FlushConflictPolicy, HyperFileConfig, HyperFileConfigBuilder,
85 HyperFileRuntimeConfig};
86use hyperfile::file::file::CompactionAttached;
87use hyperfile::file::flags::{FileFlags, HyperFileFlags};
88use hyperfile::file::hyper::Hyper;
89use hyperfile::meta_format::BlockPtrFormat;
90use hyperfile::staging::config::StagingConfig;
91use crate::s3::{ContainerView, S3Cleaner};
92
93
94/// How much of one stored object is still pointed into.
95#[derive(Clone, Debug, PartialEq, Eq)]
96pub struct ObjectUsage {
97 /// The object, `<cno>` or `<cno>.<part>`.
98 pub segid: SegmentId,
99 /// Its size on storage.
100 pub bytes: u64,
101 /// Live data blocks in it.
102 pub live_data_blocks: usize,
103 /// Live metadata nodes in it.
104 pub live_meta_blocks: usize,
105 /// What those occupy, by the container's own block sizes.
106 pub live_bytes: u64,
107 /// Whether the live metadata in it is rewritten by rewriting its live data.
108 ///
109 /// True for a data part — any part above 0 — of a `StreamSegment` container.
110 /// That layout puts every leaf of the block map in the object holding the data
111 /// under it, and the nodes above the leaves in part 0 only. A leaf in a data
112 /// part is therefore live only while none of the blocks it was written with
113 /// has changed, so those blocks are live and in the same object; relocating
114 /// them dirties the leaf, and the flush writes it elsewhere with them.
115 /// Under every other format the nodes sit in part 0 or the one object, apart
116 /// from the data they describe, and this is false.
117 pub meta_moves_with_data: bool,
118}
119
120impl ObjectUsage {
121 /// Nothing points into it: `prune` can take it, compaction has no work here.
122 pub fn is_dead(&self) -> bool {
123 self.live_data_blocks == 0 && self.live_meta_blocks == 0
124 }
125
126 /// Live bytes over stored bytes. Zero for a dead object, and the cost side
127 /// of the trade: rewriting this much frees [`Self::bytes`].
128 pub fn live_ratio(&self) -> f64 {
129 if self.bytes == 0 {
130 return 0.0;
131 }
132 self.live_bytes as f64 / self.bytes as f64
133 }
134
135 /// Whether a live metadata node is in it.
136 ///
137 /// Such an object cannot in general be freed by rewriting data blocks: a node
138 /// is relocated when the flush rewrites it, which happens because something
139 /// under it was dirtied, and which node covers which keys is not visible from
140 /// outside the map. Compaction skips these rather than rewriting their data
141 /// blocks for nothing — unless [`Self::meta_moves_with_data`] says the node
142 /// goes with the data.
143 pub fn holds_metadata(&self) -> bool {
144 self.live_meta_blocks > 0
145 }
146}
147
148/// What a container's storage is made of, and what could be given back.
149#[derive(Clone, Debug)]
150pub struct Report {
151 /// The checkpoint the walk was taken from.
152 pub high_cno: Cno,
153 /// Whether the newest published checkpoint is itself a compaction
154 /// (`InodeRaw::FLAG_COMPACTED`). `compact` will not run on top of one unless
155 /// asked to, since a second pass in a row moves `content_cno`.
156 pub latest_compacted: bool,
157 /// A compaction attached to the stored inode beside the writer and not yet
158 /// taken in, as its object's metadata has it: the version it was built on,
159 /// its parts, blocks moved and its map. `Unreadable` stops `prune`.
160 pub attached: hyperfile::compaction::StoredCompaction,
161 pub data_block_size: usize,
162 pub meta_block_size: usize,
163 /// Objects whose checkpoint falls in the queried `[low, high]`, ascending.
164 pub objects: Vec<ObjectUsage>,
165 /// Objects outside that range, which no operation here will touch.
166 pub out_of_range: usize,
167 /// Keys under the container's prefix that are neither the inode nor a segment
168 /// object. A WAL root nested under the container is the usual reason, and
169 /// compaction refuses to run while any are present.
170 pub foreign_keys: usize,
171 pub foreign_sample: Option<String>,
172}
173
174impl Report {
175 /// Bytes stored by the objects in range.
176 pub fn bytes(&self) -> u64 {
177 self.objects.iter().map(|o| o.bytes).sum()
178 }
179
180 /// Bytes still pointed into.
181 pub fn bytes_live(&self) -> u64 {
182 self.objects.iter().map(|o| o.live_bytes).sum()
183 }
184
185 /// Objects nothing points into, and what they hold. This is `prune`'s work,
186 /// and it needs no rewrite.
187 pub fn dead(&self) -> (usize, u64) {
188 let dead = self.objects.iter().filter(|o| o.is_dead());
189 (dead.clone().count(), dead.map(|o| o.bytes).sum())
190 }
191
192 /// What compaction at `max_live_ratio` would cost and free: `(objects,
193 /// bytes_to_rewrite, bytes_freed)`.
194 ///
195 /// An estimate from the same numbers [`S3Cleaner::compact`] selects by, so a
196 /// dry run and this agree. The bytes freed are the objects' own, which
197 /// `prune` collects afterwards; the rewrite itself publishes a new checkpoint
198 /// holding the live blocks.
199 pub fn compactable(&self, max_live_ratio: f64) -> (usize, u64, u64) {
200 let picked = self.objects.iter()
201 .filter(|o| is_candidate(o, max_live_ratio));
202 (
203 picked.clone().count(),
204 picked.clone().map(|o| o.live_bytes).sum(),
205 picked.map(|o| o.bytes).sum(),
206 )
207 }
208}
209
210/// Which objects compaction may rewrite, and how much it may do at once.
211#[derive(Clone, Debug)]
212pub struct CompactOptions {
213 /// Rewrite an object only if at most this much of it is still live. At `1.0`
214 /// every live object is a candidate, which rewrites the whole container.
215 pub max_live_ratio: f64,
216 /// The most live bytes one pass rewrites. Objects are taken whole and
217 /// cheapest first until the next would pass this; what is left waits for the
218 /// next pass. Unlimited by default.
219 ///
220 /// On a parted container with partial segments enabled — the default for
221 /// both — this is only a size: `fs_relocate` writes the pass out as partials
222 /// once it crosses the dirty-data limits, so memory is bounded by those
223 /// limits whatever the pass. Where no partial can be written, everything the
224 /// pass relocates stays dirty until its one flush, so the pass is also capped
225 /// at [`UNBOUNDED_PASS_BYTES`]; and a container that is not parted writes a
226 /// checkpoint as one object, so its pass is capped at what one object holds.
227 pub max_pass_bytes: u64,
228 /// The runtime configuration of the handle that does the rewrite: the size of
229 /// the partials a pass is written out as (`data_cache_dirty_min_bytes_to_part`
230 /// and the dirty-data limits), `segment_buffer_size`, the read cache,
231 /// multipart uploads. Give the one the container's writer uses, and what a
232 /// compaction writes is shaped like what the writer writes.
233 ///
234 /// `None` uses [`compaction_runtime_default`]. Either way the cleaner
235 /// overrides one field, `flush_conflict_policy`, to `FailFast`: a compactor
236 /// writes bytes it read earlier, and retrying on a conflict would republish
237 /// them over another writer's newer ones.
238 pub runtime: Option<HyperFileRuntimeConfig>,
239 /// How many bytes of blocks to hand `fs_relocate` at once. It holds what it
240 /// reads in a buffer of its own before the dirty cache takes it, so this bounds
241 /// that second copy.
242 pub relocate_chunk_bytes: usize,
243 /// Select and report, write nothing. No file is opened.
244 pub dry_run: bool,
245 /// Run even though the newest checkpoint is already a compaction. The flag that
246 /// keeps `content_cno` still holds for one checkpoint only, so a second pass in
247 /// a row is seen by consumers as a change of content. Set this when that is
248 /// acceptable, for example to finish a container larger than one pass.
249 pub allow_consecutive: bool,
250 /// Run even though the prefix holds keys this crate does not recognize. Only
251 /// set this having established that they are not a WAL.
252 pub allow_foreign_keys: bool,
253 /// Compact beside the container's writer: attach the pass to the stored inode
254 /// rather than publish a checkpoint of its own (`Hyper::fs_begin_compaction`
255 /// and `fs_publish_compaction` in `hyperfile`). It keeps the checkpoint
256 /// number and the content checkpoint, and a writer of hyperfile 0.12 or later
257 /// takes it in before its next publish of data rather than failing as on a
258 /// second writer. Needs a container with parts (`PartedSegment`,
259 /// `StreamSegment`). `allow_consecutive` does not apply: an attached
260 /// compaction never moves `content_cno`.
261 ///
262 /// The pass fails with `ResourceBusy` while a compaction is attached and not
263 /// yet taken in, or while the writer's publish is on its way, and with
264 /// `TimedOut` past `max_age`; either way try again later.
265 pub beside_writer: bool,
266 /// Beside the writer, the longest a pass may take from its start to its
267 /// attach; past it the pass is given up. Its copies are referenced by
268 /// nothing until attached, and only a `prune` grace keeps them, so set this
269 /// below the grace of any [`S3Cleaner::prune_with`] that may run meanwhile,
270 /// less a margin. Ten minutes by default.
271 pub max_age: std::time::Duration,
272}
273
274impl Default for CompactOptions {
275 fn default() -> Self {
276 Self {
277 max_live_ratio: 0.5,
278 max_pass_bytes: u64::MAX,
279 runtime: None,
280 relocate_chunk_bytes: 64 * 1024 * 1024,
281 dry_run: false,
282 allow_consecutive: false,
283 allow_foreign_keys: false,
284 beside_writer: false,
285 max_age: std::time::Duration::from_secs(600),
286 }
287 }
288}
289
290/// What a compaction pass did.
291#[derive(Clone, Debug, Default)]
292pub struct CompactOutcome {
293 /// The objects whose live blocks were rewritten, ascending.
294 pub selected: Vec<SegmentId>,
295 /// Of those, the ones a walk taken afterwards still finds referenced. Empty
296 /// on a pass that did what it set out to; anything here is an object that was
297 /// not freed, and is reported rather than assumed away.
298 pub still_referenced: Vec<SegmentId>,
299 /// Blocks `fs_relocate` marked: what was done, not what was planned. A hole,
300 /// an explicit zero block or a block already dirty is not counted.
301 pub blocks_rewritten: usize,
302 pub bytes_rewritten: u64,
303 /// The selected objects' stored bytes — what `prune` can now reclaim.
304 pub bytes_freed: u64,
305 /// Checkpoints published: one for a pass that rewrote anything, none
306 /// otherwise. That is how far `last_cno` advanced. Always none beside the
307 /// writer, whose pass keeps the number: see `attached`.
308 pub checkpoints: usize,
309 /// Beside the writer, what was attached: the version it was built on and the
310 /// one it was attached to, the parts its objects are in, blocks moved, and
311 /// how many conditional writes of the inode it took.
312 pub attached: Option<CompactionAttached>,
313 /// `fs_content_cno` before and after the pass. Equal when consumers will not
314 /// see the pass as a change of content, which is the ordinary case. They
315 /// differ when the publish could not be marked: its number did not follow the
316 /// previous one — an earlier flush failed and gave its number up — or the
317 /// newest checkpoint was already a compaction and the pass ran anyway.
318 pub content_cno_before: Cno,
319 pub content_cno_after: Cno,
320 /// Whether the candidates outlasted the pass budget. Another pass would find
321 /// more to do, at the cost of `content_cno` moving once (see
322 /// [`CompactOptions::allow_consecutive`]).
323 pub more_remaining: bool,
324 pub dry_run: bool,
325}
326
327impl CompactOutcome {
328 /// Whether a consumer comparing `content_cno` will see this pass as a change.
329 pub fn content_moved(&self) -> bool {
330 self.content_cno_before != self.content_cno_after
331 }
332}
333
334impl S3Cleaner {
335 /// What the container's storage is made of: per-object liveness for every
336 /// object whose checkpoint falls in `[low, high]`.
337 ///
338 /// Read-only, and the same walk `collect` does plus one `ListObjectsV2` for
339 /// sizes. `high == 0` means the latest published checkpoint.
340 pub async fn report(&self, low: Cno, high: Cno) -> Result<Report> {
341 let view = self.view(low, high).await?;
342 Ok(build_report(&view))
343 }
344
345 /// Rewrite the live blocks of mostly-dead objects so `prune` can reclaim
346 /// them.
347 ///
348 /// The caller must be the container's only writer for the duration; see the
349 /// [module docs](self) for why that is not a nicety. Nothing is deleted here
350 /// — a compaction makes objects reclaimable and `prune` reclaims them, which
351 /// keeps the destructive step in one place.
352 pub async fn compact(
353 &self,
354 low: Cno,
355 high: Cno,
356 opts: &CompactOptions,
357 ) -> Result<CompactOutcome> {
358 let view = self.view(low, high).await?;
359 if view.listing.foreign > 0 && !opts.allow_foreign_keys {
360 return Err(Error::new(
361 ErrorKind::InvalidInput,
362 format!(
363 "{} key(s) under the container's prefix are neither the inode nor \
364 segment objects (for example {:?}); a WAL is the usual reason, and \
365 compacting past a log this pass does not read would publish over \
366 entries recovery has not replayed. Set allow_foreign_keys once you \
367 have established otherwise",
368 view.listing.foreign,
369 view.listing.foreign_sample.as_deref().unwrap_or("?"),
370 ),
371 ));
372 }
373
374 if view.latest_compacted && !opts.allow_consecutive && !opts.dry_run && !opts.beside_writer {
375 return Err(Error::new(
376 ErrorKind::InvalidInput,
377 format!(
378 "checkpoint {} is already a compaction, and the flag that keeps \
379 content_cno still holds for one checkpoint only: a second pass in a \
380 row would be seen by consumers as a change of content. Write to the \
381 file first, or set allow_consecutive",
382 view.high_cno,
383 ),
384 ));
385 }
386
387 let report = build_report(&view);
388 let mut runtime = compaction_runtime(opts.runtime.as_ref());
389 if opts.beside_writer {
390 // Not the writer: a checkpoint the writer has on storage and has not
391 // yet pointed the inode at is the writer's to publish. Taking it up
392 // here would publish it under the writer and fail its own inode write.
393 runtime.take_up_on_open = false;
394 }
395 let pass_cap = pass_cap(opts.max_pass_bytes, &view.fmt, &runtime);
396 let (selected, bytes_planned, bytes_freed, more_remaining) =
397 plan_pass(&report.objects, opts.max_live_ratio, pass_cap);
398 debug!("compact: {} object(s) selected of {} in range, {} live bytes (cap {})",
399 selected.len(), report.objects.len(), bytes_planned, pass_cap);
400
401 let mut outcome = CompactOutcome {
402 selected: selected.iter().copied().collect(),
403 bytes_freed,
404 more_remaining,
405 dry_run: opts.dry_run,
406 ..Default::default()
407 };
408 if opts.dry_run {
409 // The plan's own numbers, which is all a dry run can report: nothing
410 // is opened, so nothing can be counted as done.
411 outcome.blocks_rewritten =
412 (bytes_planned / view.data_block_size.max(1) as u64) as usize;
413 outcome.bytes_rewritten = bytes_planned;
414 return Ok(outcome);
415 }
416
417 let config = compaction_config(&self.root_uri(), &runtime);
418 let mut file = Hyper::open(
419 self.client().clone(), config, HyperFileFlags::from_flags(FileFlags::rdwr()),
420 ).await?;
421 let result = self.relocate_and_publish(&mut file, &view, &selected, opts,
422 &mut outcome).await;
423 match result {
424 // Published, or nothing to publish: close as usual.
425 Ok(()) => { let _ = file.fs_release().await; },
426 // A failed pass publishes nothing. `fs_release` would flush what was
427 // marked, so a pass reported failed could still have published part
428 // of itself. The handle is dropped instead; any partials it wrote sit
429 // above the published checkpoint, where the next open steps past
430 // them and a later `prune` collects them once a publish passes them.
431 Err(e) => {
432 drop(file);
433 return Err(e);
434 },
435 }
436 if outcome.checkpoints == 0 && outcome.attached.is_none_or(|p| p.moved == 0) {
437 return Ok(outcome);
438 }
439
440 // Say what was actually freed rather than what should have been. A
441 // selected object that is still referenced means a block was not
442 // relocated, which is worth reporting where it can be seen. Beside the
443 // writer, by the attached map while it is attached: the stored one still
444 // points at every selected object until the writer takes it in.
445 let after = self.view(low, 0).await?;
446 let referenced = match (after.attached_root()?, outcome.attached) {
447 (Some(root), Some(_)) => self.reachable_objects(&after, root).await?,
448 _ => after.referenced_objects(),
449 };
450 outcome.still_referenced = outcome.selected.iter()
451 .filter(|segid| referenced.contains(segid))
452 .copied()
453 .collect();
454 if !outcome.still_referenced.is_empty() {
455 warn!("compact: {} of {} selected object(s) are still referenced: {:?}",
456 outcome.still_referenced.len(), outcome.selected.len(),
457 outcome.still_referenced);
458 }
459 if outcome.content_moved() {
460 warn!("compact: content_cno moved {} -> {}; consumers will see this pass \
461 as a change of content", outcome.content_cno_before,
462 outcome.content_cno_after);
463 }
464 info!("compact: rewrote {} block(s) ({} bytes) in {} checkpoint(s), freeing {} bytes \
465 across {} object(s)", outcome.blocks_rewritten, outcome.bytes_rewritten,
466 outcome.checkpoints, outcome.bytes_freed, outcome.selected.len());
467 Ok(outcome)
468 }
469
470 /// Mark every block of the selected objects for relocation, then publish once.
471 async fn relocate_and_publish(
472 &self,
473 file: &mut Hyper<'static>,
474 view: &ContainerView,
475 selected: &BTreeSet<SegmentId>,
476 opts: &CompactOptions,
477 outcome: &mut CompactOutcome,
478 ) -> Result<()> {
479 outcome.content_cno_before = file.fs_content_cno()?;
480 outcome.content_cno_after = outcome.content_cno_before;
481 if selected.is_empty() {
482 return Ok(());
483 }
484 if opts.beside_writer {
485 // Takes up the version stored now: the placements below are its.
486 file.fs_begin_compaction().await?;
487 let res = self.relocate_selected(file, view, selected, opts, outcome).await;
488 if res.is_err() || outcome.blocks_rewritten == 0 {
489 let _ = file.fs_abort_compaction().await;
490 return res;
491 }
492 let attached = file.fs_publish_compaction(opts.max_age).await?;
493 outcome.content_cno_after = file.fs_content_cno()?;
494 debug!("compact: attached a compaction of checkpoint {} to checkpoint {}: {} block(s) moved \
495 into parts {}..{}, in {} attempt(s)", attached.base, attached.attached_to, attached.moved,
496 attached.parts.0, attached.parts.1, attached.attempts);
497 outcome.attached = Some(attached);
498 return Ok(());
499 }
500 self.relocate_selected(file, view, selected, opts, outcome).await?;
501 if outcome.blocks_rewritten == 0 {
502 return Ok(());
503 }
504
505 let cno = file.fs_flush().await?;
506 outcome.checkpoints = 1;
507 outcome.content_cno_after = file.fs_content_cno()?;
508 debug!("compact: published checkpoint {} with {} relocated block(s); content_cno \
509 {} -> {}", cno, outcome.blocks_rewritten, outcome.content_cno_before,
510 outcome.content_cno_after);
511 Ok(())
512 }
513
514 /// Mark every block of the selected objects for relocation.
515 async fn relocate_selected(
516 &self,
517 file: &mut Hyper<'static>,
518 view: &ContainerView,
519 selected: &BTreeSet<SegmentId>,
520 opts: &CompactOptions,
521 outcome: &mut CompactOutcome,
522 ) -> Result<()> {
523 let bs = view.data_block_size;
524 let chunk_blocks = (opts.relocate_chunk_bytes / bs.max(1)).max(1);
525
526 // Which blocks are in the selected objects. `block_placement` answers
527 // exactly that, per index, from the live map — so it needs no agreement
528 // with a second reading of the container's layout. It is asked only about
529 // the windows the walk found mapped blocks in, never about the holes
530 // between them. Runs go to `fs_relocate` in index order, which is also
531 // the order the flush writes them in.
532 let mut pending: Vec<BlockIndex> = Vec::new();
533 for (start, n) in view.live.windows.iter().copied() {
534 let places = file.fs_block_placement(start, n).await?;
535 for (i, place) in places.iter().enumerate() {
536 if let Some((segid, _)) = place
537 && selected.contains(segid)
538 {
539 pending.push(start + i as BlockIndex);
540 if pending.len() >= chunk_blocks {
541 outcome.blocks_rewritten += relocate_runs(file, &pending).await?;
542 pending.clear();
543 }
544 }
545 }
546 }
547 if !pending.is_empty() {
548 outcome.blocks_rewritten += relocate_runs(file, &pending).await?;
549 }
550 outcome.bytes_rewritten = (outcome.blocks_rewritten * bs) as u64;
551 Ok(())
552 }
553}
554
555/// Hand `indices` to `fs_relocate` as runs of consecutive blocks.
556async fn relocate_runs(file: &mut Hyper<'static>, indices: &[BlockIndex]) -> Result<usize> {
557 let runs = into_runs(indices);
558 let marked = file.fs_relocate(&runs).await?;
559 debug!("compact: relocate {} block(s) in {} run(s) marked {}",
560 indices.len(), runs.len(), marked);
561 Ok(marked)
562}
563
564/// The most a pass holds in memory when nothing can write it out early: a parted
565/// container with partial segments turned off, or a container that is not parted.
566/// There `fs_relocate` keeps every block dirty until the pass's one flush.
567pub const UNBOUNDED_PASS_BYTES: u64 = 512 * 1024 * 1024;
568
569/// What `hyperfile` takes the most bytes of one object to be, for this format and
570/// configuration. A copy of `HyperTrait::max_object_bytes`, which is private, with
571/// its 16 KiB floor.
572fn max_object_bytes(fmt: &BlockPtrFormat, runtime: &HyperFileRuntimeConfig) -> u64 {
573 runtime.segment_max_object_bytes.clamp(16 * 1024, fmt.max_object_bytes()) as u64
574}
575
576/// What one pass may rewrite.
577///
578/// Where `fs_relocate` can write partials, the caller's budget alone: memory is
579/// bounded by the dirty-data limits. Where it cannot, also
580/// [`UNBOUNDED_PASS_BYTES`]. And on a container that is not parted, a checkpoint
581/// is one object and `hyperfile` refuses a flush larger, so the pass also fits
582/// one object, with room for the summary and index nodes it carries.
583fn pass_cap(max_pass_bytes: u64, fmt: &BlockPtrFormat, runtime: &HyperFileRuntimeConfig) -> u64 {
584 if fmt.is_parted() {
585 if runtime.parted_segment_enabled {
586 return max_pass_bytes;
587 }
588 return max_pass_bytes.min(UNBOUNDED_PASS_BYTES);
589 }
590 let one_object = max_object_bytes(fmt, runtime);
591 max_pass_bytes.min(UNBOUNDED_PASS_BYTES).min(one_object / 4 * 3)
592}
593
594/// Candidates, cheapest first, taken whole while they fit `cap`.
595///
596/// Returns `(selected, live_bytes, stored_bytes, more_remaining)`. An object is
597/// never split across passes: one that does not fit ends the selection, so the
598/// cheapest work is done first and nothing is half-moved.
599fn plan_pass(
600 objects: &[ObjectUsage],
601 max_live_ratio: f64,
602 cap: u64,
603) -> (BTreeSet<SegmentId>, u64, u64, bool) {
604 let mut selected = BTreeSet::new();
605 let mut live = 0u64;
606 let mut stored = 0u64;
607 let candidates = select_candidates(objects, max_live_ratio);
608 let total = candidates.len();
609 for usage in candidates {
610 if live.saturating_add(usage.live_bytes) > cap {
611 break;
612 }
613 live += usage.live_bytes;
614 stored += usage.bytes;
615 selected.insert(usage.segid);
616 }
617 let more = selected.len() < total;
618 (selected, live, stored, more)
619}
620
621/// Per-object liveness, from one view.
622///
623/// A location's size depends on what it holds, and a pointer does not say: a
624/// metadata node is `meta_block_size` and a data block is `data_block_size`. The
625/// metadata subset of the walk is what separates them.
626pub(crate) fn build_report(view: &ContainerView) -> Report {
627 let mut live_data: BTreeMap<SegmentId, usize> = BTreeMap::new();
628 let mut live_meta: BTreeMap<SegmentId, usize> = BTreeMap::new();
629 for blkptr in &view.live.all {
630 let segid = view.object_of(blkptr);
631 if view.live.meta.contains(blkptr) {
632 *live_meta.entry(segid).or_default() += 1;
633 } else {
634 *live_data.entry(segid).or_default() += 1;
635 }
636 }
637
638 let mut objects = Vec::new();
639 let mut out_of_range = 0;
640 for (segid, bytes) in &view.listing.objects {
641 let cno = segid.as_cno();
642 if cno < view.low || cno > view.high_cno {
643 out_of_range += 1;
644 continue;
645 }
646 let live_data_blocks = live_data.get(segid).copied().unwrap_or(0);
647 let live_meta_blocks = live_meta.get(segid).copied().unwrap_or(0);
648 objects.push(ObjectUsage {
649 segid: *segid,
650 bytes: *bytes,
651 live_data_blocks,
652 live_meta_blocks,
653 live_bytes: (live_data_blocks * view.data_block_size
654 + live_meta_blocks * view.meta_block_size) as u64,
655 meta_moves_with_data: view.fmt.is_stream() && segid.is_partial(),
656 });
657 }
658
659 Report {
660 high_cno: view.high_cno,
661 latest_compacted: view.latest_compacted,
662 attached: view.compaction.clone(),
663 data_block_size: view.data_block_size,
664 meta_block_size: view.meta_block_size,
665 objects,
666 out_of_range,
667 foreign_keys: view.listing.foreign,
668 foreign_sample: view.listing.foreign_sample.clone(),
669 }
670}
671
672/// Whether rewriting this object's live data would free it.
673///
674/// Three exclusions, each for its own reason: a dead object is `prune`'s work and
675/// rewriting nothing frees nothing; an object holding a live metadata node cannot
676/// in general be freed by rewriting data blocks, so its data would be rewritten
677/// for no return — except where the layout puts that node with the data
678/// ([`ObjectUsage::meta_moves_with_data`]); and one above the ratio costs more to
679/// rewrite than the policy accepts.
680///
681/// Getting the middle one wrong costs a rewrite, never data: compaction deletes
682/// nothing, `prune` keeps whatever the walk afterwards finds live, and
683/// [`CompactOutcome::still_referenced`] reports an object selected and not freed.
684fn is_candidate(usage: &ObjectUsage, max_live_ratio: f64) -> bool {
685 !usage.is_dead()
686 && (!usage.holds_metadata() || usage.meta_moves_with_data)
687 && usage.live_ratio() <= max_live_ratio
688}
689
690/// Candidates, cheapest first.
691///
692/// Ordered by live ratio ascending, which is the same as bytes freed per byte
693/// rewritten descending, with the larger object first among equals so a budget
694/// spent early buys more.
695fn select_candidates(
696 objects: &[ObjectUsage],
697 max_live_ratio: f64,
698) -> Vec<&ObjectUsage> {
699 let mut picked: Vec<&ObjectUsage> = objects.iter()
700 .filter(|o| is_candidate(o, max_live_ratio))
701 .collect();
702 picked.sort_by(|a, b| {
703 a.live_ratio().total_cmp(&b.live_ratio())
704 .then(b.bytes.cmp(&a.bytes))
705 .then(a.segid.cmp(&b.segid))
706 });
707 picked
708}
709
710/// Consecutive indices, as `(first, count)`.
711///
712/// What `read_ahead` wants: it takes a byte range and plans it the way a read of
713/// the same range would, so one call per run of adjacent blocks turns into a few
714/// large requests. Asking per block instead would be one request per block, which
715/// for scattered blocks it will be anyway.
716fn into_runs(indices: &[BlockIndex]) -> Vec<(BlockIndex, usize)> {
717 let mut runs: Vec<(BlockIndex, usize)> = Vec::new();
718 for idx in indices {
719 match runs.last_mut() {
720 Some((start, count)) if *start + *count as BlockIndex == *idx => *count += 1,
721 _ => runs.push((*idx, 1)),
722 }
723 }
724 runs
725}
726
727/// The runtime configuration a compaction uses when the caller gives none.
728///
729/// Partials of 64 MiB: large enough that a compacted range reads in few requests,
730/// small enough to bound what a pass holds. The flush interval is an hour so that
731/// partials are cut by size rather than by how long the reads took.
732pub fn compaction_runtime_default() -> HyperFileRuntimeConfig {
733 const PART: usize = 64 * 1024 * 1024;
734 HyperFileRuntimeConfig {
735 data_cache_dirty_max_bytes_threshold: PART,
736 data_cache_dirty_min_bytes_to_part: PART,
737 segment_buffer_size: PART,
738 data_cache_dirty_max_flush_interval: 3_600_000,
739 ..HyperFileRuntimeConfig::default()
740 }
741}
742
743/// The caller's runtime configuration, or the default, with the one field a
744/// compaction cannot leave to the caller.
745///
746/// Nothing else needs forcing. A pass never publishes before its flush — since
747/// `hyperfile 0.10.4`, `fs_relocate` answers the dirty-data limits with a partial
748/// or not at all, never with a publish — so the limits can be whatever the
749/// caller's writer uses.
750fn compaction_runtime(given: Option<&HyperFileRuntimeConfig>) -> HyperFileRuntimeConfig {
751 let mut runtime = given.cloned().unwrap_or_else(compaction_runtime_default);
752 // A compactor writes bytes it read earlier. Refreshing and re-publishing on a
753 // conflict, which `RetryLastWriterWins` does, would republish a stale copy
754 // over another writer's newer one.
755 runtime.flush_conflict_policy = FlushConflictPolicy::FailFast;
756 runtime
757}
758
759fn compaction_config(root_uri: &str, runtime: &HyperFileRuntimeConfig) -> HyperFileConfig {
760 let staging = StagingConfig::new_s3_uri(root_uri, None);
761 HyperFileConfigBuilder::new()
762 .with_staging_config(&staging)
763 .with_runtime_config(runtime)
764 .build()
765}
766
767#[cfg(test)]
768mod tests {
769 use super::*;
770
771 fn usage(segid: SegmentId, bytes: u64, data: usize, meta: usize) -> ObjectUsage {
772 ObjectUsage {
773 segid,
774 bytes,
775 live_data_blocks: data,
776 live_meta_blocks: meta,
777 live_bytes: (data * 4096 + meta * 4096) as u64,
778 meta_moves_with_data: false,
779 }
780 }
781
782 #[test]
783 fn a_dead_object_is_pruned_not_compacted() {
784 let o = usage(SegmentId::new(1), 1 << 20, 0, 0);
785 assert!(o.is_dead());
786 assert!(!is_candidate(&o, 1.0), "nothing to rewrite frees nothing");
787 }
788
789 /// An object a live metadata node sits in cannot be freed by rewriting data
790 /// blocks: which keys that node covers is not visible from out here, so
791 /// nothing this crate can do will relocate it.
792 #[test]
793 fn an_object_holding_metadata_is_not_a_candidate() {
794 let o = usage(SegmentId::with_part(3, 0), 1 << 20, 1, 1);
795 assert!(o.holds_metadata());
796 assert!(!is_candidate(&o, 1.0));
797 }
798
799 /// A stream data part keeps its leaves with its data, so it is a candidate
800 /// despite them; its part 0, which holds the nodes above the leaves, is not.
801 #[test]
802 fn a_stream_data_part_is_a_candidate_despite_its_leaves() {
803 let part = ObjectUsage { meta_moves_with_data: true,
804 ..usage(SegmentId::with_part(3, 2), 1 << 20, 20, 1) };
805 assert!(part.holds_metadata());
806 assert!(is_candidate(&part, 0.5));
807 let part0 = usage(SegmentId::with_part(3, 0), 1 << 20, 20, 1);
808 assert!(!is_candidate(&part0, 1.0));
809 }
810
811 #[test]
812 fn the_ratio_is_a_ceiling() {
813 let low = usage(SegmentId::new(1), 4096 * 100, 5, 0); // 5% live
814 let high = usage(SegmentId::new(2), 4096 * 100, 90, 0); // 90% live
815 assert!(is_candidate(&low, 0.5));
816 assert!(!is_candidate(&high, 0.5));
817 assert!(is_candidate(&high, 0.9));
818 }
819
820 /// Cheapest first, and the bigger object first among equals: the budget is
821 /// finite and the point is how much it buys.
822 #[test]
823 fn candidates_come_cheapest_first() {
824 let objects = vec![
825 usage(SegmentId::new(1), 4096 * 100, 40, 0), // 40%
826 usage(SegmentId::new(2), 4096 * 100, 5, 0), // 5%
827 usage(SegmentId::new(3), 4096 * 200, 10, 0), // 5%, larger
828 ];
829 let picked = select_candidates(&objects, 0.5);
830 let order: Vec<SegmentId> = picked.iter().map(|o| o.segid).collect();
831 assert_eq!(order, vec![SegmentId::new(3), SegmentId::new(2), SegmentId::new(1)]);
832 }
833
834 #[test]
835 fn runs_are_maximal() {
836 assert_eq!(into_runs(&[]), vec![]);
837 assert_eq!(into_runs(&[7]), vec![(7, 1)]);
838 assert_eq!(into_runs(&[0, 1, 2, 5, 6, 9]), vec![(0, 3), (5, 2), (9, 1)]);
839 }
840
841 /// A pass takes objects whole and cheapest first, and stops at the first one
842 /// that does not fit rather than skipping to a smaller one: the cheapest work
843 /// is done first, and the pass says there is more.
844 #[test]
845 fn a_pass_takes_whole_objects_until_the_budget() {
846 let objects = vec![
847 usage(SegmentId::new(1), 4096 * 100, 10, 0), // 10%, 40 KiB live
848 usage(SegmentId::new(2), 4096 * 100, 20, 0), // 20%, 80 KiB live
849 usage(SegmentId::new(3), 4096 * 100, 1, 0), // 1%, 4 KiB live
850 ];
851 let (picked, live, stored, more) = plan_pass(&objects, 0.5, 4096 * 15);
852 assert_eq!(picked.into_iter().collect::<Vec<_>>(),
853 vec![SegmentId::new(1), SegmentId::new(3)]);
854 assert_eq!(live, 4096 * 11);
855 assert_eq!(stored, 4096 * 200);
856 assert!(more, "object 2 did not fit and is still a candidate");
857
858 let (picked, _, _, more) = plan_pass(&objects, 0.5, u64::MAX);
859 assert_eq!(picked.len(), 3);
860 assert!(!more);
861 }
862
863 /// Where partials can be written, memory is bounded by the dirty-data limits
864 /// and only the caller's budget applies. Where they cannot, the pass is held
865 /// in memory whole and is capped; and a container that is not parted also has
866 /// to fit its pass in one object, as `hyperfile` reckons one — the caller's
867 /// `segment_max_object_bytes` included.
868 #[test]
869 fn the_pass_cap_follows_what_can_write_it_out() {
870 let big = u64::MAX;
871 let parts_on = compaction_runtime(None);
872 assert!(parts_on.parted_segment_enabled);
873 assert_eq!(pass_cap(big, &BlockPtrFormat::PartedSegment, &parts_on), big);
874
875 let parts_off = HyperFileRuntimeConfig { parted_segment_enabled: false, ..parts_on.clone() };
876 assert_eq!(pass_cap(big, &BlockPtrFormat::PartedSegment, &parts_off), UNBOUNDED_PASS_BYTES);
877
878 let mg = BlockPtrFormat::MicroGroup.max_object_bytes() as u64;
879 let cap = pass_cap(big, &BlockPtrFormat::MicroGroup, &parts_on);
880 assert!(cap < mg && cap <= UNBOUNDED_PASS_BYTES);
881
882 let small_objects = HyperFileRuntimeConfig {
883 segment_max_object_bytes: 1024 * 1024, ..parts_on.clone()
884 };
885 assert_eq!(pass_cap(big, &BlockPtrFormat::MicroGroup, &small_objects), 1024 * 1024 / 4 * 3,
886 "the caller's object size is what hyperfile refuses past");
887 assert_eq!(pass_cap(4096, &BlockPtrFormat::MicroGroup, &parts_on), 4096,
888 "a smaller budget is kept as given");
889 }
890
891 /// The caller's configuration is used as given, except for the conflict
892 /// policy, which a compactor cannot leave at retry-and-overwrite.
893 #[test]
894 fn a_given_runtime_is_kept_but_for_the_conflict_policy() {
895 use hyperfile::config::FlushConflictPolicy;
896 let given = HyperFileRuntimeConfig {
897 data_cache_dirty_min_bytes_to_part: 16 * 1024 * 1024,
898 data_cache_dirty_max_bytes_threshold: 16 * 1024 * 1024,
899 segment_mpu_chunk_size: 16 * 1024 * 1024,
900 flush_conflict_policy: FlushConflictPolicy::RetryLastWriterWins,
901 ..HyperFileRuntimeConfig::default()
902 };
903 let used = compaction_runtime(Some(&given));
904 assert_eq!(used.flush_conflict_policy, FlushConflictPolicy::FailFast);
905 assert_eq!(HyperFileRuntimeConfig {
906 flush_conflict_policy: FlushConflictPolicy::RetryLastWriterWins, ..used
907 }, given);
908 assert_eq!(compaction_runtime(None).flush_conflict_policy, FlushConflictPolicy::FailFast);
909 }
910
911 /// An object whose live data alone is over the budget is never taken, and
912 /// the pass reports that it left work behind instead of doing nothing quietly.
913 #[test]
914 fn an_object_larger_than_the_budget_is_reported_not_split() {
915 let objects = vec![usage(SegmentId::new(1), 4096 * 1000, 100, 0)];
916 let (picked, live, _, more) = plan_pass(&objects, 0.5, 4096 * 10);
917 assert!(picked.is_empty());
918 assert_eq!(live, 0);
919 assert!(more);
920 }
921
922 /// `compactable` must answer from the same rule the pass selects by,
923 /// otherwise a dry run and the pass disagree about what will happen.
924 #[test]
925 fn the_estimate_agrees_with_the_selection() {
926 let report = Report {
927 high_cno: 9,
928 latest_compacted: false,
929 attached: Default::default(),
930 data_block_size: 4096,
931 meta_block_size: 4096,
932 objects: vec![
933 usage(SegmentId::new(1), 4096 * 100, 5, 0),
934 usage(SegmentId::new(2), 4096 * 100, 80, 0),
935 usage(SegmentId::new(3), 4096 * 100, 0, 0),
936 usage(SegmentId::new(4), 4096 * 100, 2, 3),
937 ],
938 out_of_range: 0,
939 foreign_keys: 0,
940 foreign_sample: None,
941 };
942 let (n, rewrite, freed) = report.compactable(0.5);
943 assert_eq!(n, 1, "only the 5% object qualifies");
944 assert_eq!(rewrite, 5 * 4096);
945 assert_eq!(freed, 4096 * 100);
946 assert_eq!(report.dead(), (1, 4096 * 100));
947 let picked = select_candidates(&report.objects, CompactOptions::default().max_live_ratio);
948 assert_eq!(picked.len(), n);
949 }
950}