zeppelin_embed/lifecycle/
lock.rs1use std::collections::BTreeSet;
21use std::fs::{File, OpenOptions};
22#[cfg(unix)]
23use std::os::fd::AsRawFd;
24#[cfg(unix)]
25use std::os::unix::fs::MetadataExt as _;
26use std::path::{Path, PathBuf};
27use std::sync::{Mutex, MutexGuard};
28
29pub const STORE_LOCK_FILE: &str = "writer.lock";
31
32type StoreKey = (u64, u64);
34
35static HELD_STORES: Mutex<BTreeSet<StoreKey>> = Mutex::new(BTreeSet::new());
36
37#[derive(Debug)]
39pub struct StoreLock {
40 file: Option<File>,
41 registry_key: StoreKey,
42}
43
44impl StoreLock {
45 pub fn acquire(directory: &Path) -> Result<Self, StoreLockError> {
52 let path = directory.join(STORE_LOCK_FILE);
53 let registry_key = store_identity(directory).map_err(|source| StoreLockError::Io {
54 path: directory.to_path_buf(),
55 source,
56 })?;
57 if !held_stores().insert(registry_key) {
58 return Err(StoreLockError::Io {
59 path,
60 source: std::io::Error::from(std::io::ErrorKind::WouldBlock),
61 });
62 }
63 let file = open_lock_file(&path).map_err(|source| {
64 held_stores().remove(®istry_key);
65 StoreLockError::Io {
66 path: path.clone(),
67 source,
68 }
69 })?;
70 if let Err(source) = lock_exclusive(&file) {
71 drop(file);
72 held_stores().remove(®istry_key);
73 return Err(StoreLockError::Io { path, source });
74 }
75 Ok(Self {
76 file: Some(file),
77 registry_key,
78 })
79 }
80}
81
82fn held_stores() -> MutexGuard<'static, BTreeSet<StoreKey>> {
83 match HELD_STORES.lock() {
84 Ok(held) => held,
85 Err(poisoned) => poisoned.into_inner(),
86 }
87}
88
89#[cfg(unix)]
91fn store_identity(directory: &Path) -> std::io::Result<StoreKey> {
92 let metadata = std::fs::metadata(directory)?;
93 Ok((metadata.dev(), metadata.ino()))
94}
95
96#[cfg(windows)]
97fn store_identity(directory: &Path) -> std::io::Result<StoreKey> {
98 crate::sys::windows::directory_identity(directory)
99}
100
101#[cfg(not(any(unix, windows)))]
102fn store_identity(_directory: &Path) -> std::io::Result<StoreKey> {
103 Err(std::io::Error::new(
104 std::io::ErrorKind::Unsupported,
105 "store identity requires a supported platform",
106 ))
107}
108
109#[cfg(not(windows))]
111fn open_lock_file(path: &Path) -> std::io::Result<File> {
112 OpenOptions::new()
113 .create(true)
114 .truncate(false)
115 .read(true)
116 .write(true)
117 .open(path)
118}
119
120#[cfg(windows)]
130fn open_lock_file(path: &Path) -> std::io::Result<File> {
131 use std::os::windows::fs::OpenOptionsExt as _;
132
133 OpenOptions::new()
134 .create(true)
135 .truncate(false)
136 .read(true)
137 .write(true)
138 .share_mode(crate::sys::windows::FILE_SHARE_READ | crate::sys::windows::FILE_SHARE_WRITE)
139 .open(path)
140}
141
142#[cfg(unix)]
143fn lock_exclusive(file: &File) -> std::io::Result<()> {
144 let mut lock = unsafe {
145 std::mem::zeroed::<libc::flock>()
147 };
148 lock.l_type = libc::F_WRLCK as libc::c_short;
149 lock.l_whence = libc::SEEK_SET as libc::c_short;
150 lock.l_start = 0;
151 lock.l_len = 0;
152 let result = unsafe {
153 libc::fcntl(file.as_raw_fd(), libc::F_SETLK, &lock)
156 };
157 if result == -1 {
158 let error = std::io::Error::last_os_error();
159 if error
160 .raw_os_error()
161 .is_some_and(|errno| errno == libc::EAGAIN || errno == libc::EACCES)
162 {
163 Err(std::io::Error::from(std::io::ErrorKind::WouldBlock))
164 } else {
165 Err(error)
166 }
167 } else {
168 Ok(())
169 }
170}
171
172#[cfg(windows)]
178fn lock_exclusive(file: &File) -> std::io::Result<()> {
179 crate::sys::windows::lock_file_exclusive(file)
180}
181
182#[cfg(not(any(unix, windows)))]
183fn lock_exclusive(_: &File) -> std::io::Result<()> {
184 Err(std::io::Error::new(
185 std::io::ErrorKind::Unsupported,
186 "store locks require a supported file-locking platform",
187 ))
188}
189
190impl Drop for StoreLock {
191 fn drop(&mut self) {
192 drop(self.file.take());
195 held_stores().remove(&self.registry_key);
196 }
197}
198
199#[derive(Debug)]
201pub enum StoreLockError {
202 Io {
204 path: PathBuf,
206 source: std::io::Error,
208 },
209}
210
211impl std::fmt::Display for StoreLockError {
212 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
213 match self {
214 Self::Io { path, source } => {
215 write!(formatter, "store lock {}: {source}", path.display())
216 }
217 }
218 }
219}
220
221impl std::error::Error for StoreLockError {
222 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
223 match self {
224 Self::Io { source, .. } => Some(source),
225 }
226 }
227}