#![cfg_attr(
all(debug_assertions, feature = "fault-injection"),
doc = "# Fault injection (`fault-injection` feature)",
doc = "",
doc = "This build has the dev/test-only `fault-injection` feature active, so",
doc = "`BStack` I/O can be made to fail on demand. Implement [`FaultPolicy`] and arm",
doc = "it with [`BStack::with_fault_policy`] (at construction) or",
doc = "[`BStack::set_fault_policy`] (arm, re-arm, or disarm mid-test); every I/O",
doc = "method then consults the policy once, **after** validating its arguments. This",
doc = "exercises error-handling and rollback paths that a successful sequence of calls",
doc = "can never reach. The whole mechanism is gated on `all(debug_assertions, feature",
doc = "= \"fault-injection\")`, so a `--release` build carries none of it and its",
doc = "performance is unaffected. See the [`fault`] module for details."
)]
mod io_core;
use io_core::*;
pub mod fault;
use fault::fault_point;
#[cfg(all(debug_assertions, feature = "fault-injection"))]
pub use fault::{FaultPolicy, FaultState};
#[cfg(all(test, feature = "alloc", feature = "set"))]
mod alloc_fuzz_tests;
mod test;
#[cfg(feature = "alloc")]
mod alloc;
#[cfg(feature = "alloc")]
pub use alloc::{
BStackAllocError, BStackAllocator, BStackBulkAllocError, BStackBulkAllocator, BStackOwnedSlice,
BStackOwnedSliceAllocator, BStackRange, BStackSlice, BStackSliceReader, BStackUninitAllocator,
LinearBStackAllocator,
};
#[cfg(all(feature = "alloc", feature = "set"))]
pub use alloc::{
BStackByteVec, BStackByteVecIter, BStackSliceWriter, CheckedSlabBStackAllocator,
FirstFitBStackAllocator, GhostTreeBstackAllocator, SlabBStackAllocator,
};
#[cfg(all(feature = "guarded", feature = "atomic"))]
pub use alloc::{BStackAtomicGuardedSlice, BStackAtomicGuardedSliceSubview};
#[cfg(feature = "guarded")]
pub use alloc::{BStackGuardedSlice, BStackGuardedSliceSubview};
use std::fmt;
use std::fs::{File, OpenOptions};
use std::hash::{Hash, Hasher};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, RwLock};
#[cfg(unix)]
use std::os::unix::io::AsRawFd;
#[cfg(unix)]
use std::os::unix::io::RawFd;
#[cfg(windows)]
use std::os::windows::io::AsRawHandle;
#[cfg(windows)]
use windows_sys::Win32::Foundation::HANDLE;
#[cfg(windows)]
use windows_sys::Win32::Storage::FileSystem::{
LOCKFILE_EXCLUSIVE_LOCK, LOCKFILE_FAIL_IMMEDIATELY, LockFileEx, ReadFile,
};
#[cfg(windows)]
use windows_sys::Win32::System::IO::OVERLAPPED;
const FORMAT_MAJOR: u8 = 0;
const FORMAT_MINOR: u8 = 4;
const FORMAT_PATCH: u8 = 0;
const MAGIC: [u8; 8] = [
b'B',
b'S',
b'T',
b'K',
FORMAT_MAJOR,
FORMAT_MINOR,
FORMAT_PATCH,
0,
];
const MAGIC_PREFIX: [u8; 6] = [b'B', b'S', b'T', b'K', FORMAT_MAJOR, FORMAT_MINOR];
const LEGACY_MAGIC_PREFIX: [u8; 6] = [b'B', b'S', b'T', b'K', 0, 1];
const LEGACY_HEADER_SIZE: u64 = 16;
#[cfg(any(feature = "set", feature = "atomic"))]
pub(crate) fn checked_end(base: u64, len: u64, msg: &'static str) -> io::Result<u64> {
base.checked_add(len)
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, msg))
}
#[cfg(feature = "set")]
pub(crate) fn check_offset_unlocked(
op: &str,
offset: u64,
end: u64,
locked: u64,
) -> io::Result<()> {
if offset < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("{op}: range [{offset}, {end}) overlaps locked region [0, {locked})"),
));
}
Ok(())
}
pub struct BStack {
lock: RwLock<(File, u64)>,
locked: AtomicU64,
#[cfg(unix)]
fd: RawFd,
#[cfg(windows)]
handle: isize,
cache_enabled: bool,
cache: Mutex<Vec<u8>>,
#[cfg(all(debug_assertions, feature = "fault-injection"))]
fault: fault::FaultState,
}
impl BStack {
fn init_header(file: &mut File) -> io::Result<()> {
file.seek(SeekFrom::Start(0))?;
file.write_all(&MAGIC)?;
file.write_all(&[0u8; (HEADER_SIZE - 8) as usize])
}
fn read_header(file: &mut File) -> io::Result<(u64, u64, u64)> {
file.seek(SeekFrom::Start(0))?;
let mut hdr = [0u8; HEADER_SIZE as usize];
file.read_exact(&mut hdr)?;
if hdr[0..6] != MAGIC_PREFIX {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"bstack: bad magic number — not a bstack file or incompatible version",
));
}
let committed_len = u64::from_le_bytes(hdr[8..16].try_into().unwrap());
let wip_ptr = u64::from_le_bytes(hdr[16..24].try_into().unwrap());
let wip_aux = u64::from_le_bytes(hdr[24..32].try_into().unwrap());
Ok((committed_len, wip_ptr, wip_aux))
}
pub fn open(path: impl AsRef<Path>) -> io::Result<Self> {
let mut file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)?;
#[cfg(unix)]
flock_exclusive(&file)?;
#[cfg(windows)]
lock_file_exclusive(&file)?;
let raw_size = file.metadata()?.len();
let mut clen = 0u64;
if raw_size == 0 {
Self::init_header(&mut file)?;
durable_sync(&file)?;
} else if raw_size < HEADER_SIZE {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"bstack: file is {raw_size} bytes — too small to contain the {HEADER_SIZE}-byte header"
),
));
} else {
let (committed_len, wip_ptr, wip_aux) = Self::read_header(&mut file)?;
clen = committed_len;
if wip_ptr != 0 {
clen = recover_wip(&mut file, committed_len, wip_ptr, wip_aux, raw_size)?;
} else if wip_aux == u64::from(WipAux::MultiWrite) {
clen = recover_multi_write(&mut file, committed_len, raw_size)?;
} else {
let actual_data_len = raw_size - HEADER_SIZE;
if actual_data_len != committed_len {
let correct_len = committed_len.min(actual_data_len);
file.set_len(HEADER_SIZE + correct_len)?;
write_committed_len(&mut file, &mut clen, correct_len)?;
durable_sync(&file)?;
}
}
}
#[cfg(unix)]
let fd = file.as_raw_fd();
#[cfg(windows)]
let handle = file.as_raw_handle() as isize;
Ok(BStack {
#[cfg(unix)]
fd,
#[cfg(windows)]
handle,
lock: RwLock::new((file, clen)),
locked: AtomicU64::new(0),
cache_enabled: false,
cache: Mutex::new(Vec::new()),
#[cfg(all(debug_assertions, feature = "fault-injection"))]
fault: fault::FaultState::new(),
})
}
pub fn migrate(path: impl AsRef<Path>) -> io::Result<()> {
let path = path.as_ref();
let mut old = OpenOptions::new().read(true).open(path)?;
let old_size = old.metadata()?.len();
if old_size < LEGACY_HEADER_SIZE {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"bstack: file is {old_size} bytes — too small to be a legacy {LEGACY_HEADER_SIZE}-byte-header file"
),
));
}
let mut hdr = [0u8; LEGACY_HEADER_SIZE as usize];
old.read_exact(&mut hdr)?;
if hdr[0..6] != LEGACY_MAGIC_PREFIX {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"bstack: not a legacy 0.1.x file — nothing to migrate",
));
}
let clen =
u64::from_le_bytes(hdr[8..16].try_into().unwrap()).min(old_size - LEGACY_HEADER_SIZE);
let mut tmp = path.as_os_str().to_owned();
tmp.push(".migrating");
let tmp = PathBuf::from(tmp);
{
let mut new = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(true)
.open(&tmp)?;
new.write_all(&MAGIC)?; new.write_all(&clen.to_le_bytes())?; new.write_all(&[0u8; 16])?; old.seek(SeekFrom::Start(LEGACY_HEADER_SIZE))?;
let mut src = (&mut old).take(clen);
let copied = io::copy(&mut src, &mut new)?;
if copied != clen {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"bstack: legacy payload shorter than committed length during migration",
));
}
new.sync_all()?;
}
drop(old);
std::fs::rename(&tmp, path)?;
Ok(())
}
pub fn push(&self, data: impl AsRef<[u8]>) -> io::Result<u64> {
let data = data.as_ref();
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let file_end = file.seek(SeekFrom::End(0))?;
let logical_offset = file_end - HEADER_SIZE;
if data.is_empty() {
return Ok(logical_offset);
}
fault_point!(self, "push");
if let Err(e) = file.write_all(data) {
let _ = file.set_len(file_end);
return Err(e);
}
let new_len = logical_offset + data.len() as u64;
commit_grow(file, clen, new_len, logical_offset, file_end)?;
Ok(logical_offset)
}
pub fn extend(&self, n: u64) -> io::Result<u64> {
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let file_end = file.seek(SeekFrom::End(0))?;
let logical_offset = file_end - HEADER_SIZE;
if n == 0 {
return Ok(logical_offset);
}
fault_point!(self, "extend");
let new_file_end = file_end + n;
file.set_len(new_file_end)?;
let new_len = logical_offset + n;
commit_grow(file, clen, new_len, logical_offset, file_end)?;
Ok(logical_offset)
}
pub fn pop(&self, n: u64) -> io::Result<Vec<u8>> {
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let raw_size = file.seek(SeekFrom::End(0))?;
let data_size = raw_size - HEADER_SIZE;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("pop({n}) exceeds payload size ({data_size})"),
));
}
let new_data_len = data_size - n;
let locked = self.locked.load(Ordering::Acquire);
if new_data_len < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("pop({n}) would shrink payload below locked length ({locked})"),
));
}
let mut buf = vec![0u8; n as usize];
fault_point!(self, "pop");
read_at(file, new_data_len, &mut buf)?;
commit_shrink(file, clen, new_data_len)?;
Ok(buf)
}
pub fn peek(&self, offset: u64) -> io::Result<Vec<u8>> {
#[cfg(any(unix, windows))]
{
let guard = self.lock.read().unwrap();
let file = &guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
if offset > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("peek offset ({offset}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "peek");
pread_exact(file, HEADER_SIZE + offset, (data_size - offset) as usize)
}
#[cfg(not(any(unix, windows)))]
{
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let raw_size = file.seek(SeekFrom::End(0))?;
let data_size = raw_size.saturating_sub(HEADER_SIZE);
if offset > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("peek offset ({offset}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "peek");
file.seek(SeekFrom::Start(HEADER_SIZE + offset))?;
let mut buf = vec![0u8; (data_size - offset) as usize];
file.read_exact(&mut buf)?;
Ok(buf)
}
}
pub fn get(&self, start: u64, end: u64) -> io::Result<Vec<u8>> {
if end < start {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get: end ({end}) < start ({start})"),
));
}
#[cfg(any(unix, windows))]
{
let locked = self.locked.load(Ordering::Acquire);
if end <= locked {
if self.cache_enabled {
let len = (end - start) as usize;
let mut buf = vec![0u8; len];
let cache = self.cache.lock().unwrap();
buf.copy_from_slice(&cache[start as usize..end as usize]);
return Ok(buf);
}
#[cfg(unix)]
{
let mut buf = vec![0u8; (end - start) as usize];
pread_exact_raw(self.fd, HEADER_SIZE + start, &mut buf)?;
return Ok(buf);
}
#[cfg(windows)]
{
let mut buf = vec![0u8; (end - start) as usize];
pread_exact_raw_handle(self.handle, HEADER_SIZE + start, &mut buf)?;
return Ok(buf);
}
}
}
#[cfg(any(unix, windows))]
{
let guard = self.lock.read().unwrap();
let file = &guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get: end ({end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "get");
pread_exact(file, HEADER_SIZE + start, (end - start) as usize)
}
#[cfg(not(any(unix, windows)))]
{
let locked = self.locked.load(Ordering::Acquire);
if end <= locked && self.cache_enabled {
let cache = self.cache.lock().unwrap();
return Ok(cache[start as usize..end as usize].to_vec());
}
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let raw_size = file.seek(SeekFrom::End(0))?;
let data_size = raw_size.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get: end ({end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "get");
file.seek(SeekFrom::Start(HEADER_SIZE + start))?;
let mut buf = vec![0u8; (end - start) as usize];
file.read_exact(&mut buf)?;
Ok(buf)
}
}
pub fn peek_into(&self, offset: u64, buf: &mut [u8]) -> io::Result<()> {
if buf.is_empty() {
return Ok(());
}
let len = buf.len() as u64;
let end = offset.checked_add(len).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"peek_into: offset + len overflows u64",
)
})?;
#[cfg(any(unix, windows))]
{
let guard = self.lock.read().unwrap();
let file = &guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"peek_into: range [{offset}, {end}) exceeds payload size ({data_size})"
),
));
}
fault_point!(self, "peek_into");
pread_exact_into(file, HEADER_SIZE + offset, buf)
}
#[cfg(not(any(unix, windows)))]
{
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"peek_into: range [{offset}, {end}) exceeds payload size ({data_size})"
),
));
}
fault_point!(self, "peek_into");
file.seek(SeekFrom::Start(HEADER_SIZE + offset))?;
file.read_exact(buf)
}
}
pub fn get_into(&self, start: u64, buf: &mut [u8]) -> io::Result<()> {
if buf.is_empty() {
return Ok(());
}
let len = buf.len() as u64;
let end = start.checked_add(len).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"get_into: start + len overflows u64",
)
})?;
#[cfg(any(unix, windows))]
{
let locked = self.locked.load(Ordering::Acquire);
if end <= locked {
if self.cache_enabled {
let cache = self.cache.lock().unwrap();
buf.copy_from_slice(&cache[start as usize..end as usize]);
return Ok(());
}
#[cfg(unix)]
return pread_exact_raw(self.fd, HEADER_SIZE + start, buf);
#[cfg(windows)]
return pread_exact_raw_handle(self.handle, HEADER_SIZE + start, buf);
}
}
#[cfg(any(unix, windows))]
{
let guard = self.lock.read().unwrap();
let file = &guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get_into: end ({end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "get_into");
pread_exact_into(file, HEADER_SIZE + start, buf)
}
#[cfg(not(any(unix, windows)))]
{
let locked = self.locked.load(Ordering::Acquire);
if end <= locked && self.cache_enabled {
let cache = self.cache.lock().unwrap();
buf.copy_from_slice(&cache[start as usize..end as usize]);
return Ok(());
}
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get_into: end ({end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "get_into");
file.seek(SeekFrom::Start(HEADER_SIZE + start))?;
file.read_exact(buf)
}
}
pub fn pop_into(&self, buf: &mut [u8]) -> io::Result<()> {
if buf.is_empty() {
return Ok(());
}
let n = buf.len() as u64;
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let raw_size = file.seek(SeekFrom::End(0))?;
let data_size = raw_size - HEADER_SIZE;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("pop_into({n}) exceeds payload size ({data_size})"),
));
}
let new_data_len = data_size - n;
let locked = self.locked.load(Ordering::Acquire);
if new_data_len < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("pop_into({n}) would shrink payload below locked length ({locked})"),
));
}
fault_point!(self, "pop_into");
read_at(file, new_data_len, buf)?;
commit_shrink(file, clen, new_data_len)?;
Ok(())
}
pub fn discard(&self, n: u64) -> io::Result<()> {
if n == 0 {
return Ok(());
}
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let raw_size = file.seek(SeekFrom::End(0))?;
let data_size = raw_size - HEADER_SIZE;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("discard({n}) exceeds payload size ({data_size})"),
));
}
let new_data_len = data_size - n;
let locked = self.locked.load(Ordering::Acquire);
if new_data_len < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("discard({n}) would shrink payload below locked length ({locked})"),
));
}
fault_point!(self, "discard");
commit_shrink(file, clen, new_data_len)?;
Ok(())
}
#[cfg(feature = "set")]
pub fn set(&self, offset: u64, data: impl AsRef<[u8]>) -> io::Result<()> {
let data = data.as_ref();
if data.is_empty() {
return Ok(());
}
let end = checked_end(offset, data.len() as u64, "set: offset + len overflows u64")?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
check_offset_unlocked("set", offset, end, locked)?;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("set: write end ({end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "set");
set_in_place(file, data_size, offset, data)
}
#[cfg(feature = "set")]
pub fn zero(&self, offset: u64, n: u64) -> io::Result<()> {
if n == 0 {
return Ok(());
}
let end = checked_end(offset, n, "zero: offset + n overflows u64")?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
check_offset_unlocked("zero", offset, end, locked)?;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("zero: write end ({end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "zero");
repeat_fill(file, data_size, offset, &[0u8], n)
}
#[cfg(feature = "set")]
pub fn repeat(&self, offset: u64, pattern: impl AsRef<[u8]>, count: u64) -> io::Result<()> {
let pattern = pattern.as_ref();
if pattern.is_empty() || count == 0 {
return Ok(());
}
let total = (pattern.len() as u64).checked_mul(count).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"repeat: count * pattern.len() overflows u64",
)
})?;
let end = checked_end(
offset,
total,
"repeat: offset + count*pattern.len() overflows u64",
)?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
check_offset_unlocked("repeat", offset, end, locked)?;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("repeat: write end ({end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "repeat");
repeat_fill(file, data_size, offset, pattern, count)
}
}
#[cfg(feature = "atomic")]
impl BStack {
#[cfg(feature = "atomic")]
pub fn atrunc(&self, n: u64, buf: impl AsRef<[u8]>) -> io::Result<()> {
let buf = buf.as_ref();
let buf_len = buf.len() as u64;
if n == 0 && buf_len == 0 {
return Ok(());
}
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let file_end = file.seek(SeekFrom::End(0))?;
let data_size = file_end - HEADER_SIZE;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("atrunc: n ({n}) exceeds payload size ({data_size})"),
));
}
let locked = self.locked.load(Ordering::Acquire);
let new_tail_start = data_size - n;
if new_tail_start < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("atrunc: operation would modify locked region [0, {locked})"),
));
}
fault_point!(self, "atrunc");
commit_tail_replace(file, clen, new_tail_start, n, buf, file_end)
}
#[cfg(feature = "atomic")]
pub fn splice(&self, n: u64, buf: impl AsRef<[u8]>) -> io::Result<Vec<u8>> {
let buf = buf.as_ref();
let buf_len = buf.len() as u64;
if n == 0 && buf_len == 0 {
return Ok(Vec::new());
}
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let file_end = file.seek(SeekFrom::End(0))?;
let data_size = file_end - HEADER_SIZE;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("splice: n ({n}) exceeds payload size ({data_size})"),
));
}
let locked = self.locked.load(Ordering::Acquire);
let new_tail_start = data_size - n;
if new_tail_start < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("splice: operation would modify locked region [0, {locked})"),
));
}
fault_point!(self, "splice");
let mut removed = vec![0u8; n as usize];
read_at(file, new_tail_start, &mut removed)?;
commit_tail_replace(file, clen, new_tail_start, n, buf, file_end)?;
Ok(removed)
}
#[cfg(feature = "atomic")]
pub fn splice_into(&self, old: &mut [u8], new: impl AsRef<[u8]>) -> io::Result<()> {
let new = new.as_ref();
let n = old.len() as u64;
let new_len = new.len() as u64;
if n == 0 && new_len == 0 {
return Ok(());
}
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let file_end = file.seek(SeekFrom::End(0))?;
let data_size = file_end - HEADER_SIZE;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("splice_into: n ({n}) exceeds payload size ({data_size})"),
));
}
let locked = self.locked.load(Ordering::Acquire);
let new_tail_start = data_size - n;
if new_tail_start < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("splice_into: operation would modify locked region [0, {locked})"),
));
}
fault_point!(self, "splice_into");
read_at(file, new_tail_start, old)?;
commit_tail_replace(file, clen, new_tail_start, n, new, file_end)
}
#[cfg(feature = "atomic")]
pub fn try_extend(&self, s: u64, buf: impl AsRef<[u8]>) -> io::Result<bool> {
let buf = buf.as_ref();
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let file_end = file.seek(SeekFrom::End(0))?;
let data_size = file_end - HEADER_SIZE;
if data_size != s {
return Ok(false);
}
if buf.is_empty() {
return Ok(true);
}
fault_point!(self, "try_extend");
if let Err(e) = file.write_all(buf) {
let _ = file.set_len(file_end);
return Err(e);
}
let new_len = data_size + buf.len() as u64;
commit_grow(file, clen, new_len, data_size, file_end)?;
Ok(true)
}
#[cfg(feature = "atomic")]
pub fn try_extend_zeros(&self, s: u64, n: u64) -> io::Result<bool> {
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let file_end = file.seek(SeekFrom::End(0))?;
let data_size = file_end - HEADER_SIZE;
if data_size != s {
return Ok(false);
}
if n == 0 {
return Ok(true);
}
let new_len = checked_end(
data_size,
n,
"try_extend_zeros: data_size + n overflows u64",
)?;
fault_point!(self, "try_extend_zeros");
file.set_len(HEADER_SIZE + new_len)?;
commit_grow(file, clen, new_len, data_size, file_end)?;
Ok(true)
}
#[cfg(feature = "atomic")]
pub fn try_discard(&self, s: u64, n: u64) -> io::Result<bool> {
if n == 0 {
let guard = self.lock.read().unwrap();
let file = &guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
return Ok(data_size == s);
}
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let raw_size = file.seek(SeekFrom::End(0))?;
let data_size = raw_size - HEADER_SIZE;
if data_size != s {
return Ok(false);
}
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("try_discard: n ({n}) exceeds payload size ({data_size})"),
));
}
let new_data_len = data_size - n;
let locked = self.locked.load(Ordering::Acquire);
if new_data_len < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("try_discard: would shrink payload below locked length ({locked})"),
));
}
fault_point!(self, "try_discard");
commit_shrink(file, clen, new_data_len)?;
Ok(true)
}
#[cfg(feature = "atomic")]
pub fn get_batched<I>(&self, ranges: I) -> io::Result<Vec<Vec<u8>>>
where
I: IntoIterator<Item = std::ops::Range<u64>>,
{
let ranges: Vec<std::ops::Range<u64>> = ranges.into_iter().collect();
if ranges.is_empty() {
return Ok(Vec::new());
}
for r in &ranges {
if r.end < r.start {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get_batched: end ({}) < start ({})", r.end, r.start),
));
}
}
#[cfg(any(unix, windows))]
{
let guard = self.lock.read().unwrap();
let file = &guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
fault_point!(self, "get_batched");
let mut results = Vec::with_capacity(ranges.len());
for r in &ranges {
if r.end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"get_batched: end ({}) exceeds payload size ({data_size})",
r.end
),
));
}
results.push(pread_exact(
file,
HEADER_SIZE + r.start,
(r.end - r.start) as usize,
)?);
}
Ok(results)
}
#[cfg(not(any(unix, windows)))]
{
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
fault_point!(self, "get_batched");
let mut results = Vec::with_capacity(ranges.len());
for r in &ranges {
if r.end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"get_batched: end ({}) exceeds payload size ({data_size})",
r.end
),
));
}
file.seek(SeekFrom::Start(HEADER_SIZE + r.start))?;
let mut buf = vec![0u8; (r.end - r.start) as usize];
file.read_exact(&mut buf)?;
results.push(buf);
}
Ok(results)
}
}
#[cfg(feature = "atomic")]
pub fn get_batched_into<'a, I>(&self, bufs: I) -> io::Result<()>
where
I: IntoIterator<Item = (u64, &'a mut [u8])>,
{
let bufs: Vec<(u64, &'a mut [u8])> = bufs.into_iter().collect();
if bufs.is_empty() {
return Ok(());
}
#[cfg(any(unix, windows))]
{
let guard = self.lock.read().unwrap();
let file = &guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
fault_point!(self, "get_batched_into");
for (ptr, buf) in bufs {
let end = ptr.checked_add(buf.len() as u64).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"get_batched_into: offset + buf.len() overflows u64",
)
})?;
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get_batched_into: end ({end}) exceeds payload size ({data_size})",),
));
}
pread_exact_into(file, HEADER_SIZE + ptr, buf)?;
}
Ok(())
}
#[cfg(not(any(unix, windows)))]
{
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
fault_point!(self, "get_batched_into");
for (ptr, buf) in bufs {
let end = ptr.checked_add(buf.len() as u64).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"get_batched_into: offset + buf.len() overflows u64",
)
})?;
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get_batched_into: end ({end}) exceeds payload size ({data_size})",),
));
}
file.seek(SeekFrom::Start(HEADER_SIZE + ptr))?;
file.read_exact(buf)?;
}
Ok(())
}
}
#[cfg(feature = "atomic")]
pub fn get_batched_gen<'a, F>(&self, mut f: F) -> io::Result<()>
where
F: FnMut() -> Option<(u64, &'a mut [u8])>,
{
#[cfg(any(unix, windows))]
{
let guard = self.lock.read().unwrap();
let file = &guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
fault_point!(self, "get_batched_gen");
while let Some((offset, buf)) = f() {
let end = offset.checked_add(buf.len() as u64).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"get_batched_gen: offset + buf.len() overflows u64",
)
})?;
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get_batched_gen: end ({end}) exceeds payload size ({data_size})"),
));
}
pread_exact_into(file, HEADER_SIZE + offset, buf)?;
}
Ok(())
}
#[cfg(not(any(unix, windows)))]
{
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
fault_point!(self, "get_batched_gen");
while let Some((offset, buf)) = f() {
let end = offset.checked_add(buf.len() as u64).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"get_batched_gen: offset + buf.len() overflows u64",
)
})?;
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("get_batched_gen: end ({end}) exceeds payload size ({data_size})"),
));
}
file.seek(SeekFrom::Start(HEADER_SIZE + offset))?;
file.read_exact(buf)?;
}
Ok(())
}
}
#[cfg(feature = "atomic")]
pub fn replace<F>(&self, n: u64, f: F) -> io::Result<()>
where
F: FnOnce(&[u8]) -> Vec<u8>,
{
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let file_end = file.seek(SeekFrom::End(0))?;
let data_size = file_end - HEADER_SIZE;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("replace: n ({n}) exceeds payload size ({data_size})"),
));
}
let locked = self.locked.load(Ordering::Acquire);
let new_tail_start = data_size - n;
if new_tail_start < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("replace: operation would modify locked region [0, {locked})"),
));
}
fault_point!(self, "replace");
let mut old_tail = vec![0u8; n as usize];
read_at(file, new_tail_start, &mut old_tail)?;
let new_tail = f(&old_tail);
commit_tail_replace(file, clen, new_tail_start, n, &new_tail, file_end)
}
}
#[cfg(all(feature = "set", feature = "atomic"))]
#[non_exhaustive]
#[derive(Debug)]
pub enum BStackGenOp<'a> {
Read {
offset: u64,
buf: &'a mut [u8],
},
Write {
offset: u64,
data: &'a [u8],
},
Swap {
a_offset: u64,
b_offset: u64,
len: u64,
},
Push {
data: &'a [u8],
},
Pop {
buf: &'a mut [u8],
},
Discard {
len: u64,
},
Atrunc {
n: u64,
data: &'a [u8],
},
Splice {
old: &'a mut [u8],
new: &'a [u8],
},
Len {
out: &'a mut u64,
},
}
#[cfg(all(feature = "set", feature = "atomic"))]
impl BStack {
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn swap(&self, offset: u64, buf: impl AsRef<[u8]>) -> io::Result<Vec<u8>> {
let buf = buf.as_ref();
if buf.is_empty() {
return Ok(Vec::new());
}
let end = checked_end(offset, buf.len() as u64, "swap: offset + len overflows u64")?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
check_offset_unlocked("swap", offset, end, locked)?;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("swap: range [{offset}, {end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "swap");
let mut old = vec![0u8; buf.len()];
read_at(file, offset, &mut old)?;
set_in_place(file, data_size, offset, buf)?;
Ok(old)
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn swap_into(&self, offset: u64, buf: &mut [u8]) -> io::Result<()> {
if buf.is_empty() {
return Ok(());
}
let end = checked_end(
offset,
buf.len() as u64,
"swap_into: offset + len overflows u64",
)?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
check_offset_unlocked("swap_into", offset, end, locked)?;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("swap_into: range [{offset}, {end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "swap_into");
let mut tmp = vec![0u8; buf.len()];
read_at(file, offset, &mut tmp)?;
set_in_place(file, data_size, offset, buf)?;
buf.copy_from_slice(&tmp);
Ok(())
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn cas(
&self,
offset: u64,
old: impl AsRef<[u8]>,
new: impl AsRef<[u8]>,
) -> io::Result<bool> {
let old = old.as_ref();
let new = new.as_ref();
if old.len() != new.len() {
return Ok(false);
}
if old.is_empty() {
return Ok(true);
}
let end = checked_end(offset, old.len() as u64, "cas: offset + len overflows u64")?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
check_offset_unlocked("cas", offset, end, locked)?;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("cas: range [{offset}, {end}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "cas");
let mut current = vec![0u8; old.len()];
read_at(file, offset, &mut current)?;
if current != old {
return Ok(false);
}
set_in_place(file, data_size, offset, new)?;
Ok(true)
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn cross_exchange(&self, a: u64, b: u64, n: u64) -> io::Result<()> {
let a_end = checked_end(a, n, "cross_exchange: a + n overflows u64")?;
let b_end = checked_end(b, n, "cross_exchange: b + n overflows u64")?;
if n > 0 {
let (lo, hi) = if a < b { (a, b) } else { (b, a) };
if lo + n > hi {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("cross_exchange: regions [{a}, {a_end}) and [{b}, {b_end}) overlap"),
));
}
}
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
if a < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"cross_exchange: region [{a}, {a_end}) overlaps locked region [0, {locked})"
),
));
}
if b < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"cross_exchange: region [{b}, {b_end}) overlaps locked region [0, {locked})"
),
));
}
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if a_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("cross_exchange: region [{a}, {a_end}) exceeds payload size ({data_size})"),
));
}
if b_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("cross_exchange: region [{b}, {b_end}) exceeds payload size ({data_size})"),
));
}
if n == 0 {
return Ok(());
}
fault_point!(self, "cross_exchange");
journaled_exchange(file, data_size, a, b, n)
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn copy(&self, from: u64, to: u64, n: u64) -> io::Result<()> {
let from_end = checked_end(from, n, "copy: from + n overflows u64")?;
let to_end = checked_end(to, n, "copy: to + n overflows u64")?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
if to < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("copy: destination [{to}, {to_end}) overlaps locked region [0, {locked})"),
));
}
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if from_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("copy: source [{from}, {from_end}) exceeds payload size ({data_size})"),
));
}
if to_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("copy: destination [{to}, {to_end}) exceeds payload size ({data_size})"),
));
}
if n == 0 {
return Ok(());
}
if from == to {
return Ok(());
}
fault_point!(self, "copy");
if is_atomic_write(to, n) {
let mut buf = vec![0u8; n as usize];
read_at(file, from, &mut buf)?;
write_at(file, to, &buf)?;
durable_sync(file)
} else if from < to_end && to < from_end {
journaled_move(file, data_size, from, to, n)
} else {
journaled_copy(file, data_size, from, to, n)
}
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn process<F>(&self, start: u64, end: u64, f: F) -> io::Result<()>
where
F: FnOnce(&mut [u8]),
{
if end < start {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("process: end ({end}) < start ({start})"),
));
}
let n = end - start;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("process: end ({end}) exceeds payload size ({data_size})"),
));
}
let locked = self.locked.load(Ordering::Acquire);
if start < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("process: range [{start}, {end}) overlaps locked region [0, {locked})"),
));
}
fault_point!(self, "process");
let mut buf = vec![0u8; n as usize];
if n > 0 {
read_at(file, start, &mut buf)?;
}
f(&mut buf);
if n > 0 {
set_in_place(file, data_size, start, &buf)?;
}
Ok(())
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn process_gen<'a, F>(&self, mut f: F) -> io::Result<()>
where
F: FnMut() -> Option<BStackGenOp<'a>>,
{
let mut guard = self.lock.write().unwrap();
let (file, clen) = &mut *guard;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
let locked = self.locked.load(Ordering::Acquire);
fault_point!(self, "process_gen");
loop {
match f() {
Some(BStackGenOp::Read { offset, buf }) => {
let end = checked_end(
offset,
buf.len() as u64,
"process_gen: read offset + buf.len() overflows u64",
)?;
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: read range [{offset}, {end}) exceeds payload size ({data_size})"
),
));
}
#[cfg(any(unix, windows))]
{
if end <= locked {
if self.cache_enabled {
let cache = self.cache.lock().unwrap();
buf.copy_from_slice(&cache[offset as usize..end as usize]);
} else {
#[cfg(unix)]
pread_exact_raw(self.fd, HEADER_SIZE + offset, buf)?;
#[cfg(windows)]
pread_exact_raw_handle(self.handle, HEADER_SIZE + offset, buf)?;
}
} else {
pread_exact_into(file, HEADER_SIZE + offset, buf)?;
}
}
#[cfg(not(any(unix, windows)))]
{
if end <= locked && self.cache_enabled {
let cache = self.cache.lock().unwrap();
buf.copy_from_slice(&cache[offset as usize..end as usize]);
} else {
read_at(file, offset, buf)?;
}
}
}
Some(BStackGenOp::Write { offset, data }) => {
let end = checked_end(
offset,
data.len() as u64,
"process_gen: write offset + data.len() overflows u64",
)?;
if offset < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: write range [{offset}, {end}) overlaps locked region [0, {locked})"
),
));
}
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: write range [{offset}, {end}) exceeds payload size ({data_size})"
),
));
}
if !data.is_empty() {
set_in_place(file, data_size, offset, data)?;
}
return Ok(());
}
Some(BStackGenOp::Swap {
a_offset,
b_offset,
len,
}) => {
let a_end =
checked_end(a_offset, len, "process_gen: a_offset + len overflows u64")?;
let b_end =
checked_end(b_offset, len, "process_gen: b_offset + len overflows u64")?;
if len > 0 {
let (lo, hi) = if a_offset < b_offset {
(a_offset, b_offset)
} else {
(b_offset, a_offset)
};
if lo + len > hi {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: swap regions [{a_offset}, {a_end}) and [{b_offset}, {b_end}) overlap"
),
));
}
}
if a_offset < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: swap region [{a_offset}, {a_end}) overlaps locked region [0, {locked})"
),
));
}
if b_offset < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: swap region [{b_offset}, {b_end}) overlaps locked region [0, {locked})"
),
));
}
if a_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: swap region [{a_offset}, {a_end}) exceeds payload size ({data_size})"
),
));
}
if b_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: swap region [{b_offset}, {b_end}) exceeds payload size ({data_size})"
),
));
}
if len > 0 {
journaled_exchange(file, data_size, a_offset, b_offset, len)?;
}
return Ok(());
}
Some(BStackGenOp::Push { data }) => {
if !data.is_empty() {
let file_end = file.seek(SeekFrom::End(0))?;
let logical_offset = file_end - HEADER_SIZE;
if let Err(e) = file.write_all(data) {
let _ = file.set_len(file_end);
return Err(e);
}
let new_len = logical_offset + data.len() as u64;
commit_grow(file, clen, new_len, logical_offset, file_end)?;
}
return Ok(());
}
Some(BStackGenOp::Pop { buf }) => {
let n = buf.len() as u64;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("process_gen: pop({n}) exceeds payload size ({data_size})"),
));
}
let new_data_len = data_size - n;
if new_data_len < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: pop({n}) would shrink payload below locked length ({locked})"
),
));
}
if n > 0 {
read_at(file, new_data_len, buf)?;
commit_shrink(file, clen, new_data_len)?;
}
return Ok(());
}
Some(BStackGenOp::Discard { len }) => {
if len > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: discard({len}) exceeds payload size ({data_size})"
),
));
}
let new_data_len = data_size - len;
if new_data_len < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: discard({len}) would shrink payload below locked length ({locked})"
),
));
}
if len > 0 {
commit_shrink(file, clen, new_data_len)?;
}
return Ok(());
}
Some(BStackGenOp::Atrunc { n, data }) => {
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: atrunc n ({n}) exceeds payload size ({data_size})"
),
));
}
let new_tail_start = data_size - n;
if new_tail_start < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("process_gen: atrunc would modify locked region [0, {locked})"),
));
}
if n != 0 || !data.is_empty() {
let file_end = HEADER_SIZE + data_size;
commit_tail_replace(file, clen, new_tail_start, n, data, file_end)?;
}
return Ok(());
}
Some(BStackGenOp::Splice { old, new }) => {
let n = old.len() as u64;
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"process_gen: splice n ({n}) exceeds payload size ({data_size})"
),
));
}
let new_tail_start = data_size - n;
if new_tail_start < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("process_gen: splice would modify locked region [0, {locked})"),
));
}
if n != 0 || !new.is_empty() {
read_at(file, new_tail_start, old)?;
let file_end = HEADER_SIZE + data_size;
commit_tail_replace(file, clen, new_tail_start, n, new, file_end)?;
}
return Ok(());
}
Some(BStackGenOp::Len { out }) => {
*out = data_size;
}
None => return Ok(()),
}
}
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn set_batched<I, D>(&self, writes: I) -> io::Result<()>
where
I: IntoIterator<Item = (u64, D)>,
D: AsRef<[u8]>,
{
let owned: Vec<(u64, D)> = writes.into_iter().collect();
let mut blocks: Vec<(u64, &[u8])> = owned
.iter()
.map(|(off, d)| (*off, d.as_ref()))
.filter(|(_, d)| !d.is_empty())
.collect();
if blocks.is_empty() {
return Ok(());
}
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
for (off, data) in &blocks {
let end = checked_end(
*off,
data.len() as u64,
"set_batched: offset + len overflows u64",
)?;
if *off < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"set_batched: write range [{off}, {end}) overlaps locked region [0, {locked})"
),
));
}
if end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"set_batched: write range [{off}, {end}) exceeds payload size ({data_size})"
),
));
}
}
fault_point!(self, "set_batched");
if blocks.len() == 1 {
let (off, data) = blocks[0];
return set_in_place(file, data_size, off, data);
}
blocks.sort_by_key(|(off, _)| *off);
for pair in blocks.windows(2) {
let (a_off, a_data) = pair[0];
let (b_off, _) = pair[1];
let a_end = a_off + a_data.len() as u64;
if a_end > b_off {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("set_batched: write range [{a_off}, {a_end}) overlaps [{b_off}, ...)"),
));
}
}
journaled_multi_set(file, data_size, &blocks)
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn inplace_gen<'a, F>(&self, mut f: F) -> io::Result<()>
where
F: FnMut(io::Result<()>) -> Option<BStackGenOp<'a>>,
{
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
let locked = self.locked.load(Ordering::Acquire);
fault_point!(self, "inplace_gen");
let mut overlay: Vec<(u64, &'a [u8])> = Vec::new();
let mut feedback: io::Result<()> = Ok(());
loop {
match f(feedback) {
Some(BStackGenOp::Read { offset, buf }) => {
feedback = inplace_overlay_read(file, data_size, offset, buf, &overlay);
}
Some(BStackGenOp::Write { offset, data }) => {
feedback = inplace_validate_write(offset, data, data_size, locked);
if feedback.is_ok() && !data.is_empty() {
inplace_overlay_insert(&mut overlay, offset, data);
}
}
Some(BStackGenOp::Len { out }) => {
*out = data_size;
feedback = Ok(());
}
Some(BStackGenOp::Swap { .. }) => {
feedback = Err(io::Error::new(
io::ErrorKind::InvalidInput,
"inplace_gen: Swap is not permitted (Read/Write/Len only)",
));
}
Some(BStackGenOp::Push { .. }) => {
feedback = Err(io::Error::new(
io::ErrorKind::InvalidInput,
"inplace_gen: Push is not permitted (in-place writes only)",
));
}
Some(BStackGenOp::Pop { .. }) => {
feedback = Err(io::Error::new(
io::ErrorKind::InvalidInput,
"inplace_gen: Pop is not permitted (in-place writes only)",
));
}
Some(BStackGenOp::Discard { .. }) => {
feedback = Err(io::Error::new(
io::ErrorKind::InvalidInput,
"inplace_gen: Discard is not permitted (in-place writes only)",
));
}
Some(BStackGenOp::Atrunc { .. }) => {
feedback = Err(io::Error::new(
io::ErrorKind::InvalidInput,
"inplace_gen: Atrunc is not permitted (in-place writes only)",
));
}
Some(BStackGenOp::Splice { .. }) => {
feedback = Err(io::Error::new(
io::ErrorKind::InvalidInput,
"inplace_gen: Splice is not permitted (in-place writes only)",
));
}
None => break,
}
}
match overlay.len() {
0 => Ok(()),
1 => {
let (offset, data) = overlay[0];
set_in_place(file, data_size, offset, data)
}
_ => journaled_multi_set(file, data_size, &overlay),
}
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn eq_crds(
&self,
a_offset: u64,
a_expected: impl AsRef<[u8]>,
b_offset: u64,
b_buf: impl AsRef<[u8]>,
) -> io::Result<Option<Vec<u8>>> {
let a_expected = a_expected.as_ref();
let b_buf = b_buf.as_ref();
let a_len = a_expected.len() as u64;
let b_len = b_buf.len() as u64;
let a_end = checked_end(a_offset, a_len, "eq_crds: a_offset + a_len overflows u64")?;
let b_end = checked_end(b_offset, b_len, "eq_crds: b_offset + b_len overflows u64")?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
if !b_buf.is_empty() && b_offset < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"eq_crds: B range [{b_offset}, {b_end}) overlaps locked region [0, {locked})"
),
));
}
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if !a_expected.is_empty() && a_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"eq_crds: A range [{a_offset}, {a_end}) exceeds payload size ({data_size})"
),
));
}
if !b_buf.is_empty() && b_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"eq_crds: B range [{b_offset}, {b_end}) exceeds payload size ({data_size})"
),
));
}
fault_point!(self, "eq_crds");
let mut a_current = vec![0u8; a_expected.len()];
if !a_expected.is_empty() {
read_at(file, a_offset, &mut a_current)?;
}
if a_current != a_expected {
return Ok(None);
}
if b_buf.is_empty() {
return Ok(Some(Vec::new()));
}
let mut old_b = vec![0u8; b_buf.len()];
read_at(file, b_offset, &mut old_b)?;
set_in_place(file, data_size, b_offset, b_buf)?;
Ok(Some(old_b))
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn ne_crds(
&self,
a_offset: u64,
a_expected: impl AsRef<[u8]>,
b_offset: u64,
b_buf: impl AsRef<[u8]>,
) -> io::Result<Option<Vec<u8>>> {
let a_expected = a_expected.as_ref();
let b_buf = b_buf.as_ref();
let a_len = a_expected.len() as u64;
let b_len = b_buf.len() as u64;
let a_end = checked_end(a_offset, a_len, "ne_crds: a_offset + a_len overflows u64")?;
let b_end = checked_end(b_offset, b_len, "ne_crds: b_offset + b_len overflows u64")?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
if !b_buf.is_empty() && b_offset < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"ne_crds: B range [{b_offset}, {b_end}) overlaps locked region [0, {locked})"
),
));
}
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if !a_expected.is_empty() && a_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"ne_crds: A range [{a_offset}, {a_end}) exceeds payload size ({data_size})"
),
));
}
if !b_buf.is_empty() && b_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"ne_crds: B range [{b_offset}, {b_end}) exceeds payload size ({data_size})"
),
));
}
fault_point!(self, "ne_crds");
let mut a_current = vec![0u8; a_expected.len()];
if !a_expected.is_empty() {
read_at(file, a_offset, &mut a_current)?;
}
if a_current == a_expected {
return Ok(None);
}
if b_buf.is_empty() {
return Ok(Some(Vec::new()));
}
let mut old_b = vec![0u8; b_buf.len()];
read_at(file, b_offset, &mut old_b)?;
set_in_place(file, data_size, b_offset, b_buf)?;
Ok(Some(old_b))
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn masked_eq_crds(
&self,
a_offset: u64,
mask: impl AsRef<[u8]>,
a_expected: impl AsRef<[u8]>,
b_offset: u64,
b_buf: impl AsRef<[u8]>,
) -> io::Result<Option<Vec<u8>>> {
let mask = mask.as_ref();
let a_expected = a_expected.as_ref();
let b_buf = b_buf.as_ref();
if mask.len() != a_expected.len() {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"masked_eq_crds: mask length ({}) != a_expected length ({})",
mask.len(),
a_expected.len()
),
));
}
let a_len = a_expected.len() as u64;
let b_len = b_buf.len() as u64;
let a_end = checked_end(
a_offset,
a_len,
"masked_eq_crds: a_offset + a_len overflows u64",
)?;
let b_end = checked_end(
b_offset,
b_len,
"masked_eq_crds: b_offset + b_len overflows u64",
)?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
if !b_buf.is_empty() && b_offset < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"masked_eq_crds: B range [{b_offset}, {b_end}) overlaps locked region [0, {locked})"
),
));
}
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if !a_expected.is_empty() && a_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"masked_eq_crds: A range [{a_offset}, {a_end}) exceeds payload size ({data_size})"
),
));
}
if !b_buf.is_empty() && b_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"masked_eq_crds: B range [{b_offset}, {b_end}) exceeds payload size ({data_size})"
),
));
}
fault_point!(self, "masked_eq_crds");
let mut a_current = vec![0u8; a_expected.len()];
if !a_expected.is_empty() {
read_at(file, a_offset, &mut a_current)?;
}
let masked_match = a_current
.iter()
.zip(mask.iter())
.zip(a_expected.iter())
.all(|((&a, &m), &e)| (a & m) == (e & m));
if !masked_match {
return Ok(None);
}
if b_buf.is_empty() {
return Ok(Some(Vec::new()));
}
let mut old_b = vec![0u8; b_buf.len()];
read_at(file, b_offset, &mut old_b)?;
set_in_place(file, data_size, b_offset, b_buf)?;
Ok(Some(old_b))
}
#[cfg(all(feature = "set", feature = "atomic"))]
pub fn masked_ne_crds(
&self,
a_offset: u64,
mask: impl AsRef<[u8]>,
a_expected: impl AsRef<[u8]>,
b_offset: u64,
b_buf: impl AsRef<[u8]>,
) -> io::Result<Option<Vec<u8>>> {
let mask = mask.as_ref();
let a_expected = a_expected.as_ref();
let b_buf = b_buf.as_ref();
if mask.len() != a_expected.len() {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"masked_ne_crds: mask length ({}) != a_expected length ({})",
mask.len(),
a_expected.len()
),
));
}
let a_len = a_expected.len() as u64;
let b_len = b_buf.len() as u64;
let a_end = checked_end(
a_offset,
a_len,
"masked_ne_crds: a_offset + a_len overflows u64",
)?;
let b_end = checked_end(
b_offset,
b_len,
"masked_ne_crds: b_offset + b_len overflows u64",
)?;
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let locked = self.locked.load(Ordering::Acquire);
if !b_buf.is_empty() && b_offset < locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"masked_ne_crds: B range [{b_offset}, {b_end}) overlaps locked region [0, {locked})"
),
));
}
let data_size = file.seek(SeekFrom::End(0))?.saturating_sub(HEADER_SIZE);
if !a_expected.is_empty() && a_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"masked_ne_crds: A range [{a_offset}, {a_end}) exceeds payload size ({data_size})"
),
));
}
if !b_buf.is_empty() && b_end > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"masked_ne_crds: B range [{b_offset}, {b_end}) exceeds payload size ({data_size})"
),
));
}
fault_point!(self, "masked_ne_crds");
let mut a_current = vec![0u8; a_expected.len()];
if !a_expected.is_empty() {
read_at(file, a_offset, &mut a_current)?;
}
let masked_match = a_current
.iter()
.zip(mask.iter())
.zip(a_expected.iter())
.all(|((&a, &m), &e)| (a & m) == (e & m));
if masked_match {
return Ok(None);
}
if b_buf.is_empty() {
return Ok(Some(Vec::new()));
}
let mut old_b = vec![0u8; b_buf.len()];
read_at(file, b_offset, &mut old_b)?;
set_in_place(file, data_size, b_offset, b_buf)?;
Ok(Some(old_b))
}
}
impl BStack {
pub fn len(&self) -> io::Result<u64> {
fault_point!(self, "len");
Ok(self.lock.read().unwrap().1)
}
pub fn is_empty(&self) -> io::Result<bool> {
fault_point!(self, "is_empty");
Ok(self.lock.read().unwrap().1 == 0)
}
pub fn locked_len(&self) -> u64 {
self.locked.load(Ordering::Acquire)
}
pub fn lock_up_to(&self, n: u64) -> io::Result<()> {
#[allow(unused_mut)]
let mut guard = self.lock.write().unwrap();
let file = &mut guard.0;
let data_size = file.metadata()?.len().saturating_sub(HEADER_SIZE);
let current_locked = self.locked.load(Ordering::Relaxed);
if n < current_locked {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"lock_up_to: n ({n}) is less than the current locked length ({current_locked})"
),
));
}
if n > data_size {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("lock_up_to: n ({n}) exceeds payload size ({data_size})"),
));
}
fault_point!(self, "lock_up_to");
if self.cache_enabled && n > current_locked {
if n > usize::MAX as u64 {
return Err(io::Error::new(
io::ErrorKind::OutOfMemory,
"lock_up_to: locked region too large to cache on this platform",
));
}
let ol = current_locked as usize; let nl = n as usize; if nl > isize::MAX as usize {
return Err(io::Error::new(
io::ErrorKind::OutOfMemory,
"lock_up_to: locked region too large to cache on this platform",
));
}
let mut cache = self.cache.lock().unwrap();
if nl > cache.capacity() {
let new_cap = nl.next_power_of_two();
let mut new_cache = Vec::with_capacity(new_cap);
new_cache.extend_from_slice(&cache[..ol]);
new_cache.resize(nl, 0u8);
#[cfg(unix)]
pread_exact_raw(self.fd, HEADER_SIZE + ol as u64, &mut new_cache[ol..nl])?;
#[cfg(windows)]
pread_exact_raw_handle(
self.handle,
HEADER_SIZE + ol as u64,
&mut new_cache[ol..nl],
)?;
#[cfg(not(any(unix, windows)))]
{
file.seek(SeekFrom::Start(HEADER_SIZE + ol as u64))?;
file.read_exact(&mut new_cache[ol..nl])?;
}
*cache = new_cache;
} else {
cache.resize(nl, 0u8);
#[cfg(unix)]
if let Err(e) =
pread_exact_raw(self.fd, HEADER_SIZE + ol as u64, &mut cache[ol..nl])
{
cache.truncate(ol);
return Err(e);
}
#[cfg(windows)]
if let Err(e) =
pread_exact_raw_handle(self.handle, HEADER_SIZE + ol as u64, &mut cache[ol..nl])
{
cache.truncate(ol);
return Err(e);
}
#[cfg(not(any(unix, windows)))]
if let Err(e) = file
.seek(SeekFrom::Start(HEADER_SIZE + ol as u64))
.and_then(|_| file.read_exact(&mut cache[ol..nl]))
{
cache.truncate(ol);
return Err(e);
}
}
}
self.locked.store(n, Ordering::Release);
drop(guard);
Ok(())
}
pub fn open_locked_up_to(path: impl AsRef<Path>, n: u64) -> io::Result<Self> {
let stack = Self::open(path)?;
stack.lock_up_to(n)?;
Ok(stack)
}
pub fn open_cached(path: impl AsRef<Path>) -> io::Result<Self> {
let mut stack = Self::open(path)?;
stack.cache_enabled = true;
Ok(stack)
}
pub fn open_locked_up_to_cached(path: impl AsRef<Path>, n: u64) -> io::Result<Self> {
let stack = Self::open_cached(path)?;
stack.lock_up_to(n)?;
Ok(stack)
}
}
#[cfg(all(debug_assertions, feature = "fault-injection"))]
impl BStack {
pub fn with_fault_policy(self, policy: std::sync::Arc<dyn fault::FaultPolicy>) -> Self {
self.fault.set(Some(policy));
self
}
pub fn set_fault_policy(&self, policy: Option<std::sync::Arc<dyn fault::FaultPolicy>>) {
self.fault.set(policy);
}
pub fn fault_policy(&self) -> Option<std::sync::Arc<dyn fault::FaultPolicy>> {
self.fault.get()
}
}
impl io::Write for BStack {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.push(buf)?;
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl io::Write for &BStack {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.push(buf)?;
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl fmt::Debug for BStack {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("BStack")
.field(
"version",
&format!("{}.{}.{}", MAGIC[4], MAGIC[5], MAGIC[6]),
)
.field("len", &self.len().ok())
.finish_non_exhaustive()
}
}
impl Eq for BStack {}
impl PartialEq for BStack {
fn eq(&self, other: &Self) -> bool {
std::ptr::eq(self, other)
}
}
impl Hash for BStack {
fn hash<H: Hasher>(&self, state: &mut H) {
(self as *const BStack).hash(state);
}
}
#[derive(Clone, Copy, PartialEq, Eq, Hash)]
pub struct BStackReader<'a> {
stack: &'a BStack,
offset: u64,
}
impl BStack {
pub fn reader(&self) -> BStackReader<'_> {
BStackReader {
stack: self,
offset: 0,
}
}
pub fn reader_at(&self, offset: u64) -> BStackReader<'_> {
BStackReader {
stack: self,
offset,
}
}
}
impl<'a> BStackReader<'a> {
pub fn position(&self) -> u64 {
self.offset
}
}
impl<'a> From<&'a BStack> for BStackReader<'a> {
fn from(stack: &'a BStack) -> Self {
stack.reader()
}
}
impl<'a> From<BStackReader<'a>> for &'a BStack {
fn from(val: BStackReader<'a>) -> Self {
val.stack
}
}
impl<'a> PartialOrd for BStackReader<'a> {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl<'a> Ord for BStackReader<'a> {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
let self_ptr = self.stack as *const BStack as usize;
let other_ptr = other.stack as *const BStack as usize;
self_ptr
.cmp(&other_ptr)
.then(self.offset.cmp(&other.offset))
}
}
impl<'a> io::Read for BStackReader<'a> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
if buf.is_empty() {
return Ok(0);
}
let data_size = self.stack.len()?;
if self.offset >= data_size {
return Ok(0);
}
let available = (data_size - self.offset) as usize;
let n = buf.len().min(available);
self.stack.get_into(self.offset, &mut buf[..n])?;
self.offset += n as u64;
Ok(n)
}
}
impl<'a> io::Seek for BStackReader<'a> {
fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
let data_size = self.stack.len()? as i128;
let new_offset = match pos {
SeekFrom::Start(n) => n as i128,
SeekFrom::End(n) => data_size + n as i128,
SeekFrom::Current(n) => self.offset as i128 + n as i128,
};
if new_offset < 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"seek before beginning of payload",
));
}
self.offset = new_offset as u64;
Ok(self.offset)
}
}