use std::collections::{btree_map::Values, BTreeMap, BTreeSet};
use std::error::Error;
use std::fmt;
use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex, MutexGuard};
use crate::vfs::VirtualStat;
pub const MAX_FDS_PER_PROCESS: usize = 256;
pub const O_RDONLY: u32 = 0;
pub const O_WRONLY: u32 = 1;
pub const O_RDWR: u32 = 2;
pub const O_CREAT: u32 = 0o100;
pub const O_EXCL: u32 = 0o200;
pub const O_TRUNC: u32 = 0o1000;
pub const O_APPEND: u32 = 0o2000;
pub const O_NONBLOCK: u32 = 0o4000;
pub const O_DIRECT: u32 = 0o40000;
pub const O_DIRECTORY: u32 = 0o200000;
pub const O_NOFOLLOW: u32 = 0o400000;
pub const F_DUPFD: u32 = 0;
pub const F_GETFD: u32 = 1;
pub const F_SETFD: u32 = 2;
pub const F_GETFL: u32 = 3;
pub const F_SETFL: u32 = 4;
pub const FD_CLOEXEC: u32 = 1;
pub const LOCK_SH: u32 = 1;
pub const LOCK_EX: u32 = 2;
pub const LOCK_NB: u32 = 4;
pub const LOCK_UN: u32 = 8;
const DEFAULT_MAX_RECORD_LOCKS: usize = 4096;
pub const FILETYPE_UNKNOWN: u8 = 0;
pub const FILETYPE_BLOCK_DEVICE: u8 = 1;
pub const FILETYPE_CHARACTER_DEVICE: u8 = 2;
pub const FILETYPE_DIRECTORY: u8 = 3;
pub const FILETYPE_REGULAR_FILE: u8 = 4;
pub const FILETYPE_SOCKET_DGRAM: u8 = 5;
pub const FILETYPE_SOCKET_STREAM: u8 = 6;
pub const FILETYPE_PIPE: u8 = FILETYPE_SOCKET_STREAM;
pub const FILETYPE_SYMBOLIC_LINK: u8 = 7;
pub type FdResult<T> = Result<T, FdTableError>;
pub type SharedFileDescription = Arc<FileDescription>;
static NEXT_FILE_DESCRIPTION_ID: AtomicU64 = AtomicU64::new(1);
pub(crate) fn allocate_file_description_id() -> u64 {
NEXT_FILE_DESCRIPTION_ID
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |id| id.checked_add(1))
.expect("open-file-description id space exhausted")
}
#[derive(Debug, Default)]
pub struct AnonymousFileUsage {
bytes: AtomicU64,
inodes: AtomicUsize,
}
impl AnonymousFileUsage {
pub fn bytes(&self) -> u64 {
self.bytes.load(Ordering::SeqCst)
}
pub fn inodes(&self) -> usize {
self.inodes.load(Ordering::SeqCst)
}
fn add_file(&self, size: u64) {
self.bytes.fetch_add(size, Ordering::SeqCst);
self.inodes.fetch_add(1, Ordering::SeqCst);
}
fn remove_file(&self, size: u64) {
self.bytes.fetch_sub(size, Ordering::SeqCst);
self.inodes.fetch_sub(1, Ordering::SeqCst);
}
fn resize_file(&self, old_size: u64, new_size: u64) {
if new_size >= old_size {
self.bytes
.fetch_add(new_size.saturating_sub(old_size), Ordering::SeqCst);
} else {
self.bytes
.fetch_sub(old_size.saturating_sub(new_size), Ordering::SeqCst);
}
}
}
#[derive(Debug)]
pub struct AnonymousFile {
pub data: Vec<u8>,
pub stat: VirtualStat,
usage: Arc<AnonymousFileUsage>,
}
impl AnonymousFile {
pub fn new(data: Vec<u8>, stat: VirtualStat, usage: Arc<AnonymousFileUsage>) -> Self {
usage.add_file(stat.size);
Self { data, stat, usage }
}
}
impl Drop for AnonymousFile {
fn drop(&mut self) {
self.usage.remove_file(self.stat.size);
}
}
pub type SharedAnonymousFile = Arc<Mutex<AnonymousFile>>;
#[derive(Debug)]
enum FileBacking {
Path(String),
LinkedAlias {
former_path: String,
live_path: String,
},
Anonymous {
former_path: String,
file: SharedAnonymousFile,
},
DetachedDirectory {
former_path: String,
stat: VirtualStat,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FdTableError {
code: &'static str,
message: String,
}
impl FdTableError {
pub fn code(&self) -> &'static str {
self.code
}
fn bad_file_descriptor(fd: u32) -> Self {
Self {
code: "EBADF",
message: format!("bad file descriptor {fd}"),
}
}
fn too_many_open_files() -> Self {
Self {
code: "EMFILE",
message: String::from("too many open files"),
}
}
fn no_memory(message: impl Into<String>) -> Self {
Self {
code: "ENOMEM",
message: message.into(),
}
}
fn invalid_argument(message: impl Into<String>) -> Self {
Self {
code: "EINVAL",
message: message.into(),
}
}
fn would_block(message: impl Into<String>) -> Self {
Self {
code: "EWOULDBLOCK",
message: message.into(),
}
}
fn deadlock(message: impl Into<String>) -> Self {
Self {
code: "EDEADLK",
message: message.into(),
}
}
}
impl fmt::Display for FdTableError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}: {}", self.code, self.message)
}
}
impl Error for FdTableError {}
#[derive(Debug)]
pub struct FileDescription {
id: u64,
backing: Mutex<FileBacking>,
lock_target: Option<FileLockTarget>,
cursor: AtomicU64,
flags: AtomicU32,
ref_count: AtomicUsize,
}
impl FileDescription {
pub fn new(id: u64, path: impl Into<String>, flags: u32) -> Self {
Self::with_ref_count_and_lock(id, path, flags, 1, None)
}
pub fn new_with_lock(
id: u64,
path: impl Into<String>,
flags: u32,
lock_target: Option<FileLockTarget>,
) -> Self {
Self::with_ref_count_and_lock(id, path, flags, 1, lock_target)
}
pub fn with_ref_count(id: u64, path: impl Into<String>, flags: u32, ref_count: usize) -> Self {
Self::with_ref_count_and_lock(id, path, flags, ref_count, None)
}
pub fn with_ref_count_and_lock(
id: u64,
path: impl Into<String>,
flags: u32,
ref_count: usize,
lock_target: Option<FileLockTarget>,
) -> Self {
Self {
id,
backing: Mutex::new(FileBacking::Path(path.into())),
lock_target,
cursor: AtomicU64::new(0),
flags: AtomicU32::new(flags),
ref_count: AtomicUsize::new(ref_count),
}
}
pub fn id(&self) -> u64 {
self.id
}
pub fn path(&self) -> String {
match &*lock_or_recover(&self.backing) {
FileBacking::Path(path) => path.clone(),
FileBacking::LinkedAlias { live_path, .. } => live_path.clone(),
FileBacking::Anonymous { former_path, .. } => former_path.clone(),
FileBacking::DetachedDirectory { former_path, .. } => former_path.clone(),
}
}
pub fn proc_display_path(&self) -> String {
match &*lock_or_recover(&self.backing) {
FileBacking::Path(path) => path.clone(),
FileBacking::LinkedAlias { former_path, .. }
| FileBacking::Anonymous { former_path, .. }
| FileBacking::DetachedDirectory { former_path, .. } => {
format!("{former_path} (deleted)")
}
}
}
pub fn is_path_backed_by(&self, expected: &str) -> bool {
match &*lock_or_recover(&self.backing) {
FileBacking::Path(path) => path == expected,
FileBacking::LinkedAlias { live_path, .. } => live_path == expected,
FileBacking::Anonymous { .. } | FileBacking::DetachedDirectory { .. } => false,
}
}
pub fn rename_path_prefix(&self, old_path: &str, new_path: &str) {
let mut backing = lock_or_recover(&self.backing);
let path = match &mut *backing {
FileBacking::Path(path) => path,
FileBacking::LinkedAlias { live_path, .. } => live_path,
FileBacking::Anonymous { .. } | FileBacking::DetachedDirectory { .. } => return,
};
if path == old_path {
*path = new_path.to_string();
} else if let Some(suffix) = path
.strip_prefix(old_path)
.filter(|value| value.starts_with('/'))
{
*path = format!("{new_path}{suffix}");
}
}
pub fn detach_path(&self, expected: &str, file: SharedAnonymousFile) -> bool {
let mut backing = lock_or_recover(&self.backing);
let former_path = match &*backing {
FileBacking::Path(path) if path == expected => expected.to_string(),
FileBacking::LinkedAlias {
former_path,
live_path,
} if live_path == expected => former_path.clone(),
_ => return false,
};
*backing = FileBacking::Anonymous { former_path, file };
true
}
pub fn rebind_deleted_path(&self, expected: &str, live_path: &str) -> bool {
let mut backing = lock_or_recover(&self.backing);
let former_path = match &*backing {
FileBacking::Path(path) if path == expected => expected.to_string(),
FileBacking::LinkedAlias {
former_path,
live_path: current,
} if current == expected => former_path.clone(),
_ => return false,
};
*backing = FileBacking::LinkedAlias {
former_path,
live_path: live_path.to_string(),
};
true
}
pub fn detach_directory(&self, _expected: &str, stat: VirtualStat) -> bool {
let mut backing = lock_or_recover(&self.backing);
let FileBacking::Path(former_path) = &*backing else {
return false;
};
let former_path = former_path.clone();
*backing = FileBacking::DetachedDirectory { former_path, stat };
true
}
pub fn detached_directory_stat(&self) -> Option<VirtualStat> {
match &*lock_or_recover(&self.backing) {
FileBacking::DetachedDirectory { stat, .. } => Some(stat.clone()),
FileBacking::Path(_)
| FileBacking::LinkedAlias { .. }
| FileBacking::Anonymous { .. } => None,
}
}
pub fn anonymous_stat(&self) -> Option<VirtualStat> {
let file = match &*lock_or_recover(&self.backing) {
FileBacking::Anonymous { file, .. } => Arc::clone(file),
FileBacking::Path(_)
| FileBacking::LinkedAlias { .. }
| FileBacking::DetachedDirectory { .. } => return None,
};
let stat = lock_or_recover(&file).stat.clone();
Some(stat)
}
pub fn anonymous_pread(&self, offset: u64, length: usize) -> Option<Vec<u8>> {
let file = match &*lock_or_recover(&self.backing) {
FileBacking::Anonymous { file, .. } => Arc::clone(file),
FileBacking::Path(_)
| FileBacking::LinkedAlias { .. }
| FileBacking::DetachedDirectory { .. } => return None,
};
let file = lock_or_recover(&file);
let start = usize::try_from(offset).unwrap_or(usize::MAX);
if start >= file.data.len() {
return Some(Vec::new());
}
let end = start.saturating_add(length).min(file.data.len());
Some(file.data[start..end].to_vec())
}
pub fn anonymous_pwrite(&self, offset: u64, data: &[u8]) -> Option<FdResult<u64>> {
let file = match &*lock_or_recover(&self.backing) {
FileBacking::Anonymous { file, .. } => Arc::clone(file),
FileBacking::Path(_)
| FileBacking::LinkedAlias { .. }
| FileBacking::DetachedDirectory { .. } => return None,
};
let mut file = lock_or_recover(&file);
let Ok(start) = usize::try_from(offset) else {
return Some(Err(FdTableError::invalid_argument(
"anonymous file offset exceeds host address space",
)));
};
let Some(end) = start.checked_add(data.len()) else {
return Some(Err(FdTableError::invalid_argument(
"anonymous file write range overflows",
)));
};
if end > file.data.len() {
let additional = end - file.data.len();
if file.data.try_reserve(additional).is_err() {
return Some(Err(FdTableError::no_memory(
"anonymous file write allocation failed",
)));
}
file.data.resize(end, 0);
}
file.data[start..end].copy_from_slice(data);
let old_size = file.stat.size;
file.stat.size = file.data.len() as u64;
file.stat.blocks = file.stat.size.div_ceil(512);
file.usage.resize_file(old_size, file.stat.size);
Some(Ok(file.stat.size))
}
pub fn anonymous_truncate(&self, length: u64) -> Option<FdResult<()>> {
let file = match &*lock_or_recover(&self.backing) {
FileBacking::Anonymous { file, .. } => Arc::clone(file),
FileBacking::Path(_)
| FileBacking::LinkedAlias { .. }
| FileBacking::DetachedDirectory { .. } => return None,
};
let Ok(length) = usize::try_from(length) else {
return Some(Err(FdTableError::invalid_argument(
"anonymous file length exceeds host address space",
)));
};
let mut file = lock_or_recover(&file);
let additional = length.saturating_sub(file.data.len());
if additional > 0 && file.data.try_reserve(additional).is_err() {
return Some(Err(FdTableError::no_memory(
"anonymous file truncate allocation failed",
)));
}
file.data.resize(length, 0);
let old_size = file.stat.size;
file.stat.size = length as u64;
file.stat.blocks = file.stat.size.div_ceil(512);
file.usage.resize_file(old_size, file.stat.size);
Some(Ok(()))
}
pub fn detached_chmod(&self, mode: u32) -> bool {
let mut backing = lock_or_recover(&self.backing);
match &mut *backing {
FileBacking::Anonymous { file, .. } => {
let mut file = lock_or_recover(file);
file.stat.mode = (file.stat.mode & !0o7777) | (mode & 0o7777);
true
}
FileBacking::DetachedDirectory { stat, .. } => {
stat.mode = (stat.mode & !0o7777) | (mode & 0o7777);
true
}
FileBacking::Path(_) | FileBacking::LinkedAlias { .. } => false,
}
}
pub fn detached_chown(&self, uid: u32, gid: u32, changed_mode: Option<u32>) -> bool {
let mut backing = lock_or_recover(&self.backing);
match &mut *backing {
FileBacking::Anonymous { file, .. } => {
let mut file = lock_or_recover(file);
file.stat.uid = uid;
file.stat.gid = gid;
if let Some(mode) = changed_mode {
file.stat.mode = mode;
}
true
}
FileBacking::DetachedDirectory { stat, .. } => {
stat.uid = uid;
stat.gid = gid;
true
}
FileBacking::Path(_) | FileBacking::LinkedAlias { .. } => false,
}
}
pub fn lock_target(&self) -> Option<FileLockTarget> {
self.lock_target
}
pub fn cursor(&self) -> u64 {
self.cursor.load(Ordering::SeqCst)
}
pub fn set_cursor(&self, cursor: u64) {
self.cursor.store(cursor, Ordering::SeqCst);
}
pub fn flags(&self) -> u32 {
self.flags.load(Ordering::SeqCst)
}
pub fn update_flags(&self, mask: u32, flags: u32) -> u32 {
let mut current = self.flags();
loop {
let next = (current & !mask) | (flags & mask);
match self
.flags
.compare_exchange(current, next, Ordering::SeqCst, Ordering::SeqCst)
{
Ok(_) => return next,
Err(observed) => current = observed,
}
}
}
pub fn ref_count(&self) -> usize {
self.ref_count.load(Ordering::SeqCst)
}
pub fn increment_ref_count(&self) -> usize {
self.ref_count.fetch_add(1, Ordering::SeqCst) + 1
}
pub fn decrement_ref_count(&self) -> usize {
let mut current = self.ref_count.load(Ordering::SeqCst);
loop {
let next = current.saturating_sub(1);
match self
.ref_count
.compare_exchange(current, next, Ordering::SeqCst, Ordering::SeqCst)
{
Ok(_) => return next,
Err(observed) => current = observed,
}
}
}
}
#[derive(Debug, Clone)]
pub struct FdEntry {
pub fd: u32,
pub description: SharedFileDescription,
pub status_flags: u32,
pub fd_flags: u32,
pub rights: u64,
pub filetype: u8,
}
#[derive(Debug)]
pub struct TransferredFd {
description: SharedFileDescription,
status_flags: u32,
rights: u64,
filetype: u8,
}
impl Clone for TransferredFd {
fn clone(&self) -> Self {
self.description.increment_ref_count();
Self {
description: Arc::clone(&self.description),
status_flags: self.status_flags,
rights: self.rights,
filetype: self.filetype,
}
}
}
impl PartialEq for TransferredFd {
fn eq(&self, other: &Self) -> bool {
self.description.id() == other.description.id()
&& self.status_flags == other.status_flags
&& self.rights == other.rights
&& self.filetype == other.filetype
}
}
impl Eq for TransferredFd {}
impl Drop for TransferredFd {
fn drop(&mut self) {
self.description.decrement_ref_count();
}
}
impl TransferredFd {
pub(crate) fn description(&self) -> SharedFileDescription {
Arc::clone(&self.description)
}
pub fn description_id(&self) -> u64 {
self.description.id()
}
pub fn status_flags(&self) -> u32 {
self.status_flags
}
pub fn rights(&self) -> u64 {
self.rights
}
pub fn filetype(&self) -> u8 {
self.filetype
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FdStat {
pub filetype: u8,
pub flags: u32,
pub rights: u64,
}
#[derive(Debug, Clone)]
pub struct StdioOverride {
pub description: SharedFileDescription,
pub filetype: u8,
}
#[derive(Debug, Clone)]
struct DescriptionFactory {}
impl DescriptionFactory {
fn new(_starting_id: u64) -> Self {
Self {}
}
fn allocate(&self, path: &str, flags: u32) -> SharedFileDescription {
self.allocate_with_lock(path, flags, None)
}
fn allocate_with_lock(
&self,
path: &str,
flags: u32,
lock_target: Option<FileLockTarget>,
) -> SharedFileDescription {
let next_id = allocate_file_description_id();
Arc::new(FileDescription::new_with_lock(
next_id,
path,
flags,
lock_target,
))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct FileLockTarget {
dev: u64,
ino: u64,
}
impl FileLockTarget {
pub const fn new(dev: u64, ino: u64) -> Self {
Self { dev, ino }
}
pub const fn dev(self) -> u64 {
self.dev
}
pub const fn ino(self) -> u64 {
self.ino
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FileLockMode {
Shared,
Exclusive,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RecordLockType {
Read,
Write,
Unlock,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RecordLock {
pub lock_type: RecordLockType,
pub start: u64,
pub end: Option<u64>,
pub pid: u32,
}
impl RecordLock {
pub fn new(lock_type: RecordLockType, start: u64, length: u64, pid: u32) -> FdResult<Self> {
let end =
if length == 0 {
None
} else {
Some(start.checked_add(length).ok_or_else(|| {
FdTableError::invalid_argument("record lock range exceeds u64")
})?)
};
Ok(Self {
lock_type,
start,
end,
pid,
})
}
pub fn length(self) -> u64 {
self.end.map_or(0, |end| end - self.start)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FlockOperation {
Shared { nonblocking: bool },
Exclusive { nonblocking: bool },
Unlock,
}
impl FlockOperation {
pub fn from_bits(operation: u32) -> FdResult<Self> {
let nonblocking = operation & LOCK_NB != 0;
match operation & !LOCK_NB {
LOCK_SH => Ok(Self::Shared { nonblocking }),
LOCK_EX => Ok(Self::Exclusive { nonblocking }),
LOCK_UN => Ok(Self::Unlock),
_ => Err(FdTableError::invalid_argument(format!(
"invalid flock operation {operation:#x}"
))),
}
}
}
#[derive(Debug, Clone)]
pub struct ProcessFdTable {
entries: BTreeMap<u32, FdEntry>,
next_fd: u32,
alloc_desc: DescriptionFactory,
max_fds: usize,
}
impl ProcessFdTable {
fn new(alloc_desc: DescriptionFactory, max_fds: usize) -> Self {
Self {
entries: BTreeMap::new(),
next_fd: 3,
alloc_desc,
max_fds,
}
}
pub fn max_fds(&self) -> usize {
self.max_fds
}
pub fn available_fd_capacity(&self) -> usize {
self.max_fds.saturating_sub(self.entries.len())
}
pub fn init_stdio(
&mut self,
stdin_desc: SharedFileDescription,
stdout_desc: SharedFileDescription,
stderr_desc: SharedFileDescription,
) {
self.entries.insert(
0,
FdEntry {
fd: 0,
description: stdin_desc,
status_flags: 0,
fd_flags: 0,
rights: 0,
filetype: FILETYPE_CHARACTER_DEVICE,
},
);
self.entries.insert(
1,
FdEntry {
fd: 1,
description: stdout_desc,
status_flags: 0,
fd_flags: 0,
rights: 0,
filetype: FILETYPE_CHARACTER_DEVICE,
},
);
self.entries.insert(
2,
FdEntry {
fd: 2,
description: stderr_desc,
status_flags: 0,
fd_flags: 0,
rights: 0,
filetype: FILETYPE_CHARACTER_DEVICE,
},
);
}
pub fn init_stdio_with_types(
&mut self,
stdin_desc: SharedFileDescription,
stdin_type: u8,
stdout_desc: SharedFileDescription,
stdout_type: u8,
stderr_desc: SharedFileDescription,
stderr_type: u8,
) {
stdin_desc.increment_ref_count();
stdout_desc.increment_ref_count();
stderr_desc.increment_ref_count();
self.entries.insert(
0,
FdEntry {
fd: 0,
description: stdin_desc,
status_flags: 0,
fd_flags: 0,
rights: 0,
filetype: stdin_type,
},
);
self.entries.insert(
1,
FdEntry {
fd: 1,
description: stdout_desc,
status_flags: 0,
fd_flags: 0,
rights: 0,
filetype: stdout_type,
},
);
self.entries.insert(
2,
FdEntry {
fd: 2,
description: stderr_desc,
status_flags: 0,
fd_flags: 0,
rights: 0,
filetype: stderr_type,
},
);
}
pub fn open(&mut self, path: &str, flags: u32) -> FdResult<u32> {
self.open_with_details(path, flags, FILETYPE_REGULAR_FILE, None)
}
pub fn open_with_filetype(&mut self, path: &str, flags: u32, filetype: u8) -> FdResult<u32> {
self.open_with_details(path, flags, filetype, None)
}
pub fn open_with_details(
&mut self,
path: &str,
flags: u32,
filetype: u8,
lock_target: Option<FileLockTarget>,
) -> FdResult<u32> {
let fd = self.allocate_fd()?;
let description =
self.alloc_desc
.allocate_with_lock(path, description_flags(flags), lock_target);
self.entries.insert(
fd,
FdEntry {
fd,
description,
status_flags: status_flags(flags),
fd_flags: 0,
rights: 0,
filetype,
},
);
Ok(fd)
}
pub fn open_with(
&mut self,
description: SharedFileDescription,
filetype: u8,
target_fd: Option<u32>,
) -> FdResult<u32> {
let entry_status_flags = status_flags(description.flags());
let fd = match target_fd {
Some(fd) => {
self.validate_fd_bounds(fd)?;
if self.entries.contains_key(&fd) {
self.close(fd);
}
fd
}
None => self.allocate_fd()?,
};
description.increment_ref_count();
self.entries.insert(
fd,
FdEntry {
fd,
description,
status_flags: entry_status_flags,
fd_flags: 0,
rights: 0,
filetype,
},
);
Ok(fd)
}
pub fn open_pair_with_details(
&mut self,
first_path: &str,
second_path: &str,
status_flags: u32,
fd_flags: u32,
filetype: u8,
) -> FdResult<(u32, u32, SharedFileDescription, SharedFileDescription)> {
if self.entries.len().saturating_add(2) > self.max_fds {
return Err(FdTableError::too_many_open_files());
}
let first_fd = self.allocate_fd()?;
let second_fd = match self.allocate_fd() {
Ok(fd) => fd,
Err(error) => {
self.next_fd = first_fd;
return Err(error);
}
};
let first = self.alloc_desc.allocate(first_path, O_RDWR);
let second = self.alloc_desc.allocate(second_path, O_RDWR);
for (fd, description) in [
(first_fd, Arc::clone(&first)),
(second_fd, Arc::clone(&second)),
] {
self.entries.insert(
fd,
FdEntry {
fd,
description,
status_flags: status_flags & ENTRY_STATUS_FLAG_MASK,
fd_flags: fd_flags & FD_CLOEXEC,
rights: 0,
filetype,
},
);
}
Ok((first_fd, second_fd, first, second))
}
pub fn transfer(&self, fd: u32) -> FdResult<TransferredFd> {
let entry = self
.entries
.get(&fd)
.ok_or_else(|| FdTableError::bad_file_descriptor(fd))?;
entry.description.increment_ref_count();
Ok(TransferredFd {
description: Arc::clone(&entry.description),
status_flags: entry.status_flags,
rights: entry.rights,
filetype: entry.filetype,
})
}
pub fn create_transfer(&self, path: &str, flags: u32, filetype: u8) -> TransferredFd {
TransferredFd {
description: self.alloc_desc.allocate(path, description_flags(flags)),
status_flags: status_flags(flags),
rights: 0,
filetype,
}
}
pub fn install_transferred(
&mut self,
transfers: &[TransferredFd],
close_on_exec: bool,
) -> FdResult<Vec<u32>> {
if self.entries.len().saturating_add(transfers.len()) > self.max_fds {
return Err(FdTableError::too_many_open_files());
}
let mut candidates = Vec::with_capacity(transfers.len());
let mut cursor = self.next_fd;
for _ in transfers {
let start = usize::try_from(cursor).unwrap_or(0) % self.max_fds;
let candidate = (0..self.max_fds)
.map(|offset| ((start + offset) % self.max_fds) as u32)
.find(|fd| !self.entries.contains_key(fd) && !candidates.contains(fd))
.ok_or_else(FdTableError::too_many_open_files)?;
candidates.push(candidate);
cursor = candidate.saturating_add(1);
}
for (fd, transfer) in candidates.iter().copied().zip(transfers) {
transfer.description.increment_ref_count();
self.entries.insert(
fd,
FdEntry {
fd,
description: Arc::clone(&transfer.description),
status_flags: transfer.status_flags,
fd_flags: if close_on_exec { FD_CLOEXEC } else { 0 },
rights: transfer.rights,
filetype: transfer.filetype,
},
);
}
self.next_fd = cursor;
Ok(candidates)
}
pub(crate) fn install_transferred_at(
&mut self,
transfer: &TransferredFd,
fd: u32,
fd_flags: u32,
) -> FdResult<()> {
self.validate_fd_bounds(fd)?;
if self.entries.contains_key(&fd) {
self.close(fd);
}
transfer.description.increment_ref_count();
self.entries.insert(
fd,
FdEntry {
fd,
description: Arc::clone(&transfer.description),
status_flags: transfer.status_flags,
fd_flags: fd_flags & FD_CLOEXEC,
rights: transfer.rights,
filetype: transfer.filetype,
},
);
Ok(())
}
pub fn get(&self, fd: u32) -> Option<&FdEntry> {
self.entries.get(&fd)
}
pub fn values(&self) -> Values<'_, u32, FdEntry> {
self.entries.values()
}
pub fn close(&mut self, fd: u32) -> bool {
let Some(entry) = self.entries.remove(&fd) else {
return false;
};
entry.description.decrement_ref_count();
true
}
pub fn close_on_exec_fds(&self) -> Vec<u32> {
self.entries
.iter()
.filter_map(|(fd, entry)| (entry.fd_flags & FD_CLOEXEC != 0).then_some(*fd))
.collect()
}
pub fn dup(&mut self, fd: u32) -> FdResult<u32> {
self.dup_with_status_flags(fd, None)
}
pub fn dup_with_status_flags(
&mut self,
fd: u32,
status_flags_override: Option<u32>,
) -> FdResult<u32> {
let entry = self
.entries
.get(&fd)
.cloned()
.ok_or_else(|| FdTableError::bad_file_descriptor(fd))?;
let new_fd = self.allocate_fd()?;
self.duplicate_entry(
&entry,
new_fd,
status_flags_override.unwrap_or(entry.status_flags),
0,
)
}
pub fn dup2(&mut self, old_fd: u32, new_fd: u32) -> FdResult<()> {
let entry = self
.entries
.get(&old_fd)
.cloned()
.ok_or_else(|| FdTableError::bad_file_descriptor(old_fd))?;
self.validate_fd_bounds(new_fd)?;
if old_fd == new_fd {
return Ok(());
}
if self.entries.contains_key(&new_fd) {
self.close(new_fd);
}
self.duplicate_entry(&entry, new_fd, entry.status_flags, 0)?;
Ok(())
}
pub fn stat(&self, fd: u32) -> FdResult<FdStat> {
let entry = self
.entries
.get(&fd)
.ok_or_else(|| FdTableError::bad_file_descriptor(fd))?;
Ok(FdStat {
filetype: entry.filetype,
flags: visible_fd_flags(entry.description.flags(), entry.status_flags),
rights: entry.rights,
})
}
pub fn fcntl(&mut self, fd: u32, command: u32, arg: u32) -> FdResult<u32> {
match command {
F_DUPFD => {
let entry = self
.entries
.get(&fd)
.cloned()
.ok_or_else(|| FdTableError::bad_file_descriptor(fd))?;
let min_fd = self.validate_fcntl_dup_min(arg)?;
let new_fd = self.allocate_fd_from(min_fd)?;
self.duplicate_entry(&entry, new_fd, entry.status_flags, 0)
}
F_GETFD => {
let entry = self
.entries
.get(&fd)
.ok_or_else(|| FdTableError::bad_file_descriptor(fd))?;
Ok(entry.fd_flags & FD_CLOEXEC)
}
F_SETFD => {
let entry = self
.entries
.get_mut(&fd)
.ok_or_else(|| FdTableError::bad_file_descriptor(fd))?;
entry.fd_flags = arg & FD_CLOEXEC;
Ok(0)
}
F_GETFL => {
let entry = self
.entries
.get(&fd)
.ok_or_else(|| FdTableError::bad_file_descriptor(fd))?;
Ok(visible_fd_flags(
entry.description.flags(),
entry.status_flags,
))
}
F_SETFL => {
let entry = self
.entries
.get_mut(&fd)
.ok_or_else(|| FdTableError::bad_file_descriptor(fd))?;
entry.status_flags = arg & ENTRY_STATUS_FLAG_MASK;
entry.description.update_flags(SHARED_STATUS_FLAG_MASK, arg);
Ok(0)
}
_ => Err(FdTableError::invalid_argument(format!(
"unsupported fcntl command {command}"
))),
}
}
pub fn fork(&self) -> Self {
self.fork_with_cloexec(false)
}
pub fn fork_preserving_cloexec(&self) -> Self {
self.fork_with_cloexec(true)
}
fn fork_with_cloexec(&self, preserve_cloexec: bool) -> Self {
let mut child = Self::new(self.alloc_desc.clone(), self.max_fds);
child.next_fd = self.next_fd;
for (fd, entry) in &self.entries {
if !preserve_cloexec && entry.fd_flags & FD_CLOEXEC != 0 {
continue;
}
entry.description.increment_ref_count();
child.entries.insert(
*fd,
FdEntry {
fd: *fd,
description: Arc::clone(&entry.description),
status_flags: entry.status_flags,
fd_flags: entry.fd_flags,
rights: entry.rights,
filetype: entry.filetype,
},
);
}
child
}
pub fn len_for_exec(&self) -> usize {
self.entries
.values()
.filter(|entry| entry.fd_flags & FD_CLOEXEC == 0)
.count()
}
pub fn close_all(&mut self) {
let fds: Vec<u32> = self.entries.keys().copied().collect();
for fd in fds {
self.close(fd);
}
}
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn iter(&self) -> Values<'_, u32, FdEntry> {
self.entries.values()
}
fn allocate_fd(&mut self) -> FdResult<u32> {
if self.entries.len() >= self.max_fds {
return Err(FdTableError::too_many_open_files());
}
let start = usize::try_from(self.next_fd).unwrap_or(0) % self.max_fds;
for offset in 0..self.max_fds {
let candidate = ((start + offset) % self.max_fds) as u32;
if !self.entries.contains_key(&candidate) {
self.next_fd = candidate.saturating_add(1);
return Ok(candidate);
}
}
Err(FdTableError::too_many_open_files())
}
fn allocate_fd_from(&mut self, min_fd: u32) -> FdResult<u32> {
if self.entries.len() >= self.max_fds {
return Err(FdTableError::too_many_open_files());
}
if min_fd as usize >= self.max_fds {
return Err(FdTableError::invalid_argument(format!(
"fd {min_fd} exceeds process fd limit"
)));
}
for candidate in min_fd..self.max_fds as u32 {
if !self.entries.contains_key(&candidate) {
self.next_fd = candidate.saturating_add(1);
return Ok(candidate);
}
}
Err(FdTableError::too_many_open_files())
}
fn duplicate_entry(
&mut self,
entry: &FdEntry,
new_fd: u32,
status_flags: u32,
fd_flags: u32,
) -> FdResult<u32> {
entry.description.increment_ref_count();
self.entries.insert(
new_fd,
FdEntry {
fd: new_fd,
description: Arc::clone(&entry.description),
status_flags,
fd_flags,
rights: entry.rights,
filetype: entry.filetype,
},
);
Ok(new_fd)
}
fn validate_fd_bounds(&self, fd: u32) -> FdResult<()> {
if fd as usize >= self.max_fds {
return Err(FdTableError::bad_file_descriptor(fd));
}
Ok(())
}
fn validate_fcntl_dup_min(&self, min_fd: u32) -> FdResult<u32> {
if min_fd as usize >= self.max_fds {
return Err(FdTableError::invalid_argument(format!(
"fd {min_fd} exceeds process fd limit"
)));
}
Ok(min_fd)
}
}
fn description_flags(flags: u32) -> u32 {
flags & !status_flags(flags)
}
fn status_flags(flags: u32) -> u32 {
flags & ENTRY_STATUS_FLAG_MASK
}
fn visible_fd_flags(description_flags: u32, entry_status_flags: u32) -> u32 {
(description_flags & (0b11 | SHARED_STATUS_FLAG_MASK))
| (entry_status_flags & ENTRY_STATUS_FLAG_MASK)
}
const SHARED_STATUS_FLAG_MASK: u32 = O_APPEND;
const ENTRY_STATUS_FLAG_MASK: u32 = O_NONBLOCK;
impl<'a> IntoIterator for &'a ProcessFdTable {
type Item = &'a FdEntry;
type IntoIter = Values<'a, u32, FdEntry>;
fn into_iter(self) -> Self::IntoIter {
self.entries.values()
}
}
#[derive(Debug, Clone)]
pub struct FdTableManager {
tables: BTreeMap<u32, ProcessFdTable>,
alloc_desc: DescriptionFactory,
max_fds: usize,
}
impl Default for FdTableManager {
fn default() -> Self {
Self {
tables: BTreeMap::new(),
alloc_desc: DescriptionFactory::new(1),
max_fds: MAX_FDS_PER_PROCESS,
}
}
}
impl FdTableManager {
pub fn new() -> Self {
Self::default()
}
pub fn with_max_fds(max_fds: usize) -> Self {
Self {
max_fds,
..Self::default()
}
}
pub fn create(&mut self, pid: u32) -> &mut ProcessFdTable {
let mut table = ProcessFdTable::new(self.alloc_desc.clone(), self.max_fds);
table.init_stdio(
self.alloc_desc.allocate("/dev/stdin", O_RDONLY),
self.alloc_desc.allocate("/dev/stdout", O_WRONLY),
self.alloc_desc.allocate("/dev/stderr", O_WRONLY),
);
self.remove(pid);
self.tables.insert(pid, table);
self.tables
.get_mut(&pid)
.expect("newly created FD table should be stored")
}
pub fn create_with_stdio(
&mut self,
pid: u32,
stdin_override: Option<StdioOverride>,
stdout_override: Option<StdioOverride>,
stderr_override: Option<StdioOverride>,
) -> &mut ProcessFdTable {
let mut table = ProcessFdTable::new(self.alloc_desc.clone(), self.max_fds);
let stdin_desc = stdin_override
.as_ref()
.map(|entry| Arc::clone(&entry.description))
.unwrap_or_else(|| self.alloc_desc.allocate("/dev/stdin", O_RDONLY));
let stdout_desc = stdout_override
.as_ref()
.map(|entry| Arc::clone(&entry.description))
.unwrap_or_else(|| self.alloc_desc.allocate("/dev/stdout", O_WRONLY));
let stderr_desc = stderr_override
.as_ref()
.map(|entry| Arc::clone(&entry.description))
.unwrap_or_else(|| self.alloc_desc.allocate("/dev/stderr", O_WRONLY));
table.init_stdio_with_types(
stdin_desc,
stdin_override
.as_ref()
.map(|entry| entry.filetype)
.unwrap_or(FILETYPE_CHARACTER_DEVICE),
stdout_desc,
stdout_override
.as_ref()
.map(|entry| entry.filetype)
.unwrap_or(FILETYPE_CHARACTER_DEVICE),
stderr_desc,
stderr_override
.as_ref()
.map(|entry| entry.filetype)
.unwrap_or(FILETYPE_CHARACTER_DEVICE),
);
self.remove(pid);
self.tables.insert(pid, table);
self.tables
.get_mut(&pid)
.expect("newly created FD table should be stored")
}
pub fn fork(&mut self, parent_pid: u32, child_pid: u32) -> &mut ProcessFdTable {
self.fork_with_cloexec(parent_pid, child_pid, false)
}
pub fn fork_preserving_cloexec(
&mut self,
parent_pid: u32,
child_pid: u32,
) -> &mut ProcessFdTable {
self.fork_with_cloexec(parent_pid, child_pid, true)
}
fn fork_with_cloexec(
&mut self,
parent_pid: u32,
child_pid: u32,
preserve_cloexec: bool,
) -> &mut ProcessFdTable {
if !self.tables.contains_key(&parent_pid) {
return self.create(child_pid);
}
let parent = self
.tables
.get(&parent_pid)
.expect("parent table presence was checked");
let child = if preserve_cloexec {
parent.fork_preserving_cloexec()
} else {
parent.fork()
};
self.remove(child_pid);
self.tables.insert(child_pid, child);
self.tables
.get_mut(&child_pid)
.expect("forked FD table should be stored")
}
pub fn get(&self, pid: u32) -> Option<&ProcessFdTable> {
self.tables.get(&pid)
}
pub fn get_mut(&mut self, pid: u32) -> Option<&mut ProcessFdTable> {
self.tables.get_mut(&pid)
}
pub fn has(&self, pid: u32) -> bool {
self.tables.contains_key(&pid)
}
pub fn len(&self) -> usize {
self.tables.len()
}
pub fn is_empty(&self) -> bool {
self.tables.is_empty()
}
pub fn total_open_fds(&self) -> usize {
self.tables.values().map(ProcessFdTable::len).sum()
}
pub fn pids(&self) -> Vec<u32> {
self.tables.keys().copied().collect()
}
pub fn remove(&mut self, pid: u32) {
if let Some(mut table) = self.tables.remove(&pid) {
table.close_all();
}
}
}
#[derive(Debug, Clone, Default)]
pub struct FileLockManager {
inner: Arc<FileLockManagerInner>,
}
#[derive(Debug)]
struct FileLockManagerInner {
state: Mutex<FileLockState>,
wake: Condvar,
max_record_locks: usize,
}
impl Default for FileLockManagerInner {
fn default() -> Self {
Self {
state: Mutex::new(FileLockState::default()),
wake: Condvar::new(),
max_record_locks: DEFAULT_MAX_RECORD_LOCKS,
}
}
}
#[derive(Debug, Default)]
struct FileLockState {
entries: BTreeMap<FileLockTarget, FileLockEntry>,
record_locks: Vec<(FileLockTarget, RecordLock)>,
record_lock_waits: BTreeMap<u32, RecordLockWait>,
warned_near_record_lock_limit: bool,
warned_near_record_lock_wait_limit: bool,
}
#[derive(Debug, Clone)]
struct RecordLockWait {
target: FileLockTarget,
request: RecordLock,
blockers: BTreeSet<u32>,
}
#[derive(Debug, Default)]
struct FileLockEntry {
shared: BTreeSet<u64>,
exclusive: Option<u64>,
}
impl FileLockManager {
pub fn new() -> Self {
Self::with_record_lock_limit(DEFAULT_MAX_RECORD_LOCKS)
}
pub fn with_record_lock_limit(max_record_locks: usize) -> Self {
Self {
inner: Arc::new(FileLockManagerInner {
state: Mutex::new(FileLockState::default()),
wake: Condvar::new(),
max_record_locks: max_record_locks.max(1),
}),
}
}
pub fn apply(
&self,
owner_id: u64,
target: FileLockTarget,
operation: FlockOperation,
) -> FdResult<()> {
match operation {
FlockOperation::Shared { nonblocking } => {
self.acquire(owner_id, target, FileLockMode::Shared, nonblocking)
}
FlockOperation::Exclusive { nonblocking } => {
self.acquire(owner_id, target, FileLockMode::Exclusive, nonblocking)
}
FlockOperation::Unlock => {
self.release_owner(owner_id);
Ok(())
}
}
}
pub fn release_owner(&self, owner_id: u64) -> bool {
let mut state = lock_or_recover(&self.inner.state);
let mut released = false;
state.entries.retain(|_, entry| {
let entry_changed = entry.shared.remove(&owner_id) || entry.exclusive == Some(owner_id);
if entry.exclusive == Some(owner_id) {
entry.exclusive = None;
}
released |= entry_changed;
!entry.is_empty()
});
drop(state);
if released {
self.inner.wake.notify_all();
}
released
}
pub fn query_record_lock(
&self,
target: FileLockTarget,
request: RecordLock,
) -> Option<RecordLock> {
let state = lock_or_recover(&self.inner.state);
state
.record_locks
.iter()
.filter_map(|(candidate_target, candidate)| {
(*candidate_target == target
&& candidate.pid != request.pid
&& record_locks_conflict(*candidate, request))
.then_some(*candidate)
})
.min_by_key(|candidate| (candidate.start, candidate.pid))
}
pub fn set_record_lock(&self, target: FileLockTarget, request: RecordLock) -> FdResult<()> {
self.set_record_lock_inner(target, request, false)
}
pub fn set_blocking_record_lock(
&self,
target: FileLockTarget,
request: RecordLock,
) -> FdResult<()> {
self.set_record_lock_inner(target, request, true)
}
fn set_record_lock_inner(
&self,
target: FileLockTarget,
request: RecordLock,
blocking: bool,
) -> FdResult<()> {
let mut state = lock_or_recover(&self.inner.state);
let blockers = if request.lock_type == RecordLockType::Unlock {
BTreeSet::new()
} else {
conflicting_record_lock_pids(&state.record_locks, target, request)
};
if !blockers.is_empty() {
if blocking {
self.register_record_lock_wait(&mut state, target, request, blockers)?;
} else {
state.record_lock_waits.remove(&request.pid);
}
return Err(FdTableError::would_block(
"POSIX record lock is held by another process",
));
}
state.record_lock_waits.remove(&request.pid);
let mut next = Vec::with_capacity(state.record_locks.len().saturating_add(2));
for (candidate_target, candidate) in state.record_locks.iter().copied() {
if candidate_target != target
|| candidate.pid != request.pid
|| !record_ranges_overlap(candidate, request)
{
next.push((candidate_target, candidate));
continue;
}
if candidate.start < request.start {
next.push((
target,
RecordLock {
end: Some(request.start),
..candidate
},
));
}
if let Some(request_end) = request.end {
if candidate
.end
.is_none_or(|candidate_end| candidate_end > request_end)
{
next.push((
target,
RecordLock {
start: request_end,
..candidate
},
));
}
}
}
if request.lock_type != RecordLockType::Unlock {
next.push((target, request));
}
coalesce_record_locks(&mut next);
let warning_threshold = (self.inner.max_record_locks.saturating_mul(4) / 5).max(1);
if !state.warned_near_record_lock_limit && next.len() >= warning_threshold {
state.warned_near_record_lock_limit = true;
eprintln!(
"[agentos] POSIX record lock usage {}/{} is near the limit derived from limits.resources.maxOpenFds; raise limits.resources.maxOpenFds if needed",
next.len(), self.inner.max_record_locks
);
}
if next.len() > self.inner.max_record_locks {
return Err(FdTableError {
code: "ENOLCK",
message: format!(
"POSIX record lock table limit ({}) derived from limits.resources.maxOpenFds reached; raise limits.resources.maxOpenFds if needed",
self.inner.max_record_locks
),
});
}
state.record_locks = next;
refresh_record_lock_waits(&mut state);
drop(state);
self.inner.wake.notify_all();
Ok(())
}
pub fn cancel_record_lock_wait(&self, pid: u32) -> bool {
let mut state = lock_or_recover(&self.inner.state);
state.record_lock_waits.remove(&pid).is_some()
}
pub fn release_process_target(&self, pid: u32, target: FileLockTarget) -> bool {
let mut state = lock_or_recover(&self.inner.state);
let previous = state.record_locks.len();
state
.record_locks
.retain(|(candidate_target, lock)| *candidate_target != target || lock.pid != pid);
let wait_cancelled = state.record_lock_waits.remove(&pid).is_some();
let released = state.record_locks.len() != previous;
if released || wait_cancelled {
refresh_record_lock_waits(&mut state);
}
drop(state);
if released || wait_cancelled {
self.inner.wake.notify_all();
}
released || wait_cancelled
}
pub fn release_process(&self, pid: u32) -> bool {
let mut state = lock_or_recover(&self.inner.state);
let previous = state.record_locks.len();
state.record_locks.retain(|(_, lock)| lock.pid != pid);
let wait_cancelled = state.record_lock_waits.remove(&pid).is_some();
let released = state.record_locks.len() != previous;
if released || wait_cancelled {
refresh_record_lock_waits(&mut state);
}
drop(state);
if released || wait_cancelled {
self.inner.wake.notify_all();
}
released || wait_cancelled
}
fn register_record_lock_wait(
&self,
state: &mut FileLockState,
target: FileLockTarget,
request: RecordLock,
blockers: BTreeSet<u32>,
) -> FdResult<()> {
let new_waiter = !state.record_lock_waits.contains_key(&request.pid);
let waiter_count = state.record_lock_waits.len() + usize::from(new_waiter);
if new_waiter && waiter_count > self.inner.max_record_locks {
return Err(FdTableError {
code: "ENOLCK",
message: format!(
"POSIX record lock waiter limit ({}) derived from limits.resources.maxOpenFds reached; raise limits.resources.maxOpenFds if needed",
self.inner.max_record_locks
),
});
}
let warning_threshold = (self.inner.max_record_locks.saturating_mul(4) / 5).max(1);
if !state.warned_near_record_lock_wait_limit && waiter_count >= warning_threshold {
state.warned_near_record_lock_wait_limit = true;
eprintln!(
"[agentos] POSIX record lock waiter usage {}/{} is near the limit derived from limits.resources.maxOpenFds; raise limits.resources.maxOpenFds if needed",
waiter_count,
self.inner.max_record_locks
);
}
state.record_lock_waits.insert(
request.pid,
RecordLockWait {
target,
request,
blockers,
},
);
if record_lock_wait_cycle(state, request.pid) {
state.record_lock_waits.remove(&request.pid);
return Err(FdTableError::deadlock(
"POSIX record lock wait would create a deadlock",
));
}
Ok(())
}
fn acquire(
&self,
owner_id: u64,
target: FileLockTarget,
mode: FileLockMode,
nonblocking: bool,
) -> FdResult<()> {
let mut state = lock_or_recover(&self.inner.state);
loop {
let entry = state.entries.entry(target).or_default();
if entry.can_grant(owner_id, mode) {
entry.grant(owner_id, mode);
return Ok(());
}
if nonblocking {
return Err(FdTableError::would_block(
"advisory file lock is unavailable",
));
}
state = wait_or_recover(&self.inner.wake, state);
}
}
}
fn record_ranges_overlap(left: RecordLock, right: RecordLock) -> bool {
left.end.is_none_or(|end| right.start < end) && right.end.is_none_or(|end| left.start < end)
}
fn record_locks_conflict(left: RecordLock, right: RecordLock) -> bool {
record_ranges_overlap(left, right)
&& (left.lock_type == RecordLockType::Write || right.lock_type == RecordLockType::Write)
}
fn conflicting_record_lock_pids(
locks: &[(FileLockTarget, RecordLock)],
target: FileLockTarget,
request: RecordLock,
) -> BTreeSet<u32> {
locks
.iter()
.filter_map(|(candidate_target, candidate)| {
(*candidate_target == target
&& candidate.pid != request.pid
&& record_locks_conflict(*candidate, request))
.then_some(candidate.pid)
})
.collect()
}
fn refresh_record_lock_waits(state: &mut FileLockState) {
let FileLockState {
record_locks,
record_lock_waits,
..
} = state;
for wait in record_lock_waits.values_mut() {
wait.blockers = conflicting_record_lock_pids(record_locks, wait.target, wait.request);
}
}
fn record_lock_wait_cycle(state: &FileLockState, requester: u32) -> bool {
let Some(wait) = state.record_lock_waits.get(&requester) else {
return false;
};
let mut pending = wait.blockers.iter().copied().collect::<Vec<_>>();
let mut visited = BTreeSet::new();
while let Some(pid) = pending.pop() {
if pid == requester {
return true;
}
if !visited.insert(pid) {
continue;
}
if let Some(wait) = state.record_lock_waits.get(&pid) {
pending.extend(wait.blockers.iter().copied());
}
}
false
}
fn coalesce_record_locks(locks: &mut Vec<(FileLockTarget, RecordLock)>) {
locks.sort_by_key(|(target, lock)| (*target, lock.pid, lock.lock_type as u8, lock.start));
let mut merged: Vec<(FileLockTarget, RecordLock)> = Vec::with_capacity(locks.len());
for (target, lock) in locks.drain(..) {
if let Some((previous_target, previous)) = merged.last_mut() {
let touches = previous.end.is_none_or(|end| lock.start <= end);
if *previous_target == target
&& previous.pid == lock.pid
&& previous.lock_type == lock.lock_type
&& touches
{
previous.end = match (previous.end, lock.end) {
(None, _) | (_, None) => None,
(Some(left), Some(right)) => Some(left.max(right)),
};
continue;
}
}
merged.push((target, lock));
}
*locks = merged;
}
impl FileLockEntry {
fn can_grant(&self, owner_id: u64, mode: FileLockMode) -> bool {
match mode {
FileLockMode::Shared => self.exclusive.is_none_or(|owner| owner == owner_id),
FileLockMode::Exclusive => {
self.exclusive.is_none_or(|owner| owner == owner_id)
&& self.shared.iter().all(|owner| *owner == owner_id)
}
}
}
fn grant(&mut self, owner_id: u64, mode: FileLockMode) {
match mode {
FileLockMode::Shared => {
self.exclusive = None;
self.shared.insert(owner_id);
}
FileLockMode::Exclusive => {
self.shared.retain(|owner| *owner != owner_id);
self.exclusive = Some(owner_id);
}
}
}
fn is_empty(&self) -> bool {
self.exclusive.is_none() && self.shared.is_empty()
}
}
fn lock_or_recover<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
fn wait_or_recover<'a, T>(condvar: &Condvar, guard: MutexGuard<'a, T>) -> MutexGuard<'a, T> {
condvar
.wait(guard)
.unwrap_or_else(|poisoned| poisoned.into_inner())
}