use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use crate::status::{Result, Status};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SyncMode {
Full,
#[default]
Data,
}
pub trait WritableFile {
fn append(&mut self, data: &[u8]) -> Result<()>;
fn flush(&mut self) -> Result<()>;
fn sync(&mut self) -> Result<()>;
fn close(&mut self) -> Result<()>;
}
pub trait RandomAccessFile {
fn read_at(&self, offset: u64, n: usize) -> Result<Vec<u8>>;
}
pub trait SequentialFile {
fn read(&mut self, n: usize) -> Result<Vec<u8>>;
fn skip(&mut self, mut n: u64) -> Result<()> {
const CHUNK: usize = 8192;
while n > 0 {
let take = (n as usize).min(CHUNK);
let got = self.read(take)?;
if got.is_empty() {
break;
}
n = n.saturating_sub(got.len() as u64);
}
Ok(())
}
}
pub trait FileLock: Send + 'static {}
pub trait Env {
type Writable: WritableFile + Send + 'static;
type RandomAccess: RandomAccessFile + Clone + Send + Sync + 'static;
type Sequential: SequentialFile + Send + 'static;
type Lock: FileLock;
fn new_writable_file(&self, path: &Path) -> Result<Self::Writable>;
fn new_appendable_file(&self, path: &Path) -> Result<Self::Writable> {
let existing = if self.file_exists(path) {
self.read_file(path)?
} else {
Vec::new()
};
let mut file = self.new_writable_file(path)?;
if !existing.is_empty() {
file.append(&existing)?;
}
Ok(file)
}
fn set_sync_mode(&mut self, _mode: SyncMode) {}
fn new_random_access_file(&self, path: &Path) -> Result<Self::RandomAccess>;
fn new_sequential_file(&self, path: &Path) -> Result<Self::Sequential>;
fn lock_file(&self, path: &Path) -> Result<Self::Lock>;
fn read_file(&self, path: &Path) -> Result<Vec<u8>>;
fn write_file(&self, path: &Path, data: &[u8]) -> Result<()>;
fn append_file(&self, path: &Path, data: &[u8]) -> Result<()>;
fn file_exists(&self, path: &Path) -> bool;
fn delete_file(&self, path: &Path) -> Result<()>;
fn rename_file(&self, src: &Path, dst: &Path) -> Result<()>;
fn get_file_size(&self, path: &Path) -> Result<u64>;
fn list_dir(&self, path: &Path) -> Result<Vec<PathBuf>>;
fn create_dir(&self, path: &Path) -> Result<()>;
fn sync_file(&self, _path: &Path) -> Result<()> {
Ok(())
}
fn delete_dir(&self, path: &Path) -> Result<()> {
let entries = self.list_dir(path)?;
for entry in entries {
if entry == *path {
continue;
}
if self.delete_dir(&entry).is_err() {
let _ = self.delete_file(&entry);
}
}
Ok(())
}
}
fn io_err(prefix: &str, e: std::io::Error) -> Status {
Status::io_error(format!("{prefix}: {e}"))
}
fn ensure_parent_dir(path: &Path) -> Result<()> {
if let Some(parent) = path.parent() {
if !parent.as_os_str().is_empty() {
fs::create_dir_all(parent)
.map_err(|e| io_err(&format!("mkdir {}", parent.display()), e))?;
}
}
Ok(())
}
#[derive(Debug, Default, Clone, Copy)]
pub struct StdEnv {
sync_mode: SyncMode,
}
#[cfg(unix)]
extern "C" {
fn fdatasync(fd: i32) -> i32;
}
#[derive(Debug)]
pub struct StdWritableFile {
file: std::io::BufWriter<fs::File>,
sync_mode: SyncMode,
closed: bool,
}
impl StdWritableFile {
fn ensure_open(&self, op: &str) -> Result<()> {
if self.closed {
Err(Status::io_error(format!("{op}: writable file is closed")))
} else {
Ok(())
}
}
}
impl WritableFile for StdWritableFile {
fn append(&mut self, data: &[u8]) -> Result<()> {
self.ensure_open("append")?;
self.file
.write_all(data)
.map_err(|e| io_err("writable append", e))
}
fn flush(&mut self) -> Result<()> {
self.ensure_open("flush")?;
self.file.flush().map_err(|e| io_err("writable flush", e))
}
fn sync(&mut self) -> Result<()> {
self.ensure_open("sync")?;
self.file
.flush()
.map_err(|e| io_err("writable sync flush", e))?;
match self.sync_mode {
SyncMode::Full => self
.file
.get_ref()
.sync_all()
.map_err(|e| io_err("writable sync", e)),
SyncMode::Data => {
#[cfg(unix)]
{
use std::os::fd::AsRawFd;
let fd = self.file.get_ref().as_raw_fd();
if unsafe { fdatasync(fd) } != 0 {
return Err(io_err(
"writable fdatasync",
std::io::Error::last_os_error(),
));
}
Ok(())
}
#[cfg(not(unix))]
{
self.file
.get_ref()
.sync_all()
.map_err(|e| io_err("writable sync", e))
}
}
}
}
fn close(&mut self) -> Result<()> {
if self.closed {
return Ok(());
}
self.file.flush().map_err(|e| io_err("writable close", e))?;
self.closed = true;
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct StdRandomAccessFile {
path: PathBuf,
file: Arc<fs::File>,
}
impl RandomAccessFile for StdRandomAccessFile {
fn read_at(&self, offset: u64, n: usize) -> Result<Vec<u8>> {
let mut buf = vec![0u8; n];
#[cfg(unix)]
{
use std::os::unix::fs::FileExt;
self.file.read_exact_at(&mut buf, offset).map_err(|e| {
io_err(
&format!("random read {} @{}+{}", self.path.display(), offset, n),
e,
)
})?;
}
#[cfg(windows)]
{
use std::os::windows::fs::FileExt;
let mut filled = 0usize;
while filled < n {
let got = self
.file
.seek_read(&mut buf[filled..], offset + filled as u64)
.map_err(|e| {
io_err(
&format!("random read {} @{}+{}", self.path.display(), offset, n),
e,
)
})?;
if got == 0 {
return Err(Status::io_error(format!(
"random read {} @{}+{}: unexpected end of file",
self.path.display(),
offset,
n
)));
}
filled += got;
}
}
Ok(buf)
}
}
#[derive(Debug)]
pub struct StdSequentialFile {
path: PathBuf,
file: fs::File,
}
impl SequentialFile for StdSequentialFile {
fn read(&mut self, n: usize) -> Result<Vec<u8>> {
let mut buf = Vec::with_capacity(n);
Read::by_ref(&mut self.file)
.take(n as u64)
.read_to_end(&mut buf)
.map_err(|e| io_err(&format!("sequential read {}", self.path.display()), e))?;
Ok(buf)
}
fn skip(&mut self, n: u64) -> Result<()> {
let offset = i64::try_from(n).map_err(|_| {
Status::io_error(format!("sequential skip overflow: {}", self.path.display()))
})?;
self.file
.seek(SeekFrom::Current(offset))
.map_err(|e| io_err(&format!("sequential skip {}", self.path.display()), e))?;
Ok(())
}
}
#[derive(Debug)]
pub struct StdFileLock {
path: PathBuf,
_file: fs::File,
}
impl FileLock for StdFileLock {}
static STD_LOCKS: OnceLock<Mutex<BTreeSet<PathBuf>>> = OnceLock::new();
fn std_lock_table() -> &'static Mutex<BTreeSet<PathBuf>> {
STD_LOCKS.get_or_init(|| Mutex::new(BTreeSet::new()))
}
#[cfg(windows)]
fn open_std_lock_file(path: &Path) -> std::io::Result<fs::File> {
use std::os::windows::fs::OpenOptionsExt;
fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.share_mode(0)
.open(path)
}
#[cfg(unix)]
fn open_std_lock_file(path: &Path) -> std::io::Result<fs::File> {
let file = fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)?;
lock_std_file(&file)?;
Ok(file)
}
#[cfg(unix)]
fn lock_std_file(file: &fs::File) -> std::io::Result<()> {
use std::os::fd::AsRawFd;
const LOCK_EX: i32 = 2;
const LOCK_NB: i32 = 4;
unsafe extern "C" {
fn flock(fd: i32, operation: i32) -> i32;
}
let fd = file.as_raw_fd();
if unsafe { flock(fd, LOCK_EX | LOCK_NB) } == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
}
#[cfg(all(not(windows), not(unix)))]
fn open_std_lock_file(path: &Path) -> std::io::Result<fs::File> {
fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.open(path)
}
impl Drop for StdFileLock {
fn drop(&mut self) {
if let Ok(mut locks) = std_lock_table().lock() {
locks.remove(&self.path);
}
}
}
impl Env for StdEnv {
type Writable = StdWritableFile;
type RandomAccess = StdRandomAccessFile;
type Sequential = StdSequentialFile;
type Lock = StdFileLock;
fn set_sync_mode(&mut self, mode: SyncMode) {
self.sync_mode = mode;
}
fn new_writable_file(&self, path: &Path) -> Result<Self::Writable> {
ensure_parent_dir(path)?;
let file = fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(path)
.map_err(|e| io_err(&format!("open writable {}", path.display()), e))?;
Ok(StdWritableFile {
file: std::io::BufWriter::with_capacity(64 * 1024, file),
sync_mode: self.sync_mode,
closed: false,
})
}
fn new_appendable_file(&self, path: &Path) -> Result<Self::Writable> {
ensure_parent_dir(path)?;
let file = fs::OpenOptions::new()
.create(true)
.append(true)
.open(path)
.map_err(|e| io_err(&format!("open appendable {}", path.display()), e))?;
Ok(StdWritableFile {
file: std::io::BufWriter::with_capacity(64 * 1024, file),
sync_mode: self.sync_mode,
closed: false,
})
}
fn new_random_access_file(&self, path: &Path) -> Result<Self::RandomAccess> {
let file = fs::OpenOptions::new()
.read(true)
.open(path)
.map_err(|e| io_err(&format!("open random {}", path.display()), e))?;
Ok(StdRandomAccessFile {
path: path.to_path_buf(),
file: Arc::new(file),
})
}
fn new_sequential_file(&self, path: &Path) -> Result<Self::Sequential> {
let file = fs::OpenOptions::new()
.read(true)
.open(path)
.map_err(|e| io_err(&format!("open sequential {}", path.display()), e))?;
Ok(StdSequentialFile {
path: path.to_path_buf(),
file,
})
}
fn lock_file(&self, path: &Path) -> Result<Self::Lock> {
ensure_parent_dir(path)?;
let path_buf = path.to_path_buf();
{
let mut locks = std_lock_table().lock().unwrap();
if !locks.insert(path_buf.clone()) {
return Err(Status::io_error(format!(
"lock {}: already held by process",
path.display()
)));
}
}
match open_std_lock_file(path) {
Ok(file) => Ok(StdFileLock {
path: path_buf,
_file: file,
}),
Err(error) => {
std_lock_table().lock().unwrap().remove(&path_buf);
Err(io_err(&format!("lock {}", path.display()), error))
}
}
}
fn read_file(&self, path: &Path) -> Result<Vec<u8>> {
fs::read(path).map_err(|e| io_err(&format!("read {}", path.display()), e))
}
fn write_file(&self, path: &Path, data: &[u8]) -> Result<()> {
ensure_parent_dir(path)?;
fs::write(path, data).map_err(|e| io_err(&format!("write {}", path.display()), e))
}
fn append_file(&self, path: &Path, data: &[u8]) -> Result<()> {
ensure_parent_dir(path)?;
let mut f = fs::OpenOptions::new()
.create(true)
.append(true)
.open(path)
.map_err(|e| io_err(&format!("open {}", path.display()), e))?;
f.write_all(data)
.map_err(|e| io_err(&format!("append {}", path.display()), e))
}
fn file_exists(&self, path: &Path) -> bool {
path.exists()
}
fn delete_file(&self, path: &Path) -> Result<()> {
fs::remove_file(path).map_err(|e| io_err(&format!("delete {}", path.display()), e))
}
fn rename_file(&self, src: &Path, dst: &Path) -> Result<()> {
ensure_parent_dir(dst)?;
fs::rename(src, dst)
.map_err(|e| io_err(&format!("rename {} -> {}", src.display(), dst.display()), e))
}
fn get_file_size(&self, path: &Path) -> Result<u64> {
let meta =
fs::metadata(path).map_err(|e| io_err(&format!("stat {}", path.display()), e))?;
Ok(meta.len())
}
fn list_dir(&self, path: &Path) -> Result<Vec<PathBuf>> {
let entries =
fs::read_dir(path).map_err(|e| io_err(&format!("read_dir {}", path.display()), e))?;
let mut out = Vec::new();
for entry in entries {
let entry = entry.map_err(|e| io_err(&format!("iterate {}", path.display()), e))?;
out.push(entry.path());
}
Ok(out)
}
fn create_dir(&self, path: &Path) -> Result<()> {
fs::create_dir_all(path).map_err(|e| io_err(&format!("mkdir {}", path.display()), e))
}
fn sync_file(&self, path: &Path) -> Result<()> {
let f = fs::OpenOptions::new()
.read(true)
.write(true)
.open(path)
.map_err(|e| io_err(&format!("sync open {}", path.display()), e))?;
f.sync_all()
.map_err(|e| io_err(&format!("sync {}", path.display()), e))
}
fn delete_dir(&self, path: &Path) -> Result<()> {
if !path.exists() {
return Ok(());
}
fs::remove_dir_all(path).map_err(|e| io_err(&format!("rmdir -r {}", path.display()), e))
}
}
#[derive(Debug, Default, Clone)]
pub struct MemEnv {
files: SharedMemFiles,
locks: Arc<Mutex<BTreeSet<PathBuf>>>,
}
type MemFile = Arc<Mutex<Vec<u8>>>;
type SharedMemFiles = Arc<Mutex<BTreeMap<PathBuf, MemFile>>>;
impl MemEnv {
pub fn new() -> Self {
Self {
files: Arc::new(Mutex::new(BTreeMap::new())),
locks: Arc::new(Mutex::new(BTreeSet::new())),
}
}
pub fn clone_handle(&self) -> Self {
self.clone()
}
}
#[derive(Debug)]
pub struct MemWritableFile {
buf: Arc<Mutex<Vec<u8>>>,
closed: bool,
}
impl MemWritableFile {
fn ensure_open(&self, op: &str) -> Result<()> {
if self.closed {
Err(Status::io_error(format!("{op}: writable file is closed")))
} else {
Ok(())
}
}
}
impl WritableFile for MemWritableFile {
fn append(&mut self, data: &[u8]) -> Result<()> {
self.ensure_open("append")?;
self.buf.lock().unwrap().extend_from_slice(data);
Ok(())
}
fn flush(&mut self) -> Result<()> {
self.ensure_open("flush")
}
fn sync(&mut self) -> Result<()> {
self.ensure_open("sync")
}
fn close(&mut self) -> Result<()> {
self.closed = true;
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct MemRandomAccessFile {
path: PathBuf,
data: Arc<Vec<u8>>,
}
impl RandomAccessFile for MemRandomAccessFile {
fn read_at(&self, offset: u64, n: usize) -> Result<Vec<u8>> {
let start = offset as usize;
if start as u64 != offset {
return Err(Status::io_error(format!(
"random read offset overflow: {}",
self.path.display()
)));
}
let end = start.checked_add(n).ok_or_else(|| {
Status::io_error(format!(
"random read length overflow: {}",
self.path.display()
))
})?;
if end > self.data.len() {
return Err(Status::io_error(format!(
"short random read {} @{}+{}",
self.path.display(),
offset,
n
)));
}
Ok(self.data[start..end].to_vec())
}
}
#[derive(Debug)]
pub struct MemSequentialFile {
path: PathBuf,
data: Arc<Vec<u8>>,
offset: usize,
}
impl SequentialFile for MemSequentialFile {
fn read(&mut self, n: usize) -> Result<Vec<u8>> {
let end = self.offset.saturating_add(n).min(self.data.len());
let out = self.data[self.offset..end].to_vec();
self.offset = end;
Ok(out)
}
fn skip(&mut self, n: u64) -> Result<()> {
let n = usize::try_from(n).map_err(|_| {
Status::io_error(format!("sequential skip overflow: {}", self.path.display()))
})?;
self.offset = self.offset.saturating_add(n).min(self.data.len());
Ok(())
}
}
#[derive(Debug)]
pub struct MemFileLock {
locks: Arc<Mutex<BTreeSet<PathBuf>>>,
path: PathBuf,
}
impl FileLock for MemFileLock {}
impl Drop for MemFileLock {
fn drop(&mut self) {
if let Ok(mut locks) = self.locks.lock() {
locks.remove(&self.path);
}
}
}
impl Env for MemEnv {
type Writable = MemWritableFile;
type RandomAccess = MemRandomAccessFile;
type Sequential = MemSequentialFile;
type Lock = MemFileLock;
fn new_writable_file(&self, path: &Path) -> Result<Self::Writable> {
let mut files = self.files.lock().unwrap();
let buf = files
.entry(path.to_path_buf())
.and_modify(|b| b.lock().unwrap().clear())
.or_insert_with(|| Arc::new(Mutex::new(Vec::new())))
.clone();
Ok(MemWritableFile { buf, closed: false })
}
fn new_appendable_file(&self, path: &Path) -> Result<Self::Writable> {
let mut files = self.files.lock().unwrap();
let buf = files
.entry(path.to_path_buf())
.or_insert_with(|| Arc::new(Mutex::new(Vec::new())))
.clone();
Ok(MemWritableFile { buf, closed: false })
}
fn new_random_access_file(&self, path: &Path) -> Result<Self::RandomAccess> {
let files = self.files.lock().unwrap();
let data = files
.get(path)
.map(|b| b.lock().unwrap().clone())
.ok_or_else(|| Status::io_error(format!("not found: {}", path.display())))?;
Ok(MemRandomAccessFile {
path: path.to_path_buf(),
data: Arc::new(data),
})
}
fn new_sequential_file(&self, path: &Path) -> Result<Self::Sequential> {
let files = self.files.lock().unwrap();
let data = files
.get(path)
.map(|b| b.lock().unwrap().clone())
.ok_or_else(|| Status::io_error(format!("not found: {}", path.display())))?;
Ok(MemSequentialFile {
path: path.to_path_buf(),
data: Arc::new(data),
offset: 0,
})
}
fn lock_file(&self, path: &Path) -> Result<Self::Lock> {
let path_buf = path.to_path_buf();
{
let mut locks = self.locks.lock().unwrap();
if !locks.insert(path_buf.clone()) {
return Err(Status::io_error(format!(
"lock {}: already held by process",
path.display()
)));
}
}
self.files
.lock()
.unwrap()
.entry(path_buf.clone())
.or_default();
Ok(MemFileLock {
locks: self.locks.clone(),
path: path_buf,
})
}
fn read_file(&self, path: &Path) -> Result<Vec<u8>> {
let files = self.files.lock().unwrap();
files
.get(path)
.map(|b| b.lock().unwrap().clone())
.ok_or_else(|| Status::io_error(format!("not found: {}", path.display())))
}
fn write_file(&self, path: &Path, data: &[u8]) -> Result<()> {
self.files
.lock()
.unwrap()
.insert(path.to_path_buf(), Arc::new(Mutex::new(data.to_vec())));
Ok(())
}
fn append_file(&self, path: &Path, data: &[u8]) -> Result<()> {
let mut files = self.files.lock().unwrap();
files
.entry(path.to_path_buf())
.or_default()
.lock()
.unwrap()
.extend_from_slice(data);
Ok(())
}
fn file_exists(&self, path: &Path) -> bool {
self.files.lock().unwrap().contains_key(path)
}
fn delete_file(&self, path: &Path) -> Result<()> {
if self.files.lock().unwrap().remove(path).is_none() {
return Err(Status::io_error(format!("not found: {}", path.display())));
}
Ok(())
}
fn rename_file(&self, src: &Path, dst: &Path) -> Result<()> {
let mut files = self.files.lock().unwrap();
let data = files
.remove(src)
.ok_or_else(|| Status::io_error(format!("rename: src missing: {}", src.display())))?;
files.insert(dst.to_path_buf(), data);
Ok(())
}
fn get_file_size(&self, path: &Path) -> Result<u64> {
self.files
.lock()
.unwrap()
.get(path)
.map(|b| b.lock().unwrap().len() as u64)
.ok_or_else(|| Status::io_error(format!("not found: {}", path.display())))
}
fn list_dir(&self, path: &Path) -> Result<Vec<PathBuf>> {
let files = self.files.lock().unwrap();
let prefix = path.to_path_buf();
let mut out = BTreeSet::new();
for p in files.keys() {
if !p.starts_with(&prefix) || p == &prefix {
continue;
}
let rel = match p.strip_prefix(&prefix) {
Ok(r) => r,
Err(_) => continue,
};
if let Some(first) = rel.components().next() {
out.insert(prefix.join(first.as_os_str()));
}
}
Ok(out.into_iter().collect())
}
fn create_dir(&self, _path: &Path) -> Result<()> {
Ok(())
}
fn delete_dir(&self, path: &Path) -> Result<()> {
let mut files = self.files.lock().unwrap();
files.retain(|p, _| !p.starts_with(path));
Ok(())
}
}
#[cfg(all(test, unix))]
mod tests {
use super::*;
use std::time::{SystemTime, UNIX_EPOCH};
fn unique_temp_path(name: &str) -> PathBuf {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos();
std::env::temp_dir().join(format!("novakv-{name}-{}-{nanos}.lock", std::process::id()))
}
#[test]
fn std_lock_file_rejects_a_second_unix_lock_holder() {
let path = unique_temp_path("std-lock");
let first = open_std_lock_file(&path).expect("first lock should succeed");
let second = open_std_lock_file(&path);
assert!(
second.is_err(),
"second exclusive nonblocking flock should fail while first handle is held"
);
drop(first);
let third = open_std_lock_file(&path).expect("lock should be released when handle drops");
drop(third);
let _ = fs::remove_file(path);
}
}