alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::net::SocketAddr;
2use std::path::Path;
3use std::process::{Child, Stdio};
4use std::time::Duration;
5
6use axum::Router;
7use knot_bench::{ChurnCount, CommitCount, HistorySpec, PathCount, write_history};
8use knot_git::Layout;
9use knot_pack::{RepoLookup, RepoResolver, RepoTarget};
10use knot_types::RepoDid;
11
12mod common;
13use common::must;
14
15fn serve_dids() -> std::sync::Arc<dyn RepoResolver> {
16 std::sync::Arc::new(|target: &RepoTarget| match target {
17 RepoTarget::Did(did) => RepoLookup::Hosted(did.clone()),
18 RepoTarget::OwnerPath(_, _) => RepoLookup::Unhosted,
19 })
20}
21
22const BLOB_BYTES: usize = 16 * 1024 * 1024;
23const CONCURRENCY: usize = 10;
24const ROUNDS: usize = 5;
25const PAGE_BYTES: u64 = 4096;
26const OOM_CEILING: u64 = 1024 * 1024 * 1024;
27const GROWTH_SLACK: u64 = 64 * 1024 * 1024;
28const CURVE_LEVELS: [usize; 5] = [1, 2, 4, 8, 16];
29const PER_CONNECTION_CEILING: u64 = 96 * 1024 * 1024;
30const SUBLINEAR_SLACK: u64 = 32 * 1024 * 1024;
31
32static RSS_GATE: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
33
34fn incompressible(len: usize) -> Vec<u8> {
35 let mut state = 0x9e37_79b9_7f4a_7c15u64;
36 (0..len)
37 .map(|_| {
38 state ^= state << 13;
39 state ^= state >> 7;
40 state ^= state << 17;
41 (state & 0xff) as u8
42 })
43 .collect()
44}
45
46async fn spawn(router: Router) -> SocketAddr {
47 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
48 let addr = listener.local_addr().unwrap();
49 tokio::spawn(async move {
50 axum::serve(listener, router).await.unwrap();
51 });
52 addr
53}
54
55fn rss_bytes() -> u64 {
56 let statm = std::fs::read_to_string("/proc/self/statm").expect("/proc/self/statm is readable");
57 statm
58 .split_whitespace()
59 .nth(1)
60 .and_then(|pages| pages.parse::<u64>().ok())
61 .map(|pages| pages * PAGE_BYTES)
62 .expect("statm lists the resident page count")
63}
64
65fn clone_child(remote: &str, dest: &Path) -> Child {
66 knot_fixtures::command(dest.parent().unwrap_or(dest))
67 .args(["clone", "--bare", "--quiet", remote, dest.to_str().unwrap()])
68 .stdout(Stdio::null())
69 .stderr(Stdio::piped())
70 .spawn()
71 .expect("git clone spawns")
72}
73
74struct Soak {
75 _scan: tempfile::TempDir,
76 scratch: tempfile::TempDir,
77 remote: String,
78 tip: String,
79}
80
81async fn serve_large_repo() -> Soak {
82 let scan = tempfile::tempdir().unwrap();
83 let layout = Layout::new(scan.path());
84 let did = RepoDid::new("did:plc:squid").unwrap();
85 layout.create(&did).unwrap();
86 let bare = layout.repo_path(&did).unwrap();
87
88 let scratch = tempfile::tempdir().unwrap();
89 let work = scratch.path().join("work");
90 std::fs::create_dir_all(&work).unwrap();
91 must(&work, &["init", "-q", "-b", "main"]);
92 std::fs::write(work.join("big.bin"), incompressible(BLOB_BYTES)).unwrap();
93 must(&work, &["add", "-A"]);
94 must(&work, &["commit", "-q", "-m", "large"]);
95 must(&work, &["push", "-q", bare.to_str().unwrap(), "main"]);
96 must(bare.as_path(), &["symbolic-ref", "HEAD", "refs/heads/main"]);
97 let tip = must(&work, &["rev-parse", "HEAD"]);
98
99 let addr = spawn(knot_pack::router(
100 layout,
101 serve_dids(),
102 std::sync::Arc::new(knot_runtime::SystemClock),
103 ))
104 .await;
105 let remote = format!("http://{addr}/{}", did.as_str());
106
107 Soak {
108 _scan: scan,
109 scratch,
110 remote,
111 tip,
112 }
113}
114
115async fn serve_wide_history() -> Soak {
116 let scan = tempfile::tempdir().unwrap();
117 let layout = Layout::new(scan.path());
118 let did = RepoDid::new("did:plc:squid").unwrap();
119 let repo = layout.create(&did).unwrap();
120 let tip = write_history(
121 &repo,
122 HistorySpec {
123 commits: CommitCount::new(256),
124 paths: PathCount::new(4096),
125 churn: ChurnCount::new(16),
126 },
127 )
128 .to_hex();
129
130 let scratch = tempfile::tempdir().unwrap();
131 let addr = spawn(knot_pack::router(
132 layout,
133 serve_dids(),
134 std::sync::Arc::new(knot_runtime::SystemClock),
135 ))
136 .await;
137 let remote = format!("http://{addr}/{}", did.as_str());
138
139 Soak {
140 _scan: scan,
141 scratch,
142 remote,
143 tip,
144 }
145}
146
147fn drain_storm(remote: &str, scratch: &Path, round: usize, concurrency: usize, tip: &str) -> u64 {
148 let dests: Vec<std::path::PathBuf> = (0..concurrency)
149 .map(|index| scratch.join(format!("clone-{round}-{index}")))
150 .collect();
151 let mut children: Vec<Child> = dests.iter().map(|dest| clone_child(remote, dest)).collect();
152
153 let mut peak = rss_bytes();
154 let mut pending = true;
155 while pending {
156 peak = peak.max(rss_bytes());
157 std::thread::sleep(Duration::from_millis(3));
158 pending = children
159 .iter_mut()
160 .any(|child| matches!(child.try_wait(), Ok(None)));
161 }
162
163 children.into_iter().enumerate().for_each(|(index, child)| {
164 let out = child.wait_with_output().unwrap();
165 assert!(
166 out.status.success(),
167 "round {round} clone {index} failed:\n{}",
168 String::from_utf8_lossy(&out.stderr)
169 );
170 });
171
172 dests.iter().for_each(|dest| {
173 assert_eq!(
174 must(dest, &["rev-parse", "HEAD"]),
175 tip,
176 "soak clone must reproduce repo tip"
177 );
178 must(dest, &["fsck", "--connectivity-only", "--no-progress"]);
179 std::fs::remove_dir_all(dest).unwrap();
180 });
181
182 peak.max(rss_bytes())
183}
184
185#[tokio::test(flavor = "multi_thread")]
186async fn concurrent_clone_memory_cost_curve() {
187 let _rss_gate = RSS_GATE.lock().await;
188 let soak = serve_wide_history().await;
189
190 let baseline = rss_bytes();
191 let mut peak = baseline;
192 let curve: Vec<(usize, u64)> = CURVE_LEVELS
193 .into_iter()
194 .map(|concurrency| {
195 peak = peak.max(drain_storm(
196 &soak.remote,
197 soak.scratch.path(),
198 concurrency,
199 concurrency,
200 &soak.tip,
201 ));
202 (concurrency, peak.saturating_sub(baseline))
203 })
204 .collect();
205
206 curve.iter().for_each(|(concurrency, delta)| {
207 let per_connection = delta / *concurrency as u64;
208 println!(
209 "{concurrency} concurrent clones: cumulative +{} MiB, ~{} MiB per connection",
210 delta / (1024 * 1024),
211 per_connection / (1024 * 1024)
212 );
213 assert!(
214 per_connection <= PER_CONNECTION_CEILING,
215 "per-connection high-water for {concurrency} clones is {} MiB, past {} MiB ceiling",
216 per_connection / (1024 * 1024),
217 PER_CONNECTION_CEILING / (1024 * 1024)
218 );
219 });
220
221 let (_, single) = curve[0];
222 let (top, top_delta) = *curve.last().unwrap();
223 let top_per_connection = top_delta / top as u64;
224 assert!(
225 top_per_connection <= single + SUBLINEAR_SLACK,
226 "memory grows super-linearly with concurrency. {top} clones cost {} MiB per connection \
227 against {} MiB for single clone, past the {} MiB slack. Pack is shared, so the \
228 per-connection high-water must stay flat as connections rise",
229 top_per_connection / (1024 * 1024),
230 single / (1024 * 1024),
231 SUBLINEAR_SLACK / (1024 * 1024)
232 );
233}
234
235#[tokio::test(flavor = "multi_thread")]
236async fn concurrent_clones_of_a_large_repo_stay_bounded() {
237 let _rss_gate = RSS_GATE.lock().await;
238 let soak = serve_large_repo().await;
239 let scratch = &soak.scratch;
240 let remote = soak.remote.clone();
241 let tip = soak.tip.clone();
242
243 let baseline = rss_bytes();
244 let mut peak = baseline;
245 let after_round: Vec<u64> = (0..ROUNDS)
246 .map(|round| {
247 peak = peak.max(drain_storm(
248 &remote,
249 scratch.path(),
250 round,
251 CONCURRENCY,
252 &tip,
253 ));
254 rss_bytes()
255 })
256 .collect();
257
258 assert!(
259 peak < OOM_CEILING,
260 "serving {CONCURRENCY} concurrent clones mustn't balloon resident memory: peak {} MiB exceeds {} MiB ceiling",
261 peak / (1024 * 1024),
262 OOM_CEILING / (1024 * 1024)
263 );
264
265 let settled = after_round[..ROUNDS - 1].iter().copied().max().unwrap();
266 let last = after_round[ROUNDS - 1];
267 assert!(
268 last <= settled + GROWTH_SLACK,
269 "resident memory is still climbing at final round, a leak: settled at {} MiB, round {ROUNDS} left {} MiB",
270 settled / (1024 * 1024),
271 last / (1024 * 1024)
272 );
273}