1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
//! What:
//! The inotify watch on the directory of an index lock,
//! and the reading of the events it delivers.
//! Why:
//! The service waits for a holder of `index.lock` without polling:
//! the watch wakes it when the lock is removed or moved away,
//! when the kernel reports an overflow that may have hidden that,
//! or when the watch itself ended (the directory was unmounted or removed).
//! Platforms without inotify have no watch,
//! and a held lock fails there at once with the plain message.
//! The module is split from `service_index_lock.rs`,
//! which re-exports everything here under its old names.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! const watch = watchDirectory(dir); const stillWatching = watch.wait(timeout, 'index.lock');
//! ```
use std::path::Path;
use std::time::{Duration, Instant};
#[cfg(test)]
thread_local! {
/// How many watches the current thread holds open,
/// so a control can tell that a watch was dropped without reading descriptor numbers another thread may reuse.
static LIVE_ON_THIS_THREAD: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
/// How many watches the current thread holds open.
#[cfg(test)]
pub fn live_watches_on_this_thread() -> usize {
return LIVE_ON_THIS_THREAD.with(std::cell::Cell::get);
}
/// An inotify watch on the directory of the lock;
/// nothing on platforms without inotify.
pub struct Watch {
#[cfg(target_os = "linux")]
descriptor: i32,
}
/// What:
/// The events of an inotify read as their masks and names.
/// Why:
/// Bytes are parsed,
/// not cast,
/// so an unaligned buffer is read as safely as an aligned one;
/// a record cut off by the end of the buffer ends the list.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// parseEvents(buffer) // [{ mask, name }]
/// ```
#[cfg(target_os = "linux")]
fn parse_events(buffer: &[u8]) -> Vec<(u32, &[u8])> {
let header: usize = std::mem::size_of::<libc::inotify_event>();
let mut events: Vec<(u32, &[u8])> = Vec::new();
let mut at: usize = 0;
while at + header <= buffer.len() {
let field = |offset: usize| -> u32 {
let mut bytes: [u8; 4] = [0; 4];
bytes.copy_from_slice(&buffer[at + offset..at + offset + 4]);
return u32::from_ne_bytes(bytes);
};
let length: usize = usize::try_from(field(12)).unwrap_or(usize::MAX);
let start: usize = at + header;
let end: usize = start.saturating_add(length).min(buffer.len());
let name: &[u8] = buffer[start..end].split(|byte| { return *byte == 0; }).next().unwrap_or(&[]);
events.push((field(4), name));
at = start.saturating_add(length);
}
return events;
}
/// What:
/// Whether a buffer of inotify events reports that the watch itself ended:
/// its directory was unmounted (`IN_UNMOUNT`),
/// the kernel dropped the watch (`IN_IGNORED`,
/// which also follows the directory's removal),
/// or the directory was renamed (`IN_MOVE_SELF`).
/// Why:
/// None of these names a file.
/// A watch that has ended reports nothing more,
/// and one on a directory that was renamed aside stays on the directory at its new name,
/// so a wait that kept polling either would sleep to its timer instead of looking at the lock at the old path again.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// events.some((event) => event.unmount || event.ignored || event.moveSelf)
/// ```
#[cfg(target_os = "linux")]
pub fn events_end_watch(buffer: &[u8]) -> bool {
return parse_events(buffer).iter().any(|(mask, _)| { return mask & (libc::IN_UNMOUNT | libc::IN_IGNORED | libc::IN_MOVE_SELF) != 0; });
}
/// What:
/// Whether a buffer of inotify events names `wanted`,
/// or reports an overflow that may have hidden it,
/// or the end of the watch.
/// Why:
/// The watch covers the whole Git directory,
/// where Git and the service create and remove other files all the time;
/// only the lock's own removal is a reason to try again,
/// so a wait is not woken by its own staging file or a neighbour's.
/// An overflow may have hidden that removal,
/// and a watch that ended can report none,
/// so those wake the caller to look.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// events.some((event) => event.overflow || event.unmount || event.ignored || event.name === wanted)
/// ```
#[cfg(target_os = "linux")]
pub fn events_name(buffer: &[u8], wanted: &[u8]) -> bool {
return events_end_watch(buffer)
|| parse_events(buffer).iter().any(|(mask, name)| { return mask & libc::IN_Q_OVERFLOW != 0 || *name == wanted; });
}
#[cfg(target_os = "linux")]
impl Watch {
pub fn on(directory: &Path) -> Option<Watch> {
use std::os::unix::ffi::OsStrExt;
let path = std::ffi::CString::new(directory.as_os_str().as_bytes()).ok()?;
// SAFETY: plain system calls with valid arguments; the descriptor is closed in `Drop`.
let descriptor: i32 = unsafe { libc::inotify_init1(libc::IN_CLOEXEC) };
if descriptor < 0 {
return None;
}
// SAFETY: `path` is a valid NUL-terminated string for the call.
let added: i32 = unsafe { libc::inotify_add_watch(descriptor, path.as_ptr(), libc::IN_DELETE | libc::IN_MOVED_FROM | libc::IN_MOVE_SELF) };
if added < 0 {
// SAFETY: the descriptor was just opened and nothing else owns it.
unsafe { libc::close(descriptor) };
return None;
}
#[cfg(test)]
LIVE_ON_THIS_THREAD.with(|count| { count.set(count.get() + 1); });
return Some(Watch { descriptor });
}
/// Blocks until the file `name` in the directory was removed or moved away,
/// or `timeout` passed;
/// either way the caller looks again.
/// Events for other files are drained and the wait goes on.
/// False when the watch has ended (the directory was unmounted,
/// removed or moved),
/// so the caller sets a new one before it waits again.
pub fn wait(&self, timeout: Duration, name: &std::ffi::OsStr) -> bool {
use std::os::unix::ffi::OsStrExt;
let deadline: Instant = Instant::now() + timeout;
loop {
let left: Duration = deadline.saturating_duration_since(Instant::now());
if left.is_zero() {
return true;
}
let mut entry = libc::pollfd { fd: self.descriptor, events: libc::POLLIN, revents: 0 };
let milliseconds: i32 = i32::try_from(left.as_millis()).unwrap_or(i32::MAX).max(1);
// SAFETY: `entry` is valid for the one descriptor named by the count.
let ready: i32 = unsafe { libc::poll(&raw mut entry, 1, milliseconds) };
if ready <= 0 {
return true;
}
let mut buffer: [u8; 4096] = [0; 4096];
// SAFETY: the buffer is valid for its length.
let read: isize = unsafe { libc::read(self.descriptor, buffer.as_mut_ptr().cast(), buffer.len()) };
let Ok(count) = usize::try_from(read) else { return true };
if count == 0 {
return true;
}
if events_name(&buffer[..count], name.as_bytes()) {
return !events_end_watch(&buffer[..count]);
}
}
}
}
#[cfg(target_os = "linux")]
impl Drop for Watch {
fn drop(&mut self) {
#[cfg(test)]
LIVE_ON_THIS_THREAD.with(|count| { count.set(count.get().saturating_sub(1)); });
// SAFETY: the descriptor belongs to this watch and is closed once.
unsafe { libc::close(self.descriptor) };
}
}
#[cfg(not(target_os = "linux"))]
impl Watch {
pub fn on(_directory: &Path) -> Option<Watch> {
return None;
}
pub fn wait(&self, _timeout: Duration, _name: &std::ffi::OsStr) -> bool {
return true;
}
}