//! repo locks are the innermost locks: take them in ascending `repo_key` order,
//! and never hold one next to an output-root publication lock.
use std::cell::RefCell;
use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use anyhow::Context as _;
/// never unlink it, a new file is a new inode that old holders don't exclude.
const LOCK_FILE: &str = ".build.lock";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LockMode {
Exclusive,
Lifetime,
}
impl LockMode {
pub const fn describe(self) -> &'static str {
match self {
Self::Exclusive => "exclusive (source mutation)",
Self::Lifetime => "lifetime (source reader)",
}
}
}
#[derive(Debug)]
pub(crate) struct RepoLock {
key: String,
path: PathBuf,
mode: LockMode,
_file: File,
}
impl RepoLock {
pub fn key(&self) -> &str {
&self.key
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn mode(&self) -> LockMode {
self.mode
}
}
impl Drop for RepoLock {
fn drop(&mut self) {
forget_held(&self.key);
}
}
pub fn repo_dir(output_root: &Path, key: &str) -> PathBuf {
output_root.join("repos").join(key)
}
pub fn lock_path(output_root: &Path, key: &str) -> PathBuf {
repo_dir(output_root, key).join(LOCK_FILE)
}
fn open_lock_file(output_root: &Path, key: &str) -> anyhow::Result<(PathBuf, File)> {
let path = lock_path(output_root, key);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("create repo directory {}", parent.display()))?;
}
let file = OpenOptions::new()
.create(true)
.read(true)
.write(true)
.open(&path)
.with_context(|| format!("open repo lock {}", path.display()))?;
Ok((path, file))
}
pub fn acquire(output_root: &Path, key: &str, mode: LockMode) -> anyhow::Result<RepoLock> {
let (path, file) = open_lock_file(output_root, key)?;
let file = lock_file(file, key, mode, None)?.expect("unbounded acquisition cannot fail");
Ok(RepoLock {
key: key.to_string(),
path,
mode,
_file: file,
})
}
pub fn try_acquire(
output_root: &Path,
key: &str,
mode: LockMode,
) -> anyhow::Result<Option<RepoLock>> {
acquire_within(output_root, key, mode, Duration::ZERO)
}
pub fn acquire_within(
output_root: &Path,
key: &str,
mode: LockMode,
budget: Duration,
) -> anyhow::Result<Option<RepoLock>> {
let (path, file) = open_lock_file(output_root, key)?;
let file = lock_file(file, key, mode, Some(budget))?;
Ok(file.map(|_file| RepoLock {
key: key.to_string(),
path,
mode,
_file,
}))
}
fn lock_file(
file: File,
key: &str,
mode: LockMode,
budget: Option<Duration>,
) -> anyhow::Result<Option<File>> {
check_acquisition_allowed(key, mode, budget.is_none())?;
let deadline = budget.map(|b| Instant::now() + b);
loop {
let attempt = match mode {
LockMode::Exclusive => file.try_lock(),
LockMode::Lifetime => file.try_lock_shared(),
};
if attempt.is_ok() {
remember_held(key, mode);
return Ok(Some(file));
}
match deadline {
None => {
match mode {
LockMode::Exclusive => file.lock(),
LockMode::Lifetime => file.lock_shared(),
}
.with_context(|| format!("lock repo {key}"))?;
remember_held(key, mode);
return Ok(Some(file));
}
Some(d) if Instant::now() < d => {
std::thread::sleep(Duration::from_millis(20));
}
Some(_) => return Ok(None),
}
}
}
#[derive(Debug, Default)]
pub(crate) struct RepoLockSet {
locks: Vec<RepoLock>,
}
impl RepoLockSet {
pub(crate) fn get(&self, key: &str) -> Option<&RepoLock> {
self.locks.iter().find(|lock| lock.key() == key)
}
pub fn keys(&self) -> Vec<&str> {
self.locks.iter().map(|lock| lock.key()).collect()
}
pub fn len(&self) -> usize {
self.locks.len()
}
}
pub fn acquire_dids<'a, I>(
output_root: &Path,
repo_dids: I,
mode: LockMode,
) -> anyhow::Result<RepoLockSet>
where
I: IntoIterator<Item = &'a str>,
{
let keys: Vec<String> = repo_dids.into_iter().map(crate::plan::repo_key).collect();
acquire_keys(output_root, &keys, mode)
}
fn acquire_keys(
output_root: &Path,
keys: &[String],
mode: LockMode,
) -> anyhow::Result<RepoLockSet> {
let mut sorted: Vec<&String> = keys.iter().collect();
sorted.sort();
sorted.dedup();
let mut locks = Vec::with_capacity(sorted.len());
for key in sorted {
locks.push(acquire(output_root, key, mode)?);
}
Ok(RepoLockSet { locks })
}
#[derive(Debug, Clone, Copy)]
struct Held {
mode: LockMode,
guards: usize,
}
thread_local! {
static HELD: RefCell<BTreeMap<String, Held>> = RefCell::new(BTreeMap::new());
}
fn remember_held(key: &str, mode: LockMode) {
HELD.with(|held| {
let mut held = held.borrow_mut();
match held.get_mut(key) {
Some(entry) => {
entry.guards += 1;
if mode == LockMode::Exclusive {
entry.mode = LockMode::Exclusive;
}
}
None => {
held.insert(key.to_string(), Held { mode, guards: 1 });
}
}
});
}
fn forget_held(key: &str) {
HELD.with(|held| {
let mut held = held.borrow_mut();
if let Some(entry) = held.get_mut(key) {
entry.guards -= 1;
if entry.guards == 0 {
held.remove(key);
}
}
});
}
// flock is per open file: taking a held key again blocks forever unless both are shared.
fn check_acquisition_allowed(key: &str, mode: LockMode, blocking: bool) -> anyhow::Result<()> {
HELD.with(|held| {
let held = held.borrow();
if let Some(existing) = held.get(key)
&& blocking
&& (existing.mode == LockMode::Exclusive || mode == LockMode::Exclusive)
{
anyhow::bail!(
"repo lock {key} is already held by this thread as {}; flock is not recursive, \
request {mode:?} after dropping the existing guard instead",
existing.mode.describe()
);
}
if let Some((highest, _)) = held.iter().next_back()
&& highest.as_str() > key
{
anyhow::bail!(
"repo lock order violation: {key} must be acquired before {highest}; repo locks are \
acquired in ascending repo-key order (use acquire_keys/acquire_dids, which sort)"
);
}
Ok(())
})
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Barrier};
fn root() -> tempfile::TempDir {
tempfile::tempdir().expect("tempdir")
}
/// flock release lags close on macos, which is why it retries up to 5 seconds.
fn wait_mode(root: &Path, key: &str, mode: LockMode) -> RepoLock {
let deadline = Instant::now() + Duration::from_secs(5);
loop {
if let Some(lock) = try_acquire(root, key, mode).unwrap() {
return lock;
}
assert!(Instant::now() < deadline, "{mode:?} lock never became free");
std::thread::sleep(Duration::from_millis(5));
}
}
#[test]
fn lifetime_holders_share_and_block_mutation() {
let dir = root();
let root_path = dir.path();
let barrier = Arc::new(Barrier::new(2));
let holder = {
let barrier = Arc::clone(&barrier);
let root_path = root_path.to_path_buf();
std::thread::spawn(move || {
let first = acquire(&root_path, "k", LockMode::Lifetime).unwrap();
let second = acquire(&root_path, "k", LockMode::Lifetime).unwrap();
barrier.wait();
barrier.wait();
drop(second);
drop(first);
})
};
barrier.wait();
assert!(
try_acquire(root_path, "k", LockMode::Exclusive)
.unwrap()
.is_none(),
"exclusive acquisition must be refused while lifetime holders exist"
);
barrier.wait();
holder.join().unwrap();
let lock = wait_mode(root_path, "k", LockMode::Exclusive);
assert_eq!(lock.key(), "k");
assert_eq!(lock.mode(), LockMode::Exclusive);
}
#[test]
fn exclusive_holder_blocks_readers() {
let dir = root();
let root_path = dir.path();
let exclusive = acquire(root_path, "k", LockMode::Exclusive).unwrap();
assert!(
try_acquire(root_path, "k", LockMode::Lifetime)
.unwrap()
.is_none(),
"a lifetime reader must not start under a mutation holder"
);
drop(exclusive);
let lock = wait_mode(root_path, "k", LockMode::Lifetime);
assert_eq!(lock.mode(), LockMode::Lifetime);
}
#[test]
fn acquire_within_gives_up_without_blocking_forever() {
let dir = root();
let root_path = dir.path();
let _held = acquire(root_path, "k", LockMode::Exclusive).unwrap();
let started = Instant::now();
let got = acquire_within(
root_path,
"k",
LockMode::Lifetime,
Duration::from_millis(150),
)
.unwrap();
assert!(got.is_none());
assert!(started.elapsed() >= Duration::from_millis(100));
}
#[test]
fn acquire_set_sorts_dedups_and_creates_lock_files() {
let dir = root();
let root_path = dir.path();
let set = acquire_keys(
root_path,
&[
"z%2F".to_string(),
"a".to_string(),
"m".to_string(),
"a".to_string(),
],
LockMode::Lifetime,
)
.unwrap();
assert_eq!(set.keys(), vec!["a", "m", "z%2F"]);
assert_eq!(set.len(), 3);
for key in set.keys() {
assert!(lock_path(root_path, key).is_file(), "{key} lock file");
}
drop(set);
assert!(lock_path(root_path, "a").is_file());
}
#[test]
fn acquire_dids_uses_stable_repo_keys() {
let dir = root();
let root_path = dir.path();
let set = acquire_dids(
root_path,
["did:plc:zzz", "did:plc:aaa", "did:plc:zzz"],
LockMode::Lifetime,
)
.unwrap();
assert_eq!(set.keys(), vec!["did%3Aplc%3Aaaa", "did%3Aplc%3Azzz"]);
}
#[test]
fn reacquiring_a_held_key_is_an_error_not_a_hang() {
let dir = root();
let root_path = dir.path();
let _held = acquire(root_path, "k", LockMode::Lifetime).unwrap();
let error = acquire(root_path, "k", LockMode::Exclusive)
.expect_err("re-acquiring a held key must be refused");
assert!(
format!("{error:#}").contains("already held by this thread"),
"{error:#}"
);
}
#[test]
fn out_of_order_acquisition_is_refused() {
let dir = root();
let root_path = dir.path();
let _first = acquire(root_path, "m", LockMode::Lifetime).unwrap();
let error = acquire(root_path, "a", LockMode::Lifetime)
.expect_err("descending order must be refused");
assert!(
format!("{error:#}").contains("lock order violation"),
"{error:#}"
);
let _second = acquire(root_path, "z", LockMode::Lifetime).unwrap();
drop(_first);
drop(_second);
let _any = acquire(root_path, "a", LockMode::Lifetime).unwrap();
}
#[test]
fn unsorted_acquisition_between_two_threads_would_cycle() {
let dir = root();
let root_path = dir.path().to_path_buf();
let barrier = Arc::new(Barrier::new(2));
let want = |held: &'static str, wanted: &'static str| {
let root_path = root_path.clone();
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
let _held = acquire(&root_path, held, LockMode::Lifetime).unwrap();
barrier.wait();
let outcome = match try_acquire(&root_path, wanted, LockMode::Exclusive) {
Ok(Some(_)) => "granted".to_string(),
Ok(None) => "refused by the kernel".to_string(),
Err(error) => format!("refused by the order guard: {error:#}"),
};
barrier.wait();
outcome
})
};
let first = want("a", "b");
let second = want("b", "a");
let first = first.join().unwrap();
let second = second.join().unwrap();
assert_eq!(first, "refused by the kernel", "{first}");
assert!(second.starts_with("refused by the order guard"), "{second}");
}
#[test]
fn same_key_from_another_thread_is_a_real_lock_not_a_local_hang() {
let dir = root();
let root_path = dir.path().to_path_buf();
let barrier = Arc::new(Barrier::new(2));
let holder = {
let barrier = Arc::clone(&barrier);
let root_path = root_path.clone();
std::thread::spawn(move || {
let lock = acquire(&root_path, "k", LockMode::Exclusive).unwrap();
barrier.wait();
barrier.wait();
drop(lock);
})
};
barrier.wait();
assert!(
try_acquire(&root_path, "k", LockMode::Lifetime)
.unwrap()
.is_none()
);
barrier.wait();
holder.join().unwrap();
wait_mode(&root_path, "k", LockMode::Exclusive);
}
}