src/tools/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 anyhow::{bail, Error};
  11
  12 use std::sync::{Arc, Mutex};
  13 use std::os::unix::io::AsRawFd;
  14 use std::collections::HashMap;
  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 { panic!("unexpected ProcessLocker state"); }
  42
  43         data.shared_guard_list.remove(&self.guard_id);
  44
  45         if data.writers == 1 && !data.exclusive {
  46
  47             let op = libc::flock {
  48                 l_type: libc::F_UNLCK as i16,
  49                 l_whence: libc::SEEK_SET as i16,
  50                 l_start: 0,
  51                 l_len: 0,
  52                 l_pid: 0,
  53             };
  54
  55             if let Err(err) = nix::fcntl::fcntl(data.file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLKW(&op)) {
  56                 panic!("unable to drop writer lock - {}", err);
  57             }
  58         }
  59         if data.writers > 0 {
  60             data.writers -= 1;
  61         }
  62     }
  63 }
  64
  65 /// Lock guard for exclusive locks
  66 ///
  67 /// Release the lock when it goes out of scope.
  68 pub struct ProcessLockExclusiveGuard {
  69     locker: Arc<Mutex<ProcessLocker>>,
  70 }
  71
  72 impl  Drop for ProcessLockExclusiveGuard {
  73     fn drop(&mut self) {
  74         let mut data = self.locker.lock().unwrap();
  75
  76         if !data.exclusive { panic!("unexpected ProcessLocker state"); }
  77
  78         let ltype = if data.writers != 0 { libc::F_RDLCK } else { libc::F_UNLCK };
  79         let op = libc::flock {
  80             l_type: ltype as i16,
  81             l_whence: libc::SEEK_SET as i16,
  82             l_start: 0,
  83             l_len: 0,
  84             l_pid: 0,
  85         };
  86
  87         if let Err(err) = nix::fcntl::fcntl(data.file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLKW(&op)) {
  88             panic!("unable to drop exclusive lock - {}", err);
  89         }
  90
  91         data.exclusive = false;
  92     }
  93 }
  94
  95 impl ProcessLocker {
  96
  97     /// Create a new instance for the specified file.
  98     ///
  99     /// This simply creates the file if it does not exist.
 100     pub fn new<P: AsRef<std::path::Path>>(lockfile: P) -> Result<Arc<Mutex<Self>>, Error> {
 101
 102         let file = std::fs::OpenOptions::new()
 103             .create(true)
 104             .read(true)
 105             .write(true)
 106             .open(lockfile)?;
 107
 108         Ok(Arc::new(Mutex::new(Self {
 109             file,
 110             exclusive: false,
 111             writers: 0,
 112             next_guard_id: 0,
 113             shared_guard_list: HashMap::new(),
 114         })))
 115     }
 116
 117     fn try_lock(file: &std::fs::File, ltype: i32) -> Result<(), Error> {
 118
 119         let op = libc::flock {
 120             l_type: ltype as i16,
 121             l_whence: libc::SEEK_SET as i16,
 122             l_start: 0,
 123             l_len: 0,
 124             l_pid: 0,
 125         };
 126
 127         nix::fcntl::fcntl(file.as_raw_fd(), nix::fcntl::FcntlArg::F_SETLK(&op))?;
 128
 129         Ok(())
 130     }
 131
 132     /// Try to acquire a shared lock
 133     ///
 134     /// On success, this makes sure that no other process can get an exclusive lock for the file.
 135     pub fn try_shared_lock(locker: Arc<Mutex<Self>>) -> Result<ProcessLockSharedGuard, Error> {
 136
 137         let mut data = locker.lock().unwrap();
 138
 139         if data.writers == 0 && !data.exclusive {
 140             if let Err(err) = Self::try_lock(&data.file, libc::F_RDLCK) {
 141                 bail!("unable to get shared lock - {}", err);
 142             }
 143         }
 144
 145         data.writers += 1;
 146
 147         let guard = ProcessLockSharedGuard { locker: locker.clone(), guard_id: data.next_guard_id };
 148         data.next_guard_id += 1;
 149
 150         let now = unsafe { libc::time(std::ptr::null_mut()) };
 151
 152         data.shared_guard_list.insert(guard.guard_id, now);
 153
 154         Ok(guard)
 155     }
 156
 157     /// Get oldest shared lock timestamp
 158     pub fn oldest_shared_lock(locker: Arc<Mutex<Self>>) -> Option<i64> {
 159         let mut result = None;
 160
 161         let data = locker.lock().unwrap();
 162
 163         for v in data.shared_guard_list.values() {
 164             result = match result {
 165                 None => Some(*v),
 166                 Some(x) => if x < *v { Some(x) } else { Some(*v) },
 167             };
 168         }
 169
 170         result
 171     }
 172
 173     /// Try to acquire a exclusive lock
 174     ///
 175     /// Make sure the we are the only process which has locks for this file (shared or exclusive).
 176     pub fn try_exclusive_lock(locker: Arc<Mutex<Self>>) -> Result<ProcessLockExclusiveGuard, Error> {
 177
 178         let mut data = locker.lock().unwrap();
 179
 180         if data.exclusive {
 181             bail!("already locked exclusively");
 182         }
 183
 184         if let Err(err) = Self::try_lock(&data.file, libc::F_WRLCK) {
 185             bail!("unable to get exclusive lock - {}", err);
 186         }
 187
 188         data.exclusive = true;
 189
 190         Ok(ProcessLockExclusiveGuard { locker: locker.clone() })
 191     }
 192 }