pbs-tools/src/process_locker.rs

   1 //! Inter-process reader-writer lock builder.
   2 //!
   3 //! This implementation uses fcntl record locks with non-blocking
   4 //! F_SETLK command (never blocks).
   5 //!
   6 //! We maintain a map of shared locks with time stamps, so you can get
   7 //! the timestamp for the oldest open lock with
   8 //! `oldest_shared_lock()`.
   9
  10 use std::collections::HashMap;
  11 use std::os::unix::io::AsRawFd;
  12 use std::sync::{Arc, Mutex};
  13
  14 use anyhow::{bail, Error};
  15
  16 // fixme: use F_OFD_ locks when implemented with nix::fcntl
  17
  18 // Note: flock lock conversion is not atomic, so we need to use fcntl
  19
  20 /// Inter-process reader-writer lock
  21 pub struct ProcessLocker {
  22     file: std::fs::File,
  23     exclusive: bool,
  24     writers: usize,
  25     next_guard_id: u64,
  26     shared_guard_list: HashMap<u64, i64>, // guard_id => timestamp
  27 }
  28
  29 /// Lock guard for shared locks
  30 ///
  31 /// Release the lock when it goes out of scope.
  32 pub struct ProcessLockSharedGuard {
  33     guard_id: u64,
  34     locker: Arc<Mutex<ProcessLocker>>,
  35 }
  36
  37 impl Drop for ProcessLockSharedGuard {
  38     fn drop(&mut self) {
  39         let mut data = self.locker.lock().unwrap();
  40
  41         if data.writers == 0 {
  42             panic!("unexpected ProcessLocker state");
  43         }
  44
  45         data.shared_guard_list.remove(&self.guard_id);
  46
  47         if data.writers == 1 && !data.exclusive {
  48             let op = libc::flock {
  49                 l_type: libc::F_UNLCK as i16,
  50                 l_whence: libc::SEEK_SET as i16,
  51                 l_start: 0,
  52                 l_len: 0,
  53                 l_pid: 0,
  54             };
  55
  56             if let Err(err) =
  57                 nix::fcntl::fcntl(data.file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLKW(&op))
  58             {
  59                 panic!("unable to drop writer lock - {}", err);
  60             }
  61         }
  62         if data.writers > 0 {
  63             data.writers -= 1;
  64         }
  65     }
  66 }
  67
  68 /// Lock guard for exclusive locks
  69 ///
  70 /// Release the lock when it goes out of scope.
  71 pub struct ProcessLockExclusiveGuard {
  72     locker: Arc<Mutex<ProcessLocker>>,
  73 }
  74
  75 impl Drop for ProcessLockExclusiveGuard {
  76     fn drop(&mut self) {
  77         let mut data = self.locker.lock().unwrap();
  78
  79         if !data.exclusive {
  80             panic!("unexpected ProcessLocker state");
  81         }
  82
  83         let ltype = if data.writers != 0 {
  84             libc::F_RDLCK
  85         } else {
  86             libc::F_UNLCK
  87         };
  88         let op = libc::flock {
  89             l_type: ltype as i16,
  90             l_whence: libc::SEEK_SET as i16,
  91             l_start: 0,
  92             l_len: 0,
  93             l_pid: 0,
  94         };
  95
  96         if let Err(err) =
  97             nix::fcntl::fcntl(data.file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLKW(&op))
  98         {
  99             panic!("unable to drop exclusive lock - {}", err);
 100         }
 101
 102         data.exclusive = false;
 103     }
 104 }
 105
 106 impl ProcessLocker {
 107     /// Create a new instance for the specified file.
 108     ///
 109     /// This simply creates the file if it does not exist.
 110     pub fn new<P: AsRef<std::path::Path>>(lockfile: P) -> Result<Arc<Mutex<Self>>, Error> {
 111         let file = std::fs::OpenOptions::new()
 112             .create(true)
 113             .read(true)
 114             .write(true)
 115             .open(lockfile)?;
 116
 117         Ok(Arc::new(Mutex::new(Self {
 118             file,
 119             exclusive: false,
 120             writers: 0,
 121             next_guard_id: 0,
 122             shared_guard_list: HashMap::new(),
 123         })))
 124     }
 125
 126     fn try_lock(file: &std::fs::File, ltype: i32) -> Result<(), Error> {
 127         let op = libc::flock {
 128             l_type: ltype as i16,
 129             l_whence: libc::SEEK_SET as i16,
 130             l_start: 0,
 131             l_len: 0,
 132             l_pid: 0,
 133         };
 134
 135         nix::fcntl::fcntl(file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLK(&op))?;
 136
 137         Ok(())
 138     }
 139
 140     /// Try to acquire a shared lock
 141     ///
 142     /// On success, this makes sure that no other process can get an exclusive lock for the file.
 143     pub fn try_shared_lock(locker: Arc<Mutex<Self>>) -> Result<ProcessLockSharedGuard, Error> {
 144         let mut data = locker.lock().unwrap();
 145
 146         if data.writers == 0 && !data.exclusive {
 147             if let Err(err) = Self::try_lock(&data.file, libc::F_RDLCK) {
 148                 bail!("unable to get shared lock - {}", err);
 149             }
 150         }
 151
 152         data.writers += 1;
 153
 154         let guard = ProcessLockSharedGuard {
 155             locker: locker.clone(),
 156             guard_id: data.next_guard_id,
 157         };
 158         data.next_guard_id += 1;
 159
 160         let now = unsafe { libc::time(std::ptr::null_mut()) };
 161
 162         data.shared_guard_list.insert(guard.guard_id, now);
 163
 164         Ok(guard)
 165     }
 166
 167     /// Get oldest shared lock timestamp
 168     pub fn oldest_shared_lock(locker: Arc<Mutex<Self>>) -> Option<i64> {
 169         let mut result = None;
 170
 171         let data = locker.lock().unwrap();
 172
 173         for v in data.shared_guard_list.values() {
 174             result = match result {
 175                 None => Some(*v),
 176                 Some(x) => {
 177                     if x < *v {
 178                         Some(x)
 179                     } else {
 180                         Some(*v)
 181                     }
 182                 }
 183             };
 184         }
 185
 186         result
 187     }
 188
 189     /// Try to acquire a exclusive lock
 190     ///
 191     /// Make sure the we are the only process which has locks for this file (shared or exclusive).
 192     pub fn try_exclusive_lock(
 193         locker: Arc<Mutex<Self>>,
 194     ) -> Result<ProcessLockExclusiveGuard, Error> {
 195         let mut data = locker.lock().unwrap();
 196
 197         if data.exclusive {
 198             bail!("already locked exclusively");
 199         }
 200
 201         if let Err(err) = Self::try_lock(&data.file, libc::F_WRLCK) {
 202             bail!("unable to get exclusive lock - {}", err);
 203         }
 204
 205         data.exclusive = true;
 206
 207         Ok(ProcessLockExclusiveGuard {
 208             locker: locker.clone(),
 209         })
 210     }
 211 }