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
//! What:
//! The drain of several repositories in one round of the service's after-commit work, side by side.
//! Why:
//! Repositories share nothing: each has its own work directory, its own claims and its own pushes, so one repository's slow
//! push or review never needs to hold up another's. The round's thread hands each repository to a worker and takes the answers
//! back in the order the repositories were named (`parallel_runner.rs`), so the retry list is the one a drain of one after
//! another gives. No lock is taken (`service_lock_inventory_tests.rs`): the handlers are shared read-only, every claim is made
//! by an atomic rename as before, and the directories of one repository (a main checkout and its linked worktrees) share a lane
//! and are drained one after another there, because they share a work directory and its upkeep.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! const retries = await drainDirectories({ directories, handlers, now, host });
//! ```
use super::claim_time::ClaimTime;
use super::parallel_budget::{BudgetRequest, HostResources, worker_budget};
use super::parallel_runner::{Event, OrderedRun, run_ordered};
use super::push_retry::{RetryChange, Retrying};
use super::service_work::{CLAIM_LEASE, Drained, WorkHandlers, common_directory_of, drain_common_records_planned, drain_planned};
use std::path::{Path, PathBuf};
use std::time::Duration;
/// The bytes the drain of one repository holds at its peak, estimated: the items of a burst, the output of a push, and a review's
/// patch and answer, which the review bounds; about 64 MiB.
pub const REPOSITORY_MEMORY: u64 = 64 << 20;
/// What:
/// One directory of a round, with the pushes of it that are due again.
/// Why:
/// The service loop owns the retry plan; it takes what is due for the directory out of the plan before the round and hands it to
/// the directory's worker inside the request, as an owned value, so no worker reads the plan.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type Assignment = { directory: string; retrying: Retrying };
/// ```
#[derive(Clone, Debug)]
pub struct Assignment {
/// The Git directory to drain.
pub directory: PathBuf,
/// The first pause and the pushes due again for this directory.
pub retrying: Retrying,
}
/// Methods over an assignment.
impl Assignment {
/// A directory with no plan: a push that could pass is reported as failed.
pub fn plain(directory: PathBuf) -> Self {
return Self { directory, retrying: Retrying::none() };
}
}
/// What:
/// What a round gives back: when to look again at the directories that have a claim another instance holds, and the changes the
/// workers ask of the retry plan.
/// Why:
/// The waits keep the order the directories were named; the changes are in that order too, so applying them in sequence gives
/// the plan a drain of one directory after another would.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type RoundAnswer = { waits: [string, number][]; changes: RetryChange[] };
/// ```
#[derive(Debug, Default)]
pub struct RoundAnswer {
/// For each directory that has such a claim, how long until its lease runs out, in the order the directories were named.
pub waits: Vec<(PathBuf, Duration)>,
/// What the workers ask of the retry plan.
pub changes: Vec<RetryChange>,
}
/// What:
/// What a round needs to drain its repositories.
/// Why:
/// One value names the inputs, so the call reads as a list of facts.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type RoundDrain<H> = { directories: string[]; handlers: H; now: ClaimTime; host: HostResources };
/// ```
pub struct RoundDrain<'a, H: WorkHandlers + Sync> {
/// The Git directories named by the round's wakes, each once, in the order they were first named, each with what is due again.
pub directories: &'a [Assignment],
/// What runs each kind of work; shared read-only by every worker.
pub handlers: &'a H,
/// The reading of the clock the round started at.
pub now: &'a ClaimTime,
/// The CPUs and memory the budget is derived from.
pub host: &'a HostResources,
}
/// What:
/// The lane of a directory: the common directory of its repository when it names one, else the directory itself.
/// Why:
/// The directories of one repository share a work directory and its upkeep, so they are drained one after another.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// const lane = commonDirectoryOf(gitDir) ?? gitDir;
/// ```
fn lane_key(directory: &Path) -> PathBuf {
return common_directory_of(directory).unwrap_or_else(|| { return directory.to_path_buf(); });
}
/// What:
/// Drain every directory, repositories side by side, and give, for each directory that has a claim another instance holds, how
/// long until that claim's lease runs out, in the order the directories were named.
/// Why:
/// See the module description.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// const answer = await drainDirectories({ directories, handlers, now, host });
/// ```
pub fn drain_directories<H: WorkHandlers + Sync>(request: &RoundDrain<'_, H>) -> RoundAnswer {
let RoundDrain { directories, handlers, now, host } = *request;
let mut lanes: Vec<Vec<usize>> = Vec::new();
let mut lane_of: std::collections::HashMap<PathBuf, usize> = std::collections::HashMap::new();
for (index, assignment) in directories.iter().enumerate() {
match lane_of.get(&lane_key(assignment.directory.as_path())) {
Some(&lane) => lanes[lane].push(index),
None => {
lane_of.insert(lane_key(assignment.directory.as_path()), lanes.len());
lanes.push(vec![index]);
}
}
}
let workers: usize = worker_budget(&BudgetRequest { items: lanes.len(), per_item_memory: REPOSITORY_MEMORY, ceiling: None, cap: None, host });
let mut answer: RoundAnswer = RoundAnswer::default();
run_ordered(OrderedRun {
items: directories,
lanes: lanes.as_slice(),
workers,
work: &|_index: usize, assignment: &Assignment, _say: &mut dyn FnMut(&str)| {
let own: Drained = drain_planned(assignment.directory.as_path(), handlers, now, &assignment.retrying);
let common: Drained = drain_common_records_planned(assignment.directory.as_path(), handlers, now, CLAIM_LEASE, assignment.retrying.first);
let wait: Option<Duration> = [own.wait, common.wait].into_iter().flatten().min();
return (wait, own.changes.into_iter().chain(common.changes).collect::<Vec<RetryChange>>());
},
on_event: &mut |event: Event<'_, (Option<Duration>, Vec<RetryChange>)>| {
if let Event::End(index, (wait, changes)) = event {
if let Some(lease_wait) = wait {
answer.waits.push((directories[index].directory.clone(), lease_wait));
}
answer.changes.extend(changes);
}
},
});
return answer;
}
/// Round controls stay out of the release executable.
#[cfg(test)]
#[path = "service_round_tests.rs"]
mod tests;