mirror of
https://github.com/LibreQoE/LibreQoS.git
synced 2026-10-24 13:57:40 -05:00
Merge pull request #1114 from LibreQoE/chore/dupe_file_lock
Refactor Rust file locks into shared helper
This commit is contained in:
6 files changed
+1300
-570
No files matched your search
Generated
+1
-1
@@ -2332,7 +2332,7 @@ dependencies = [
|
||||
"clap",
|
||||
"lqos_bus",
|
||||
"lqos_config",
|
||||
"nix",
|
||||
"lqos_utils",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tokio",
|
||||
|
||||
@@ -7,8 +7,8 @@ edition = "2024"
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
anyhow.workspace = true
|
||||
nix.workspace = true
|
||||
lqos_config = { path = "../lqos_config" }
|
||||
lqos_bus = { path = "../lqos_bus" }
|
||||
lqos_utils = { path = "../lqos_utils" }
|
||||
clap = { workspace = true, features = ["derive"] }
|
||||
tokio.workspace = true
|
||||
@@ -1,339 +1,63 @@
|
||||
use anyhow::{Error, Result};
|
||||
use nix::{
|
||||
errno::Errno,
|
||||
libc::{getpid, mode_t},
|
||||
};
|
||||
use std::{
|
||||
ffi::CString,
|
||||
fs::{File, OpenOptions, hard_link, remove_file},
|
||||
io::{ErrorKind, Read, Write},
|
||||
os::fd::AsRawFd,
|
||||
path::{Path, PathBuf},
|
||||
sync::atomic::{AtomicU64, Ordering},
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
use anyhow::Result;
|
||||
use lqos_utils::process_lock::{ProcessFileLock, ProcessLockConfig};
|
||||
#[cfg(test)]
|
||||
use std::cell::RefCell;
|
||||
|
||||
const LOCK_PATH: &str = "/run/lqos/lqos_overrides.lock";
|
||||
const LOCK_DIR: &str = "/run/lqos";
|
||||
const LOCK_DIR_PERMS: &str = "/run/lqos";
|
||||
const LOCK_GUARD_PATH: &str = "/run/lqos/lqos_overrides.lock.guard";
|
||||
const STALE_LOCK_REPLACE_ATTEMPTS: usize = 3;
|
||||
const LOCK_TEMP_CREATE_ATTEMPTS: usize = 8;
|
||||
const LOCK_CONTENTION_CODE: &str = "LQOS_OVERRIDES_LOCKED";
|
||||
|
||||
static TEMP_LOCK_SEQUENCE: AtomicU64 = AtomicU64::new(0);
|
||||
#[cfg(test)]
|
||||
thread_local! {
|
||||
static TEST_LOCK_CONFIG: RefCell<Option<ProcessLockConfig>> = const { RefCell::new(None) };
|
||||
}
|
||||
|
||||
/// Cross-process lock used while mutating operator-owned override files.
|
||||
///
|
||||
/// Dropping this guard removes `/run/lqos/lqos_overrides.lock`.
|
||||
#[derive(Debug)]
|
||||
pub struct FileLock {
|
||||
lock_path: PathBuf,
|
||||
}
|
||||
|
||||
struct AcquisitionGuard {
|
||||
file: File,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
struct LockMetadata {
|
||||
pid: i32,
|
||||
process: Option<String>,
|
||||
operation: Option<String>,
|
||||
created_unix: Option<u64>,
|
||||
}
|
||||
|
||||
impl LockMetadata {
|
||||
fn current(operation: &str) -> Self {
|
||||
Self {
|
||||
pid: unsafe { getpid() },
|
||||
process: current_process_name(),
|
||||
operation: Some(sanitize_lock_field(operation)),
|
||||
created_unix: SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.ok()
|
||||
.map(|duration| duration.as_secs()),
|
||||
}
|
||||
}
|
||||
|
||||
fn parse(contents: &str) -> Result<Self> {
|
||||
let trimmed = contents.trim();
|
||||
if let Ok(pid) = trimmed.parse::<i32>() {
|
||||
return Ok(Self {
|
||||
pid,
|
||||
process: None,
|
||||
operation: None,
|
||||
created_unix: None,
|
||||
});
|
||||
}
|
||||
|
||||
let mut pid = None;
|
||||
let mut process = None;
|
||||
let mut operation = None;
|
||||
let mut created_unix = None;
|
||||
for line in contents.lines() {
|
||||
let Some((key, value)) = line.split_once('=') else {
|
||||
continue;
|
||||
};
|
||||
match key.trim() {
|
||||
"pid" => pid = Some(value.trim().parse::<i32>()?),
|
||||
"process" => process = Some(value.trim().to_string()),
|
||||
"operation" => operation = Some(value.trim().to_string()),
|
||||
"created_unix" => created_unix = Some(value.trim().parse::<u64>()?),
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
let Some(pid) = pid else {
|
||||
return Err(Error::msg(
|
||||
"The LibreQoS overrides lock file does not contain a process id.",
|
||||
));
|
||||
};
|
||||
|
||||
Ok(Self {
|
||||
pid,
|
||||
process,
|
||||
operation,
|
||||
created_unix,
|
||||
})
|
||||
}
|
||||
|
||||
fn serialize(&self) -> String {
|
||||
let process = self.process.as_deref().unwrap_or("unknown");
|
||||
let operation = self.operation.as_deref().unwrap_or("unknown");
|
||||
let created_unix = self
|
||||
.created_unix
|
||||
.map(|seconds| seconds.to_string())
|
||||
.unwrap_or_else(|| "unknown".to_string());
|
||||
format!(
|
||||
"pid={}\nprocess={}\noperation={}\ncreated_unix={}\n",
|
||||
self.pid, process, operation, created_unix
|
||||
)
|
||||
}
|
||||
|
||||
fn describe_holder(&self) -> String {
|
||||
let mut parts = vec![format!("pid={}", self.pid)];
|
||||
if let Some(process) = self.process.as_deref().filter(|value| !value.is_empty()) {
|
||||
parts.push(format!("process={process}"));
|
||||
}
|
||||
if let Some(operation) = self.operation.as_deref().filter(|value| !value.is_empty()) {
|
||||
parts.push(format!("operation={operation}"));
|
||||
}
|
||||
if let Some(created_unix) = self.created_unix {
|
||||
parts.push(format!("created_unix={created_unix}"));
|
||||
}
|
||||
parts.join(", ")
|
||||
}
|
||||
_lock: ProcessFileLock,
|
||||
}
|
||||
|
||||
impl FileLock {
|
||||
/// Acquires the lock and records the caller operation for diagnostics.
|
||||
pub fn new_for_operation(operation: &str) -> Result<Self> {
|
||||
Self::new_at(
|
||||
operation,
|
||||
Path::new(LOCK_PATH),
|
||||
Path::new(LOCK_DIR),
|
||||
Path::new(LOCK_DIR_PERMS),
|
||||
Path::new(LOCK_GUARD_PATH),
|
||||
)
|
||||
let config = lock_config(operation);
|
||||
Self::new_with_config(&config)
|
||||
}
|
||||
|
||||
fn new_at(
|
||||
operation: &str,
|
||||
lock_path: &Path,
|
||||
lock_dir: &Path,
|
||||
lock_dir_perms: &Path,
|
||||
guard_path: &Path,
|
||||
) -> Result<Self> {
|
||||
Self::check_directory_at(lock_dir, lock_dir_perms)?;
|
||||
let _guard = Self::acquire_guard(guard_path)?;
|
||||
for _ in 0..STALE_LOCK_REPLACE_ATTEMPTS {
|
||||
if Self::create_lock(lock_path, lock_dir, operation)? {
|
||||
return Ok(Self {
|
||||
lock_path: lock_path.to_path_buf(),
|
||||
});
|
||||
}
|
||||
|
||||
let metadata = match Self::read_lock_metadata(lock_path) {
|
||||
Ok(metadata) => metadata,
|
||||
Err(err) if Self::is_not_found_error(&err) => continue,
|
||||
Err(err) => {
|
||||
return Err(Error::msg(format!(
|
||||
"{LOCK_CONTENTION_CODE}: The LibreQoS overrides files are locked by another process (lock metadata unreadable: {err})."
|
||||
)));
|
||||
}
|
||||
};
|
||||
if Self::is_lock_valid(&metadata) {
|
||||
return Err(Error::msg(format!(
|
||||
"{LOCK_CONTENTION_CODE}: The LibreQoS overrides files are locked by another process ({}).",
|
||||
metadata.describe_holder()
|
||||
)));
|
||||
}
|
||||
|
||||
match remove_file(lock_path) {
|
||||
Ok(()) => {}
|
||||
Err(err) if err.kind() == ErrorKind::NotFound => {}
|
||||
Err(err) => return Err(err.into()),
|
||||
}
|
||||
}
|
||||
|
||||
Err(Error::msg(
|
||||
"Unable to acquire the LibreQoS overrides lock after replacing a stale lock.",
|
||||
))
|
||||
}
|
||||
|
||||
fn acquire_guard(guard_path: &Path) -> Result<AcquisitionGuard> {
|
||||
let file = OpenOptions::new()
|
||||
.read(true)
|
||||
.write(true)
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.open(guard_path)?;
|
||||
let unix_path = CString::new(guard_path.to_string_lossy().as_bytes())?;
|
||||
unsafe {
|
||||
nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(0o666));
|
||||
}
|
||||
let ret = unsafe { nix::libc::flock(file.as_raw_fd(), nix::libc::LOCK_EX) };
|
||||
if ret == 0 {
|
||||
Ok(AcquisitionGuard { file })
|
||||
} else {
|
||||
Err(Error::msg(format!(
|
||||
"Unable to acquire LibreQoS overrides lock guard: {}",
|
||||
Errno::last()
|
||||
)))
|
||||
}
|
||||
}
|
||||
|
||||
fn read_lock_metadata(lock_path: &Path) -> Result<LockMetadata> {
|
||||
let mut f = File::open(lock_path)?;
|
||||
let mut contents = String::new();
|
||||
f.read_to_string(&mut contents)?;
|
||||
LockMetadata::parse(&contents)
|
||||
}
|
||||
|
||||
fn is_not_found_error(err: &Error) -> bool {
|
||||
err.downcast_ref::<std::io::Error>()
|
||||
.is_some_and(|io_err| io_err.kind() == ErrorKind::NotFound)
|
||||
}
|
||||
|
||||
fn is_lock_valid(metadata: &LockMetadata) -> bool {
|
||||
let ret = unsafe { nix::libc::kill(metadata.pid, 0) };
|
||||
if ret == 0 {
|
||||
return true;
|
||||
}
|
||||
let err = Errno::last();
|
||||
err != Errno::ESRCH
|
||||
}
|
||||
|
||||
fn create_lock(lock_path: &Path, lock_dir: &Path, operation: &str) -> Result<bool> {
|
||||
let metadata = LockMetadata::current(operation);
|
||||
let (mut f, temp_path) = Self::create_temp_lock_file(lock_dir)?;
|
||||
f.write_all(metadata.serialize().as_bytes())?;
|
||||
f.sync_all()?;
|
||||
drop(f);
|
||||
let link_result = hard_link(&temp_path, lock_path);
|
||||
let _ = remove_file(&temp_path);
|
||||
match link_result {
|
||||
Ok(()) => {}
|
||||
Err(err) if err.kind() == ErrorKind::AlreadyExists => return Ok(false),
|
||||
Err(err) => return Err(err.into()),
|
||||
}
|
||||
let unix_path = CString::new(lock_path.to_string_lossy().as_bytes())?;
|
||||
unsafe {
|
||||
nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(0o666));
|
||||
}
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
fn create_temp_lock_file(lock_dir: &Path) -> Result<(File, PathBuf)> {
|
||||
let nanos = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.ok()
|
||||
.map(|duration| duration.as_nanos())
|
||||
.unwrap_or_default();
|
||||
for _ in 0..LOCK_TEMP_CREATE_ATTEMPTS {
|
||||
let sequence = TEMP_LOCK_SEQUENCE.fetch_add(1, Ordering::Relaxed);
|
||||
let temp_path = lock_dir.join(format!(
|
||||
".lqos_overrides.lock.{}.{}.{}.tmp",
|
||||
std::process::id(),
|
||||
nanos,
|
||||
sequence
|
||||
));
|
||||
match OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(&temp_path)
|
||||
{
|
||||
Ok(file) => return Ok((file, temp_path)),
|
||||
Err(err) if err.kind() == ErrorKind::AlreadyExists => continue,
|
||||
Err(err) => return Err(err.into()),
|
||||
}
|
||||
}
|
||||
|
||||
Err(Error::msg(format!(
|
||||
"Unable to create a unique temporary LibreQoS overrides lock file after {LOCK_TEMP_CREATE_ATTEMPTS} attempts."
|
||||
)))
|
||||
}
|
||||
|
||||
fn check_directory_at(lock_dir: &Path, lock_dir_perms: &Path) -> Result<()> {
|
||||
if lock_dir.exists() && lock_dir.is_dir() {
|
||||
Ok(())
|
||||
} else {
|
||||
std::fs::create_dir(lock_dir)?;
|
||||
let unix_path = CString::new(lock_dir_perms.to_string_lossy().as_bytes())?;
|
||||
unsafe {
|
||||
nix::libc::chmod(unix_path.as_ptr(), 0o777);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn new_with_config(config: &ProcessLockConfig) -> Result<Self> {
|
||||
let lock = ProcessFileLock::acquire(config)?;
|
||||
Ok(Self { _lock: lock })
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for FileLock {
|
||||
fn drop(&mut self) {
|
||||
let _ = remove_file(&self.lock_path);
|
||||
fn lock_config(operation: &str) -> ProcessLockConfig {
|
||||
#[cfg(test)]
|
||||
if let Some(config) = TEST_LOCK_CONFIG.with(|config| config.borrow().clone()) {
|
||||
return config;
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for AcquisitionGuard {
|
||||
fn drop(&mut self) {
|
||||
let _ = unsafe { nix::libc::flock(self.file.as_raw_fd(), nix::libc::LOCK_UN) };
|
||||
}
|
||||
}
|
||||
|
||||
fn current_process_name() -> Option<String> {
|
||||
std::fs::read_to_string("/proc/self/comm")
|
||||
.ok()
|
||||
.and_then(|name| {
|
||||
let trimmed = sanitize_lock_field(name.trim());
|
||||
if trimmed.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(trimmed)
|
||||
}
|
||||
})
|
||||
.or_else(|| {
|
||||
std::env::args().next().and_then(|arg| {
|
||||
let trimmed = sanitize_lock_field(&arg);
|
||||
if trimmed.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(trimmed)
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
fn sanitize_lock_field(value: &str) -> String {
|
||||
value
|
||||
.chars()
|
||||
.filter(|character| *character != '\n' && *character != '\r')
|
||||
.collect()
|
||||
ProcessLockConfig::new(
|
||||
LOCK_PATH,
|
||||
LOCK_DIR,
|
||||
LOCK_GUARD_PATH,
|
||||
operation,
|
||||
LOCK_CONTENTION_CODE,
|
||||
"The LibreQoS overrides file lock",
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{FileLock, LockMetadata};
|
||||
use super::{
|
||||
FileLock, LOCK_CONTENTION_CODE, LOCK_DIR, LOCK_GUARD_PATH, LOCK_PATH, lock_config,
|
||||
};
|
||||
use lqos_utils::process_lock::ProcessLockConfig;
|
||||
use std::{
|
||||
fs::{create_dir_all, read_to_string, remove_dir_all, write},
|
||||
fs::{read_to_string, remove_dir_all},
|
||||
path::PathBuf,
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
@@ -344,209 +68,74 @@ mod tests {
|
||||
.expect("system clock before UNIX_EPOCH")
|
||||
.as_nanos();
|
||||
std::env::temp_dir().join(format!(
|
||||
"lqos-overrides-lock-test-{}-{nanos}",
|
||||
"libreqos-overrides-lock-test-{}-{nanos}",
|
||||
std::process::id()
|
||||
))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_legacy_pid_only_lock_file() {
|
||||
let metadata = LockMetadata::parse("12345\n").expect("legacy pid lock should parse");
|
||||
|
||||
assert_eq!(metadata.pid, 12345);
|
||||
assert_eq!(metadata.process, None);
|
||||
assert_eq!(metadata.operation, None);
|
||||
assert_eq!(metadata.created_unix, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_metadata_lock_file() {
|
||||
let metadata = LockMetadata::parse(
|
||||
"pid=12345\nprocess=lqosd\noperation=load effective overrides\ncreated_unix=1800000000\n",
|
||||
fn test_config_for(dir: &std::path::Path, operation: &str) -> ProcessLockConfig {
|
||||
ProcessLockConfig::new(
|
||||
dir.join("lqos_overrides.lock"),
|
||||
dir,
|
||||
dir.join("lqos_overrides.lock.guard"),
|
||||
operation,
|
||||
LOCK_CONTENTION_CODE,
|
||||
"The LibreQoS overrides file lock",
|
||||
)
|
||||
.expect("metadata lock should parse");
|
||||
}
|
||||
|
||||
assert_eq!(metadata.pid, 12345);
|
||||
assert_eq!(metadata.process.as_deref(), Some("lqosd"));
|
||||
assert_eq!(
|
||||
metadata.operation.as_deref(),
|
||||
Some("load effective overrides")
|
||||
);
|
||||
assert_eq!(metadata.created_unix, Some(1_800_000_000));
|
||||
struct TestLockConfigGuard;
|
||||
|
||||
impl TestLockConfigGuard {
|
||||
fn set(config: ProcessLockConfig) -> Self {
|
||||
super::TEST_LOCK_CONFIG.with(|stored| {
|
||||
*stored.borrow_mut() = Some(config);
|
||||
});
|
||||
Self
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TestLockConfigGuard {
|
||||
fn drop(&mut self) {
|
||||
super::TEST_LOCK_CONFIG.with(|stored| {
|
||||
*stored.borrow_mut() = None;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn holder_description_includes_available_metadata() {
|
||||
let metadata = LockMetadata::parse(
|
||||
"pid=12345\nprocess=node_manager\noperation=save overrides\ncreated_unix=1800000000\n",
|
||||
)
|
||||
.expect("metadata lock should parse");
|
||||
fn lock_config_preserves_overrides_contract() {
|
||||
let config = lock_config("save operator overrides");
|
||||
|
||||
assert_eq!(
|
||||
metadata.describe_holder(),
|
||||
"pid=12345, process=node_manager, operation=save overrides, created_unix=1800000000"
|
||||
);
|
||||
assert_eq!(config.lock_path.to_string_lossy(), LOCK_PATH);
|
||||
assert_eq!(config.lock_dir.to_string_lossy(), LOCK_DIR);
|
||||
assert_eq!(config.guard_path.to_string_lossy(), LOCK_GUARD_PATH);
|
||||
assert_eq!(config.operation, "save operator overrides");
|
||||
assert_eq!(config.contention_code, LOCK_CONTENTION_CODE);
|
||||
assert_eq!(config.resource_name, "The LibreQoS overrides file lock");
|
||||
assert_eq!(config.valid_process_name_contains, None);
|
||||
assert!(!config.write_pid_only);
|
||||
assert!(config.metadata_errors_are_contention);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lock_file_records_operation_metadata_and_cleans_up_on_drop() {
|
||||
fn file_lock_new_records_structured_operation_metadata() {
|
||||
let dir = unique_test_dir();
|
||||
create_dir_all(&dir).expect("failed to create temp test dir");
|
||||
let path = dir.join("lqos_overrides.lock");
|
||||
let _config_guard =
|
||||
TestLockConfigGuard::set(test_config_for(&dir, "save operator overrides"));
|
||||
|
||||
{
|
||||
let _lock = FileLock::new_at(
|
||||
"load effective overrides",
|
||||
&path,
|
||||
&dir,
|
||||
&dir,
|
||||
&dir.join("lqos_overrides.lock.guard"),
|
||||
)
|
||||
.expect("failed to create lock in temp dir");
|
||||
let contents = read_to_string(&path).expect("failed to read temp lock");
|
||||
let _lock = FileLock::new_for_operation("save operator overrides")
|
||||
.expect("failed to create overrides temp lock");
|
||||
let contents = read_to_string(&path).expect("failed to read overrides temp lock");
|
||||
|
||||
assert!(contents.contains(&format!("pid={}", std::process::id())));
|
||||
assert!(contents.contains("operation=load effective overrides"));
|
||||
assert!(contents.contains("operation=save operator overrides"));
|
||||
assert!(contents.contains("created_unix="));
|
||||
}
|
||||
|
||||
assert!(!path.exists());
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn live_lock_error_reports_existing_holder_metadata() {
|
||||
let dir = unique_test_dir();
|
||||
create_dir_all(&dir).expect("failed to create temp test dir");
|
||||
let path = dir.join("lqos_overrides.lock");
|
||||
write(
|
||||
&path,
|
||||
format!(
|
||||
"pid={}\nprocess=test-holder\noperation=save overrides\ncreated_unix=1800000000\n",
|
||||
std::process::id()
|
||||
),
|
||||
)
|
||||
.expect("failed to write temp lock");
|
||||
|
||||
let error = FileLock::new_at(
|
||||
"load effective overrides",
|
||||
&path,
|
||||
&dir,
|
||||
&dir,
|
||||
&dir.join("lqos_overrides.lock.guard"),
|
||||
)
|
||||
.expect_err("live lock should reject a second holder");
|
||||
let message = error.to_string();
|
||||
|
||||
assert!(message.contains("pid="));
|
||||
assert!(message.contains("process=test-holder"));
|
||||
assert!(message.contains("operation=save overrides"));
|
||||
assert!(message.contains("created_unix=1800000000"));
|
||||
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn malformed_lock_file_is_reported_as_lock_contention() {
|
||||
let dir = unique_test_dir();
|
||||
create_dir_all(&dir).expect("failed to create temp test dir");
|
||||
let path = dir.join("lqos_overrides.lock");
|
||||
write(&path, "pid=").expect("failed to write malformed temp lock");
|
||||
|
||||
let error = FileLock::new_at(
|
||||
"load effective overrides",
|
||||
&path,
|
||||
&dir,
|
||||
&dir,
|
||||
&dir.join("lqos_overrides.lock.guard"),
|
||||
)
|
||||
.expect_err("malformed lock metadata should be reported as contention");
|
||||
let message = error.to_string();
|
||||
|
||||
assert!(message.contains("locked by another process"));
|
||||
assert!(message.contains("lock metadata unreadable"));
|
||||
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncated_metadata_lock_missing_pid_is_reported_as_contention() {
|
||||
let dir = unique_test_dir();
|
||||
create_dir_all(&dir).expect("failed to create temp test dir");
|
||||
let path = dir.join("lqos_overrides.lock");
|
||||
write(
|
||||
&path,
|
||||
"process=test-holder\noperation=save overrides\ncreated_unix=1800000000\n",
|
||||
)
|
||||
.expect("failed to write truncated temp lock");
|
||||
|
||||
let error = FileLock::new_at(
|
||||
"load effective overrides",
|
||||
&path,
|
||||
&dir,
|
||||
&dir,
|
||||
&dir.join("lqos_overrides.lock.guard"),
|
||||
)
|
||||
.expect_err("truncated lock metadata should be reported as contention");
|
||||
let message = error.to_string();
|
||||
|
||||
assert!(message.contains("locked by another process"));
|
||||
assert!(message.contains("lock metadata unreadable"));
|
||||
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stale_legacy_lock_is_replaced_with_metadata_lock() {
|
||||
let dir = unique_test_dir();
|
||||
create_dir_all(&dir).expect("failed to create temp test dir");
|
||||
let path = dir.join("lqos_overrides.lock");
|
||||
write(&path, "2147483647\n").expect("failed to write stale temp lock");
|
||||
|
||||
{
|
||||
let _lock = FileLock::new_at(
|
||||
"load effective overrides",
|
||||
&path,
|
||||
&dir,
|
||||
&dir,
|
||||
&dir.join("lqos_overrides.lock.guard"),
|
||||
)
|
||||
.expect("stale legacy lock should be replaceable");
|
||||
let contents = read_to_string(&path).expect("failed to read temp lock");
|
||||
|
||||
assert!(contents.contains(&format!("pid={}", std::process::id())));
|
||||
assert!(contents.contains("operation=load effective overrides"));
|
||||
}
|
||||
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stale_structured_lock_is_replaced_with_metadata_lock() {
|
||||
let dir = unique_test_dir();
|
||||
create_dir_all(&dir).expect("failed to create temp test dir");
|
||||
let path = dir.join("lqos_overrides.lock");
|
||||
write(
|
||||
&path,
|
||||
"pid=2147483647\nprocess=old-holder\noperation=save overrides\ncreated_unix=1800000000\n",
|
||||
)
|
||||
.expect("failed to write stale structured temp lock");
|
||||
|
||||
{
|
||||
let _lock = FileLock::new_at(
|
||||
"load effective overrides",
|
||||
&path,
|
||||
&dir,
|
||||
&dir,
|
||||
&dir.join("lqos_overrides.lock.guard"),
|
||||
)
|
||||
.expect("stale structured lock should be replaceable");
|
||||
let contents = read_to_string(&path).expect("failed to read temp lock");
|
||||
|
||||
assert!(contents.contains(&format!("pid={}", std::process::id())));
|
||||
assert!(contents.contains("operation=load effective overrides"));
|
||||
}
|
||||
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
}
|
||||
@@ -18,6 +18,8 @@ pub mod hex_string;
|
||||
|
||||
/// Utilities for scaling bits and packets to human-readable format
|
||||
pub mod packet_scale;
|
||||
/// Cross-process PID-file locking helpers.
|
||||
pub mod process_lock;
|
||||
mod string_table_enum;
|
||||
|
||||
/// Rolling heatmap data storage for executive summary views.
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
+156
-79
@@ -1,93 +1,170 @@
|
||||
use anyhow::{Error, Result};
|
||||
use nix::libc::{getpid, mode_t};
|
||||
use std::{
|
||||
ffi::CString,
|
||||
fs::{File, remove_file},
|
||||
io::{Read, Write},
|
||||
path::Path,
|
||||
};
|
||||
use sysinfo::System;
|
||||
use anyhow::Result;
|
||||
use lqos_utils::process_lock::{ProcessFileLock, ProcessLockConfig};
|
||||
#[cfg(test)]
|
||||
use std::cell::RefCell;
|
||||
|
||||
const LOCK_PATH: &str = "/run/lqos/lqosd.lock";
|
||||
const LOCK_DIR: &str = "/run/lqos";
|
||||
const LOCK_DIR_PERMS: &str = "/run/lqos";
|
||||
const LOCK_GUARD_PATH: &str = "/run/lqos/lqosd.lock.guard";
|
||||
|
||||
pub struct FileLock {}
|
||||
#[cfg(test)]
|
||||
thread_local! {
|
||||
static TEST_LOCK_CONFIG: RefCell<Option<ProcessLockConfig>> = const { RefCell::new(None) };
|
||||
}
|
||||
|
||||
/// Process lock used to prevent multiple `lqosd` instances from running.
|
||||
///
|
||||
/// Dropping this guard removes `/run/lqos/lqosd.lock`.
|
||||
#[derive(Debug)]
|
||||
pub struct FileLock {
|
||||
_lock: ProcessFileLock,
|
||||
}
|
||||
|
||||
impl FileLock {
|
||||
/// Acquires the `lqosd` process lock.
|
||||
///
|
||||
/// A stale lock is replaced unless the recorded PID still belongs to a
|
||||
/// process whose command name contains `lqosd`.
|
||||
pub fn new() -> Result<Self> {
|
||||
Self::check_directory()?;
|
||||
let lock_path = Path::new(LOCK_PATH);
|
||||
if lock_path.exists() {
|
||||
if Self::is_lock_valid()? {
|
||||
return Err(Error::msg("lqosd is already running"));
|
||||
}
|
||||
|
||||
// It's a stale pid, so we need to replace it
|
||||
Self::create_lock()?;
|
||||
|
||||
Ok(Self {})
|
||||
} else {
|
||||
Self::create_lock()?;
|
||||
Ok(Self {})
|
||||
}
|
||||
let config = lock_config();
|
||||
Self::new_with_config(&config)
|
||||
}
|
||||
|
||||
fn is_lock_valid() -> Result<bool> {
|
||||
let mut f = File::open(LOCK_PATH)?;
|
||||
let mut contents = String::new();
|
||||
f.read_to_string(&mut contents)?;
|
||||
let pid: i32 = contents.parse()?;
|
||||
|
||||
let sys = System::new_all();
|
||||
let pid = sysinfo::Pid::from(pid as usize);
|
||||
if let Some(process) = sys.processes().get(&pid)
|
||||
&& process
|
||||
.name()
|
||||
.to_str()
|
||||
.unwrap_or_default()
|
||||
.contains("lqosd")
|
||||
{
|
||||
return Ok(true);
|
||||
}
|
||||
Ok(false)
|
||||
}
|
||||
|
||||
fn create_lock() -> Result<()> {
|
||||
let pid = unsafe { getpid() };
|
||||
let pid_format = format!("{pid}");
|
||||
{
|
||||
let mut f = File::create(LOCK_PATH)?;
|
||||
f.write_all(pid_format.as_bytes())?;
|
||||
}
|
||||
let unix_path = CString::new(LOCK_PATH)?;
|
||||
unsafe {
|
||||
nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(666));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn check_directory() -> Result<()> {
|
||||
let dir_path = std::path::Path::new(LOCK_DIR);
|
||||
if dir_path.exists() && dir_path.is_dir() {
|
||||
Ok(())
|
||||
} else {
|
||||
std::fs::create_dir(dir_path)?;
|
||||
let unix_path = CString::new(LOCK_DIR_PERMS)?;
|
||||
unsafe {
|
||||
nix::libc::chmod(unix_path.as_ptr(), 777);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub fn remove_lock() {
|
||||
let _ = remove_file(LOCK_PATH); // Ignore result
|
||||
fn new_with_config(config: &ProcessLockConfig) -> Result<Self> {
|
||||
let lock = ProcessFileLock::acquire(config)?;
|
||||
Ok(Self { _lock: lock })
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for FileLock {
|
||||
fn drop(&mut self) {
|
||||
Self::remove_lock();
|
||||
fn lock_config() -> ProcessLockConfig {
|
||||
#[cfg(test)]
|
||||
if let Some(config) = TEST_LOCK_CONFIG.with(|config| config.borrow().clone()) {
|
||||
return config;
|
||||
}
|
||||
|
||||
ProcessLockConfig::new(
|
||||
LOCK_PATH,
|
||||
LOCK_DIR,
|
||||
LOCK_GUARD_PATH,
|
||||
"start lqosd",
|
||||
"LQOSD_LOCKED",
|
||||
"lqosd",
|
||||
)
|
||||
.with_valid_process_name_contains("lqosd")
|
||||
.with_pid_only_lock_file()
|
||||
.with_strict_metadata_errors()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{FileLock, LOCK_DIR, LOCK_GUARD_PATH, LOCK_PATH, lock_config};
|
||||
use lqos_utils::process_lock::ProcessLockConfig;
|
||||
use std::{
|
||||
fs::{create_dir_all, read_to_string, remove_dir_all, write},
|
||||
path::PathBuf,
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
fn unique_test_dir() -> PathBuf {
|
||||
let nanos = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.expect("system clock before UNIX_EPOCH")
|
||||
.as_nanos();
|
||||
std::env::temp_dir().join(format!(
|
||||
"libreqos-lqosd-lock-test-{}-{nanos}",
|
||||
std::process::id()
|
||||
))
|
||||
}
|
||||
|
||||
fn test_config_for(dir: &std::path::Path) -> ProcessLockConfig {
|
||||
ProcessLockConfig::new(
|
||||
dir.join("lqosd.lock"),
|
||||
dir,
|
||||
dir.join("lqosd.lock.guard"),
|
||||
"start lqosd",
|
||||
"LQOSD_LOCKED",
|
||||
"lqosd",
|
||||
)
|
||||
.with_valid_process_name_contains("lqosd")
|
||||
.with_pid_only_lock_file()
|
||||
.with_strict_metadata_errors()
|
||||
}
|
||||
|
||||
struct TestLockConfigGuard;
|
||||
|
||||
impl TestLockConfigGuard {
|
||||
fn set(config: ProcessLockConfig) -> Self {
|
||||
super::TEST_LOCK_CONFIG.with(|stored| {
|
||||
*stored.borrow_mut() = Some(config);
|
||||
});
|
||||
Self
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TestLockConfigGuard {
|
||||
fn drop(&mut self) {
|
||||
super::TEST_LOCK_CONFIG.with(|stored| {
|
||||
*stored.borrow_mut() = None;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lock_config_preserves_lqosd_contract() {
|
||||
let config = lock_config();
|
||||
|
||||
assert_eq!(config.lock_path.to_string_lossy(), LOCK_PATH);
|
||||
assert_eq!(config.lock_dir.to_string_lossy(), LOCK_DIR);
|
||||
assert_eq!(config.guard_path.to_string_lossy(), LOCK_GUARD_PATH);
|
||||
assert_eq!(config.operation, "start lqosd");
|
||||
assert_eq!(config.contention_code, "LQOSD_LOCKED");
|
||||
assert_eq!(config.resource_name, "lqosd");
|
||||
assert_eq!(config.valid_process_name_contains.as_deref(), Some("lqosd"));
|
||||
assert!(config.write_pid_only);
|
||||
assert!(!config.metadata_errors_are_contention);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn file_lock_new_preserves_pid_only_lock_format() {
|
||||
let dir = unique_test_dir();
|
||||
let path = dir.join("lqosd.lock");
|
||||
let _config_guard = TestLockConfigGuard::set(test_config_for(&dir));
|
||||
|
||||
{
|
||||
let _lock = FileLock::new().expect("failed to create lqosd temp lock");
|
||||
let contents = read_to_string(&path).expect("failed to read lqosd temp lock");
|
||||
|
||||
assert_eq!(contents, format!("{}\n", std::process::id()));
|
||||
}
|
||||
|
||||
assert!(!path.exists());
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn file_lock_new_preserves_strict_metadata_errors() {
|
||||
let dir = unique_test_dir();
|
||||
create_dir_all(&dir).expect("failed to create temp test dir");
|
||||
let _config_guard = TestLockConfigGuard::set(test_config_for(&dir));
|
||||
write(dir.join("lqosd.lock"), "pid=").expect("failed to write malformed temp lock");
|
||||
|
||||
let error = FileLock::new().expect_err("malformed lqosd lock should remain a strict error");
|
||||
|
||||
assert!(error.to_string().contains("Invalid process lock metadata"));
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn file_lock_new_reports_live_lqosd_holder() {
|
||||
let dir = unique_test_dir();
|
||||
create_dir_all(&dir).expect("failed to create temp test dir");
|
||||
let _config_guard = TestLockConfigGuard::set(test_config_for(&dir));
|
||||
write(dir.join("lqosd.lock"), format!("{}\n", std::process::id()))
|
||||
.expect("failed to write live-holder temp lock");
|
||||
|
||||
let error = FileLock::new().expect_err("live lqosd lock should reject a second holder");
|
||||
|
||||
assert!(error.to_string().contains("LQOSD_LOCKED"));
|
||||
remove_dir_all(&dir).expect("failed to clean up temp test dir");
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user