#[cfg(target_os = "linux")]
mod linux;
#[cfg(target_os = "macos")]
mod macos;
#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
mod other;
#[cfg(target_os = "windows")]
mod windows;
#[cfg(target_os = "linux")]
use linux as platform;
#[cfg(target_os = "macos")]
use macos as platform;
#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
use other as platform;
#[cfg(target_os = "windows")]
use windows as platform;
use std::fs::{File, OpenOptions};
use std::io::{self, Seek, SeekFrom, Write};
use std::path::Path;
use super::writeback::WritebackPipeline;
const WRITEBACK_CHUNK_BYTES_DEFAULT: u64 = 32 * 1024 * 1024;
const WRITEBACK_CHUNK_MIB_MAX: u64 = 64 * 1024;
fn writeback_chunk_bytes() -> u64 {
std::env::var("FREEMKV_WRITEBACK_CHUNK_MIB")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|&n| n > 0 && n <= WRITEBACK_CHUNK_MIB_MAX)
.map(|n| n * 1024 * 1024)
.unwrap_or(WRITEBACK_CHUNK_BYTES_DEFAULT)
}
pub(crate) struct WritebackFile {
file: File,
pipeline: WritebackPipeline,
pos: u64,
}
impl WritebackFile {
pub(crate) fn new(mut file: File) -> io::Result<Self> {
let pos = file.stream_position()?;
let pipeline = WritebackPipeline::new(&file, pos, writeback_chunk_bytes());
Ok(Self {
file,
pipeline,
pos,
})
}
#[allow(dead_code)]
pub(crate) fn create(path: &Path) -> io::Result<Self> {
let file = File::create(path)?;
Self::new(file)
}
pub(crate) fn create_with_size_hint(path: &Path, size_bytes: u64) -> io::Result<Self> {
let file = File::create(path)?;
platform::preallocate(&file, size_bytes);
Self::new(file)
}
pub(crate) fn open(path: &Path) -> io::Result<Self> {
let file = OpenOptions::new().write(true).open(path)?;
Self::new(file)
}
pub(crate) fn sync_all(&mut self) -> io::Result<()> {
self.pipeline.finalize();
platform::durable_sync(&self.file)
}
}
impl Write for WritebackFile {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let n = self.file.write(buf)?;
self.pos += n as u64;
self.pipeline.note_progress(self.pos);
Ok(n)
}
fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
self.file.write_all(buf)?;
self.pos += buf.len() as u64;
self.pipeline.note_progress(self.pos);
Ok(())
}
fn flush(&mut self) -> io::Result<()> {
self.file.flush()
}
}
impl Seek for WritebackFile {
fn seek(&mut self, from: SeekFrom) -> io::Result<u64> {
let p = self.file.seek(from)?;
if p != self.pos {
let from_pos = self.pos;
let to_pos = p;
let delta: i64 = (to_pos as i64).wrapping_sub(from_pos as i64);
tracing::debug!(
target: "mux",
"WritebackFile seek from={from_pos} to={to_pos} delta={delta}"
);
self.pipeline.handle_seek(p);
self.pos = p;
}
Ok(p)
}
}
impl super::sink::SequentialSink for WritebackFile {
fn finish(&mut self) -> io::Result<()> {
self.sync_all()
}
}
impl super::sink::RandomAccessSink for WritebackFile {}
impl Drop for WritebackFile {
fn drop(&mut self) {
self.pipeline.finalize();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Read;
fn read_back(path: &Path) -> Vec<u8> {
let mut f = File::open(path).unwrap();
let mut v = Vec::new();
f.read_to_end(&mut v).unwrap();
v
}
#[test]
fn write_then_drop_persists_bytes() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("a.bin");
{
let mut w = WritebackFile::create(&p).unwrap();
w.write_all(b"hello world").unwrap();
}
assert_eq!(read_back(&p), b"hello world");
}
#[test]
fn sync_all_drains_and_flushes() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("b.bin");
let mut w = WritebackFile::create(&p).unwrap();
for _ in 0..32 {
w.write_all(&[0x5au8; 1024]).unwrap();
}
w.sync_all().unwrap();
let bytes = read_back(&p);
assert_eq!(bytes.len(), 32 * 1024);
assert!(bytes.iter().all(|&b| b == 0x5a));
drop(w);
}
#[test]
fn seek_then_patch_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("c.bin");
let mut w = WritebackFile::create(&p).unwrap();
let big = vec![b'A'; 4096];
w.write_all(&big).unwrap();
w.seek(SeekFrom::Start(1000)).unwrap();
w.write_all(b"PATCHED!").unwrap();
w.sync_all().unwrap();
drop(w);
let bytes = read_back(&p);
assert_eq!(bytes.len(), 4096);
assert_eq!(&bytes[1000..1008], b"PATCHED!");
assert_eq!(bytes[999], b'A');
assert_eq!(bytes[1008], b'A');
}
#[test]
fn flush_is_observed_in_order() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("f.bin");
let mut w = WritebackFile::create(&p).unwrap();
w.write_all(b"one").unwrap();
w.flush().unwrap();
w.write_all(b"two").unwrap();
w.flush().unwrap();
w.write_all(b"three").unwrap();
w.sync_all().unwrap();
drop(w);
assert_eq!(read_back(&p), b"onetwothree");
}
#[test]
fn finish_through_trait_object_persists() {
use crate::io::sink::RandomAccessSink;
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("finish-dyn.bin");
let w = WritebackFile::create(&p).unwrap();
let mut boxed: Box<dyn RandomAccessSink> = Box::new(w);
boxed.write_all(b"durable-tail").unwrap();
boxed.finish().unwrap();
assert_eq!(read_back(&p), b"durable-tail");
}
#[test]
fn write_returns_byte_count_and_advances_pos() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("wc.bin");
let mut w = WritebackFile::create(&p).unwrap();
let n = w.write(b"twelve bytes").unwrap();
assert_eq!(n, 12, "write must report bytes written");
let pos = w.stream_position().unwrap();
assert_eq!(pos, 12, "pos not advanced by write count");
w.sync_all().unwrap();
drop(w);
assert_eq!(read_back(&p), b"twelve bytes");
}
#[test]
fn seek_to_current_position_is_noop_for_data() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("noop-seek.bin");
let mut w = WritebackFile::create(&p).unwrap();
w.write_all(b"AAAA").unwrap();
let off = w.seek(SeekFrom::Start(4)).unwrap();
assert_eq!(off, 4);
w.write_all(b"BBBB").unwrap();
w.sync_all().unwrap();
drop(w);
assert_eq!(
read_back(&p),
b"AAAABBBB",
"redundant seek corrupted contiguous write"
);
}
#[test]
fn open_preserves_existing_contents() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("reopen.bin");
std::fs::write(&p, b"ORIGINAL-CONTENT").unwrap();
let mut w = WritebackFile::open(&p).unwrap();
w.write_all(b"PATCHED!").unwrap();
w.sync_all().unwrap();
drop(w);
assert_eq!(read_back(&p), b"PATCHED!-CONTENT");
}
#[test]
fn new_tracks_initial_position() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("pos-init.bin");
std::fs::write(&p, b"0123456789").unwrap();
let mut w = WritebackFile::open(&p).unwrap();
let start = w.stream_position().unwrap();
assert_eq!(start, 0, "freshly opened file should start at offset 0");
w.write_all(b"XY").unwrap();
let after = w.stream_position().unwrap();
assert_eq!(after, 2, "pos must advance by written length");
}
#[test]
fn seek_past_eof_creates_zero_hole() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("hole.bin");
let mut w = WritebackFile::create(&p).unwrap();
w.write_all(b"head").unwrap(); w.seek(SeekFrom::Start(20)).unwrap(); w.write_all(b"tail").unwrap(); w.sync_all().unwrap();
drop(w);
let bytes = read_back(&p);
assert_eq!(
bytes.len(),
24,
"file should extend to the last written byte"
);
assert_eq!(&bytes[0..4], b"head");
assert!(bytes[4..20].iter().all(|&b| b == 0), "hole not zero-filled");
assert_eq!(&bytes[20..24], b"tail");
}
#[test]
fn seek_from_end_resolves_against_length() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("end-seek.bin");
let mut w = WritebackFile::create(&p).unwrap();
w.write_all(b"0123456789").unwrap();
let landed = w.seek(SeekFrom::End(-2)).unwrap();
assert_eq!(landed, 8, "End(-2) of a 10-byte file is offset 8");
w.write_all(b"XY").unwrap();
w.sync_all().unwrap();
drop(w);
assert_eq!(read_back(&p), b"01234567XY");
}
#[test]
fn create_with_size_hint_does_not_inflate_logical_length() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("hint-len.bin");
let mut w = WritebackFile::create_with_size_hint(&p, 1024 * 1024).unwrap();
w.write_all(b"hello").unwrap();
w.sync_all().unwrap();
drop(w);
let bytes = read_back(&p);
assert_eq!(bytes.len(), 5, "size hint must not inflate logical length");
assert_eq!(&bytes, b"hello");
}
#[test]
fn double_sync_all_is_idempotent() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("double-sync.bin");
let mut w = WritebackFile::create(&p).unwrap();
w.write_all(b"idempotent").unwrap();
w.sync_all().unwrap();
w.sync_all().unwrap(); drop(w); assert_eq!(read_back(&p), b"idempotent");
}
#[test]
fn writeback_chunk_constants_and_conversion() {
assert_eq!(WRITEBACK_CHUNK_BYTES_DEFAULT, 32 * 1024 * 1024);
assert_eq!(WRITEBACK_CHUNK_MIB_MAX, 64 * 1024);
let max_bytes = (WRITEBACK_CHUNK_MIB_MAX as u128) * 1024 * 1024;
assert!(
max_bytes <= u64::MAX as u128,
"max chunk MiB * 1MiB must fit in u64"
);
}
#[test]
fn writeback_chunk_env_override_branches() {
let set = |v: &str| unsafe { std::env::set_var("FREEMKV_WRITEBACK_CHUNK_MIB", v) };
let clear = || unsafe { std::env::remove_var("FREEMKV_WRITEBACK_CHUNK_MIB") };
set("8");
assert_eq!(
writeback_chunk_bytes(),
8 * 1024 * 1024,
"in-range mis-converted"
);
set("0");
assert_eq!(
writeback_chunk_bytes(),
WRITEBACK_CHUNK_BYTES_DEFAULT,
"zero must fall back (n > 0 filter)"
);
set("not-a-number");
assert_eq!(
writeback_chunk_bytes(),
WRITEBACK_CHUNK_BYTES_DEFAULT,
"unparseable must fall back"
);
set(&(WRITEBACK_CHUNK_MIB_MAX + 1).to_string());
assert_eq!(
writeback_chunk_bytes(),
WRITEBACK_CHUNK_BYTES_DEFAULT,
"over-max must fall back (n <= MAX filter)"
);
set(&WRITEBACK_CHUNK_MIB_MAX.to_string());
assert_eq!(
writeback_chunk_bytes(),
WRITEBACK_CHUNK_MIB_MAX * 1024 * 1024,
"max boundary must be accepted (inclusive)"
);
clear();
assert_eq!(writeback_chunk_bytes(), WRITEBACK_CHUNK_BYTES_DEFAULT);
}
}