use std::fs::{File, OpenOptions};
use std::io::{BufWriter, Seek, SeekFrom, Write as _};
use std::path::{Path, PathBuf};
use memmap2::Mmap;
use plugmem_core::snapshot::SnapshotSink;
use plugmem_core::{Error, Scratch, Storage};
use crate::error::HostError;
fn sink_io(e: std::io::Error) -> Error {
Error::Storage(format!("{e}"))
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum FsyncPolicy {
#[default]
EachOp,
OnSnapshot,
}
#[derive(Debug)]
pub struct FileStorage {
base: PathBuf,
journal_path: PathBuf,
manifest_tmp: PathBuf,
current_gen: u64,
_lock: File,
journal: File,
fsync: FsyncPolicy,
batch: bool,
}
impl FileStorage {
pub fn open(base: impl Into<PathBuf>, fsync: FsyncPolicy) -> Result<Self, HostError> {
let base = base.into();
let lock_path = suffixed(&base, "lock");
let journal_path = suffixed(&base, "journal");
let manifest_tmp = suffixed(&base, "manifest.tmp");
let lock = OpenOptions::new()
.create(true)
.write(true)
.truncate(false)
.open(&lock_path)
.map_err(|e| HostError::io(&lock_path, e))?;
match lock.try_lock() {
Ok(()) => {}
Err(std::fs::TryLockError::WouldBlock) => {
return Err(HostError::Locked { path: base });
}
Err(std::fs::TryLockError::Error(e)) => {
return Err(HostError::io(&lock_path, e));
}
}
let current_gen = read_manifest(&base)?.unwrap_or(0);
sweep_generations(&base, current_gen)?;
let journal = OpenOptions::new()
.create(true)
.append(true)
.open(&journal_path)
.map_err(|e| HostError::io(&journal_path, e))?;
Ok(Self {
base,
journal_path,
manifest_tmp,
current_gen,
_lock: lock,
journal,
fsync,
batch: false,
})
}
pub(crate) fn set_batch(&mut self, on: bool) {
self.batch = on;
}
pub(crate) fn sync_journal(&mut self) -> Result<(), HostError> {
self.journal
.sync_data()
.map_err(|e| HostError::io(&self.journal_path, e))
}
pub fn path(&self) -> &Path {
&self.base
}
pub fn journal_bytes(&self) -> u64 {
self.journal.metadata().map(|m| m.len()).unwrap_or(0)
}
pub(crate) fn current_snapshot_path(&self) -> Result<Option<PathBuf>, HostError> {
Ok(read_manifest(&self.base)?.map(|g| gen_path(&self.base, g)))
}
fn next_gen(&self) -> u64 {
self.current_gen + 1
}
pub(crate) fn stage_snapshot(
&mut self,
write: impl FnOnce(&mut FileSink) -> Result<(), HostError>,
) -> Result<(), HostError> {
let tmp = gen_tmp_path(&self.base, self.next_gen());
let file = File::create(&tmp).map_err(|e| HostError::io(&tmp, e))?;
let mut sink = FileSink::new(file, tmp.clone());
write(&mut sink)?;
let file = sink.finish()?;
file.sync_all().map_err(|e| HostError::io(&tmp, e))?;
Ok(())
}
pub(crate) fn commit_snapshot(&mut self) -> Result<(), HostError> {
let next = self.next_gen();
let tmp = gen_tmp_path(&self.base, next);
let genp = gen_path(&self.base, next);
std::fs::rename(&tmp, &genp).map_err(|e| HostError::io(&genp, e))?;
sync_dir(&self.base)?;
publish_manifest(&self.base, &self.manifest_tmp, next)?;
self.current_gen = next;
let _ = sweep_generations(&self.base, self.current_gen);
Ok(())
}
}
pub(crate) struct FileSink {
buf: BufWriter<File>,
path: PathBuf,
}
impl FileSink {
fn new(file: File, path: PathBuf) -> Self {
Self {
buf: BufWriter::new(file),
path,
}
}
fn finish(self) -> Result<File, HostError> {
self.buf
.into_inner()
.map_err(|e| HostError::io(&self.path, e.into_error()))
}
}
impl SnapshotSink for &mut FileSink {
fn write(&mut self, bytes: &[u8]) -> Result<(), Error> {
self.buf.write_all(bytes).map_err(sink_io)
}
fn patch(&mut self, at: u64, bytes: &[u8]) -> Result<(), Error> {
self.buf.flush().map_err(sink_io)?;
let file = self.buf.get_mut();
file.seek(SeekFrom::Start(at)).map_err(sink_io)?;
file.write_all(bytes).map_err(sink_io)?;
file.seek(SeekFrom::End(0)).map_err(sink_io)?;
Ok(())
}
}
fn suffixed(base: &Path, ext: &str) -> PathBuf {
let mut s = base.as_os_str().to_os_string();
s.push(".");
s.push(ext);
PathBuf::from(s)
}
const MANIFEST_MAGIC: u32 = 0x504D_474C;
const MANIFEST_VERSION: u16 = 1;
const MANIFEST_LEN: usize = 24;
#[cfg(windows)]
const SHARE_RETRIES: u32 = 100;
fn is_transient_share_error(e: &std::io::Error) -> bool {
#[cfg(windows)]
{
matches!(e.raw_os_error(), Some(5) | Some(32))
}
#[cfg(not(windows))]
{
let _ = e;
false
}
}
fn read_across_rename(path: &Path) -> std::io::Result<Vec<u8>> {
#[cfg(windows)]
{
let mut attempts = 0u32;
loop {
match std::fs::read(path) {
Err(e) if is_transient_share_error(&e) && attempts < SHARE_RETRIES => {
attempts += 1;
std::thread::sleep(std::time::Duration::from_millis(1));
}
other => return other,
}
}
}
#[cfg(not(windows))]
{
std::fs::read(path)
}
}
fn fnv1a(bytes: &[u8]) -> u64 {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for &b in bytes {
h ^= u64::from(b);
h = h.wrapping_mul(0x0000_0100_0000_01b3);
}
h
}
fn gen_path(base: &Path, n: u64) -> PathBuf {
suffixed(base, &format!("snap.{n}"))
}
fn gen_tmp_path(base: &Path, n: u64) -> PathBuf {
suffixed(base, &format!("snap.{n}.tmp"))
}
pub(crate) fn read_manifest(base: &Path) -> Result<Option<u64>, HostError> {
let bytes = match read_across_rename(base) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(HostError::io(base, e)),
};
let ok = bytes.len() == MANIFEST_LEN
&& u32::from_le_bytes(bytes[0..4].try_into().unwrap()) == MANIFEST_MAGIC
&& u16::from_le_bytes(bytes[4..6].try_into().unwrap()) == MANIFEST_VERSION
&& fnv1a(&bytes[0..16]) == u64::from_le_bytes(bytes[16..24].try_into().unwrap());
if !ok {
return Err(HostError::Engine(Error::Corrupt("manifest is corrupt")));
}
Ok(Some(u64::from_le_bytes(bytes[8..16].try_into().unwrap())))
}
fn publish_manifest(base: &Path, manifest_tmp: &Path, generation: u64) -> Result<(), HostError> {
let mut buf = [0u8; MANIFEST_LEN];
buf[0..4].copy_from_slice(&MANIFEST_MAGIC.to_le_bytes());
buf[4..6].copy_from_slice(&MANIFEST_VERSION.to_le_bytes());
buf[8..16].copy_from_slice(&generation.to_le_bytes());
let sum = fnv1a(&buf[0..16]);
buf[16..24].copy_from_slice(&sum.to_le_bytes());
let mut f = File::create(manifest_tmp).map_err(|e| HostError::io(manifest_tmp, e))?;
f.write_all(&buf)
.and_then(|()| f.sync_all())
.map_err(|e| HostError::io(manifest_tmp, e))?;
drop(f);
std::fs::rename(manifest_tmp, base).map_err(|e| HostError::io(base, e))?;
sync_dir(base)
}
fn sync_dir(base: &Path) -> Result<(), HostError> {
#[cfg(unix)]
{
let dir = base.parent().filter(|p| !p.as_os_str().is_empty());
let dir = dir.unwrap_or_else(|| Path::new("."));
File::open(dir)
.and_then(|d| d.sync_all())
.map_err(|e| HostError::io(dir, e))?;
}
#[cfg(not(unix))]
let _ = base;
Ok(())
}
fn try_reclaim_generation(genp: &Path) {
if let Ok(f) = File::open(genp)
&& f.try_lock().is_ok()
{
let _ = std::fs::remove_file(genp);
}
}
fn sweep_generations(base: &Path, current: u64) -> Result<(), HostError> {
let dir = base
.parent()
.filter(|p| !p.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
let name = base
.file_name()
.and_then(|n| n.to_str())
.unwrap_or_default();
let prefix = format!("{name}.snap.");
let entries = match std::fs::read_dir(dir) {
Ok(e) => e,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(e) => return Err(HostError::io(dir, e)),
};
for entry in entries {
let entry = entry.map_err(|e| HostError::io(dir, e))?;
let fname = entry.file_name();
let Some(fname) = fname.to_str() else {
continue;
};
let Some(rest) = fname.strip_prefix(&prefix) else {
continue;
};
if rest.ends_with(".tmp") {
let _ = std::fs::remove_file(entry.path()); continue;
}
match rest.parse::<u64>() {
Ok(n) if n == current => {} Ok(n) if n > current => {
let _ = std::fs::remove_file(entry.path()); }
Ok(_) => try_reclaim_generation(&entry.path()), Err(_) => {}
}
}
let _ = std::fs::remove_file(suffixed(base, "manifest.tmp"));
Ok(())
}
pub(crate) fn pin_current_generation(
base: &Path,
) -> Result<Option<(File, PathBuf, u64)>, HostError> {
loop {
let Some(generation) = read_manifest(base)? else {
return Ok(None);
};
let genp = gen_path(base, generation);
let file = match File::open(&genp) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
Err(e) if is_transient_share_error(&e) => {
std::thread::sleep(std::time::Duration::from_millis(1));
continue;
}
Err(e) => return Err(HostError::io(&genp, e)),
};
match file.try_lock_shared() {
Ok(()) => {}
Err(std::fs::TryLockError::WouldBlock) => continue,
Err(std::fs::TryLockError::Error(e)) => return Err(HostError::io(&genp, e)),
}
if genp.exists() {
return Ok(Some((file, genp, generation)));
}
}
}
impl Storage for FileStorage {
type Error = HostError;
fn read_snapshot(&mut self) -> Result<Option<Vec<u8>>, HostError> {
match self.current_snapshot_path()? {
Some(p) => Ok(Some(std::fs::read(&p).map_err(|e| HostError::io(&p, e))?)),
None => Ok(None),
}
}
fn write_snapshot(&mut self, bytes: &[u8]) -> Result<(), HostError> {
let next = self.next_gen();
let tmp = gen_tmp_path(&self.base, next);
let mut f = File::create(&tmp).map_err(|e| HostError::io(&tmp, e))?;
f.write_all(bytes)
.and_then(|()| f.sync_all())
.map_err(|e| HostError::io(&tmp, e))?;
drop(f);
let genp = gen_path(&self.base, next);
std::fs::rename(&tmp, &genp).map_err(|e| HostError::io(&genp, e))?;
sync_dir(&self.base)?;
publish_manifest(&self.base, &self.manifest_tmp, next)?;
self.current_gen = next;
let _ = sweep_generations(&self.base, self.current_gen);
Ok(())
}
fn read_journal(&mut self) -> Result<Vec<u8>, HostError> {
std::fs::read(&self.journal_path).map_err(|e| HostError::io(&self.journal_path, e))
}
fn append_journal(&mut self, entry: &[u8]) -> Result<(), HostError> {
self.journal
.write_all(entry)
.map_err(|e| HostError::io(&self.journal_path, e))?;
if self.fsync == FsyncPolicy::EachOp && !self.batch {
self.journal
.sync_data()
.map_err(|e| HostError::io(&self.journal_path, e))?;
}
Ok(())
}
fn clear_journal(&mut self) -> Result<(), HostError> {
let truncated = OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(&self.journal_path)
.map_err(|e| HostError::io(&self.journal_path, e))?;
truncated
.sync_data()
.map_err(|e| HostError::io(&self.journal_path, e))?;
drop(truncated);
self.journal = OpenOptions::new()
.create(true)
.append(true)
.open(&self.journal_path)
.map_err(|e| HostError::io(&self.journal_path, e))?;
Ok(())
}
}
pub struct FileScratch {
path: PathBuf,
writer: Option<BufWriter<File>>,
map: Option<Mmap>,
len: u64,
}
impl FileScratch {
pub fn create(path: impl Into<PathBuf>) -> Result<Self, HostError> {
let path = path.into();
let file = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(&path)
.map_err(|e| HostError::io(&path, e))?;
Ok(Self {
path,
writer: Some(BufWriter::new(file)),
map: None,
len: 0,
})
}
}
impl Scratch for FileScratch {
type Error = HostError;
fn write(&mut self, bytes: &[u8]) -> Result<(), HostError> {
let Self {
writer, path, len, ..
} = self;
let w = writer.as_mut().ok_or(HostError::Engine(Error::Invalid(
"scratch write after freeze",
)))?;
w.write_all(bytes).map_err(|e| HostError::io(path, e))?;
*len += bytes.len() as u64;
Ok(())
}
fn len(&self) -> u64 {
self.len
}
fn freeze(&mut self) -> Result<&[u8], HostError> {
if self.map.is_none() {
let writer = self
.writer
.take()
.ok_or(HostError::Engine(Error::Invalid("scratch frozen twice")))?;
let file = writer
.into_inner()
.map_err(|e| HostError::io(&self.path, e.into_error()))?;
file.sync_all().map_err(|e| HostError::io(&self.path, e))?;
drop(file);
let file = File::open(&self.path).map_err(|e| HostError::io(&self.path, e))?;
let map = unsafe { Mmap::map(&file) }.map_err(|e| HostError::io(&self.path, e))?;
self.map = Some(map);
}
Ok(&self.map.as_ref().expect("just set")[..])
}
}
impl Drop for FileScratch {
fn drop(&mut self) {
self.map = None;
self.writer = None;
let _ = std::fs::remove_file(&self.path);
}
}