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
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
//! What:
//! Keeps the lease of the claims one service instance holds fresh while it works, so a claim is taken over only when its holder
//! stopped, never because its holder is slow, and tells the holder when its claim was taken over anyway.
//! Why:
//! A claim's lease starts at the clock reading in its file name and the other instances run a claim again once the lease ran
//! out. A push can take longer than any fixed lease, and an item can wait behind other claimed work; a claim that is never
//! renewed would make both look abandoned. A holder that dies stops renewing, so its claims still expire, and no lock file is
//! involved: the claim's file name is the whole protocol. A renewal renames the claim from the name the holder last wrote, so
//! it can only reach its own claim, and a claim that was taken over is found gone instead of being renewed or settled on
//! another instance's behalf.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! const renewal = startRenewal(workDir, held, lease / 3); try { await work() } finally { renewal.stop() }
//! ```
use super::claim_time::ClaimTime;
use super::work_item::{CLAIMED, Claimed, renew_claim, settle_named, state_directory};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Condvar, Mutex, PoisonError};
use std::thread::JoinHandle;
use std::time::Duration;
/// How many times inside one lease a holder renews, so a renewal that is late by up to two intervals still beats the lease.
pub const RENEWALS_PER_LEASE: u32 = 3;
/// How often a holder renews its claims for a lease of `lease`:
/// a third of it,
/// so that a renewal that is late by up to two
/// intervals still beats the lease.
/// The production lease is 120 seconds,
/// so a holder renews every 40 seconds and loses its
/// claim only when a renewal is late by more than 80.
pub fn renewal_interval(lease: Duration) -> Duration {
return lease / RENEWALS_PER_LEASE;
}
/// One claim this drain holds: the id, the name of its file as of the last renewal, and whether a renewal found it gone.
#[derive(Clone, Debug, Eq, PartialEq)]
struct Held {
/// The item's id.
id: String,
/// The claim's file name in `claimed` that this holder last wrote.
name: String,
/// Set when the file was not there under that name: another instance took the claim over.
lost: bool,
}
/// What:
/// The claims one drain holds right now, each with the file name its holder last wrote.
/// Why:
/// The renewal must refresh only what this instance holds: refreshing every file in the directory would keep the dead
/// holder's claims alive for good. The drain and the renewal thread both rename a claim, so the current name is kept here
/// and every rename happens with the list locked; a settlement can then never name a file that a renewal just renamed.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type HeldClaims = { held: { id: string; name: string; lost: boolean }[] };
/// ```
#[derive(Clone, Debug, Default)]
pub struct HeldClaims {
/// Shared with the renewal thread.
held: Arc<Mutex<Vec<Held>>>,
}
/// Methods over the held claims.
impl HeldClaims {
/// Add a claim the drain took.
pub fn hold(&self, claimed: &Claimed) {
self.held.lock().unwrap_or_else(PoisonError::into_inner).push(Held { id: claimed.id.clone(), name: claimed.name.clone(), lost: false });
}
/// Forget a claim the drain settled, so it is no longer renewed.
pub fn release(&self, id: &str) {
self.held.lock().unwrap_or_else(PoisonError::into_inner).retain(|held| { return held.id != id; });
}
/// The ids held at this moment, including claims found taken over.
pub fn snapshot(&self) -> Vec<String> {
return self.held.lock().unwrap_or_else(PoisonError::into_inner).iter().map(|held| { return held.id.clone(); }).collect();
}
/// Whether the claim of `id` is held and was not found taken over.
pub fn is_held(&self, id: &str) -> bool {
return self.held.lock().unwrap_or_else(PoisonError::into_inner).iter().any(|held| { return held.id == id && !held.lost; });
}
/// The file name the claim of `id` has now, or nothing when it is not held.
pub fn name_of(&self, id: &str) -> Option<String> {
return self.held.lock().unwrap_or_else(PoisonError::into_inner).iter().find(|held| { return held.id == id; }).map(|held| { return held.name.clone(); });
}
/// What:
/// Whether the claim of `id` is held and its file is in `claimed` under the name this holder last wrote,
/// now.
/// Why:
/// A renewal that has not run since another instance took the claim over has not found it gone,
/// so the flag of `is_held` still says held.
/// The holder looks at the directory itself just before it acts on the claim,
/// so a push is not started from a claim that is already another instance's,
/// however late the renewal is.
/// The look and the push are two steps,
/// so a takeover between them is still possible;
/// what the look removes is the long window of a renewal that is late.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// held.has(id) && !held.lost(id) && exists(join(workDir, 'claimed', held.name(id)))
/// ```
pub fn is_held_on_disk(&self, work_dir: &Path, id: &str) -> bool {
let mut held = self.held.lock().unwrap_or_else(PoisonError::into_inner);
let Some(claim) = held.iter_mut().find(|candidate| { return candidate.id == id; }) else { return false };
if claim.lost {
return false;
}
if std::fs::symlink_metadata(state_directory(work_dir, CLAIMED).join(claim.name.as_str())).is_err() {
claim.lost = true;
return false;
}
return true;
}
/// Renew every claim held at the reading `now`; a claim whose file is not there under its name is marked taken over.
pub fn renew(&self, work_dir: &Path, now: &ClaimTime) {
for held in self.held.lock().unwrap_or_else(PoisonError::into_inner).iter_mut().filter(|held| { return !held.lost; }) {
match renew_claim(work_dir, held.name.as_str(), now) {
Some(renewed) => held.name = renewed,
None => held.lost = true,
}
}
}
/// What:
/// Move the claim of `id` to `state` and stop holding it; an error says so when another instance took the claim over.
/// Why:
/// The rename names the claim's current file, so a claim that was taken over, or taken over and claimed again by
/// another instance, is not moved: that instance's work is its own to settle.
pub fn settle(&self, work_dir: &Path, id: &str, state: &str) -> Result<(), String> {
let mut held = self.held.lock().unwrap_or_else(PoisonError::into_inner);
let Some(position) = held.iter().position(|candidate| { return candidate.id == id; }) else {
return Err(format!("{id} is not held by this instance"));
};
let claim: Held = held.remove(position);
if claim.lost {
return Err(format!("{id} was taken over by another instance"));
}
return settle_named(work_dir, claim.name.as_str(), state)
.map_err(|reason| { return format!("{reason}; its lease probably ran out and another instance took the work over"); });
}
}
/// How the claims of one drain are kept fresh.
#[derive(Clone, Copy, Debug)]
pub enum Renewal<'a> {
/// A thread of the drain renews every third of the lease by the real clock.
Timed,
/// Nothing renews until the driver is told to,
/// with the reading it is given;
/// a control uses this to be independent of how
/// the host schedules threads,
/// and to let any number of leases pass without waiting.
Driven(&'a DrivenRenewal),
}
/// What:
/// The claims a drain registered with its driver,
/// renewed whenever the driver is asked to,
/// at the reading it is given.
/// Why:
/// The renewal thread's timing is real time,
/// and a control of what a renewal protects cannot be told apart from a slow host by waiting.
/// A driven renewal does the same renaming at moments the control chooses.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// const driver = new DrivenRenewal(); drain(..., { driven: driver }); driver.renew(clockAfterTwoLeases);
/// ```
#[derive(Debug, Default)]
pub struct DrivenRenewal {
/// The work directory and the claims of the drain that registered,
/// once it has.
target: Mutex<Option<(PathBuf, HeldClaims)>>,
}
/// Methods over a driver.
impl DrivenRenewal {
/// Register the claims of a drain;
/// its later claims are renewed too.
pub fn attach(&self, work_dir: PathBuf, held: HeldClaims) {
*self.target.lock().unwrap_or_else(PoisonError::into_inner) = Some((work_dir, held));
}
/// Renew every claim the registered drain holds at the reading `now`;
/// false when no drain has registered.
pub fn renew(&self, now: &ClaimTime) -> bool {
let target = self.target.lock().unwrap_or_else(PoisonError::into_inner).clone();
let Some((work_dir, held)) = target else { return false };
held.renew(work_dir.as_path(), now);
return true;
}
}
/// What:
/// A running renewal: a thread that stamps every held claim again each interval until this value is dropped.
/// Why:
/// Dropping stops and joins the thread, so a drain that returns or unwinds leaves no thread behind, and a drain that ended
/// renews nothing.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// const renewal = ClaimRenewal.start(workDir, held, every); renewal.stop();
/// ```
pub struct ClaimRenewal {
/// Set to `true` and signalled to stop the thread.
stop: Arc<(Mutex<bool>, Condvar)>,
/// The thread, taken when it is joined.
worker: Option<JoinHandle<()>>,
}
/// Methods over a renewal.
impl ClaimRenewal {
/// Start renewing the claims `held` names in `work_dir` every `every`.
pub fn start(work_dir: PathBuf, held: HeldClaims, every: Duration) -> ClaimRenewal {
let stop: Arc<(Mutex<bool>, Condvar)> = Arc::new((Mutex::new(false), Condvar::new()));
let signal: Arc<(Mutex<bool>, Condvar)> = Arc::clone(&stop);
// A host with no thread to spare renews nothing: the claims then expire with their lease, as for a service that died.
let worker: Option<JoinHandle<()>> = std::thread::Builder::new().spawn(move || {
let (flag, wake) = &*signal;
let mut stopped = flag.lock().unwrap_or_else(PoisonError::into_inner);
loop {
let (guard, _timeout) = wake.wait_timeout_while(stopped, every, |done| { return !*done; }).unwrap_or_else(PoisonError::into_inner);
stopped = guard;
if *stopped {
return;
}
// A claim that was settled is no longer held; one that was taken over is found gone and marked.
held.renew(work_dir.as_path(), &ClaimTime::read());
}
})
.map_err(|error| { crate::service_error!("no thread to renew claims: {error}"); })
.ok();
return ClaimRenewal { stop, worker };
}
}
/// Stops the thread when the renewal goes out of scope.
impl Drop for ClaimRenewal {
fn drop(&mut self) {
let (flag, wake) = &*self.stop;
*flag.lock().unwrap_or_else(PoisonError::into_inner) = true;
wake.notify_all();
if let Some(worker) = self.worker.take() {
// A renewal thread that panicked has nothing left to report: the claims it held expire with the lease.
let _joined = worker.join();
}
}
}
/// Renewal controls stay out of the release executable.
#[cfg(test)]
#[path = "service_claim_renewal_tests.rs"]
mod tests;