#![allow(unsafe_code)]
use std::os::raw::c_int;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use crate::error::{Error, LockKind, Result};
use crate::platform::FileHandle;
pub const WRITER_LOCK_OFFSET: u64 = 96;
pub const READER_LOCK_RANGE_OFFSET: u64 = 97;
pub const READER_LOCK_RANGE_LEN: u64 = 31;
const INITIAL_BACKOFF: Duration = Duration::from_millis(1);
const MAX_BACKOFF: Duration = Duration::from_millis(100);
#[derive(Debug)]
#[must_use = "WriterLock releases the OS-side lock when dropped"]
pub struct WriterLock {
fd: c_int,
released: bool,
}
impl WriterLock {
pub fn release(mut self) -> Result<()> {
if self.released {
return Ok(());
}
self.released = true;
unlock_range(self.fd, WRITER_LOCK_OFFSET, 1)
}
}
impl Drop for WriterLock {
fn drop(&mut self) {
if !self.released {
let _ = unlock_range(self.fd, WRITER_LOCK_OFFSET, 1);
}
}
}
#[derive(Debug)]
#[must_use = "ReaderLock releases the OS-side lock when dropped"]
pub struct ReaderLock {
fd: c_int,
slot: u64,
released: bool,
}
impl ReaderLock {
#[must_use]
pub fn slot(&self) -> u64 {
self.slot
}
pub fn release(mut self) -> Result<()> {
if self.released {
return Ok(());
}
self.released = true;
unlock_range(self.fd, self.slot, 1)
}
}
impl Drop for ReaderLock {
fn drop(&mut self) {
if !self.released {
let _ = unlock_range(self.fd, self.slot, 1);
}
}
}
impl FileHandle {
pub fn try_lock_writer(&self) -> Result<Option<WriterLock>> {
ensure_ofd_locks_supported()?;
let fd = self.raw_fd();
if try_lock_range(fd, WRITER_LOCK_OFFSET, 1, LockMode::Exclusive)? {
Ok(Some(WriterLock {
fd,
released: false,
}))
} else {
Ok(None)
}
}
pub fn lock_writer(&self, timeout: Duration) -> Result<WriterLock> {
ensure_ofd_locks_supported()?;
let fd = self.raw_fd();
retry_until_acquired(timeout, LockKind::Writer, || {
try_lock_range(fd, WRITER_LOCK_OFFSET, 1, LockMode::Exclusive)
})?;
Ok(WriterLock {
fd,
released: false,
})
}
pub fn lock_reader(&self, timeout: Duration) -> Result<ReaderLock> {
ensure_ofd_locks_supported()?;
let fd = self.raw_fd();
let start_slot = next_reader_slot();
let mut last_err: Option<Error> = None;
for offset in 0..READER_LOCK_RANGE_LEN {
let slot = READER_LOCK_RANGE_OFFSET + ((start_slot + offset) % READER_LOCK_RANGE_LEN);
match try_lock_range(fd, slot, 1, LockMode::Shared) {
Ok(true) => {
return Ok(ReaderLock {
fd,
slot,
released: false,
});
}
Ok(false) => {}
Err(e) => last_err = Some(e),
}
}
if let Some(err) = last_err {
return Err(err);
}
let slot = READER_LOCK_RANGE_OFFSET + start_slot;
retry_until_acquired(timeout, LockKind::Reader, || {
try_lock_range(fd, slot, 1, LockMode::Shared)
})?;
Ok(ReaderLock {
fd,
slot,
released: false,
})
}
#[cfg(unix)]
fn raw_fd(&self) -> c_int {
use std::os::unix::io::AsRawFd;
self.file_ref().as_raw_fd()
}
#[cfg(windows)]
fn raw_fd(&self) -> c_int {
use std::os::windows::io::AsRawHandle;
self.file_ref().as_raw_handle() as c_int
}
}
static READER_ROUND_ROBIN: AtomicU64 = AtomicU64::new(0);
fn next_reader_slot() -> u64 {
READER_ROUND_ROBIN.fetch_add(1, Ordering::Relaxed) % READER_LOCK_RANGE_LEN
}
#[derive(Debug, Clone, Copy)]
enum LockMode {
Exclusive,
Shared,
}
fn retry_until_acquired<F>(timeout: Duration, kind: LockKind, mut once: F) -> Result<()>
where
F: FnMut() -> Result<bool>,
{
let start = Instant::now();
let mut backoff = INITIAL_BACKOFF;
let timeout_millis = u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX);
let max_iters: u64 = timeout_millis.saturating_add(2);
let mut iters: u64 = 0;
loop {
iters = iters.saturating_add(1);
if iters > max_iters.saturating_add(64) {
return Err(Error::Busy { kind });
}
if once()? {
return Ok(());
}
if start.elapsed() >= timeout {
return Err(Error::Busy { kind });
}
std::thread::sleep(backoff);
backoff = (backoff * 2).min(MAX_BACKOFF);
}
}
#[cfg(unix)]
fn build_flock(l_type: i32, offset: u64, len: u64) -> Result<libc::flock> {
let l_type_short =
libc::c_short::try_from(l_type).map_err(|_| Error::InvalidArgument("lock l_type"))?;
let l_whence_short = libc::c_short::try_from(libc::SEEK_SET)
.map_err(|_| Error::InvalidArgument("lock l_whence"))?;
Ok(libc::flock {
l_type: l_type_short,
l_whence: l_whence_short,
l_start: offset_to_off_t(offset)?,
l_len: offset_to_off_t(len)?,
l_pid: 0,
#[cfg(target_os = "freebsd")]
l_sysid: 0,
})
}
#[cfg(unix)]
fn try_lock_range(fd: c_int, offset: u64, len: u64, mode: LockMode) -> Result<bool> {
#[allow(clippy::useless_conversion)]
let l_type: i32 = match mode {
LockMode::Exclusive => libc::F_WRLCK.into(),
LockMode::Shared => libc::F_RDLCK.into(),
};
let flock = build_flock(l_type, offset, len)?;
let ret = unsafe { libc::fcntl(fd, ofd_setlk_cmd(), &raw const flock) };
if ret == 0 {
return Ok(true);
}
let errno = unsafe { *libc_errno() };
if errno == libc::EAGAIN || errno == libc::EACCES {
return Ok(false);
}
Err(Error::Io(std::io::Error::from_raw_os_error(errno)))
}
#[cfg(unix)]
fn unlock_range(fd: c_int, offset: u64, len: u64) -> Result<()> {
#[allow(clippy::useless_conversion)]
let flock = build_flock(libc::F_UNLCK.into(), offset, len)?;
let ret = unsafe { libc::fcntl(fd, ofd_setlk_cmd(), &raw const flock) };
if ret == 0 {
return Ok(());
}
let errno = unsafe { *libc_errno() };
Err(Error::Io(std::io::Error::from_raw_os_error(errno)))
}
#[cfg(unix)]
const TARGET_HAS_OFD_LOCKS: bool = cfg!(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
));
#[cfg(unix)]
fn ensure_ofd_locks_supported() -> Result<()> {
if TARGET_HAS_OFD_LOCKS {
return Ok(());
}
Err(Error::Io(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"obj requires OFD (open-file-description) fcntl locks, which \
this target does not provide; classic POSIX F_SETLK locks are \
per-process and would silently break same-process multi-handle \
exclusion (see obj-core platform::lock supported-target matrix)",
)))
}
#[cfg(unix)]
fn ofd_setlk_cmd() -> c_int {
#[cfg(any(target_os = "linux", target_os = "android"))]
{
37 }
#[cfg(target_vendor = "apple")]
{
90 }
#[cfg(not(any(target_os = "linux", target_os = "android", target_vendor = "apple",)))]
{
-1
}
}
#[cfg(unix)]
fn offset_to_off_t(v: u64) -> Result<libc::off_t> {
libc::off_t::try_from(v).map_err(|_| Error::InvalidArgument("lock offset overflow"))
}
#[cfg(unix)]
fn libc_errno() -> *mut c_int {
#[cfg(any(target_os = "linux", target_os = "android"))]
unsafe {
libc::__errno_location()
}
#[cfg(target_vendor = "apple")]
unsafe {
libc::__error()
}
#[cfg(any(target_os = "freebsd", target_os = "dragonfly"))]
unsafe {
libc::__error()
}
#[cfg(any(target_os = "openbsd", target_os = "netbsd"))]
unsafe {
libc::__errno()
}
#[cfg(not(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
target_os = "freebsd",
target_os = "dragonfly",
target_os = "openbsd",
target_os = "netbsd",
)))]
unsafe {
libc::__errno_location()
}
}
#[cfg(windows)]
#[allow(clippy::unnecessary_wraps)]
fn ensure_ofd_locks_supported() -> Result<()> {
Ok(())
}
#[cfg(windows)]
#[allow(clippy::cast_possible_truncation)]
fn split_u64(v: u64) -> (u32, u32) {
(v as u32, (v >> 32) as u32)
}
#[cfg(windows)]
fn try_lock_range(fd: c_int, offset: u64, len: u64, mode: LockMode) -> Result<bool> {
use windows_sys::Win32::Foundation::{ERROR_IO_PENDING, ERROR_LOCK_VIOLATION, HANDLE};
use windows_sys::Win32::Storage::FileSystem::{
LockFileEx, LOCKFILE_EXCLUSIVE_LOCK, LOCKFILE_FAIL_IMMEDIATELY,
};
use windows_sys::Win32::System::IO::OVERLAPPED;
let mut flags = LOCKFILE_FAIL_IMMEDIATELY;
if matches!(mode, LockMode::Exclusive) {
flags |= LOCKFILE_EXCLUSIVE_LOCK;
}
let mut overlapped: OVERLAPPED = unsafe { std::mem::zeroed() };
let (off_lo, off_hi) = split_u64(offset);
let (len_lo, len_hi) = split_u64(len);
overlapped.Anonymous.Anonymous.Offset = off_lo;
overlapped.Anonymous.Anonymous.OffsetHigh = off_hi;
let ret = unsafe { LockFileEx(fd as HANDLE, flags, 0, len_lo, len_hi, &raw mut overlapped) };
if ret != 0 {
return Ok(true);
}
let last = unsafe { windows_sys::Win32::Foundation::GetLastError() };
if last == ERROR_LOCK_VIOLATION || last == ERROR_IO_PENDING {
return Ok(false);
}
Err(Error::Io(std::io::Error::from_raw_os_error(
last.cast_signed(),
)))
}
#[cfg(windows)]
fn unlock_range(fd: c_int, offset: u64, len: u64) -> Result<()> {
use windows_sys::Win32::Foundation::HANDLE;
use windows_sys::Win32::Storage::FileSystem::UnlockFileEx;
use windows_sys::Win32::System::IO::OVERLAPPED;
let mut overlapped: OVERLAPPED = unsafe { std::mem::zeroed() };
let (off_lo, off_hi) = split_u64(offset);
let (len_lo, len_hi) = split_u64(len);
overlapped.Anonymous.Anonymous.Offset = off_lo;
overlapped.Anonymous.Anonymous.OffsetHigh = off_hi;
let ret = unsafe { UnlockFileEx(fd as HANDLE, 0, len_lo, len_hi, &raw mut overlapped) };
if ret != 0 {
return Ok(());
}
let last = unsafe { windows_sys::Win32::Foundation::GetLastError() };
Err(Error::Io(std::io::Error::from_raw_os_error(
last.cast_signed(),
)))
}
impl FileHandle {
fn file_ref(&self) -> &std::fs::File {
&self.file
}
}
#[cfg(test)]
mod tests {
#[cfg(unix)]
use super::*;
#[cfg(unix)]
use tempfile::TempDir;
#[cfg(unix)]
fn fresh_handle(dir: &TempDir, name: &str) -> FileHandle {
let path = dir.path().join(name);
let h = FileHandle::open_or_create(&path).expect("open");
h.set_len(4096).expect("extend");
h
}
#[test]
#[cfg(unix)]
fn writer_lock_excludes_writers() {
let dir = TempDir::new().expect("tmp");
let path = dir.path().join("lock.obj");
FileHandle::open_or_create(&path)
.expect("init")
.set_len(4096)
.expect("len");
let h1 = FileHandle::open_or_create(&path).expect("h1");
let h2 = FileHandle::open_or_create(&path).expect("h2");
let guard = h1
.try_lock_writer()
.expect("try lock h1")
.expect("must acquire");
let none = h2.try_lock_writer().expect("try lock h2");
assert!(none.is_none(), "second writer lock must be refused");
drop(guard);
let _g2 = h2
.try_lock_writer()
.expect("try lock h2 again")
.expect("now acquires");
}
#[test]
#[cfg(unix)]
fn writer_busy_timeout_returns_err_busy() {
let dir = TempDir::new().expect("tmp");
let _h0 = fresh_handle(&dir, "lock.obj");
let h1 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h1");
let h2 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h2");
let _g1 = h1
.try_lock_writer()
.expect("h1 lock")
.expect("h1 must acquire");
let start = std::time::Instant::now();
let err = h2
.lock_writer(Duration::from_millis(50))
.expect_err("must time out");
let elapsed = start.elapsed();
assert!(matches!(
err,
Error::Busy {
kind: LockKind::Writer
}
));
assert!(
elapsed >= Duration::from_millis(45),
"must wait at least the timeout (~50 ms); got {elapsed:?}",
);
}
#[test]
#[cfg(unix)]
fn many_readers_can_coexist() {
let dir = TempDir::new().expect("tmp");
let _h0 = fresh_handle(&dir, "lock.obj");
let h1 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h1");
let h2 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h2");
let h3 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h3");
let g1 = h1.lock_reader(Duration::from_millis(50)).expect("r1");
let g2 = h2.lock_reader(Duration::from_millis(50)).expect("r2");
let g3 = h3.lock_reader(Duration::from_millis(50)).expect("r3");
drop((g1, g2, g3));
}
#[test]
#[cfg(unix)]
fn reader_and_writer_dont_collide_on_separate_anchors() {
let dir = TempDir::new().expect("tmp");
let _h0 = fresh_handle(&dir, "lock.obj");
let h1 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h1");
let h2 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h2");
let _wg = h1.lock_writer(Duration::from_millis(50)).expect("writer");
let _rg = h2
.lock_reader(Duration::from_millis(50))
.expect("reader must not collide");
}
#[test]
#[cfg(unix)]
fn explicit_release_returns_ok() {
let dir = TempDir::new().expect("tmp");
let _h0 = fresh_handle(&dir, "lock.obj");
let h = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h");
let g = h.lock_writer(Duration::from_millis(50)).expect("lock");
g.release().expect("release ok");
let _g2 = h.lock_writer(Duration::from_millis(50)).expect("relock");
}
#[test]
#[cfg(unix)]
fn lock_methods_compile_when_dropped() {
let dir = TempDir::new().expect("tmp");
let _h0 = fresh_handle(&dir, "lock.obj");
let h = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h");
let g = h.lock_reader(Duration::from_millis(10)).expect("rlock");
drop(g);
}
#[test]
#[cfg(unix)]
fn ofd_capability_gate_matches_target() {
assert_eq!(
TARGET_HAS_OFD_LOCKS,
cfg!(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
)),
"OFD capability constant must track the supported-target set",
);
let gate = ensure_ofd_locks_supported();
if TARGET_HAS_OFD_LOCKS {
gate.expect("supported targets must pass the gate");
} else {
match gate {
Err(Error::Io(e)) => {
assert_eq!(e.kind(), std::io::ErrorKind::Unsupported);
}
other => panic!("expected Io(Unsupported), got {other:?}"),
}
}
}
#[test]
#[cfg(unix)]
fn same_process_multi_fd_writer_exclusion_holds() {
let dir = TempDir::new().expect("tmp");
let _h0 = fresh_handle(&dir, "lock.obj");
let h1 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h1");
let h2 = FileHandle::open_or_create(dir.path().join("lock.obj")).expect("h2");
let g1 = h1.try_lock_writer().expect("h1 try").expect("h1 acquires");
assert!(
h2.try_lock_writer().expect("h2 try").is_none(),
"per-fd OFD lock must refuse a second in-process handle; a \
per-process F_SETLK fallback would wrongly grant this",
);
drop(g1);
}
}