tyler/proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

move tools::process_locker to pbs-tools

Browse Source 
Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
This commit is contained in:
Wolfgang Bumiller
2021-07-07 14:15:42 +02:00
parent 95f9d67ce9
commit 83771aa037
3 changed files with 44 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/tools/mod.rs
View File

@ -26,6 +26,9 @@ use proxmox_http::{
pub use pbs_tools::json;
pub use pbs_tools::nom;
pub use pbs_tools::{run_command, command_output, command_output_as_string};
pub use pbs_tools::process_locker::{
ProcessLocker, ProcessLockExclusiveGuard, ProcessLockSharedGuard
};
pub mod acl;
pub mod apt;
@ -70,9 +73,6 @@ pub use std_channel_writer::StdChannelWriter;
mod tokio_writer_adapter;
pub use tokio_writer_adapter::TokioWriterAdapter;
mod process_locker;
pub use process_locker::{ProcessLocker, ProcessLockExclusiveGuard, ProcessLockSharedGuard};
mod file_logger;
pub use file_logger::{FileLogger, FileLogOptions};

192
src/tools/process_locker.rs
View File

@ -1,192 +0,0 @@
//! Inter-process reader-writer lock builder.
//!
//! This implementation uses fcntl record locks with non-blocking
//! F_SETLK command (never blocks).
//!
//! We maintain a map of shared locks with time stamps, so you can get
//! the timestamp for the oldest open lock with
//! `oldest_shared_lock()`.
use anyhow::{bail, Error};
use std::sync::{Arc, Mutex};
use std::os::unix::io::AsRawFd;
use std::collections::HashMap;
// fixme: use F_OFD_ locks when implemented with nix::fcntl
// Note: flock lock conversion is not atomic, so we need to use fcntl
/// Inter-process reader-writer lock
pub struct ProcessLocker {
file: std::fs::File,
exclusive: bool,
writers: usize,
next_guard_id: u64,
shared_guard_list: HashMap<u64, i64>, // guard_id => timestamp
}
/// Lock guard for shared locks
///
/// Release the lock when it goes out of scope.
pub struct ProcessLockSharedGuard {
guard_id: u64,
locker: Arc<Mutex<ProcessLocker>>,
}
impl Drop for ProcessLockSharedGuard {
fn drop(&mut self) {
let mut data = self.locker.lock().unwrap();
if data.writers == 0 { panic!("unexpected ProcessLocker state"); }
data.shared_guard_list.remove(&self.guard_id);
if data.writers == 1 && !data.exclusive {
let op = libc::flock {
l_type: libc::F_UNLCK as i16,
l_whence: libc::SEEK_SET as i16,
l_start: 0,
l_len: 0,
l_pid: 0,
};
if let Err(err) = nix::fcntl::fcntl(data.file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLKW(&op)) {
panic!("unable to drop writer lock - {}", err);
}
}
if data.writers > 0 {
data.writers -= 1;
}
}
}
/// Lock guard for exclusive locks
///
/// Release the lock when it goes out of scope.
pub struct ProcessLockExclusiveGuard {
locker: Arc<Mutex<ProcessLocker>>,
}
impl Drop for ProcessLockExclusiveGuard {
fn drop(&mut self) {
let mut data = self.locker.lock().unwrap();
if !data.exclusive { panic!("unexpected ProcessLocker state"); }
let ltype = if data.writers != 0 { libc::F_RDLCK } else { libc::F_UNLCK };
let op = libc::flock {
l_type: ltype as i16,
l_whence: libc::SEEK_SET as i16,
l_start: 0,
l_len: 0,
l_pid: 0,
};
if let Err(err) = nix::fcntl::fcntl(data.file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLKW(&op)) {
panic!("unable to drop exclusive lock - {}", err);
}
data.exclusive = false;
}
}
impl ProcessLocker {
/// Create a new instance for the specified file.
///
/// This simply creates the file if it does not exist.
pub fn new<P: AsRef<std::path::Path>>(lockfile: P) -> Result<Arc<Mutex<Self>>, Error> {
let file = std::fs::OpenOptions::new()
.create(true)
.read(true)
.write(true)
.open(lockfile)?;
Ok(Arc::new(Mutex::new(Self {
file,
exclusive: false,
writers: 0,
next_guard_id: 0,
shared_guard_list: HashMap::new(),
})))
}
fn try_lock(file: &std::fs::File, ltype: i32) -> Result<(), Error> {
let op = libc::flock {
l_type: ltype as i16,
l_whence: libc::SEEK_SET as i16,
l_start: 0,
l_len: 0,
l_pid: 0,
};
nix::fcntl::fcntl(file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLK(&op))?;
Ok(())
}
/// Try to acquire a shared lock
///
/// On success, this makes sure that no other process can get an exclusive lock for the file.
pub fn try_shared_lock(locker: Arc<Mutex<Self>>) -> Result<ProcessLockSharedGuard, Error> {
let mut data = locker.lock().unwrap();
if data.writers == 0 && !data.exclusive {
if let Err(err) = Self::try_lock(&data.file, libc::F_RDLCK) {
bail!("unable to get shared lock - {}", err);
}
}
data.writers += 1;
let guard = ProcessLockSharedGuard { locker: locker.clone(), guard_id: data.next_guard_id };
data.next_guard_id += 1;
let now = unsafe { libc::time(std::ptr::null_mut()) };
data.shared_guard_list.insert(guard.guard_id, now);
Ok(guard)
}
/// Get oldest shared lock timestamp
pub fn oldest_shared_lock(locker: Arc<Mutex<Self>>) -> Option<i64> {
let mut result = None;
let data = locker.lock().unwrap();
for v in data.shared_guard_list.values() {
result = match result {
None => Some(*v),
Some(x) => if x < *v { Some(x) } else { Some(*v) },
};
}
result
}
/// Try to acquire a exclusive lock
///
/// Make sure the we are the only process which has locks for this file (shared or exclusive).
pub fn try_exclusive_lock(locker: Arc<Mutex<Self>>) -> Result<ProcessLockExclusiveGuard, Error> {
let mut data = locker.lock().unwrap();
if data.exclusive {
bail!("already locked exclusively");
}
if let Err(err) = Self::try_lock(&data.file, libc::F_WRLCK) {
bail!("unable to get exclusive lock - {}", err);
}
data.exclusive = true;
Ok(ProcessLockExclusiveGuard { locker: locker.clone() })
}
}