mirror of
https://github.com/freemkv/libfreemkv.git
synced 2026-10-07 18:13:10 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2dcf969ac8 |
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "libfreemkv"
|
||||
version = "0.20.3"
|
||||
version = "0.20.4"
|
||||
edition = "2024"
|
||||
rust-version = "1.86"
|
||||
license = "AGPL-3.0-only"
|
||||
|
||||
@@ -14,9 +14,28 @@
|
||||
//! a full chunk's worth of work to flush — the wait is near-instant.
|
||||
//! Dirty cache stays bounded at ~2 × `chunk_bytes` and writes drain
|
||||
//! continuously instead of in bursts.
|
||||
//!
|
||||
//! The chunk size is adaptive: we measure the elapsed time of the
|
||||
//! `WAIT_AFTER` call over a rolling window of the last 16 chunks and
|
||||
//! resize the chunk based on the p95. Slow storage (NFS, network
|
||||
//! shares, HDD) sees larger chunks to amortise per-chunk overhead;
|
||||
//! fast storage (NVMe) sees smaller chunks to keep cache pressure
|
||||
//! tight. Bounds: [4 MiB, 256 MiB].
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::fs::File;
|
||||
use std::os::unix::io::{AsRawFd, RawFd};
|
||||
use std::time::Instant;
|
||||
|
||||
const ADAPTIVE_WINDOW: usize = 16;
|
||||
const CHUNK_BYTES_MIN: u64 = 4 * 1024 * 1024;
|
||||
const CHUNK_BYTES_MAX: u64 = 256 * 1024 * 1024;
|
||||
const ADAPTIVE_GROW_MS: u64 = 200;
|
||||
const ADAPTIVE_SHRINK_MS: u64 = 20;
|
||||
/// Every N chunks, emit a `debug!` snapshot of the current chunk
|
||||
/// size so operators tailing the log can see where the autoscaler
|
||||
/// settled.
|
||||
const SIZE_LOG_INTERVAL: u64 = 32;
|
||||
|
||||
pub(crate) struct WritebackPipeline {
|
||||
/// Aliases the wrapping `WritebackFile::file`. Only valid for the
|
||||
@@ -28,6 +47,11 @@ pub(crate) struct WritebackPipeline {
|
||||
chunk_bytes: u64,
|
||||
last_flush_pos: u64,
|
||||
pending: Option<(u64, u64)>,
|
||||
/// Rolling window of recent `WAIT_AFTER` elapsed_ms measurements.
|
||||
wait_after_window: VecDeque<u64>,
|
||||
/// Count of chunks emitted (used to space out periodic
|
||||
/// `debug!` size snapshots).
|
||||
chunk_count: u64,
|
||||
}
|
||||
|
||||
impl WritebackPipeline {
|
||||
@@ -41,6 +65,8 @@ impl WritebackPipeline {
|
||||
chunk_bytes,
|
||||
last_flush_pos: start_pos,
|
||||
pending: None,
|
||||
wait_after_window: VecDeque::with_capacity(ADAPTIVE_WINDOW),
|
||||
chunk_count: 0,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -51,27 +77,83 @@ impl WritebackPipeline {
|
||||
if pos < self.last_flush_pos.saturating_add(self.chunk_bytes) {
|
||||
return;
|
||||
}
|
||||
let chunk_off = self.last_flush_pos as i64;
|
||||
let chunk_len = (pos - self.last_flush_pos) as i64;
|
||||
let mut wait_ms: u64 = 0;
|
||||
let mut fadvise_ms: u64 = 0;
|
||||
unsafe {
|
||||
let chunk_off = self.last_flush_pos as i64;
|
||||
let chunk_len = (pos - self.last_flush_pos) as i64;
|
||||
libc::sync_file_range(self.fd, chunk_off, chunk_len, libc::SYNC_FILE_RANGE_WRITE);
|
||||
if let Some((prev_off, prev_len)) = self.pending.take() {
|
||||
let t_wait = Instant::now();
|
||||
libc::sync_file_range(
|
||||
self.fd,
|
||||
prev_off as i64,
|
||||
prev_len as i64,
|
||||
libc::SYNC_FILE_RANGE_WAIT_AFTER,
|
||||
);
|
||||
wait_ms = t_wait.elapsed().as_millis() as u64;
|
||||
let t_fadv = Instant::now();
|
||||
libc::posix_fadvise(
|
||||
self.fd,
|
||||
prev_off as i64,
|
||||
prev_len as i64,
|
||||
libc::POSIX_FADV_DONTNEED,
|
||||
);
|
||||
fadvise_ms = t_fadv.elapsed().as_millis() as u64;
|
||||
self.record_wait(wait_ms);
|
||||
}
|
||||
self.pending = Some((chunk_off as u64, chunk_len as u64));
|
||||
}
|
||||
self.last_flush_pos = pos;
|
||||
self.chunk_count += 1;
|
||||
tracing::trace!(
|
||||
target: "mux",
|
||||
"WritebackPipeline chunk off={} len={} sync_file_range_ms={wait_ms} fadvise_ms={fadvise_ms} chunk_bytes={}",
|
||||
chunk_off,
|
||||
chunk_len,
|
||||
self.chunk_bytes
|
||||
);
|
||||
if self.chunk_count % SIZE_LOG_INTERVAL == 0 {
|
||||
tracing::debug!(
|
||||
target: "mux",
|
||||
"WritebackPipeline chunk_bytes={} after {} chunks",
|
||||
self.chunk_bytes,
|
||||
self.chunk_count
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Push a new `WAIT_AFTER` measurement into the rolling window
|
||||
/// and, if the window is full, adapt `chunk_bytes` based on p95.
|
||||
fn record_wait(&mut self, wait_ms: u64) {
|
||||
if self.wait_after_window.len() == ADAPTIVE_WINDOW {
|
||||
self.wait_after_window.pop_front();
|
||||
}
|
||||
self.wait_after_window.push_back(wait_ms);
|
||||
if self.wait_after_window.len() < ADAPTIVE_WINDOW {
|
||||
return;
|
||||
}
|
||||
// p95 of 16 samples ≈ sorted[14] (5 % of 16 = 0.8 ≈ 1 above).
|
||||
let mut sorted: Vec<u64> = self.wait_after_window.iter().copied().collect();
|
||||
sorted.sort_unstable();
|
||||
let p95 = sorted[14];
|
||||
let old = self.chunk_bytes;
|
||||
let new = if p95 > ADAPTIVE_GROW_MS && self.chunk_bytes < CHUNK_BYTES_MAX {
|
||||
(self.chunk_bytes * 2).min(CHUNK_BYTES_MAX)
|
||||
} else if p95 < ADAPTIVE_SHRINK_MS && self.chunk_bytes > CHUNK_BYTES_MIN {
|
||||
(self.chunk_bytes / 2).max(CHUNK_BYTES_MIN)
|
||||
} else {
|
||||
self.chunk_bytes
|
||||
};
|
||||
if new != old {
|
||||
self.chunk_bytes = new;
|
||||
tracing::info!(
|
||||
target: "mux",
|
||||
"WritebackPipeline adaptive chunk_bytes {} -> {} p95_ms={p95}",
|
||||
old,
|
||||
new
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Caller is about to seek away from the current write region.
|
||||
|
||||
@@ -54,11 +54,57 @@ impl WritebackFile {
|
||||
/// and wrap it. Convenience for the common
|
||||
/// `File::create(path)` + `WritebackFile::new(file)` pair so callers
|
||||
/// don't have to assemble a `File` first.
|
||||
///
|
||||
/// Callers that know the target output size should prefer
|
||||
/// [`Self::create_with_size_hint`] so the kernel can pre-reserve
|
||||
/// extents.
|
||||
#[allow(dead_code)]
|
||||
pub(crate) fn create(path: &Path) -> io::Result<Self> {
|
||||
let file = File::create(path)?;
|
||||
Self::new(file)
|
||||
}
|
||||
|
||||
/// Like [`Self::create`] but pre-reserves `size_bytes` of disk
|
||||
/// space via `fallocate(FALLOC_FL_KEEP_SIZE)` on Linux. The
|
||||
/// reported file size is unchanged (writes still grow the file
|
||||
/// naturally) — only the on-disk extent allocation is preallocated,
|
||||
/// which reduces extent fragmentation on large sequential writes
|
||||
/// (mux output, especially on slow storage / NFS).
|
||||
///
|
||||
/// On macOS / Windows the size hint is ignored and this is
|
||||
/// equivalent to `create`.
|
||||
pub(crate) fn create_with_size_hint(path: &Path, size_bytes: u64) -> io::Result<Self> {
|
||||
let file = File::create(path)?;
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
use std::os::unix::io::AsRawFd;
|
||||
// FALLOC_FL_KEEP_SIZE = 0x01 — keep the reported file
|
||||
// size at 0 (writes grow it normally) while still
|
||||
// pre-reserving the extents.
|
||||
let rc = unsafe {
|
||||
libc::fallocate(
|
||||
file.as_raw_fd(),
|
||||
libc::FALLOC_FL_KEEP_SIZE,
|
||||
0,
|
||||
size_bytes as i64,
|
||||
)
|
||||
};
|
||||
tracing::debug!(
|
||||
target: "mux",
|
||||
"WritebackFile fallocate size_hint={size_bytes} rc={rc} ok={}",
|
||||
rc == 0
|
||||
);
|
||||
}
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
{
|
||||
tracing::debug!(
|
||||
target: "mux",
|
||||
"WritebackFile fallocate size_hint={size_bytes} skipped (non-linux)"
|
||||
);
|
||||
}
|
||||
Self::new(file)
|
||||
}
|
||||
|
||||
/// Open an existing file at `path` for writing (no truncation) and
|
||||
/// wrap it. Mirrors `File::open` semantics for the writable case
|
||||
/// — used by patch / resume paths that mutate an existing ISO in
|
||||
|
||||
+5
-3
@@ -250,11 +250,13 @@ pub fn output(
|
||||
// writeback) so a UHD-scale MKV mux to slow / network-attached
|
||||
// staging doesn't hit the dirty-page burst pathology that
|
||||
// sweep already side-steps. BufWriter sits on top to coalesce
|
||||
// mux's many small EBML element writes.
|
||||
// mux's many small EBML element writes. Pre-reserve the
|
||||
// target's worth of extents on Linux via fallocate(KEEP_SIZE)
|
||||
// to reduce extent fragmentation during the mux.
|
||||
let writer: Box<dyn super::WriteSeek + Send> =
|
||||
Box::new(std::io::BufWriter::with_capacity(
|
||||
IO_BUF_SIZE,
|
||||
crate::io::WritebackFile::create(path)?,
|
||||
crate::io::WritebackFile::create_with_size_hint(path, title.size_bytes)?,
|
||||
));
|
||||
Ok(Box::new(MkvStream::create(writer, title)?))
|
||||
}
|
||||
@@ -262,7 +264,7 @@ pub fn output(
|
||||
validate_file_path(path, "m2ts")?;
|
||||
let writer = std::io::BufWriter::with_capacity(
|
||||
IO_BUF_SIZE,
|
||||
crate::io::WritebackFile::create(path)?,
|
||||
crate::io::WritebackFile::create_with_size_hint(path, title.size_bytes)?,
|
||||
);
|
||||
Ok(Box::new(M2tsStream::create(writer, title)?))
|
||||
}
|
||||
|
||||
+66
-1
@@ -15,6 +15,12 @@ use crate::error::{Error, Result};
|
||||
|
||||
use super::{SectorSink, SectorSource};
|
||||
|
||||
/// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop on the
|
||||
/// read side. Mirrors the writeback chunk size so the read-side
|
||||
/// page cache stays bounded the same way the write side does.
|
||||
#[cfg(target_os = "linux")]
|
||||
const READ_DROP_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
|
||||
|
||||
/// SectorSource backed by a file (ISO image).
|
||||
///
|
||||
/// Seeks to `lba * 2048`, reads `count * 2048` bytes per call. The
|
||||
@@ -27,6 +33,14 @@ use super::{SectorSink, SectorSource};
|
||||
pub struct FileSectorSource {
|
||||
file: File,
|
||||
capacity: u32,
|
||||
/// Bytes read since the last `posix_fadvise(DONTNEED)` drop.
|
||||
/// Only updated on Linux; on other targets it stays at 0.
|
||||
#[cfg(target_os = "linux")]
|
||||
bytes_read_since_drop: u64,
|
||||
/// Byte offset at which the current drop window starts (the
|
||||
/// next `posix_fadvise(DONTNEED)` call drops from here).
|
||||
#[cfg(target_os = "linux")]
|
||||
drop_window_start: u64,
|
||||
}
|
||||
|
||||
impl FileSectorSource {
|
||||
@@ -45,7 +59,26 @@ impl FileSectorSource {
|
||||
.into());
|
||||
}
|
||||
let capacity = sectors as u32;
|
||||
Ok(Self { file, capacity })
|
||||
|
||||
// Hint sequential access on Linux so the kernel's readahead
|
||||
// window widens for the ISO sweep. Best-effort: return value
|
||||
// is ignored. On macOS / Windows this is a no-op.
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
use std::os::unix::io::AsRawFd;
|
||||
unsafe {
|
||||
libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
file,
|
||||
capacity,
|
||||
#[cfg(target_os = "linux")]
|
||||
bytes_read_since_drop: 0,
|
||||
#[cfg(target_os = "linux")]
|
||||
drop_window_start: 0,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -75,6 +108,38 @@ impl SectorSource for FileSectorSource {
|
||||
self.file
|
||||
.read_exact(&mut buf[..bytes])
|
||||
.map_err(|e| Error::IoError { source: e })?;
|
||||
|
||||
// On Linux, periodically drop the just-read region from the
|
||||
// page cache to keep cache pressure bounded during multi-GB
|
||||
// sequential ISO reads. Mirrors the write-side pipeline.
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
use std::os::unix::io::AsRawFd;
|
||||
self.bytes_read_since_drop += bytes as u64;
|
||||
if self.bytes_read_since_drop >= READ_DROP_CHUNK_BYTES {
|
||||
let drop_start = self.drop_window_start;
|
||||
let drop_len = self.bytes_read_since_drop;
|
||||
let t0 = std::time::Instant::now();
|
||||
unsafe {
|
||||
libc::posix_fadvise(
|
||||
self.file.as_raw_fd(),
|
||||
drop_start as i64,
|
||||
drop_len as i64,
|
||||
libc::POSIX_FADV_DONTNEED,
|
||||
);
|
||||
}
|
||||
let elapsed_ms = t0.elapsed().as_millis();
|
||||
let start_lba = drop_start / 2048;
|
||||
let end_lba = (drop_start + drop_len) / 2048;
|
||||
tracing::trace!(
|
||||
target: "mux",
|
||||
"FileSectorSource fadvise DONTNEED lba=[{start_lba}..{end_lba}) bytes={drop_len} elapsed_ms={elapsed_ms}"
|
||||
);
|
||||
self.drop_window_start = drop_start + drop_len;
|
||||
self.bytes_read_since_drop = 0;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(bytes)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user