1//! A crowd on one shared notebook: a host, and guests that join it through a relay, half of
2//! them typing on the same page. Reports how long keys and carets take to reach the others,
3//! what the relay passes on, and what the relay, the host and the guests cost.
4//!
5//! `cargo run --release -p notebook --features live --example live_crowd -- RELAY_BINARY
6//! [PEERS=30] [TYPISTS=PEERS/2] [SECONDS=60]`, the relay's binary built with
7//! `cargo build --release -p relay`. The relay, the host and the guests each run as a process
8//! of their own, so `ps` tells their costs apart.
9//!
10//! For watching a crowd in the app: `fixture FOLDER PARAGRAPHS` writes the notebook the
11//! host shares, and `guests URL CODE PEERS SECONDS` has that many join the app's share of it
12//! and move their carets about its page.
13
14use notebook::{
15 Replica,
16 live::{
17 Caret, Guid, Hello, Presence, Spot,
18 share::{self, Guest, Host, Sharing},
19 },
20 session::{Background, Notebook, Section},
21};
22use onestore::{
23 ExGuid,
24 op::{Edit, Op, PageOp},
25};
26use std::{
27 collections::HashMap,
28 io::{BufRead, BufReader, Read, Write},
29 net::TcpStream,
30 path::Path,
31 process::{Child, Command, Stdio},
32 sync::{
33 Arc, Mutex,
34 atomic::{AtomicBool, Ordering},
35 },
36 thread,
37 time::{Duration, Instant},
38};
39
40/// Keys each typist types a second, and how often observers look.
41const KEYS_PER_SECOND: f64 = 5.0;
42const LOOK: Duration = Duration::from_millis(2);
43/// Guests that time what they see.
44const OBSERVERS: usize = 3;
45
46fn until(what: &str, limit: Duration, done: impl Fn() -> bool) {
47 let deadline = Instant::now() + limit;
48 while !done() {
49 assert!(Instant::now() < deadline, "{what}");
50 thread::sleep(Duration::from_millis(5));
51 }
52}
53
54fn hello(name: &str) -> Hello {
55 Hello::new(name.into(), None).unwrap()
56}
57
58fn main() {
59 let args: Vec<String> = std::env::args().skip(1).collect();
60 match args.first().map(String::as_str) {
61 Some("host") => return host(&args[1], args[2].parse().unwrap()),
62 Some("fixture") => {
63 fixture(Path::new(&args[1]), args[2].parse().unwrap());
64 return;
65 }
66 Some("guests") => {
67 let seconds = Duration::from_secs(args[4].parse().unwrap());
68 return guests(&args[1], &args[2], args[3].parse().unwrap(), seconds);
69 }
70 _ => {}
71 }
72 let binary = args.first().expect("the relay's binary");
73 let number = |at: usize, default: usize| args.get(at).map_or(default, |n| n.parse().unwrap());
74 let peers = number(1, 30);
75 let typists = number(2, peers / 2);
76 let seconds = number(3, 60) as u64;
77 crowd(binary, peers, typists, Duration::from_secs(seconds));
78}
79
80/// Writes the notebook `Garden` in `directory`: a section whose page has `paragraphs`
81/// paragraphs, `Typed:` then `Note 1:` and on. Its folder.
82fn fixture(directory: &Path, paragraphs: usize) -> std::path::PathBuf {
83 let folder = directory.join("Garden");
84 std::fs::create_dir_all(&folder).unwrap();
85 let mut image = onestore::create_section("Garden.one", "Typed:", "Fixture").unwrap();
86 for n in (1..paragraphs).rev() {
87 let arena = onestore::Arena::default();
88 let mut section = onestore::Section::open(&arena, image).unwrap();
89 let (space, ..) = section.pages().unwrap()[0];
90 let page = section.page(space).unwrap();
91 let (text, _) = texts(&page).swap_remove(0);
92 let right = onestore::page::text::new_id().unwrap();
93 let ops = [
94 PageOp::Split {
95 text,
96 at: 6,
97 paragraph: onestore::page::text::new_id().unwrap(),
98 right,
99 lists: Vec::new(),
100 },
101 PageOp::Text {
102 text: right,
103 range: 0..0,
104 with: format!("Note {n}:"),
105 },
106 ];
107 let edit = Edit {
108 at: 134_000_000_000_000_000,
109 ops: ops.into_iter().map(|op| Op::Page { space, op }).collect(),
110 };
111 section.apply("Fixture", &edit).unwrap();
112 section.seal().unwrap();
113 image = section.image();
114 }
115 std::fs::write(folder.join("Garden.one"), image).unwrap();
116 folder
117}
118
119/// The host's process: shares a fixture notebook of a page with `paragraphs` paragraphs
120/// through `url`, says its code, and reports what the section grew to once its input closes.
121fn host(url: &str, paragraphs: usize) {
122 let directory = tempfile::tempdir().unwrap();
123 let folder = fixture(directory.path(), paragraphs);
124 let file = folder.join("Garden.one");
125 let storage = Notebook::open(&folder, directory.path().join("cache"))
126 .unwrap()
127 .into_storage();
128 let host = Host::start(
129 storage,
130 hello("Host"),
131 Sharing::new("").unwrap(),
132 "Garden",
133 None,
134 Some(url),
135 || {},
136 |_| Ok(()),
137 )
138 .unwrap();
139 until("no code", Duration::from_secs(30), || {
140 host.code().is_some_and(|code| share::code(&code).is_some())
141 });
142 println!("code {}", host.code().unwrap());
143 let _ = std::io::stdin().read_to_end(&mut Vec::new());
144 let image = std::fs::read(&file).unwrap();
145 if let Some(output) = std::env::var_os("SNOWBOUND_LIVE_EVIDENCE") {
146 let output = Path::new(&output);
147 std::fs::create_dir_all(output).unwrap();
148 std::fs::copy(&file, output.join("Garden.one")).unwrap();
149 }
150 let size = image.len();
151 let arena = onestore::Arena::default();
152 let mut section = onestore::Section::open(&arena, image).unwrap();
153 let pages = section.pages().unwrap().len();
154 let conflicts: usize = section
155 .conflicts()
156 .unwrap()
157 .iter()
158 .map(|(_, c)| c.len())
159 .sum();
160 println!("section {size} bytes, {pages} pages, {conflicts} conflict pages");
161}
162
163/// A guest as the app holds one: its share, the notebook's background, the section open
164/// and its file's identity.
165struct Open {
166 guest: Arc<Guest>,
167 _background: Background,
168 section: Section,
169 file: [u8; 16],
170}
171
172fn join(name: &str, code: &str, url: &str, cache: &Path) -> Open {
173 let welcome = share::join(hello(name), code, "", None, Some(url)).unwrap();
174 let guest = Guest::start(
175 hello(name),
176 welcome.share,
177 welcome.secret,
178 None,
179 Some(url),
180 || {},
181 )
182 .unwrap();
183 until("no host", Duration::from_secs(120), || {
184 guest.host().is_some()
185 });
186 let mut notebook = Notebook::open_hosted(Arc::clone(&guest), cache).unwrap();
187 let background = Background::hosted(Arc::clone(&guest), || {}).unwrap();
188 background.watch(notebook.replicas());
189 let replica = notebook.replica_path("Garden.one").unwrap();
190 std::fs::create_dir_all(replica.parent().unwrap()).unwrap();
191 let image = notebook.read_section("Garden.one").unwrap();
192 let file = onestore::Header::parse(&image[..1024]).unwrap().file_id;
193 let replica = Replica::open_or_create(&replica, None, || Ok(image)).unwrap();
194 let section = Section::resume_hosted(
195 "Garden.one".into(),
196 replica,
197 Arc::clone(&guest),
198 false,
199 || {},
200 )
201 .unwrap();
202 background.hold("Garden.one", &section);
203 Open {
204 guest,
205 _background: background,
206 section,
207 file,
208 }
209}
210
211/// Where in the open page's `paragraph`th text, `offset` units in, `open`'s guest is.
212fn presence(open: &Open, paragraph: usize, offset: u32) -> Option<Presence> {
213 let (space, texts) = texts_of(&open.section)?;
214 let (text, _) = texts.get(paragraph)?;
215 let at = Spot {
216 text: Guid {
217 guid: text.guid,
218 n: text.n,
219 },
220 offset,
221 };
222 Some(Presence {
223 section: Some(open.file),
224 page: Some(Guid {
225 guid: space.guid,
226 n: space.n,
227 }),
228 caret: Some(Caret {
229 anchor: at,
230 focus: at,
231 }),
232 })
233}
234
235/// `peers` guests join the share `code` names through `url` and, for `seconds`, put their
236/// carets in the page's paragraphs, each moving every few seconds.
237fn guests(url: &str, code: &str, peers: usize, seconds: Duration) {
238 let directory = tempfile::tempdir().unwrap();
239 let names = [
240 "Ada", "Grace", "Alan", "Barbara", "Edsger", "Frances", "Donald", "Margaret", "Ken",
241 "Radia", "Dennis", "Hedy", "Linus", "Karen", "John", "Sophie",
242 ];
243 let joining: Vec<_> = (0..peers)
244 .map(|n| {
245 let name = format!("{} {n}", names[n % names.len()]);
246 let (code, url, cache) = (
247 code.to_owned(),
248 url.to_owned(),
249 directory.path().join(n.to_string()),
250 );
251 thread::sleep(Duration::from_millis(100));
252 thread::spawn(move || join(&name, &code, &url, &cache))
253 })
254 .collect();
255 let guests: Vec<Open> = joining
256 .into_iter()
257 .map(|joined| joined.join().unwrap())
258 .collect();
259 let began = Instant::now();
260 let mut turn = 0;
261 while began.elapsed() < seconds {
262 for (n, open) in guests.iter().enumerate() {
263 let paragraphs = texts_of(&open.section).map_or(1, |(_, texts)| texts.len());
264 if (n + turn) % 4 == 0 || turn == 0 {
265 let at = (n * 7 + turn) % paragraphs;
266 if let Some(presence) = presence(open, at, (turn % 5) as u32) {
267 open.guest.set_presence(presence);
268 }
269 }
270 }
271 turn += 1;
272 thread::sleep(Duration::from_secs(1));
273 }
274}
275
276/// The page's object space and its paragraphs' texts: each one's object and what it reads.
277fn texts_of(section: &Section) -> Option<(ExGuid, Vec<(ExGuid, String)>)> {
278 let (space, ..) = *section.pages().ok()?.first()?;
279 let page = section.page(space).ok()?;
280 Some((space, texts(&page)))
281}
282
283fn texts(page: &onestore::page::Page) -> Vec<(ExGuid, String)> {
284 let outlines = page.objects.iter().filter_map(|object| match object {
285 onestore::page::PageObject::Outline(outline) => Some(outline),
286 _ => None,
287 });
288 outlines
289 .flat_map(|outline| &outline.paragraphs)
290 .filter_map(|p| p.text().map(|t| (t.id, t.text.text().to_owned())))
291 .collect()
292}
293
294/// The key typist `typist` types `n`th, unique to both.
295fn key(typist: usize, n: usize) -> char {
296 char::from_u32(0x4E00 + (typist * 600 + n) as u32).unwrap()
297}
298
299fn typist_of(key: char) -> Option<(usize, usize)> {
300 let at = (key as u32).checked_sub(0x4E00)? as usize;
301 Some((at / 600, at % 600))
302}
303
304/// Seconds of CPU and kilobytes resident `ps` reports for `pid`.
305fn cost(pid: u32) -> (f64, u64) {
306 let output = Command::new("ps")
307 .args(["-o", "time=,rss=", "-p", &pid.to_string()])
308 .output()
309 .unwrap();
310 let text = String::from_utf8_lossy(&output.stdout);
311 let mut words = text.split_whitespace();
312 let time = words.next().unwrap_or("0:0");
313 let seconds = time.split(':').fold(0.0, |total, part| {
314 total * 60.0 + part.parse::<f64>().unwrap_or(0.0)
315 });
316 (
317 seconds,
318 words.next().and_then(|rss| rss.parse().ok()).unwrap_or(0),
319 )
320}
321
322/// The relay's `/health`, as numbers by name.
323fn health(address: &str) -> HashMap<String, f64> {
324 let mut stream = TcpStream::connect(address).unwrap();
325 write!(
326 stream,
327 "GET /health HTTP/1.1\r\nHost: relay\r\nConnection: close\r\n\r\n"
328 )
329 .unwrap();
330 let mut text = String::new();
331 stream.read_to_string(&mut text).unwrap();
332 let body = text.split("\r\n\r\n").nth(1).unwrap_or_default();
333 body.trim()
334 .trim_matches(['{', '}'])
335 .split(',')
336 .filter_map(|pair| {
337 let (name, value) = pair.split_once(':')?;
338 Some((name.trim_matches('"').to_owned(), value.parse().ok()?))
339 })
340 .collect()
341}
342
343fn report(what: &str, mut times: Vec<Duration>) {
344 if times.is_empty() {
345 println!("{what}: none seen");
346 return;
347 }
348 times.sort();
349 let ms = |at: usize| times[at].as_secs_f64() * 1000.0;
350 println!(
351 "{what}: median {:.0} ms, p90 {:.0} ms, p99 {:.0} ms, worst {:.0} ms ({} seen)",
352 ms(times.len() / 2),
353 ms(times.len() * 9 / 10),
354 ms(times.len() * 99 / 100),
355 ms(times.len() - 1),
356 times.len()
357 );
358}
359
360struct Spawned(Child);
361
362impl Drop for Spawned {
363 fn drop(&mut self) {
364 let _ = self.0.kill();
365 let _ = self.0.wait();
366 }
367}
368
369fn crowd(binary: &str, peers: usize, typists: usize, window: Duration) {
370 let port = std::net::TcpListener::bind("127.0.0.1:0")
371 .unwrap()
372 .local_addr()
373 .unwrap()
374 .port();
375 let address = format!("127.0.0.1:{port}");
376 // Everyone comes from one address here, which a relay would take for one abuser.
377 let relay = Spawned(
378 Command::new(binary)
379 .args(["--listen", &address])
380 .args(["--max-connections-per-address", "1000"])
381 .args(["--joins-per-minute", "10000"])
382 .args(["--failures-per-minute", "10000"])
383 .stdout(Stdio::null())
384 .spawn()
385 .unwrap(),
386 );
387 until("no relay", Duration::from_secs(10), || {
388 TcpStream::connect(&address).is_ok()
389 });
390 let url = format!("ws://{address}");
391 let mut hosting = Spawned(
392 Command::new(std::env::current_exe().unwrap())
393 .args(["host", &url, &typists.max(1).to_string()])
394 .stdin(Stdio::piped())
395 .stdout(Stdio::piped())
396 .spawn()
397 .unwrap(),
398 );
399 let mut said = BufReader::new(hosting.0.stdout.take().unwrap()).lines();
400 let code = said.next().unwrap().unwrap();
401 let code = code.strip_prefix("code ").unwrap().to_owned();
402 println!("{peers} guests, {typists} typing {KEYS_PER_SECOND} keys a second, code {code}");
403
404 let directory = tempfile::tempdir().unwrap();
405 let began = Instant::now();
406 let joining: Vec<_> = (0..peers)
407 .map(|n| {
408 let (code, url) = (code.clone(), url.clone());
409 let cache = directory.path().join(format!("guest{n}"));
410 thread::sleep(Duration::from_millis(50));
411 thread::spawn(move || join(&format!("Guest {n}"), &code, &url, &cache))
412 })
413 .collect();
414 let guests: Vec<Arc<Open>> = joining
415 .into_iter()
416 .map(|joined| Arc::new(joined.join().unwrap()))
417 .collect();
418 println!("all joined in {:.1} s", began.elapsed().as_secs_f64());
419 until("not everyone met", Duration::from_secs(300), || {
420 guests.iter().all(|open| open.guest.peers().len() >= peers)
421 });
422 println!("all met in {:.1} s", began.elapsed().as_secs_f64());
423
424 let (_, texts) = texts_of(&guests[0].section).unwrap();
425 assert!(texts.len() >= typists, "a paragraph for each typist");
426 for open in &guests {
427 open.guest.set_presence(presence(open, 0, 0).unwrap());
428 }
429
430 // When each typist typed each key, and said its caret moved past it.
431 let typed: Arc<Mutex<HashMap<(usize, usize), Instant>>> = Arc::default();
432 let moved: Arc<Mutex<HashMap<(usize, usize), Instant>>> = Arc::default();
433 let seen_keys: Arc<Mutex<Vec<Duration>>> = Arc::default();
434 let seen_carets: Arc<Mutex<Vec<Duration>>> = Arc::default();
435 let done = Arc::new(AtomicBool::new(false));
436 let (relay_before, host_before, crowd_before) = (
437 cost(relay.0.id()),
438 cost(hosting.0.id()),
439 cost(std::process::id()),
440 );
441 let bytes_before = health(&address);
442 let started = Instant::now();
443
444 let mut threads = Vec::new();
445 for (n, open) in guests.iter().enumerate().take(typists) {
446 let (open, typed, done) = (Arc::clone(open), Arc::clone(&typed), Arc::clone(&done));
447 let moved = Arc::clone(&moved);
448 threads.push(thread::spawn(move || {
449 let pause = Duration::from_secs_f64(1.0 / KEYS_PER_SECOND);
450 thread::sleep(pause.mul_f64(n as f64 / typists as f64));
451 let mut count = 0;
452 while !done.load(Ordering::Acquire) && count < 600 {
453 open.section.events();
454 let Some((space, mut texts)) = texts_of(&open.section) else {
455 continue;
456 };
457 let (text, before) = texts.swap_remove(n);
458 let end = before.encode_utf16().count() as u32;
459 let edit = Edit {
460 at: 134_000_000_000_000_000,
461 ops: vec![Op::Page {
462 space,
463 op: PageOp::Text {
464 text,
465 range: end..end,
466 with: key(n, count).to_string(),
467 },
468 }],
469 };
470 typed.lock().unwrap().insert((n, count), Instant::now());
471 open.section.replica().apply("Typist", edit).unwrap();
472 count += 1;
473 if let Some(presence) = presence(&open, 0, count as u32) {
474 moved.lock().unwrap().insert((n, count), Instant::now());
475 open.guest.set_presence(presence);
476 }
477 thread::sleep(pause);
478 }
479 }));
480 }
481 for open in guests.iter().skip(typists).take(OBSERVERS) {
482 let (open, typed, done) = (Arc::clone(open), Arc::clone(&typed), Arc::clone(&done));
483 let moved = Arc::clone(&moved);
484 let (keys, carets) = (Arc::clone(&seen_keys), Arc::clone(&seen_carets));
485 threads.push(thread::spawn(move || {
486 let mut keys_seen = std::collections::HashSet::new();
487 let mut carets_seen = HashMap::new();
488 while !done.load(Ordering::Acquire) {
489 open.section.events();
490 let now = Instant::now();
491 if let Some((_, texts)) = texts_of(&open.section) {
492 let chars = texts.iter().flat_map(|(_, text)| text.chars());
493 for (typist, n) in chars.filter_map(typist_of) {
494 if keys_seen.insert((typist, n))
495 && let Some(at) = typed.lock().unwrap().get(&(typist, n))
496 {
497 keys.lock().unwrap().push(now - *at);
498 }
499 }
500 }
501 for peer in open.guest.peers() {
502 let Some(typist) = (peer.hello.name.strip_prefix("Guest "))
503 .and_then(|n| n.parse::<usize>().ok())
504 .filter(|n| *n < typists)
505 else {
506 continue;
507 };
508 let Some(count) = peer.presence.and_then(|p| p.caret).map(|c| c.focus.offset)
509 else {
510 continue;
511 };
512 let count = count as usize;
513 if count > 0
514 && carets_seen.insert(typist, count) != Some(count)
515 && let Some(at) = moved.lock().unwrap().get(&(typist, count))
516 {
517 carets.lock().unwrap().push(now - *at);
518 }
519 }
520 thread::sleep(LOOK);
521 }
522 }));
523 }
524 thread::sleep(window);
525 done.store(true, Ordering::Release);
526 let elapsed = started.elapsed().as_secs_f64();
527 let (relay_after, host_after, crowd_after) = (
528 cost(relay.0.id()),
529 cost(hosting.0.id()),
530 cost(std::process::id()),
531 );
532 let bytes_after = health(&address);
533 for thread in threads {
534 thread.join().unwrap();
535 }
536 let keys = typed.lock().unwrap().len();
537 report(
538 "keys, typist to observer",
539 std::mem::take(&mut seen_keys.lock().unwrap()),
540 );
541 report(
542 "carets, typist to observer",
543 std::mem::take(&mut seen_carets.lock().unwrap()),
544 );
545 let delta = |name: &str| bytes_after[name] - bytes_before.get(name).copied().unwrap_or(0.0);
546 let members = (peers + 1) as f64;
547 println!(
548 "relay passed on {:.0} KB/s in, {:.0} KB/s out: per peer {:.1} KB/s in, {:.1} KB/s out",
549 delta("bytes_in") / elapsed / 1000.0,
550 delta("bytes_out") / elapsed / 1000.0,
551 delta("bytes_in") / elapsed / 1000.0 / members,
552 delta("bytes_out") / elapsed / 1000.0 / members,
553 );
554 let load = |name: &str, (before, _): (f64, u64), (after, rss): (f64, u64)| {
555 println!(
556 "{name}: {:.0}% of a core, {:.1} MB resident",
557 (after - before) / elapsed * 100.0,
558 rss as f64 / 1024.0
559 );
560 };
561 load("relay", relay_before, relay_after);
562 load("host", host_before, host_after);
563 load(&format!("{peers} guests"), crowd_before, crowd_after);
564
565 // Every key typed should reach the host's file; conflict pages show as extra pages.
566 thread::sleep(Duration::from_secs(5));
567 let observed = texts_of(&guests[typists.min(peers - 1)].section).map_or(0, |(_, texts)| {
568 let chars = texts.iter().flat_map(|(_, text)| text.chars());
569 chars.filter_map(typist_of).count()
570 });
571 println!("{keys} keys typed, {observed} on an observer's page after 5 s");
572 drop(hosting.0.stdin.take());
573 for line in said.map_while(Result::ok) {
574 println!("host: {line}");
575 }
576}