use std::{
fs::{self, File, OpenOptions, TryLockError},
io,
path::PathBuf,
time::{Duration, Instant},
};
use tracing::debug;
use crate::backend::NodeId;
const POLL_PERIOD: Duration = Duration::from_millis(20);
#[must_use]
pub fn lock_dir() -> PathBuf {
std::env::temp_dir().join("openlogi-locks")
}
#[derive(Debug)]
pub struct HostLock {
_file: File,
}
pub fn try_lock(name: &str) -> io::Result<Option<HostLock>> {
let dir = lock_dir();
fs::create_dir_all(&dir)?;
let path = dir.join(name);
let file = match OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)
{
Ok(file) => file,
Err(error) if error.kind() == io::ErrorKind::PermissionDenied => File::open(&path)?,
Err(error) => return Err(error),
};
match file.try_lock() {
Ok(()) => Ok(Some(HostLock { _file: file })),
Err(TryLockError::WouldBlock) => Ok(None),
Err(TryLockError::Error(error)) => Err(error),
}
}
pub async fn lock_within(name: &str, budget: Duration) -> io::Result<Option<HostLock>> {
let deadline = Instant::now() + budget;
loop {
match try_lock(name) {
Ok(Some(lock)) => return Ok(Some(lock)),
Ok(None) if Instant::now() < deadline => tokio::time::sleep(POLL_PERIOD).await,
Ok(None) => {
debug!(name, ?budget, "host lock still held past the budget");
return Ok(None);
}
Err(error) => {
debug!(name, %error, "host lock unusable");
return Err(error);
}
}
}
}
pub const RECEIVER_REGISTER_TIMEOUT: Duration = Duration::from_secs(8);
#[derive(Debug)]
pub struct ReceiverRegisterPhase {
_lock: Option<HostLock>,
}
pub async fn lock_receiver_registers(
node: &NodeId,
timeout: Duration,
) -> Option<ReceiverRegisterPhase> {
match lock_within(&node_lock_name(node), timeout).await {
Ok(Some(lock)) => Some(ReceiverRegisterPhase { _lock: Some(lock) }),
Ok(None) => {
debug!(
node = %node,
"another OpenLogi process still holds this receiver's register phase"
);
None
}
Err(_) => Some(ReceiverRegisterPhase { _lock: None }),
}
}
#[must_use]
pub fn node_lock_name(node: &NodeId) -> String {
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
for byte in node.to_string().bytes() {
hash ^= u64::from(byte);
hash = hash.wrapping_mul(0x0100_0000_01b3);
}
format!("node-{hash:016x}")
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use super::{lock_within, node_lock_name, try_lock};
use crate::backend::NodeId;
fn name(tag: &str) -> String {
format!("test-{}-{tag}", std::process::id())
}
#[test]
fn a_held_lock_excludes_a_second_taker_until_dropped() {
let name = name("exclusive");
let first = try_lock(&name)
.unwrap()
.expect("the first taker gets the lock");
assert!(
try_lock(&name).unwrap().is_none(),
"a held lock must not be handed out twice"
);
drop(first);
assert!(
try_lock(&name).unwrap().is_some(),
"dropping the lock releases it"
);
}
#[tokio::test]
async fn lock_within_waits_for_a_release_inside_its_budget() {
let name = name("wait");
let held = try_lock(&name)
.unwrap()
.expect("the first taker gets the lock");
let release = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(60)).await;
drop(held);
});
assert!(
lock_within(&name, Duration::from_secs(2))
.await
.unwrap()
.is_some(),
"a lock released inside the budget is taken"
);
release.await.unwrap();
}
#[tokio::test]
async fn lock_within_gives_up_after_its_budget() {
let name = name("give-up");
let _held = try_lock(&name)
.unwrap()
.expect("the first taker gets the lock");
assert!(
lock_within(&name, Duration::from_millis(50))
.await
.unwrap()
.is_none(),
"a lock held past the budget is not taken"
);
}
#[test]
fn node_lock_names_are_stable_and_distinct() {
let a = NodeId::from("RegistryEntryId(4295012345)".to_string());
let same = NodeId::from("RegistryEntryId(4295012345)".to_string());
let b = NodeId::from("RegistryEntryId(4295012346)".to_string());
assert_eq!(node_lock_name(&a), node_lock_name(&same));
assert_ne!(node_lock_name(&a), node_lock_name(&b));
}
}