diff --git a/Cargo.lock b/Cargo.lock index 0163a2b886..699863fa02 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1131,6 +1131,7 @@ dependencies = [ "toyos-gpt", "toyos-keymap", "toyos-ld", + "toyos-logstream", "toyos-manifest", "toyos-sched", "toyos-tco", @@ -1208,6 +1209,13 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "toyos-logstream" +version = "0.1.0" +dependencies = [ + "toyos-abi", +] + [[package]] name = "toyos-manifest" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 283735df1a..b3ed8e92d5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ members = [ "toyos-i219", "toyos-keymap", "toyos-ld", + "toyos-logstream", "toyos-manifest", "toyos-mixer", "toyos-pci", @@ -110,6 +111,9 @@ toyos-fat32 = { path = "toyos-fat32" } toyos-fat32-check = { path = "toyos-fat32-check" } toyos-gpt = { path = "toyos-gpt" } toyos-keymap = { path = "toyos-keymap" } +# The record stream's boot parameter, so the gate that clears a valued +# parameter by name reads the same constant the kernel and `logd` do. +toyos-logstream = { path = "toyos-logstream" } toyos-manifest = { path = "toyos-manifest" } # The watchdog's parameter name, so the gate over `kernel/src/params.rs` reads # the same constant the kernel and the bootloader do. diff --git a/issues/design-debt/a-boot-with-no-log-volume-streams-nothing.md b/issues/design-debt/a-boot-with-no-log-volume-streams-nothing.md new file mode 100644 index 0000000000..b7e63f65e6 --- /dev/null +++ b/issues/design-debt/a-boot-with-no-log-volume-streams-nothing.md @@ -0,0 +1,24 @@ +--- +status: open +kind: defect +opened: 2026-09-08 +--- + +# A boot with no `/log` streams nothing, and that is backwards for the bench + +`userland/logd/src/main.rs`'s loop offers the record stream only what it has +already written to the volume, and it reaches that offer through +`let Some(v) = volume.as_mut() else { continue }`. So a boot with no log +partition, or one whose volume `policy::fate` has given up on, streams not one +record — the second sink is a mirror of the first and dies with it. + +That is right for the rule it comes from (the file is the sink of record and the +stream may never cost it a line) and wrong for the machine the stream was built +for. On the bench's ThinkPad a dead stick is precisely when the cable is the +only channel left, and it is the case where the stream is worth most. + +The exit condition is a decision about what `logd` offers when there is nothing +to write to: the records it read, or nothing. Whichever it is, the ordering rule +the stream rests on — a line reaches the file before it reaches the wire — has +to be restated for a boot where there is no file, because today it is what makes +the answer "nothing" by construction. diff --git a/issues/design-debt/logd-holds-a-netd-connector-on-every-boot.md b/issues/design-debt/logd-holds-a-netd-connector-on-every-boot.md new file mode 100644 index 0000000000..05513d679c --- /dev/null +++ b/issues/design-debt/logd-holds-a-netd-connector-on-every-boot.md @@ -0,0 +1,23 @@ +--- +status: open +kind: defect +opened: 2026-09-08 +--- + +# `logd` holds a `netd` connector on every boot, and streams on almost none + +`system.toml`'s `[programs.logd]` carries `receives = ["netd"]` on every image +this tree builds. The record stream it is for is armed by a boot parameter +(`logstream=`), which no shipping boot carries, so the one process that reads +every record every CPU wrote holds an outbound network connector for the whole +life of a machine that will never open a connection. + +`init` builds a program's endowment from its manifest row before it spawns it +and has no way to make a row conditional on the parameter line, so the authority +is static while the feature is not. `sshd` is kept out of `[boot] start` +entirely for a weaker version of the same argument. + +The exit condition is one of: `init` learns to grant a connector only when the +boot asked for what it is for; or the manifest gains a way to say "this row is +armed by this parameter"; or the tree decides an unopened connector is not +authority worth withholding and this file is closed by that ruling. diff --git a/issues/design-debt/the-parameter-lines-value-is-read-in-two-crates.md b/issues/design-debt/the-parameter-lines-value-is-read-in-two-crates.md new file mode 100644 index 0000000000..d93de587b6 --- /dev/null +++ b/issues/design-debt/the-parameter-lines-value-is-read-in-two-crates.md @@ -0,0 +1,23 @@ +--- +status: open +kind: defect +opened: 2026-09-08 +--- + +# The boot parameter line's value is read in two crates + +`toyos-abi/src/boot.rs` reads the kernel command line twice — `root_uuid` for +`root=` and `actuators` for the token list — and +`toyos-logstream/src/lib.rs`'s `value_in` is a third reading with `root_uuid`'s +body: `cmdline.split(',').find_map(|t| t.strip_prefix(P))`. There is one +question there and it is asked in two crates. + +The reading belongs in `toyos_abi::boot` beside its two neighbours, as one +function taking the prefix, with `root_uuid` and `value_in` both calls to it. +What stopped that here is the branch rule: a commit touching `toyos-abi/src` may +not share a branch with anything else, and the record stream is not an ABI +change. + +The exit condition is `toyos_abi::boot` growing that one function on an ABI +branch of its own, and `toyos-logstream` losing `value_in` in the branch that +follows it. diff --git a/issues/hardware/the-t14-answers-only-through-a-usb-stick.md b/issues/hardware/the-t14-answers-only-through-a-usb-stick.md index ede4532503..467be96774 100644 --- a/issues/hardware/the-t14-answers-only-through-a-usb-stick.md +++ b/issues/hardware/the-t14-answers-only-through-a-usb-stick.md @@ -47,5 +47,15 @@ Constraints a reader would otherwise pay to re-derive: hand-over, so a function still holding its last holder's queue addresses can act on none of them. `release` asks for a reset where the function advertises one; the I219 does, so on the T14 both hold. +- **The record stream is `logstream=:` on the parameter line**, + copied by the kernel into `/system/bin/init`'s environment and read from there + by `logd` (`toyos-logstream`'s `PARAM` and `ENV`). What is left to build is the + metal half: arming the flashed image with the Mac's address and listening while + the T14 boots. A boot that dies before `logd` runs still needs the stick. +- **A stalled peer's backpressure reaches `logd`'s queue only after megabytes.** + Between them stand a 2 MiB kernel pipe (`kernel/src/pipe.rs`'s `PIPE_SIZE`) and + netd's 64 KiB send buffer, and a `log-storm` at `--smp 8` produces 4,213 lines + / 674 KiB — measured — which they absorb entirely. The guest arm for that path + widens the storm's records instead of narrowing the peer (`log-storm-wide`). - The metal loop is `toyos-metal` (`src/metal.rs`), and the T14 is run by the orchestrator alone. diff --git a/issues/kernel/netd-drops-what-the-clients-pipe-would-not-take.md b/issues/kernel/netd-drops-what-the-clients-pipe-would-not-take.md new file mode 100644 index 0000000000..0933630a25 --- /dev/null +++ b/issues/kernel/netd-drops-what-the-clients-pipe-would-not-take.md @@ -0,0 +1,33 @@ +--- +status: open +kind: defect +opened: 2026-09-08 +--- + +# netd drops what arrives off the wire when the client's pipe will not take it + +`userland/netd/src/main.rs`'s `bridge_piped` moves what the wire delivered into +the client's rx pipe like this: + +```rust +Ok(n) if n > 0 => { + let _ = toyos_abi::syscall::write_nonblock(pipe.as_handle(), &buf[..n]); +} +``` + +`write_nonblock` answers **how many bytes it took**, and it takes fewer than it +was offered when the pipe is short of room. The return value is discarded, so +those bytes are gone: `recv_slice` has already consumed them from the socket, +the client is never told, and the stream it reads is short in the middle with +nothing anywhere saying so. A client slower than the wire is exactly when it +fires. + +The send direction had the same shape and no longer does: it now takes out of +the pipe only what the socket has room for, so nothing is consumed from one side +without landing on the other. The same answer does not fit here yet — it needs +the pipe's remaining room, which no syscall answers today. + +Reproduced on the send side by `tests/common/logstream.rs`'s +`log_stream_stalled_peer_wide_storm`: against a peer that had stopped reading, +a record arrived cut in half with the next record's line beginning inside it. +That arm is the reproduction to point this half at once the room is askable. diff --git a/kernel/Cargo.lock b/kernel/Cargo.lock index c46e0a2e5f..bfc44e656c 100644 --- a/kernel/Cargo.lock +++ b/kernel/Cargo.lock @@ -45,6 +45,7 @@ dependencies = [ "toyos-fat32", "toyos-gpt", "toyos-hda", + "toyos-logstream", "toyos-pci", "toyos-pcid", "toyos-proclife", @@ -109,6 +110,13 @@ version = "0.1.0" name = "toyos-hda" version = "0.1.0" +[[package]] +name = "toyos-logstream" +version = "0.1.0" +dependencies = [ + "toyos-abi", +] + [[package]] name = "toyos-pci" version = "0.1.0" diff --git a/kernel/Cargo.toml b/kernel/Cargo.toml index f599e826f3..b65dc54200 100644 --- a/kernel/Cargo.toml +++ b/kernel/Cargo.toml @@ -387,6 +387,7 @@ toyos-fat32 = { path = "../toyos-fat32" } toyos-elf = { path = "../toyos-elf" } toyos-gpt = { path = "../toyos-gpt" } toyos-hda = { path = "../toyos-hda" } +toyos-logstream = { path = "../toyos-logstream" } toyos-pci = { path = "../toyos-pci" } toyos-pcid = { path = "../toyos-pcid" } toyos-tco = { path = "../toyos-tco" } diff --git a/kernel/src/actuator.rs b/kernel/src/actuator.rs index 8d4e0adc4c..3919e2b144 100644 --- a/kernel/src/actuator.rs +++ b/kernel/src/actuator.rs @@ -248,6 +248,9 @@ actuators! { /// Have every CPU emit patterned log records at once from spawned kernel threads. log_storm = "log-storm"; + /// Widen every storm record to nearly a whole record's message, so one boot offers a stalled log stream more than the buffers under it can hold. + log_storm_wide = "log-storm-wide"; + /// Remove the IF/TF bracket around shard selection through publication — the negative control on the log's interrupt-atomicity claim. log_unbracketed_reserve = "log-unbracketed-reserve"; diff --git a/kernel/src/loader/mod.rs b/kernel/src/loader/mod.rs index 38a4fc0632..9b757cdec7 100644 --- a/kernel/src/loader/mod.rs +++ b/kernel/src/loader/mod.rs @@ -915,7 +915,15 @@ pub fn spawn_init() -> Pid { }], label.as_bytes().to_vec(), ); - match spawn(&[INIT_PATH], PendingHandles::Ready(handles, endowments), String::from("/"), Vec::new()) { + // **The only channel a boot parameter has to userland**, and the one thing + // the kernel ever puts in an environment: the command line reaches no + // process. It is information and not authority — reaching the address needs + // a `netd` connector, which one manifest row grants. + let env = match crate::params::log_stream() { + Some(at) => alloc::format!("{}={at}\0", toyos_logstream::ENV).into_bytes(), + None => Vec::new(), + }; + match spawn(&[INIT_PATH], PendingHandles::Ready(handles, endowments), String::from("/"), env) { Ok(object) => object.pid(), Err(crate::object::Refusal::Error(e)) => panic!("spawn_init: failed to spawn: {e:?}"), Err(crate::object::Refusal::Handle(e)) => panic!("spawn_init: {e}"), diff --git a/kernel/src/log/storm.rs b/kernel/src/log/storm.rs index 7289c3c6b4..ef390ae5f5 100644 --- a/kernel/src/log/storm.rs +++ b/kernel/src/log/storm.rs @@ -10,6 +10,15 @@ const STORM_RECORDS: u64 = 1024; // Must exceed one machine word: a single-store payload couldn't reveal a torn write. const PAYLOAD: usize = 96; +/// What `log-storm-wide` widens the payload to: a record's whole message less +/// the `t=`/`i=`/`k=` fields in front of it. +/// +/// A storm of these is what a guest offers a log stream whose peer has stopped +/// reading and cannot outrun with narrow records — the buffers between `logd` +/// and that peer hold megabytes. The reader regenerates a payload from `t=` and +/// `i=`, so the log gate does not run on a boot carrying this. +const WIDE_PAYLOAD: usize = 900; + /// Deterministic checksum of `thread` and `index`, embedded in a record's `k=` field. pub fn checksum(thread: u64, index: u64) -> u64 { (thread.wrapping_mul(0x9E37_79B9_7F4A_7C15) ^ index.wrapping_mul(0xC2B2_AE3D_27D4_EB4F)) @@ -25,12 +34,13 @@ pub fn payload_byte(checksum: u64, offset: usize) -> u8 { /// The reader regenerates this text independently from `t=`/`i=`, so the format here must stay in sync with it. pub fn emit_patterned(thread: u64, index: u64) { let checksum = checksum(thread, index); - let mut payload = [0u8; PAYLOAD]; - for (offset, byte) in payload.iter_mut().enumerate() { + let width = if crate::actuator::log_storm_wide() { WIDE_PAYLOAD } else { PAYLOAD }; + let mut payload = [0u8; WIDE_PAYLOAD]; + for (offset, byte) in payload[..width].iter_mut().enumerate() { *byte = payload_byte(checksum, offset); } // Fallback rather than `expect`: a panic here would halt the machine over the producer's own formatting. - let payload = core::str::from_utf8(&payload).unwrap_or(""); + let payload = core::str::from_utf8(&payload[..width]).unwrap_or(""); crate::log!("logstorm t={thread} i={index} k={checksum:016x} {payload}"); } diff --git a/kernel/src/params.rs b/kernel/src/params.rs index d2edb296ed..f62dcfbac4 100644 --- a/kernel/src/params.rs +++ b/kernel/src/params.rs @@ -2,9 +2,15 @@ //! other kind of token and is test-only, so a kernel built without them refuses //! every one it is handed; a name here is claimed before that table sees it, //! and is the only way an image the owner flashes asks for anything. +//! +//! Two kinds live here: a flag, which is a name [`PARAMS`] matches whole, and a +//! parameter carrying a value after it, which [`claims`] matches as a prefix +//! and no table holds. use core::sync::atomic::{AtomicBool, Ordering}; +use crate::sync::Lock; + /// Each parameter beside the flag it sets, so a name cannot be claimed and then handled by nothing. pub const PARAMS: &[(&str, &AtomicBool)] = &[(toyos_tco::PARAM, &WATCHDOG_NAMED), ("early-panel", &EARLY_PANEL_NAMED)]; @@ -13,25 +19,68 @@ static WATCHDOG_NAMED: AtomicBool = AtomicBool::new(false); static EARLY_PANEL_NAMED: AtomicBool = AtomicBool::new(false); static PARSED: AtomicBool = AtomicBool::new(false); +/// [`toyos_logstream::PARAM`]'s value, **copied** and not borrowed. +/// +/// The parameter line is in memory no reserved region covers, so `mm::init` may +/// hand it out; nothing may hold a borrow of it past [`init`]. There is also no +/// allocator yet, so the copy goes into a fixed buffer rather than a `String`. +static LOG_STREAM: Lock<([u8; toyos_logstream::MAX_VALUE_BYTES], usize)> = + Lock::new(([0; toyos_logstream::MAX_VALUE_BYTES], 0)); + pub fn init(cmdline: &str) { for token in toyos_abi::boot::actuators(cmdline) { if let Some((_, named)) = PARAMS.iter().find(|(name, _)| *name == token) { named.store(true, Ordering::Relaxed); } } + if let Some(value) = toyos_logstream::value_in(cmdline) { + // A value this buffer cannot hold is refused whole rather than + // truncated: half an address is an address, and it is somebody else's. + if value.len() > toyos_logstream::MAX_VALUE_BYTES { + crate::log!( + "log-stream: {}{value:?} is {} bytes and an address is at most {}", + toyos_logstream::PARAM, + value.len(), + toyos_logstream::MAX_VALUE_BYTES + ); + } else { + let mut held = LOG_STREAM.lock(); + held.0[..value.len()].copy_from_slice(value.as_bytes()); + held.1 = value.len(); + } + } PARSED.store(true, Ordering::Relaxed); } /// Whether this kernel handles `token` itself, which is what stops /// `actuator::init` refusing it as a name it does not know. /// -/// **The one parameter that carries a value is not in [`PARAMS`] and is not -/// read here**: the black-box page's address is read out of the raw buffer in -/// `kernel_main`'s first statements (`crate::blackbox::arm`), because a panic -/// before this function runs still has to be able to seal. All this does is stop -/// the actuator table refusing a word it does not know. +/// **Neither parameter that carries a value is in [`PARAMS`]**, which is a +/// table of flags matched whole; both are prefixes matched here. The black-box +/// page's address is not read by [`init`] either — it comes out of the raw +/// buffer in `kernel_main`'s first statements (`crate::blackbox::arm`), because +/// a panic before [`init`] runs still has to be able to seal. The log stream's +/// address is read by [`init`], because nothing before it needs one. pub fn claims(token: &str) -> bool { - PARAMS.iter().any(|(name, _)| *name == token) || token.starts_with(toyos_blackbox::PARAM) + PARAMS.iter().any(|(name, _)| *name == token) + || token.starts_with(toyos_blackbox::PARAM) + || token.starts_with(toyos_logstream::PARAM) +} + +/// Where this boot streams its records, for the one hop the kernel makes with +/// it: into `/system/bin/init`'s environment. +/// +/// A `String` and not a borrow, because the buffer behind it is locked and the +/// bytes it copied are gone from the parameter line by now. Called once, after +/// the allocator exists. +pub fn log_stream() -> Option { + let held = LOG_STREAM.lock(); + if held.1 == 0 { + return None; + } + // Written from a `&str`, so this cannot fail; a refusal here would still be + // an empty stream rather than a boot that dies over a log's address. + core::str::from_utf8(&held.0[..held.1]).ok().map(alloc::string::ToString::to_string) } pub fn watchdog() -> bool { diff --git a/src/build.rs b/src/build.rs index 04bcd0a2a0..17523e84ca 100644 --- a/src/build.rs +++ b/src/build.rs @@ -903,13 +903,31 @@ impl Boot { /// the test kernel, and what goes on a stick is the shipping kernel — so an /// actuator name is refused here by name rather than reaching /// [`build_test_image`]'s assert, which would answer about kernel features. +/// +/// **Every valued parameter is cleared here by name, and the one this build can +/// read is parsed here too.** A stick is written, carried to the bench and +/// booted before anything else looks at what is on it, so an address this gate +/// passed and the kernel cannot use is discovered by a machine that streams to +/// nothing — the round trip the record stream exists to remove. pub fn flashable_params(root: &Path, asked: &[String]) -> Result<(), String> { let own = declared_params(root); for name in asked { - if !own.contains(name) { + if let Some(value) = name.strip_prefix(toyos_logstream::PARAM) { + toyos_logstream::endpoint(value).map_err(|why| { + format!( + "--kernel-param {name} beside --boot-config: {value:?} is not an address \ + ({})", + why.as_str() + ) + })?; + continue; + } + if !own.contains(name) && !is_valued_param(name) { return Err(format!( "--kernel-param {name} beside --boot-config: {name} is not one of the kernel's \ - boot parameters {own:?}, and a flashed image carries no actuator" + boot parameters {own:?} or one carrying a value ({}), and a flashed image \ + carries no actuator", + valued_params().join(", ") )); } } @@ -944,7 +962,7 @@ fn kernel_features( // A parameter names an actuator or one of the kernel's own boot parameters, // and only the first needs a kernel compiled with them. let own = declared_params(root); - if params.iter().any(|p| !own.contains(p)) { + if params.iter().any(|p| !own.contains(p) && !is_valued_param(p)) { features.push("boot-actuators"); } if !requested.is_empty() { @@ -979,16 +997,62 @@ fn check_params(root: &Path, params: &[String]) { let own = declared_params(root); for name in params { assert!( - declared.contains(name) || own.contains(name), + declared.contains(name) || own.contains(name) || is_valued_param(name), "--kernel-param {name}: the kernel declares no such actuator or boot parameter.\n\ Actuators it declares: {}.\n\ - Boot parameters it declares: {}.", + Boot parameters it declares: {}.\n\ + Boot parameters carrying a value: {}.", declared.join(", "), own.join(", "), + valued_params().join(", "), ); } } +/// The boot parameters that carry a value after their name, each beside the +/// path `kernel/src/params.rs`'s `claims` matches it by. +/// +/// **Not read out of `PARAMS`, because they are not in it**: a flag is a name +/// the kernel matches whole, and these are prefixes it matches with +/// `starts_with`. The path is here so the two lists can be checked against each +/// other — a prefix only one of them knows is an image the other refuses. +const VALUED_PARAMS: &[(&str, &str)] = &[ + ("toyos_blackbox::PARAM", toyos_blackbox::PARAM), + ("toyos_logstream::PARAM", toyos_logstream::PARAM), +]; + +/// The names in [`VALUED_PARAMS`], for a refusal that says what it would have +/// taken. +pub fn valued_params() -> Vec { + VALUED_PARAMS.iter().map(|(_, name)| (*name).to_string()).collect() +} + +/// Whether `param` is one of [`VALUED_PARAMS`] with its value after it. +pub fn is_valued_param(param: &str) -> bool { + VALUED_PARAMS.iter().any(|(_, prefix)| param.starts_with(prefix)) +} + +/// Every prefix `kernel/src/params.rs`'s `claims` matches with `starts_with`, +/// as the constant paths it names them by. +/// +/// The gate's own reading of the kernel, so the two lists of valued parameters +/// can be asserted equal; nothing in a build needs it. +/// +/// Anchored on the function and closed on the first line that ends it, so a +/// reflow still reads and a declaration this cannot find is empty rather than +/// guessed. +#[cfg(test)] +fn prefixes_claimed(text: &str) -> Vec { + let Some((_, body)) = text.split_once("pub fn claims") else { return Vec::new() }; + let Some((body, _)) = body.split_once("\n}") else { return Vec::new() }; + body.match_indices("starts_with(") + .filter_map(|(at, marker)| { + let rest = &body[at + marker.len()..]; + rest.split_once(')').map(|(path, _)| path.trim().to_string()) + }) + .collect() +} + /// The boot parameters the kernel itself answers to, off `kernel/src/params.rs`. pub fn declared_params(root: &Path) -> Vec { let path = root.join("kernel/src/params.rs"); @@ -1524,7 +1588,7 @@ pub fn build_test_image( // different actuators from sharing one disk. let own = declared_params(root); assert!( - kernel_params.iter().all(|p| own.contains(p)) + kernel_params.iter().all(|p| own.contains(p) || is_valued_param(p)) || kernel_features.iter().eq(TEST_KERNEL.iter().copied()), "a boot asking for {kernel_params:?} must boot the test kernel, not {kernel_features:?}" ); @@ -1936,6 +2000,62 @@ mod tests { } } + /// **The two lists of valued parameters are one list.** A prefix + /// `kernel/src/params.rs`'s `claims` matches and this file does not know is + /// a name the pre-flash gate refuses; one this file clears and `claims` does + /// not match writes an image `actuator::init` panics on. Both are read from + /// the kernel's own source, so neither can be satisfied by editing this + /// test. + #[test] + fn a_valued_parameter_is_one_the_kernel_claims_by_prefix() { + let root = Path::new(env!("CARGO_MANIFEST_DIR")); + let text = fs::read_to_string(root.join("kernel/src/params.rs")).expect("params.rs"); + let mut claimed = prefixes_claimed(&text); + claimed.sort(); + let mut declared: Vec = + VALUED_PARAMS.iter().map(|(path, _)| (*path).to_string()).collect(); + declared.sort(); + assert_eq!( + claimed, declared, + "`params::claims` matches {claimed:?} by prefix and `VALUED_PARAMS` names {declared:?}" + ); + + // Anchored on the function and closed on it: a body it cannot find + // reads as nothing rather than as the rest of the file. + assert!(prefixes_claimed("fn other(t: &str) { t.starts_with(a::B) }").is_empty()); + assert_eq!( + prefixes_claimed("pub fn claims(t: &str) -> bool {\n t.starts_with( a::B )\n}\n"), + vec!["a::B".to_string()] + ); + } + + /// The gate a stick is written behind, on the one valued parameter it can + /// read: an address it cannot parse is refused by name here, and not by a + /// machine on the bench that streams to nothing. + #[test] + fn the_pre_flash_gate_parses_the_address_it_clears() { + let root = Path::new(env!("CARGO_MANIFEST_DIR")); + let good = format!("{}10.0.2.2:41337", toyos_logstream::PARAM); + assert_eq!(flashable_params(root, &[good]), Ok(())); + for (bad, why) in [ + ("10.0.2.2", toyos_logstream::Malformed::NoPort), + ("", toyos_logstream::Malformed::NoPort), + ("t14:22", toyos_logstream::Malformed::NotAnAddress), + ("10.0.2.2:0", toyos_logstream::Malformed::NotAPort), + ("10.0.2.2:65536", toyos_logstream::Malformed::NotAPort), + ] { + let asked = format!("{}{bad}", toyos_logstream::PARAM); + let refused = flashable_params(root, std::slice::from_ref(&asked)) + .expect_err(&format!("{asked} passed the gate")); + assert!(refused.contains(why.as_str()), "{asked} was refused as {refused:?}"); + } + // The other valued parameter still passes: this gate parses the one it + // can read and clears the rest by name. + assert_eq!(flashable_params(root, &[format!("{}0x1000", toyos_blackbox::PARAM)]), Ok(())); + // And a name that is neither is still refused. + assert!(flashable_params(root, &["log-storm".to_string()]).is_err()); + } + #[test] fn params_read_the_kernels_own_list() { let root = Path::new(env!("CARGO_MANIFEST_DIR")); diff --git a/src/tiers.rs b/src/tiers.rs index b2591e1090..4826294615 100644 --- a/src/tiers.rs +++ b/src/tiers.rs @@ -1295,6 +1295,73 @@ pub const RELEGATED: &[Relegated] = &[ is a copy of the boot image: 8,079 ms on the hosted shard, over \ `FAST_COMMIT_MS`.", }, + Relegated { + test: "log_stream", + ci_ms: 24_865, + why: Why::Cost, + guards: "`logd`'s second sink judged live: a host listener, the address on the boot \ + parameter line, and the guest's own `/log` read off the FAT partition behind \ + its back as the oracle — every line the listener received equal to the file's, \ + in the file's order, with `Boot: complete` and a job's `exit:` record asserted \ + to arrive while the machine is still running. Nothing else in the tree asks \ + whether a record leaves the machine at all, so the whole of stage 3 goes dark \ + per pull request with it. What costs the price is the image: the host picks the \ + listener's port and the port is on the parameter line, so every arm builds a \ + boot image nothing can memoize — about 20 s of a 25 s run, with no assertion \ + behind it. That is what would return this name and its three siblings.", + }, + Relegated { + test: "log_stream_e1000e", + ci_ms: 25_219, + why: Why::Cost, + guards: "The same stream over netd's Intel driver rather than virtio, for the reason \ + `https_tls13_e1000e` exists: the bench's NIC is an I219 and QEMU's `e1000e` is \ + the only machine in reach that runs that driver. What still runs per pull \ + request: `https_tls13_e1000e` moves real frames through the same driver, so a \ + driver that stopped working reds there; what goes dark is the record stream \ + over it. Its price is the per-arm image build `log_stream` describes.", + }, + Relegated { + test: "log_stream_no_listener", + ci_ms: 25_182, + why: Why::Cost, + guards: "A boot told to stream to a port nothing answers on: the file carries exactly \ + one line saying so, naming the address, and the boot goes on and runs a job. \ + The refusal path — `ConnectionRefused` is final, said once, and never again — \ + has no other gate; `toyos-logstream`'s host tests cover the parse refusals but \ + no host test can reach netd's answer. Its price is the per-arm image build \ + `log_stream` describes.", + }, + Relegated { + test: "log_stream_unreachable", + ci_ms: 25_707, + why: Why::Cost, + guards: "A `log-storm` offered to a stream whose address answers nothing, which is the \ + accounting under a stream that never opened: the bounded queue refuses what it \ + cannot hold, every report says what its run of loss added and what the boot has \ + lost in total, and the two must agree — and `/log` carries every record \ + regardless. What still runs per pull request: \ + `log_stream_stalled_peer_storm_over_a_bounded_window` drives the same \ + accounting through the writer's own backpressure, so a counter that stopped \ + counting still reds; what goes dark is the arm where nothing ever drains. Its \ + price is the per-arm image build `log_stream` describes.", + }, + Relegated { + test: "log_stream_stalled_peer_delivers_whole_records", + ci_ms: 34_591, + why: Why::Cost, + guards: "A peer that accepts the stream, stops reading for a storm's worth of records \ + and then reads again: every line it receives is a whole record, `/log`'s own, \ + in `/log`'s own order. It is the only gate anywhere on netd moving a client's \ + bytes without losing the tail of a short send — reverting that hunk reds it on \ + a record cut in half with a whole one behind it — and the only place a stalled \ + peer's `write_all` blocks under a test at all. What it does not judge, and what \ + no arm may: how many lines a stall costs, which is the pipe\'s size, netd\'s and \ + whatever QEMU holds between them. What still runs per pull request: nothing on \ + the wire; `toyos-logstream`\'s host tests carry the drop accounting and red \ + under the drop-count mutation with no guest at all. Its price is the storm and \ + the per-arm image build `log_stream` describes.", + }, ]; /// The names [`RELEGATED`] holds, which is what `tests/toyos.rs` checks its own diff --git a/system.toml b/system.toml index 9c7f05cf47..b8d227a947 100644 --- a/system.toml +++ b/system.toml @@ -46,10 +46,12 @@ toyos-cc = { path = "toyos-cc" } # image does. The kernel keeps the record ring and the console and writes no # file at all, so a boot config without `logd` is a boot whose `/log` is empty — # `every_boot_config_runs_logd` is what refuses one. It claims no device and -# serves no port: its whole authority is `logread`, which is -# `Rights::LOG | Rights::WAIT` on a `SysCap` duplicate. +# serves no port. [programs.logd] syscap = ["logread"] +# The record stream's authority: the address on the boot parameter line is +# information, and this row is the whole of what can act on it. +receives = ["netd"] [programs.compositor] serves = ["compositor"] diff --git a/tests/common/https.rs b/tests/common/https.rs index c219e47222..bf089c93b0 100644 --- a/tests/common/https.rs +++ b/tests/common/https.rs @@ -15,9 +15,8 @@ use std::time::Duration; use super::compile; use super::qemu::{self, BootOptions, QemuInstance}; -/// Where the guest sees the host under QEMU's user-mode networking, and where -/// the host arm sees the same servers. The judge's certificate carries both. -const GUEST_VIEW_OF_HOST: &str = "10.0.2.2"; +/// Where the host arm sees the servers the guest reaches at +/// [`qemu::GUEST_VIEW_OF_HOST`]. The judge's certificate carries both. const HOST_VIEW_OF_HOST: &str = "127.0.0.1"; /// Where the minted CA lands on ROOT, which mounts at `/system`. @@ -113,7 +112,7 @@ pub fn tls13_judge(rust_bins: &[(String, Vec)], bench: Bench) -> Result<(), let good = server.port("ok")?; let mut lines = Vec::new(); - let guest_ok = fetch_in_guest(&mut guest, GUEST_VIEW_OF_HOST, good, true)?; + let guest_ok = fetch_in_guest(&mut guest, qemu::GUEST_VIEW_OF_HOST, good, true)?; if guest_ok != ok_line { return Err(format!("the guest fetched {guest_ok:?}, and the server served {ok_line:?}")); } @@ -121,7 +120,7 @@ pub fn tls13_judge(rust_bins: &[(String, Vec)], bench: Bench) -> Result<(), for (role, expected) in REFUSALS { let port = server.port(role)?; - let got = fetch_in_guest(&mut guest, GUEST_VIEW_OF_HOST, port, true)?; + let got = fetch_in_guest(&mut guest, qemu::GUEST_VIEW_OF_HOST, port, true)?; if got != *expected { return Err(format!("the {role} arm answered {got:?}, not {expected:?}")); } @@ -130,7 +129,7 @@ pub fn tls13_judge(rust_bins: &[(String, Vec)], bench: Bench) -> Result<(), // The CA is what makes the judge's own roots trusted, so withholding it is // the unknown-authority arm rather than a separate server. - let unknown = fetch_in_guest(&mut guest, GUEST_VIEW_OF_HOST, good, false)?; + let unknown = fetch_in_guest(&mut guest, qemu::GUEST_VIEW_OF_HOST, good, false)?; if unknown != "https_fetch: refused unknown-authority" { return Err(format!("a fetch with no extra root answered {unknown:?}")); } @@ -139,7 +138,10 @@ pub fn tls13_judge(rust_bins: &[(String, Vec)], bench: Bench) -> Result<(), let cleartext = server.port("plain")?; let plain = run_guest( &mut guest, - &format!("test_rs_https_fetch http://{GUEST_VIEW_OF_HOST}:{cleartext}/ --ca {CA_IN_GUEST}"), + &format!( + "test_rs_https_fetch http://{}:{cleartext}/ --ca {CA_IN_GUEST}", + qemu::GUEST_VIEW_OF_HOST + ), )?; if plain != "https_fetch: refused plain-http" { return Err(format!("a plain http:// fetch answered {plain:?}")); diff --git a/tests/common/logstream.rs b/tests/common/logstream.rs new file mode 100644 index 0000000000..f491b4951a --- /dev/null +++ b/tests/common/logstream.rs @@ -0,0 +1,785 @@ +//! The record stream's judge: a listener on the host, a guest booted with its +//! address on the parameter line, and the guest's own `/log` as the oracle. +//! +//! **The file is what the stream is judged against.** `logd` writes a line to +//! `/log` and then offers the same line to the stream, so a listener that kept +//! up received that file's own first lines and nothing else ([`is_prefix_of`]). +//! The file is read off the FAT volume behind the guest's back, so the two +//! readings share nothing but the boot that produced them. + +use std::io::{BufRead, BufReader, Write}; +use std::net::{Ipv4Addr, SocketAddr, TcpListener}; +use std::os::fd::AsRawFd; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use super::qemu::{self, BootOptions, QemuInstance}; +use super::{compile, serial, volumes}; + +/// A liveness guard on a guest that stopped talking, never a verdict. +const CEILING: Duration = Duration::from_secs(90); + +/// The whole of the lag the stream is allowed over the guest's own console, +/// which the host reads off a 16550 while the stream crosses a driver, a stack +/// and slirp. +const LAG: Duration = Duration::from_secs(20); + +/// Which machine the stream is judged on. **Two of them, and the driver is the +/// difference** — the bench's NIC is an Intel I219, and QEMU's `e1000e` is the +/// only machine in reach that runs netd's Intel driver. +#[derive(Clone, Copy)] +pub struct Bench { + pub profile: qemu::Profile, + /// The boot config whose netd claims this machine's card, and whose `logd` + /// row carries the `netd` connector the stream needs. + pub config: &'static str, + /// The `-device` this profile must actually carry, asked of the argv rather + /// than assumed: a profile with no NIC would make every line below arrive + /// from nowhere, or not arrive and be believed. + pub device: &'static str, +} + +pub const VIRTIO: Bench = + Bench { profile: qemu::Profile::Headless, config: "tests/netcase", device: "virtio-net" }; + +pub const E1000E: Bench = + Bench { profile: qemu::Profile::E1000e, config: "tests/e1000case", device: "e1000e" }; + +/// What a host holds of a stream whose reader has stopped taking it. +/// +/// Held rather than left to the runner, so a peer that stops reading closes its +/// window promptly on every host — which is what puts netd's send buffer under +/// its own bound and the writer inside a blocking write. **It does not bound +/// what the machine absorbs**: the pipe below netd is megabytes and what QEMU's +/// user-mode networking holds is nothing this tree names, so how much a stalled +/// peer costs is not a number any arm here may assert. +const STALLED_PEER_WINDOW: usize = 32 * 1024; + +/// An address on the guest's own network that answers nothing, ever. +/// +/// QEMU's user-mode networking answers ARP for the four addresses it hosts in +/// `10.0.2.0/24` and for nothing else, so a SYN aimed here never leaves the +/// guest's stack: no refusal, no reset, no peer. That is "the cable is out", +/// staged without a cable. +const UNREACHABLE: &str = "10.0.2.99"; + +/// The port that address does not answer on either. Any number does; a fixed +/// one keeps the parameter line readable. +const UNREACHABLE_PORT: u16 = 41337; + +/// A host listener for one boot's records. +/// +/// It accepts exactly one connection — a boot has one `logd` — appends every +/// line to a file as it arrives, and keeps them for the comparison at the end. +pub struct Listener { + /// The host port the guest is told to reach, as `10.0.2.2:`. + pub port: u16, + /// Where the lines are appended as they arrive, so a failing run leaves the + /// stream on disk beside the guest's own log. + pub path: PathBuf, + lines: Arc>>, + connected: Arc, + ended: Arc, + /// Cleared by [`Listener::stalled`]: the thread accepts the connection and + /// then reads nothing until [`Listener::release`] sets it. + reading: Arc, +} + +impl Listener { + /// A port nothing is listening on: bound and released rather than picked + /// out of the air. Something taking it in the meantime is a false red and + /// never a false green, because the line the arm asserts is the *refusal*. + pub fn silent_port() -> Result { + let socket = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)) + .map_err(|e| format!("bind a port to hand back: {e}"))?; + let port = socket + .local_addr() + .map_err(|e| format!("ask a bound socket its port: {e}"))? + .port(); + drop(socket); + Ok(port) + } + + /// A listener reading as fast as the boot writes. + pub fn start(path: &Path) -> Result { + Self::bound(path, true) + } + + /// A listener that accepts the connection and then reads nothing, so the + /// guest's own buffers are what the boot's records pile up in. + /// + /// It is released and drained later, because a peer that never reads says + /// nothing about what reached it: what arrived is the evidence the writer + /// got as far as writing at all. + pub fn stalled(path: &Path) -> Result { + Self::bound(path, false) + } + + /// Read whatever the peer piled up while this listener was stalled. + pub fn release(&self) { + self.reading.store(true, Ordering::SeqCst); + } + + fn bound(path: &Path, reading: bool) -> Result { + let socket = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + .map_err(|e| format!("bind the log stream's listener: {e}"))?; + if !reading { + clamp_receive_buffer(&socket, STALLED_PEER_WINDOW)?; + } + let port = socket + .local_addr() + .map_err(|e| format!("ask the listener its port: {e}"))? + .port(); + let lines = Arc::new(Mutex::new(Vec::new())); + let connected = Arc::new(AtomicUsize::new(0)); + let ended = Arc::new(AtomicBool::new(false)); + let reading = Arc::new(AtomicBool::new(reading)); + let file = std::fs::File::create(path) + .map_err(|e| format!("create {}: {e}", path.display()))?; + + let theirs = ( + Arc::clone(&lines), + Arc::clone(&connected), + Arc::clone(&ended), + Arc::clone(&reading), + ); + std::thread::spawn(move || { + let (lines, connected, ended, reading) = theirs; + let Ok((stream, _)) = socket.accept() else { return }; + connected.fetch_add(1, Ordering::SeqCst); + while !reading.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(50)); + } + let mut file = file; + let mut reader = BufReader::new(stream); + loop { + let mut line = String::new(); + match reader.read_line(&mut line) { + Ok(0) | Err(_) => break, + Ok(_) => { + let _ = file.write_all(line.as_bytes()); + lines.lock().expect("the stream's lines").push(line); + } + } + } + let _ = file.flush(); + // **The close is the boot's end**, and it is the one thing this + // stream says that no line carries: `logd` dies with the machine, + // netd sees its pipe hang up and closes the socket. + ended.store(true, Ordering::SeqCst); + }); + + Ok(Self { port, path: path.to_path_buf(), lines, connected, ended, reading }) + } + + pub fn connections(&self) -> usize { + self.connected.load(Ordering::SeqCst) + } + + pub fn ended(&self) -> bool { + self.ended.load(Ordering::SeqCst) + } + + pub fn lines(&self) -> Vec { + self.lines.lock().expect("the stream's lines").clone() + } + + /// Wait for a line carrying `needle`, answering how long it took. + pub fn wait_for(&self, needle: &str, by: Duration) -> Result { + let began = Instant::now(); + while began.elapsed() < by { + if self.lines().iter().any(|l| l.contains(needle)) { + return Ok(began.elapsed()); + } + std::thread::sleep(Duration::from_millis(25)); + } + Err(format!( + "{needle:?} never arrived on the stream in {by:?}: {} connection(s), {} line(s), \ + ended={}", + self.connections(), + self.lines().len(), + self.ended() + )) + } + + /// Wait until nothing new has arrived for `still`, answering how many lines + /// have. + /// + /// A liveness guard on a drain no line announces: what it waits for is the + /// backlog a released peer takes, and a guest that is merely idle goes quiet + /// long before `still`. + pub fn wait_until_quiet(&self, still: Duration, by: Duration) -> Result { + let began = Instant::now(); + let mut seen = self.lines().len(); + let mut since = Instant::now(); + while began.elapsed() < by { + std::thread::sleep(Duration::from_millis(100)); + let now = self.lines().len(); + if now != seen { + seen = now; + since = Instant::now(); + } else if since.elapsed() >= still { + return Ok(seen); + } + } + Err(format!( + "the stream was still arriving {by:?} after the peer read again; {seen} line(s) so far" + )) + } + + /// Wait for the connection to close, which is this boot saying it is over. + pub fn wait_for_end(&self, by: Duration) -> Result<(), String> { + let began = Instant::now(); + while began.elapsed() < by { + if self.ended() { + return Ok(()); + } + std::thread::sleep(Duration::from_millis(25)); + } + Err(format!( + "the stream was still open {by:?} after the guest went down; {} line(s) received", + self.lines().len() + )) + } +} + +/// Hold this socket's receive buffer to `bytes`, so what a peer that stops +/// reading can absorb is this test's number and not the runner's. +fn clamp_receive_buffer(socket: &TcpListener, bytes: usize) -> Result<(), String> { + let size = bytes as libc::c_int; + // SAFETY: `socket` owns the descriptor for the whole call, and the pointer + // and length describe the one `c_int` `SO_RCVBUF` is documented to take. + let set = unsafe { + libc::setsockopt( + socket.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_RCVBUF, + std::ptr::addr_of!(size).cast(), + std::mem::size_of::() as libc::socklen_t, + ) + }; + if set != 0 { + return Err(format!( + "hold the listener's receive buffer to {bytes} bytes: {}", + std::io::Error::last_os_error() + )); + } + Ok(()) +} + +/// One boot's image, built with the stream's address on it, and where its log +/// partition sits inside it. +struct Staged { + image: PathBuf, + start: usize, + len: usize, +} + +fn stage( + bench: Bench, + name: &str, + at: (&'static str, u16), + extra: &[&str], + c_bins: &[(String, Vec)], + rust_bins: &[(String, Vec)], +) -> Result { + let config = compile::repo_root().join(bench.config); + let param = qemu::log_stream_param(at); + let mut params: Vec<&str> = extra.to_vec(); + params.push(param.as_str()); + let bytes = qemu::build_boot_image(&config, c_bins, rust_bins, ¶ms); + let image = super::lane::dir().join(format!("{name}.img")); + std::fs::write(&image, &bytes).map_err(|e| format!("write {}: {e}", image.display()))?; + let (start, len) = volumes::log_extent(&bytes, &image)?; + Ok(Staged { image, start, len }) +} + +/// The two numbers a drop report carries: what this run of loss added, and what +/// the boot has lost in total. +/// +/// Anchored on the report's own words and not on a position, so a line that +/// grows a prefix still reads. A line that carries neither is a failing verdict +/// rather than a zero. +fn drop_report(line: &str) -> Result<(u64, u64), String> { + let words: Vec<&str> = line.split_whitespace().collect(); + let after = |word: &str| -> Option { + let at = words.iter().position(|w| *w == word)?; + words.get(at + 1)?.parse().ok() + }; + match (after("logd:"), after("and")) { + (Some(run), Some(total)) => Ok((run, total)), + _ => Err(format!("this line is not a drop report: {line:?}")), + } +} + +/// What the guest's own log volume says, read off the device behind the guest's +/// back — the oracle the stream is compared with. +/// +/// Every file this boot wrote, because a boot that logs enough to starve a +/// stream logs enough to rotate, and a rotation is not a hole. +fn on_the_volume(staged: &Staged) -> Result, String> { + volumes::whole_log(&staged.image, staged.start, staged.len) +} + +/// What this boot's own log says the stream refused, and in how many lines, or +/// `None` when it says it refused nothing. +/// +/// **The accounting is checked against itself.** Every report says what its own +/// run of loss added and what the boot has lost in total, so the first numbers +/// must sum to the last line's second one. A counter that has stopped counting +/// cannot satisfy both, and a report nobody can read is the same as no report. +/// +/// Whether a boot refuses anything at all is a fact about buffers no arm here +/// owns; whether what it says about its refusals holds together is not. +fn refusals_in(file: &[String]) -> Result, String> { + let reports: Vec<&String> = + file.iter().filter(|l| l.contains("never reached the log stream")).collect(); + let Some(last) = reports.last() else { + return Ok(None); + }; + let mut added = 0u64; + for line in &reports { + added += drop_report(line)?.0; + } + let dropped = drop_report(last)?.1; + if dropped == 0 { + return Err(format!("the stream reported dropping nothing, in {last:?}")); + } + if added != dropped { + return Err(format!( + "the drop reports add up to {added} and the last one says {dropped} for the whole \ + boot, so what is counted is not the drops:\n{}", + reports.iter().map(|l| l.trim_end()).collect::>().join("\n") + )); + } + Ok(Some((dropped, reports.len()))) +} + +/// The same, where refusing nothing is a failing verdict: a stream whose address +/// answers nothing drains never, so a boot that offered a storm to it and lost +/// no line measured nothing at all. +fn refused_in(file: &[String]) -> Result<(u64, usize), String> { + refusals_in(file)?.ok_or_else(|| { + format!( + "a log storm offered to a stream that could not take it cost it no record it admits \ + to; the file has {} line(s), ending {:?}", + file.len(), + file.iter().rev().take(3).collect::>() + ) + }) +} + +/// What the listener received is the file's own lines in the file's own order, +/// with holes where the queue refused one. +/// +/// A peer that stalls costs the stream lines; it may not reorder, duplicate, +/// invent or cut one. **Every received line is a whole record, including the +/// last** — a line short of the file's is a byte the stream consumed and did +/// not deliver, which is the failure this comparison exists to catch, and no +/// position in the stream is a place it may happen. +fn is_subsequence_of(received: &[String], file: &[String]) -> Result<(), String> { + let mut at = 0usize; + for line in received { + match file[at..].iter().position(|theirs| theirs == line) { + Some(step) => at += step + 1, + None => { + return Err(format!( + "the stream carries {line:?}, which /log does not carry after its line {at}" + )) + } + } + } + Ok(()) +} + +/// The listener received the file's own first lines, in the file's own order, +/// and nothing else. +/// +/// Reported as the first disagreement rather than as a count: a stream that lost +/// its third line and one that reordered two are different defects, and a length +/// calls them the same one. +fn is_prefix_of(received: &[String], file: &[String]) -> Result<(), String> { + if received.is_empty() { + return Err("the listener received nothing at all".to_string()); + } + for (i, line) in received.iter().enumerate() { + match file.get(i) { + Some(theirs) if theirs == line => {} + Some(theirs) => { + return Err(format!( + "the stream and /log disagree at line {i}:\n stream: {line:?}\n /log: \ + {theirs:?}" + )) + } + None => { + return Err(format!( + "the stream carries {} line(s) and /log only {}; the first line past the \ + file is {line:?}", + received.len(), + file.len() + )) + } + } + } + Ok(()) +} + +/// A boot's records arriving over the wire while it is booting, judged live and +/// then against the file. +pub fn stream( + bench: Bench, + c_bins: &[(String, Vec)], + rust_bins: &[(String, Vec)], +) -> Result<(), String> { + let name = format!("logstream-{}", bench.device); + let listener = Listener::start(&super::lane::dir().join(format!("{name}.txt")))?; + let staged = + stage(bench, &name, (qemu::GUEST_VIEW_OF_HOST, listener.port), &[], c_bins, rust_bins)?; + + let options = BootOptions { + profile: bench.profile, + boot_image: Some(staged.image.clone()), + log_stream: Some((qemu::GUEST_VIEW_OF_HOST, listener.port)), + ..Default::default() + }; + if !qemu::profile_argv(&options).iter().any(|a| a.contains(bench.device)) { + return Err(format!("this test needs a {} and the profile carries none", bench.device)); + } + let config = compile::repo_root().join(bench.config); + let mut guest = QemuInstance::boot_with_options(&config, c_bins, rust_bins, options); + let mut console = guest.boot_log().to_string(); + serial::Serial::named("boot console", console.as_str()).must_be_clean()?; + + // **The claim is "while it is booting", so it is asserted before anything + // shuts the guest down.** `Boot: complete` is a kernel record, so its only + // way here is the ring, `logd`, netd and the wire. + let took = listener.wait_for("Boot: complete", CEILING)?; + eprintln!( + " [stream] `Boot: complete` reached the host over {} {} ms into the listener's life, \ + with the guest still running", + bench.device, + took.as_millis() + ); + + // A job, and its exit record over the wire. The kernel logs `exit:` when + // the process ends, so this is a record produced *after* the stream was + // already open — the boot log alone could have been a replay of a ring. + let job = "test_rs_empty_dir_stat"; + let result = guest.run_test(job, Duration::from_secs(60)); + if result.exit_code != Some(0) { + return Err(format!("{job} exited {:?}:\n{}", result.exit_code, result.stdout)); + } + let exit = format!("exit: {job} "); + let took = listener.wait_for(&exit, LAG)?; + eprintln!(" [stream] {exit:?} reached the host {} ms after the job ended", took.as_millis()); + + // Down, and then the two readings of the same boot. + writeln!(guest.stdin_mut(), "run shutdown").map_err(|e| format!("write to QEMU stdin: {e}"))?; + guest.flush_stdin(); + let tail = guest.drain_serial(Duration::from_secs(20)); + console.push_str(&tail); + drop(guest); + for bad in ["PANIC:", "panicked at"] { + if console.contains(bad) { + return Err(format!("{bad:?} on the way down\n{tail}")); + } + } + listener.wait_for_end(LAG)?; + + let received = listener.lines(); + let file = on_the_volume(&staged)?; + is_prefix_of(&received, &file)?; + if received.iter().any(|l| l.contains("never reached the log stream")) { + return Err(format!( + "a listener that read every line was still reported as behind this machine: {:?}", + received.iter().find(|l| l.contains("never reached the log stream")) + )); + } + // Non-vacuity: a comparison over three lines proves nothing about a boot. + if received.len() < 100 { + return Err(format!( + "the stream carried {} line(s), which is fewer than a boot writes", + received.len() + )); + } + eprintln!( + " [stream] {} line(s) over the wire, every one of them the same line /log holds ({} \ + line(s) in the file); the connection closed with the machine", + received.len(), + file.len() + ); + let _ = std::fs::remove_file(&staged.image); + Ok(()) +} + +/// A boot told to stream to a port nothing is listening on: the file is whole +/// and carries the one line saying what could not be done. +pub fn no_listener( + c_bins: &[(String, Vec)], + rust_bins: &[(String, Vec)], +) -> Result<(), String> { + let bench = VIRTIO; + let port = Listener::silent_port()?; + let staged = + stage(bench, "logstream-silent", (qemu::GUEST_VIEW_OF_HOST, port), &[], c_bins, rust_bins)?; + + let options = BootOptions { + profile: bench.profile, + boot_image: Some(staged.image.clone()), + log_stream: Some((qemu::GUEST_VIEW_OF_HOST, port)), + ..Default::default() + }; + let config = compile::repo_root().join(bench.config); + let mut guest = QemuInstance::boot_with_options(&config, c_bins, rust_bins, options); + let mut console = guest.boot_log().to_string(); + serial::Serial::named("boot console", console.as_str()).must_be_clean()?; + + // The boot goes on. That is the claim: a stream nobody is listening to + // costs this machine its stream and nothing else. + let result = guest.run_test("test_rs_empty_dir_stat", Duration::from_secs(60)); + if result.exit_code != Some(0) { + return Err(format!( + "a boot whose log stream found no listener could not run a job: {:?}\n{}", + result.exit_code, result.stdout + )); + } + writeln!(guest.stdin_mut(), "run shutdown").map_err(|e| format!("write to QEMU stdin: {e}"))?; + guest.flush_stdin(); + console.push_str(&guest.drain_serial(Duration::from_secs(20))); + drop(guest); + + let file = on_the_volume(&staged)?; + let refusals: Vec<&String> = + file.iter().filter(|l| l.contains("for this boot's log stream")).collect(); + let [said] = refusals.as_slice() else { + return Err(format!( + "the file carries {} line(s) about a log stream that never opened, and it owes \ + exactly one; the file has {} line(s), ending {:?}", + refusals.len(), + file.len(), + file.iter().rev().take(3).collect::>() + )); + }; + if !said.contains(&format!("{}:{port}", qemu::GUEST_VIEW_OF_HOST)) { + return Err(format!("the refusal does not say which address it was: {said:?}")); + } + if !file.iter().any(|l| l.contains("Boot: complete")) { + return Err("the file stops before `Boot: complete`, so the stream cost it records" + .to_string()); + } + eprintln!(" [stream] no listener, and the file says so once: {}", said.trim_end()); + let _ = std::fs::remove_file(&staged.image); + Ok(()) +} + +/// A boot whose stream address is on this machine's network and answers +/// nothing: every line offered while the connection is being opened is refused +/// by the queue, counted, and reported — and `/log` is whole regardless. +/// +/// **A peer that never answers is where the queue is the first buffer to +/// fill**: nothing below it drains, so it is the only thing that can refuse. +pub fn unreachable( + c_bins: &[(String, Vec)], + rust_bins: &[(String, Vec)], +) -> Result<(), String> { + let bench = VIRTIO; + // `log-storm` is what makes this machine produce records faster than a + // stream that is going nowhere can take them, and it is baked into the + // image rather than passed to a staged one. + let staged = stage( + bench, + "logstream-unreachable", + (UNREACHABLE, UNREACHABLE_PORT), + &["log-storm"], + c_bins, + rust_bins, + )?; + + let options = BootOptions { + profile: bench.profile, + boot_image: Some(staged.image.clone()), + log_stream: Some((UNREACHABLE, UNREACHABLE_PORT)), + kernel_params: &["log-storm"], + smp: 8, + ..Default::default() + }; + let config = compile::repo_root().join(bench.config); + let mut guest = QemuInstance::boot_with_options(&config, c_bins, rust_bins, options); + let mut console = guest.boot_log().to_string(); + + // The storm starts on the first `SYS_LOG_READ`, which `logd` makes before + // this line is printed; what it produces has nowhere to go. + qemu::await_marker(&mut guest, &mut console, "logstorm done t=", "the storm to run out")?; + + // **The guest goes on working.** That is the claim the whole design turns + // on: a stream that cannot deliver a byte costs this machine its stream and + // nothing else. + let result = guest.run_test("test_rs_empty_dir_stat", Duration::from_secs(60)); + if result.exit_code != Some(0) { + return Err(format!( + "a boot whose log stream reached nothing could not run a job: {:?}\n{}", + result.exit_code, result.stdout + )); + } + writeln!(guest.stdin_mut(), "run shutdown").map_err(|e| format!("write to QEMU stdin: {e}"))?; + guest.flush_stdin(); + console.push_str(&guest.drain_serial(Duration::from_secs(20))); + drop(guest); + + let file = on_the_volume(&staged)?; + let (dropped, said_in) = refused_in(&file)?; + + // **The file goes on being written on the far side of every drop.** That is + // what the stream may never cost, and a storm with nowhere to send it is + // exactly when it would. `Boot: complete` is not the marker to ask for + // here: the storm overruns the kernel's own record ring, so an early + // record's absence is the ring's doing and says nothing about the stream — + // the job's exit record is, because it happened after the queue had already + // started refusing lines. + let owed = "exit: test_rs_empty_dir_stat "; + if !file.iter().any(|l| l.contains(owed)) { + return Err(format!( + "{owed:?} never reached /log on a boot whose stream went nowhere; the file has {} \ + line(s), ending {:?}", + file.len(), + file.iter().rev().take(3).collect::>() + )); + } + let stormed = file.iter().filter(|l| l.contains("logstorm t=")).count(); + if stormed == 0 { + return Err("the storm reached the log stream's arm and not the file".to_string()); + } + eprintln!( + " [stream] {stormed} storm record(s) in /log and a stream that reached nothing; \ + {dropped} line(s) refused by the queue and the log says so in {said_in} line(s)" + ); + let _ = std::fs::remove_file(&staged.image); + Ok(()) +} + +/// A peer that accepts the connection, stops reading for a storm's worth of +/// records, and then reads again: every line it receives is a whole record, and +/// `/log`'s own, in `/log`'s own order. +/// +/// **That is what a stall may cost and what it may not.** It may cost lines — +/// how many is the pipe's size, netd's, and whatever QEMU's user-mode +/// networking holds between them, none of which this arm owns, so it demands no +/// refusal. It may never cost half a line: a byte the machine consumed and did +/// not deliver is a record cut on the wire, which is what netd discarding the +/// tail of a short send produces and what [`is_subsequence_of`] refuses at every +/// position. +/// +/// **The peer is released while the guest is still running**, which is what +/// puts any such cut in the middle of a stream that goes on past it. Ending the +/// arm at the stall instead leaves every cut on the last line, where a +/// comparison can no longer tell a truncation from the connection's own end. +/// +/// What a boot *says* it refused is checked against itself where it says +/// anything ([`refusals_in`]); that the accounting counts what it refuses at all +/// is `toyos-logstream`'s host tests, which need no guest and no host's buffers. +pub fn stalled_peer( + c_bins: &[(String, Vec)], + rust_bins: &[(String, Vec)], +) -> Result<(), String> { + let bench = VIRTIO; + let storm: &[&str] = &["log-storm", "log-storm-wide"]; + let listener = Listener::stalled(&super::lane::dir().join("logstream-stalled.txt"))?; + let staged = stage( + bench, + "logstream-stalled", + (qemu::GUEST_VIEW_OF_HOST, listener.port), + storm, + c_bins, + rust_bins, + )?; + + let options = BootOptions { + profile: bench.profile, + boot_image: Some(staged.image.clone()), + log_stream: Some((qemu::GUEST_VIEW_OF_HOST, listener.port)), + kernel_params: storm, + smp: 8, + ..Default::default() + }; + let config = compile::repo_root().join(bench.config); + let mut guest = QemuInstance::boot_with_options(&config, c_bins, rust_bins, options); + let mut console = guest.boot_log().to_string(); + + qemu::await_marker(&mut guest, &mut console, "logstorm done t=", "the storm to run out")?; + + // **The guest goes on working**, which is the claim a peer that stopped + // reading tests and a peer that never answered does not: this one has the + // machine's own writer blocked on it. + let result = guest.run_test("test_rs_empty_dir_stat", Duration::from_secs(60)); + if result.exit_code != Some(0) { + return Err(format!( + "a boot whose log stream stalled could not run a job: {:?}\n{}", + result.exit_code, result.stdout + )); + } + + // The peer reads again, and the machine stays up until the backlog it was + // holding has gone past. Everything the stall cost is then in the middle of + // the stream rather than at its end. + listener.release(); + listener.wait_until_quiet(Duration::from_secs(2), LAG)?; + + writeln!(guest.stdin_mut(), "run shutdown").map_err(|e| format!("write to QEMU stdin: {e}"))?; + guest.flush_stdin(); + console.push_str(&guest.drain_serial(Duration::from_secs(20))); + drop(guest); + listener.wait_for_end(LAG)?; + + // The connection was opened, so the writer entered the loop this arm is + // about rather than giving up in `open`. + if listener.connections() != 1 { + return Err(format!( + "the stream opened {} time(s), so nothing was ever written into it", + listener.connections() + )); + } + let file = on_the_volume(&staged)?; + // **Whether this host's buffers made the machine refuse anything is not + // asserted** — how much a stalled peer costs is the pipe's size, netd's, + // and what QEMU holds between them, and two hosted runs proved that is not + // a number this arm may demand. What it does assert is that whatever the + // boot says about its refusals holds together. The accounting itself is + // `toyos-logstream`'s host tests, which red under the drop-count mutation. + let refused = refusals_in(&file)?; + + let received = listener.lines(); + let bytes: usize = received.iter().map(String::len).sum(); + if bytes <= toyos_logstream::MAX_BACKLOG_BYTES { + return Err(format!( + "the peer received {bytes} byte(s), which the queue alone holds ({}), so nothing \ + here says a stream went through the writer at all", + toyos_logstream::MAX_BACKLOG_BYTES + )); + } + is_subsequence_of(&received, &file)?; + + let owed = "exit: test_rs_empty_dir_stat "; + if !file.iter().any(|l| l.contains(owed)) { + return Err(format!( + "{owed:?} never reached /log on a boot whose stream stalled; the file has {} line(s), \ + ending {:?}", + file.len(), + file.iter().rev().take(3).collect::>() + )); + } + eprintln!( + " [stream] a peer that stopped reading and then read again took {bytes} byte(s) in {} \ + whole line(s), each /log's own in /log's own order; /log holds {} line(s) and {}", + received.len(), + file.len(), + match refused { + Some((dropped, said_in)) => + format!("says it refused {dropped} of them in {said_in} line(s) that agree"), + None => "says it refused none".to_string(), + } + ); + let _ = std::fs::remove_file(&staged.image); + Ok(()) +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 64336c7c4e..35932a9187 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -25,6 +25,8 @@ pub mod irqcensus; #[allow(dead_code)] pub mod logread; #[allow(dead_code)] +pub mod logstream; +#[allow(dead_code)] pub mod passcost; #[allow(dead_code)] pub mod pkg; diff --git a/tests/common/qemu.rs b/tests/common/qemu.rs index aa6da5c8d3..73800af717 100644 --- a/tests/common/qemu.rs +++ b/tests/common/qemu.rs @@ -2251,6 +2251,42 @@ pub struct BootOptions { /// the image is memoized on their names and bytes, so two boots staging /// different fixtures do not share one. pub extra_root_files: Vec<(String, Vec)>, + /// Where this boot's `logd` streams its records, as seen from inside the + /// guest — [`GUEST_VIEW_OF_HOST`] and the host port a listener took, or an + /// address on the guest's network that answers nothing. + /// + /// **It is a `kernel_params` entry in every way but its type.** The host + /// picks the port, so the parameter cannot be a `&'static str`; + /// [`BootOptions::params`] is where the two become one list, and that list + /// is what an image is built with and what a staged image is asked to + /// match. + pub log_stream: Option<(&'static str, u16)>, +} + +/// Where the guest sees the host under QEMU's user-mode networking, and where +/// the host sees the same servers. +pub const GUEST_VIEW_OF_HOST: &str = "10.0.2.2"; + +impl BootOptions { + /// The whole parameter line this boot's image is built with: the names in + /// [`BootOptions::kernel_params`] and, when this boot streams its records, + /// the address it streams them to. + /// + /// One function, called by the build and by the staged-image check, so a + /// parameter that reaches the image and not the check — or the other way + /// round — is not expressible. + pub fn params(&self) -> Vec { + let mut params: Vec = self.kernel_params.iter().map(|p| (*p).to_string()).collect(); + if let Some(at) = self.log_stream { + params.push(log_stream_param(at)); + } + params + } +} + +/// `logstream=:`, spelled once. +pub fn log_stream_param((host, port): (&str, u16)) -> String { + format!("{}{host}:{port}", toyos_logstream::PARAM) } /// The in-guest test runner's startup marker. @@ -2281,6 +2317,7 @@ impl Default for BootOptions { usb_images: Vec::new(), rtc_base: None, extra_root_files: Vec::new(), + log_stream: None, } } } @@ -2420,8 +2457,15 @@ pub fn build_boot_image( rust_tests: &[(String, Vec)], kernel_params: &[&str], ) -> Vec { + // A parameter carrying a value is one the *shipping* kernel answers to, so + // it selects no kernel: an image built for the record stream and nothing + // else must be the image a flashed stick would be. let kernel: &[&str] = - if kernel_params.is_empty() { &[] } else { toyos_build::build::TEST_KERNEL }; + if kernel_params.iter().all(|p| toyos_build::build::is_valued_param(p)) { + &[] + } else { + toyos_build::build::TEST_KERNEL + }; build_boot_image_with(test_crate, c_tests, rust_tests, &[], kernel, kernel_params, false) } @@ -2470,7 +2514,9 @@ fn refuse_a_staged_image_this_boot_did_not_ask_for(image: &Path, options: &BootO options.extra_root_files.len(), image.display(), ); - if let Some(why) = toyos_build::image::param_conflict(image, options.kernel_params) { + let params = options.params(); + let asked: Vec<&str> = params.iter().map(String::as_str).collect(); + if let Some(why) = toyos_build::image::param_conflict(image, &asked) { panic!( "[qemu] {why}. `BootOptions::boot_image` replaces the image this call would have \ built, so `kernel_params` cannot arm a guest booting one: build the staged image \ @@ -2528,7 +2574,8 @@ fn build_boot_image_with( PARAMS.get_or_init(|| toyos_build::build::declared_params(&compile::repo_root())); for name in kernel_params { assert!( - actuators.iter().chain(params).any(|a| a == name), + actuators.iter().chain(params).any(|a| a == name) + || toyos_build::build::is_valued_param(name), "{name:?} is a `kernel_params` and the kernel declares no such actuator or parameter" ); } @@ -2661,13 +2708,15 @@ impl QemuInstance { let boot_image = match &options.boot_image { Some(staged) => staged.clone(), None => { + let params = options.params(); + let params: Vec<&str> = params.iter().map(String::as_str).collect(); let disk = build_boot_image_with( test_crate, c_tests, rust_tests, &options.extra_root_files, &features, - options.kernel_params, + ¶ms, options.debug_wait, ); let path = test_dir.join(format!("boot-{seq}.img")); diff --git a/tests/common/volumes.rs b/tests/common/volumes.rs index 2190f5133d..c9380c5085 100644 --- a/tests/common/volumes.rs +++ b/tests/common/volumes.rs @@ -657,6 +657,33 @@ pub fn newest_log(image_path: &Path, start: usize, len: usize) -> Result<(String Ok((newest.clone(), need(found.pop().flatten(), newest)?)) } +/// The whole of this boot's log, oldest line first, across every file it was +/// written to. +/// +/// A boot long enough to rotate writes `.log`, then `_0002.log` and +/// up, and the names are chosen to sort in the order they were written — so the +/// boot's log is their concatenation. A reading that took only the newest would +/// call a rotation a hole. +pub fn whole_log(image_path: &Path, start: usize, len: usize) -> Result, String> { + let image = std::fs::read(image_path).map_err(|e| format!("read the image: {e}"))?; + if start + len > image.len() { + return Err(format!("the image shrank to {} bytes", image.len())); + } + let volume = &image[start..start + len]; + let mut names = log_names(volume)?; + names.sort(); + if names.is_empty() { + return Err("the log volume holds no .log file at all".to_string()); + } + let asked: Vec<&str> = names.iter().map(String::as_str).collect(); + let mut lines = Vec::new(); + for (name, found) in names.iter().zip(read_files(volume, &asked)?) { + let bytes = need(found, name)?; + lines.extend(String::from_utf8_lossy(&bytes).lines().map(|l| format!("{l}\n"))); + } + Ok(lines) +} + /// The loader's own file on the volume, line by line. pub fn loader_log_lines( image_path: &Path, diff --git a/tests/e1000case/system.toml b/tests/e1000case/system.toml index 44088a8666..fad67c164f 100644 --- a/tests/e1000case/system.toml +++ b/tests/e1000case/system.toml @@ -13,6 +13,9 @@ start = ["logd", "netd", "test-runner"] [programs.logd] syscap = ["logread"] +# The record stream's authority: the address on the boot parameter line is +# information, and this row is the whole of what can act on it. +receives = ["netd"] # netd holds the NIC's PCI function and drives it: the descriptor rings, the # register window and the interrupt are its own, and the kernel keeps only the diff --git a/tests/netcase/system.toml b/tests/netcase/system.toml index 86c7b6e1b3..b200246e3b 100644 --- a/tests/netcase/system.toml +++ b/tests/netcase/system.toml @@ -12,10 +12,12 @@ start = ["logd", "netd", "test-runner"] # `every_boot_config_runs_logd` is what refuses a boot config without it: the # kernel keeps the record ring and writes no file, so such a boot's `/log` is -# empty. It claims no device and serves no port; its whole authority is -# `logread`, which is `Rights::LOG | Rights::WAIT` on a `SysCap` duplicate. +# empty. It claims no device and serves no port. [programs.logd] syscap = ["logread"] +# The record stream's authority: the address on the boot parameter line is +# information, and this row is the whole of what can act on it. +receives = ["netd"] # netd holds the NIC's PCI function and drives it: the virtqueues, the register # window and the interrupt are its own, and the kernel keeps only the claim. diff --git a/tests/test-durations b/tests/test-durations index 080d740c1b..fc3794a077 100644 --- a/tests/test-durations +++ b/tests/test-durations @@ -266,6 +266,11 @@ log_partition_layout 477 shards=12 log_poll_outlives_a_close 4738 shards=12 log_reserve_window 7022 shards=12 log_reserve_window_negative 6791 shards=12 +log_stream 24865 shards=12 +log_stream_e1000e 25219 shards=12 +log_stream_no_listener 25182 shards=12 +log_stream_stalled_peer_delivers_whole_records 34591 shards=12 +log_stream_unreachable 25707 shards=12 lseek_past_eof 24 shards=12 machine_reboot 4551 shards=12 metal_job_reboot 2289 shards=12 diff --git a/tests/toyos.rs b/tests/toyos.rs index dc5c24f8c0..809e28a9b6 100644 --- a/tests/toyos.rs +++ b/tests/toyos.rs @@ -384,6 +384,9 @@ const RUST_SKIP: &[&str] = &[ /// what the host staged: the shipping build here, `sched_check_build`'s /// assert-carrying build there. const DRIVEN_AND_SHARED: &[&str] = &[ + // The log-stream arms drive it for the kernel's `exit:` record about it, + // not for anything it does: it is the cheapest process this tree starts. + "empty_dir_stat", // Its shared run is a whole handle-lifecycle gate with its own census; // `userdev_dma_fault` drives the same binary for a different reason // entirely — as the proof the machine still schedules and spawns after a @@ -576,6 +579,28 @@ const MACHINE_TESTS: &[(&str, Sched, Tier)] = &[ // 82574L QEMU models has the register file the T14's I219 has, so this is // where that driver moves real frames before the laptop does. ("https_tls13_e1000e", Sched::Parallel, Tier::Fast), + // `logd`'s second sink: a host listener, the boot's address on the + // parameter line, and the guest's own `/log` as the oracle the wire is + // compared with. The verdicts are a line's arrival and a line-for-line + // comparison; the clocks in it are liveness guards on a guest that stopped + // talking. Fast with the UNMEASURED bootstrap marker until CI prices it. + ("log_stream", Sched::Parallel, Tier::Nightly), + // The same stream over netd's Intel driver, for the same reason + // `https_tls13_e1000e` exists: the T14's NIC is an I219 and this is the + // only machine in reach that runs that driver. + ("log_stream_e1000e", Sched::Parallel, Tier::Nightly), + // A boot told to stream to a port nothing answers on. The verdict is the + // one line the file carries about it and the file being whole regardless. + ("log_stream_no_listener", Sched::Parallel, Tier::Nightly), + // A `log-storm` offered to a stream whose address answers nothing: the + // bounded queue refuses what it cannot hold, counts it, says so in the log, + // and `/log` still carries every record. The control on the accounting. + ("log_stream_unreachable", Sched::Parallel, Tier::Nightly), + // The other half of that: a peer that accepted, stopped reading for a + // storm's worth of records and then read again. What a stall costs in lines + // is the buffers' business and no arm's to demand; what it may never cost + // is half a line, and that is what this one judges. + ("log_stream_stalled_peer_delivers_whole_records", Sched::Parallel, Tier::Nightly), ("netd_connection_caps", Sched::Parallel, Tier::Fast), // The netcase boot again: netd must not abort a listener on a ring flag its // own client forged. Its verdict is a kernel-reported EOF or its absence; @@ -13269,6 +13294,15 @@ fn run_machine_test( } "https_tls13" => common::https::tls13_judge(rust_bins, common::https::VIRTIO), "https_tls13_e1000e" => common::https::tls13_judge(rust_bins, common::https::E1000E), + "log_stream" => common::logstream::stream(common::logstream::VIRTIO, c_bins, rust_bins), + "log_stream_e1000e" => { + common::logstream::stream(common::logstream::E1000E, c_bins, rust_bins) + } + "log_stream_no_listener" => common::logstream::no_listener(c_bins, rust_bins), + "log_stream_unreachable" => common::logstream::unreachable(c_bins, rust_bins), + "log_stream_stalled_peer_delivers_whole_records" => { + common::logstream::stalled_peer(c_bins, rust_bins) + } "netd_connection_caps" => { // The only boot that runs netd at all. Its `main` opens the NIC // first and returns on `NotFound`, so metal-sim never reaches a diff --git a/toyos-logstream/Cargo.toml b/toyos-logstream/Cargo.toml new file mode 100644 index 0000000000..b8f55dd073 --- /dev/null +++ b/toyos-logstream/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "toyos-logstream" +version = "0.1.0" +edition = "2021" +license = "MIT OR Apache-2.0" + +[dependencies] +# `MAX_RECORD_MESSAGE`, which the queue's bound is derived from. +toyos-abi = { path = "../toyos-abi" } diff --git a/toyos-logstream/src/lib.rs b/toyos-logstream/src/lib.rs new file mode 100644 index 0000000000..55d4f2d113 --- /dev/null +++ b/toyos-logstream/src/lib.rs @@ -0,0 +1,452 @@ +//! The record stream: where a boot's log goes besides the file, said once for +//! everyone who has to spell it, and the one decision a peer that will not take +//! it forces. +//! +//! `/system/bin/logd` owns every policy about where records go. The file is the +//! sink of record and this is the second sink: the same text line, in the same +//! order, over a TCP connection netd opens for it, the instant the file gets +//! it. +//! +//! Nothing here can lose a record from the file. [`Backlog::round`] offers a +//! line and never waits: a peer that stops taking bytes fills the queue, and +//! the lines that do not fit are refused, counted, and reported in one line +//! that goes into the file like any other record. A drop nobody can count is +//! the failure this type exists to make impossible, so the refusal and the +//! counter are one statement. +//! +//! Pure: `core` and `alloc`, no `unsafe`, no I/O. + +#![no_std] +#![forbid(unsafe_code)] + +extern crate alloc; + +#[cfg(test)] +extern crate std; + +use alloc::collections::VecDeque; +use alloc::string::{String, ToString}; +use alloc::vec::Vec; + +use toyos_abi::log::MAX_RECORD_MESSAGE; + +/// The boot parameter carrying the listener's address, with the address after +/// it — `logstream=10.0.2.2:41337`. +/// +/// A *valued* parameter, like `blackbox=`: the kernel matches it with +/// `starts_with` rather than whole, so it is not in `kernel/src/params.rs`'s +/// `PARAMS` table and is cleared by name in `src/build.rs`'s `VALUED_PARAMS` +/// instead. +pub const PARAM: &str = "logstream="; + +/// Where the kernel puts [`PARAM`]'s value for userland to find. +/// +/// The kernel command line reaches no process, and the kernel already builds +/// `/system/bin/init`'s environment. `init` passes its own environment on to +/// the daemons it starts at boot and clears it for anything the launcher +/// starts, so `logd` reads this and a program a user runs does not. +pub const ENV: &str = "TOYOS_LOG_STREAM"; + +/// What the kernel copies [`PARAM`]'s value into. +/// +/// The parameter line lives in memory the allocator may hand out, so the value +/// is copied out of it before `mm::init` runs and there is no heap to copy it +/// into. The widest address this can carry is `255.255.255.255:65535`, which is +/// 21 bytes. +pub const MAX_VALUE_BYTES: usize = 64; + +/// The value of [`PARAM`] on a boot parameter line, or `None` when the line +/// does not carry it. +/// +/// The line is comma-separated, as `toyos_abi::boot::actuators` reads it. +pub fn value_in(cmdline: &str) -> Option<&str> { + cmdline.split(',').find_map(|token| token.strip_prefix(PARAM)) +} + +/// Why an address the machine was handed is not one. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum Malformed { + /// No `:` at all, so nothing says which port. + NoPort, + /// The part before the `:` is not four decimal octets. + NotAnAddress, + /// The part after the `:` is not a port, or is zero — which names no + /// listener on any stack. + NotAPort, +} + +impl Malformed { + pub fn as_str(self) -> &'static str { + match self { + Self::NoPort => "it names no port", + Self::NotAnAddress => "the part before the colon is not four decimal octets", + Self::NotAPort => "the part after the colon is not a port between 1 and 65535", + } + } +} + +/// `a.b.c.d:port`, refused by name. +/// +/// A boot whose address is a typo has to say so rather than stream to whatever +/// the typo parsed as. +pub fn endpoint(value: &str) -> Result<([u8; 4], u16), Malformed> { + let (host, port) = value.rsplit_once(':').ok_or(Malformed::NoPort)?; + let mut octets = [0u8; 4]; + let mut seen = 0usize; + for (slot, text) in host.split('.').enumerate() { + let octet = octets.get_mut(slot).ok_or(Malformed::NotAnAddress)?; + // Digits asked for before `parse` is: `u8::from_str` accepts a leading + // `+`, so `10.0.2.+2` would otherwise be this machine's own address + // spelled a way no writer of it meant. + if !text.bytes().all(|b| b.is_ascii_digit()) { + return Err(Malformed::NotAnAddress); + } + *octet = text.parse::().map_err(|_| Malformed::NotAnAddress)?; + seen = slot + 1; + } + if seen != 4 { + return Err(Malformed::NotAnAddress); + } + if !port.bytes().all(|b| b.is_ascii_digit()) { + return Err(Malformed::NotAPort); + } + let port: u16 = port.parse().map_err(|_| Malformed::NotAPort)?; + if port == 0 { + return Err(Malformed::NotAPort); + } + Ok((octets, port)) +} + +/// Records `logd` asks `SYS_LOG_READ` for at once, which is also the most it +/// can offer this queue between two rounds of its own loop. +/// +/// Above `MAX_LOG_SHARDS`, which the call refuses below, and large enough that +/// an ordinary boot's burst is a handful of syscalls rather than one per line. +pub const BATCH: usize = 64; + +/// What `toyos_abi::log::Tagged` renders around a record's message — the +/// brackets, the wall-clock stamp `logd` tags it with, the monotonic +/// `{secs}.{mmm} cpuN`, and the `boot`, `tid=` and elided-byte fields a record +/// may carry — plus the newline `logd` ends the line with. +/// +/// Above every one of those at its widest, which is a claim about what `Tagged` +/// prints and is checked by rendering it: a number short here is a bound that +/// does not hold what it says it holds, and a listener losing lines on a round +/// nobody thought could overflow. +const AROUND_A_MESSAGE: usize = 128; + +/// The widest line one record renders to. +const WIDEST_LINE: usize = MAX_RECORD_MESSAGE + AROUND_A_MESSAGE; + +/// What the queue may hold before a line is refused rather than waited for: +/// one whole `SYS_LOG_READ` batch at its widest, so a listener that misses one +/// round of `logd`'s loop loses nothing. +/// +/// It is deliberately the *small* buffer in the chain. Everything downstream — +/// the pipe netd reads, netd's own send buffer, the peer's receive window — +/// absorbs a stall before this is reached at all, so a line that reaches this +/// bound is a peer that is gone rather than one that is behind. +pub const MAX_BACKLOG_BYTES: usize = BATCH * WIDEST_LINE; + +/// Whether this round may put a line in the log about what the queue refused. +/// +/// `logd` holds the clock and decides how often a run of loss is worth a line; +/// what that line may not be is decided here. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum Due { + Now, + NotYet, +} + +/// The lines waiting for a listener that is slower than the machine. +/// +/// **Bounded, and its refusals are counted.** The alternative — waiting for the +/// socket — puts a listener on the far side of a cable between `logd` and the +/// file it owns, which is the one thing the stream may never cost. +/// +/// Ordering is FIFO and drops are at the tail: nothing is ever reordered, +/// duplicated or evicted after it was taken, so what a listener received is +/// always the file's own lines in the file's own order. Evicting the head +/// instead would keep the end of a boot at the price of making the two +/// readings incomparable. +#[derive(Debug, Default)] +pub struct Backlog { + lines: VecDeque, + bytes: usize, + /// Lines this queue refused, for the life of the process. + dropped: u64, + /// How much of [`Self::dropped`] has been said out loud. + reported: u64, +} + +impl Backlog { + pub const fn new() -> Self { + Self { lines: VecDeque::new(), bytes: 0, dropped: 0, reported: 0 } + } + + /// One round: the lines the file has just taken, offered in order, and the + /// one line the file owes about what this queue refused. + /// + /// **The report is returned and never offered.** A report handed to a queue + /// that is refusing is refused too, which is a drop, which owes another + /// report — a run that never ends and a log that fills with lines about + /// itself. This is the only function that both offers lines and produces + /// the report, so it is the only place that mistake can be made. + pub fn round<'a>( + &mut self, + wrote: impl IntoIterator, + due: Due, + ) -> Option { + for line in wrote { + self.admit(line); + } + match due { + Due::Now => self.report(), + Due::NotYet => None, + } + } + + /// Offer one line, answering whether the queue took it. + /// + /// `false` is a drop and is counted; there is no third answer, and no + /// answer that waits. + fn admit(&mut self, line: &str) -> bool { + if self.bytes + line.len() > MAX_BACKLOG_BYTES { + self.dropped += 1; + return false; + } + self.bytes += line.len(); + self.lines.push_back(line.to_string()); + true + } + + /// Everything waiting, oldest first, leaving the queue empty. + /// + /// The writer takes the whole queue in one step so it holds no lock while + /// it writes: a `logd` blocked behind its own stream thread would be the + /// defect this type exists to prevent, one level in. + pub fn drain(&mut self) -> Vec { + self.bytes = 0; + self.lines.drain(..).collect() + } + + pub fn is_empty(&self) -> bool { + self.lines.is_empty() + } + + /// The one line that says what the stream lost, or `None` when it has lost + /// nothing since it last said so. + fn report(&mut self) -> Option { + let unsaid = self.dropped - self.reported; + if unsaid == 0 { + return None; + } + self.reported = self.dropped; + // **Two numbers, and they are checkable against each other**: every + // line's first number is what this run of loss added, and the second is + // the boot's running total, so a reader that adds the first numbers up + // must arrive at the last line's second one. + Some(alloc::format!( + "logd: {unsaid} record(s) never reached the log stream, and {} in this boot; \ + /log has every one of them", + self.dropped + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A round that offers the queue no line, for the reports. + const NOTHING: [&str; 0] = []; + + #[test] + fn the_value_is_read_off_a_line_that_carries_other_parameters() { + assert_eq!(value_in("logstream=10.0.2.2:41337"), Some("10.0.2.2:41337")); + assert_eq!( + value_in("root=1234,blackbox=0x1000,logstream=10.0.2.2:1,watchdog"), + Some("10.0.2.2:1") + ); + assert_eq!(value_in("root=1234,watchdog"), None); + assert_eq!(value_in(""), None); + // The name whole, not a prefix of another token. + assert_eq!(value_in("mylogstream=1.2.3.4:5"), None); + // Named and empty is not an address; `endpoint` is what refuses it. + assert_eq!(value_in("logstream="), Some("")); + } + + #[test] + fn an_address_is_four_octets_and_a_port() { + assert_eq!(endpoint("10.0.2.2:41337"), Ok(([10, 0, 2, 2], 41337))); + assert_eq!(endpoint("255.255.255.255:65535"), Ok(([255, 255, 255, 255], 65535))); + assert_eq!(endpoint("192.168.1.10:22"), Ok(([192, 168, 1, 10], 22))); + } + + /// Every way a typo reaches this function, refused by name rather than + /// parsed into some other machine's address. + #[test] + fn a_typo_is_refused_and_says_which_kind_it_is() { + assert_eq!(endpoint(""), Err(Malformed::NoPort)); + assert_eq!(endpoint("10.0.2.2"), Err(Malformed::NoPort)); + assert_eq!(endpoint("10.0.2:22"), Err(Malformed::NotAnAddress)); + assert_eq!(endpoint("10.0.2.2.2:22"), Err(Malformed::NotAnAddress)); + assert_eq!(endpoint("10.0.2.256:22"), Err(Malformed::NotAnAddress)); + assert_eq!(endpoint("10.0.2.+2:22"), Err(Malformed::NotAnAddress)); + assert_eq!(endpoint("10.0..2:22"), Err(Malformed::NotAnAddress)); + assert_eq!(endpoint("t14:22"), Err(Malformed::NotAnAddress)); + assert_eq!(endpoint("10.0.2.2:"), Err(Malformed::NotAPort)); + assert_eq!(endpoint("10.0.2.2:65536"), Err(Malformed::NotAPort)); + assert_eq!(endpoint("10.0.2.2:0"), Err(Malformed::NotAPort)); + assert_eq!(endpoint("10.0.2.2:http"), Err(Malformed::NotAPort)); + assert_eq!(endpoint("10.0.2.2:+22"), Err(Malformed::NotAPort)); + // Widest form, so the kernel's copy buffer is not the thing that refuses one. + assert!("255.255.255.255:65535".len() < MAX_VALUE_BYTES); + // Each kind says a different thing, so a boot's log names which typo it was. + let words = [Malformed::NoPort, Malformed::NotAnAddress, Malformed::NotAPort] + .map(Malformed::as_str); + for (i, word) in words.iter().enumerate() { + assert!(!word.is_empty()); + assert!(!words[..i].contains(word), "{word:?} is said by two kinds"); + } + } + + #[test] + fn a_queue_that_is_read_keeps_every_line_in_order() { + let mut q = Backlog::new(); + for i in 0..1000 { + let line = std::format!("line {i}\n"); + assert_eq!(q.round([line.as_str()], Due::Now), None); + assert_eq!(q.drain(), std::vec![line]); + } + assert!(q.is_empty()); + } + + /// **The accounting, which is the whole of what a stalled peer costs.** + /// A queue that stops taking lines and does not count them is the failure + /// no boot can see: the file is whole, the stream is short, and nothing + /// says by how much. + #[test] + fn a_starved_queue_drops_the_newest_and_counts_every_one() { + let line = "x".repeat(1024); + let mut q = Backlog::new(); + let mut admitted = 0u64; + while q.admit(&line) { + admitted += 1; + assert!(admitted < 1_000, "the bound never refused a line"); + } + for _ in 0..99 { + assert!(!q.admit(&line)); + } + + let said = q.round(NOTHING, Due::Now).expect("a queue that dropped says so"); + assert!(said.contains("100 record(s) never reached"), "{said}"); + // One line per episode: nothing new to say until something else drops. + assert_eq!(q.round(NOTHING, Due::Now), None); + let again = q.round([line.as_str()], Due::Now).expect("a second episode says so too"); + assert!(again.contains("1 record(s) never reached"), "{again}"); + assert!(again.contains("and 101 in this boot"), "{again}"); + // And a round that is not due says nothing however much it refused. + assert_eq!(q.round([line.as_str()], Due::NotYet), None); + + // What it did take is a prefix of what it was offered, in order. + let kept = q.drain(); + assert_eq!(kept.len() as u64, admitted); + assert!(kept.iter().all(|k| *k == line)); + assert!(q.is_empty()); + } + + /// **A report that is offered back is a run that never ends.** Written into + /// the file *and* handed to a queue that is refusing, the report is itself + /// refused; that drop owes another report, and the next round owes another, + /// for the life of the boot. So a queue nothing new is offered goes quiet. + #[test] + fn a_drop_report_is_never_a_line_the_queue_is_offered() { + let line = "y".repeat(1024); + let mut q = Backlog::new(); + while q.admit(&line) {} + let said = q.round(NOTHING, Due::Now).expect("a starved queue says so"); + assert!(said.contains("never reached the log stream"), "{said}"); + assert_eq!( + q.round(NOTHING, Due::Now), + None, + "a queue offered nothing new still owes a report, so it reported its own report" + ); + } + + /// A line wider than the whole queue is refused rather than admitted into a + /// queue it does not fit — the bound is on the bytes, not on the count. + #[test] + fn one_impossible_line_does_not_evict_the_boot() { + let mut q = Backlog::new(); + assert!(q.admit("first\n")); + assert!(!q.admit(&"y".repeat(MAX_BACKLOG_BYTES + 1))); + assert!(q.round(NOTHING, Due::Now).expect("a refusal is counted").contains("1 record(s)")); + assert_eq!(q.drain(), std::vec!["first\n"]); + assert!(q.is_empty()); + // And the bytes came back with it: the queue takes lines again. + assert!(q.admit(&"z".repeat(MAX_BACKLOG_BYTES))); + } + + #[test] + fn draining_gives_the_bytes_back() { + let mut q = Backlog::new(); + for _ in 0..64 { + assert!(q.admit(&"a".repeat(1000))); + } + assert_eq!(q.drain().len(), 64); + // The whole bound is available again, which a `bytes` that only ever + // grew would refuse. + assert!(q.admit(&"b".repeat(MAX_BACKLOG_BYTES))); + assert_eq!(q.round(NOTHING, Due::Now), None); + } + + /// The widest line `logd` can put in front of this queue: every field + /// `toyos_abi::log::Tagged` renders at the maximum its type allows, tagged + /// with the widest stamp, ended with the newline `logd` writes. + fn widest_line() -> String { + let mut record = toyos_abi::log::LogRecord::EMPTY; + record.at_ns = u64::MAX; + record.tid = u32::MAX; + record.cpu = u16::MAX; + record.elided = u16::MAX; + record.len = MAX_RECORD_MESSAGE as u16; + record.msg = [b'm'; MAX_RECORD_MESSAGE]; + record.flags = toyos_abi::log::FLAG_EARLY; + // The stamp `logd` tags a record with: `Civil`'s `YYYY-MM-DD HH:MM:SS`, + // or the same width in dashes on a boot with no clock. + alloc::format!("{}\n", record.tagged("9999-12-31 23:59:59")) + } + + /// **The bound holds one whole batch of the widest lines a record renders + /// to**, which is what makes "a listener that misses one round loses + /// nothing" arithmetic rather than a hope. + /// + /// The width is rendered rather than assumed: the claim + /// [`AROUND_A_MESSAGE`] makes is about what `Tagged` prints, so a number + /// short of it fails here and not on a boot. + #[test] + fn the_bound_holds_a_whole_batch_of_the_widest_lines_a_record_renders_to() { + let line = widest_line(); + assert!( + line.len() <= WIDEST_LINE, + "a record renders to {} bytes and the bound allows {WIDEST_LINE}", + line.len() + ); + let mut q = Backlog::new(); + for _ in 0..BATCH { + assert!(q.admit(&line), "the bound refused a line inside one batch"); + } + assert_eq!(q.round(NOTHING, Due::Now), None, "a whole batch was refused a line"); + } + + /// The name and the environment entry are one statement about one machine: + /// a parameter that is not `name=` cannot carry a value, and an environment + /// key with an `=` in it splits in the wrong place. + #[test] + fn the_two_spellings_are_shaped_the_way_their_readers_read_them() { + assert!(PARAM.ends_with('=')); + assert!(!ENV.contains('=')); + assert!(!ENV.contains('\0')); + } +} diff --git a/userland/Cargo.lock b/userland/Cargo.lock index 84ac7b6fd1..a1b96947fc 100644 --- a/userland/Cargo.lock +++ b/userland/Cargo.lock @@ -1801,6 +1801,7 @@ version = "0.1.0" dependencies = [ "toyos 0.6.0", "toyos-abi 0.5.0", + "toyos-logstream", "toyos-wallclock", ] @@ -3768,6 +3769,13 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06db3c1cb438056c9b9a766c8f0dbbfdf3331e723a5bc7f287badc0cfb2f7f78" +[[package]] +name = "toyos-logstream" +version = "0.1.0" +dependencies = [ + "toyos-abi 0.5.0", +] + [[package]] name = "toyos-manifest" version = "0.1.0" diff --git a/userland/logd/Cargo.toml b/userland/logd/Cargo.toml index b2e05c9bbb..0454e1b50a 100644 --- a/userland/logd/Cargo.toml +++ b/userland/logd/Cargo.toml @@ -7,4 +7,5 @@ license = "MIT OR Apache-2.0" [dependencies] toyos = { path = "../../toyos" } toyos-abi = { path = "../../toyos-abi" } +toyos-logstream = { path = "../../toyos-logstream" } toyos-wallclock = { path = "../../toyos-wallclock" } diff --git a/userland/logd/src/main.rs b/userland/logd/src/main.rs index 60f454abe6..16a1b6db3c 100644 --- a/userland/logd/src/main.rs +++ b/userland/logd/src/main.rs @@ -6,17 +6,31 @@ //! FAT volume **from the idle loop**, which is why an idle CPU on this machine //! could be found four spinlocks deep inside a USB transfer with a userland //! `println!` behind it. The kernel keeps the record ring and the console; -//! every policy about files — where they go, what they are called, how many -//! there are, what happens when the stick stops answering — is here. +//! every policy about where records go — what the files are called, how many +//! there are, what happens when the stick stops answering, and whether a copy +//! also leaves the machine over a cable — is here. +//! +//! # Two sinks, and only one of them is the sink of record +//! +//! The file is. `stream`'s is the other: the same text line, in the same order, +//! over a TCP connection netd opens to an address the boot parameter line +//! named. It is best effort in every direction — no address, no netd, no peer, +//! a peer that stopped taking bytes — and none of those may cost `/log` a +//! record or delay a write to it. `stream`'s own header is that argument; what +//! this file owes it is one rule: **a line goes to the volume first and is +//! offered to the stream after, and the offer cannot fail.** //! //! # Its whole authority //! //! One `SysCap` duplicate carrying `Rights::LOG | Rights::WAIT`, which its -//! manifest row asks for by the name `logread`. With it, it may read every -//! record every CPU wrote and park on the readiness source when there is -//! nothing new. It claims no device, opens no compositor connection and can -//! name no process. Writing files is ambient — a known residual of the -//! capability endowment, and not this program's to close. +//! manifest row asks for by the name `logread`, and one `netd` connector, which +//! the same row asks for by name. With the first it may read every record every +//! CPU wrote and park on the readiness source when there is nothing new; the +//! second is the whole of what stands between the address on the parameter line +//! and a peer, since that address is inherited by every program on the machine +//! and this is the only one endowed to act on it. It claims no device, opens no +//! compositor connection and can name no process. Writing files is ambient — a +//! known residual of the capability endowment, and not this program's to close. //! //! # What it does not do, and why the port is not here //! @@ -73,6 +87,7 @@ mod policy; mod store; +mod stream; mod wall; use std::time::Instant; @@ -81,19 +96,14 @@ use toyos::endow::{Endowments, SYSCAP_LABEL}; use toyos::log::{LogTail, Record}; use toyos::poller::{Poller, READABLE}; use toyos::syscap::SysCap; +use toyos_logstream::BATCH; use toyos_wallclock::Civil; use policy::{fate, Fate, Step, LOG_WRITE_BUDGET}; use store::{Volume, DIR, MAX_LOG_BYTES, ROTATE_FAST_BYTES}; +use stream::Stream; use wall::Wall; -/// Records per `SYS_LOG_READ`. -/// -/// Above `MAX_LOG_SHARDS`, which the call refuses below, and large enough that -/// an ordinary boot's burst is a handful of syscalls rather than one per line. -/// A `LogRecord` is a kilobyte, so this is 64 KiB of stack-adjacent buffer held -/// for the life of the process — allocated once, never grown. -const BATCH: usize = 64; /// How long a park on the log's readiness source waits before looking again. @@ -161,6 +171,11 @@ fn main() { ), } + // The second sink, opened after the volume and never before it: the file is + // the sink of record, so nothing about the stream may stand between this + // program and the first line it writes. + let stream = Stream::start(std::env::var(toyos_logstream::ENV).ok().as_deref()); + let mut tail = LogTail::new(); let mut buf = vec![Record::EMPTY; BATCH]; let poller = Poller::new(1); @@ -174,6 +189,8 @@ fn main() { poller.wait(0, 0, |_| {}); let mut lost = 0u64; + // What the stream owes this boot's log, carried until the file takes it. + let mut owed: Vec = Vec::new(); // When the current run of consecutive retries began, or `None` when the // last batch was answered. `policy::fate` bounds the run and not the round. let mut retrying_since: Option = None; @@ -203,7 +220,12 @@ fn main() { lost = tail.lost(); } - if batch.is_empty() { + // Nothing to put a report in, so nothing owes one. + if volume.is_none() { + owed.clear(); + } + + if batch.is_empty() && owed.is_empty() { // **Nothing new, so park on the readiness source rather than spin.** // `SYS_LOG_READ` never blocks by design; this is the other half of // that design. @@ -217,12 +239,43 @@ fn main() { let newest = batch.last().map_or(0, |r| r.at_ns); let began = Instant::now(); let mut refused: Option<(Step, std::io::ErrorKind, String)> = None; - for record in batch.iter() { - let line = format!("{}\n", record.tagged(&stamp(boot_local, record.at_ns))); - if let Err(e) = v.write(line.as_bytes()) { + // What the stream owed at the end of the last round — a connection it + // could not open, or the count of what a peer slower than this machine + // cost. It goes in the file, because the file is where this boot's log + // is: a `say!` reaches the console and no record, so a stream that + // failed silently on the one channel that survives the machine would be + // a failure only somebody watching the wire could see. + // **Dropped as the file takes them, one by one**: a refused write + // leaves the rest owed rather than losing the line that says what the + // stream cost, and leaves none of them to be written a second time. + let mut said_through = 0usize; + for said in &owed { + if let Err(e) = v.write(format!("{said}\n").as_bytes()) { refused = Some((Step::Append, e.kind(), e.to_string())); break; } + said_through += 1; + } + owed.drain(..said_through); + let lines: Vec = + batch.iter().map(|r| format!("{}\n", r.tagged(&stamp(boot_local, r.at_ns)))).collect(); + let mut written = 0usize; + if refused.is_none() { + for line in &lines { + if let Err(e) = v.write(line.as_bytes()) { + refused = Some((Step::Append, e.kind(), e.to_string())); + break; + } + written += 1; + } + } + // **After the file has them, and it cannot fail.** The queue either + // takes a line or counts it as dropped; there is no answer that waits, + // so no listener anywhere can slow this loop down. What comes back is + // the stream's own line about what it refused, which the next round + // writes to the file and never offers back. + if let Some(stream) = stream.as_ref() { + owed.extend(stream.round(lines[..written].iter().map(String::as_str))); } if refused.is_none() { if let Err(e) = v.sync() { diff --git a/userland/logd/src/stream.rs b/userland/logd/src/stream.rs new file mode 100644 index 0000000000..68a6ad5525 --- /dev/null +++ b/userland/logd/src/stream.rs @@ -0,0 +1,245 @@ +//! The second sink: every line the file gets, over a TCP connection netd opens, +//! the instant the file gets it. +//! +//! **The file is the sink of record and this is not.** Nothing here may delay a +//! write to `/log`, refuse one, or lose a record from one: the whole of what +//! `logd` offers the stream is a line it has already written, and the offer +//! cannot fail. +//! +//! The connection is opened, and every byte written, on a thread of its own, +//! because opening it waits on a DHCP lease, a SYN and a peer, and `logd`'s own +//! loop may wait on none of those. Between that thread and the loop stands one +//! bounded queue ([`toyos_logstream::Backlog`]): a peer that stops taking bytes +//! closes the TCP window, netd stops draining the pipe, the writer blocks, the +//! queue fills, and lines are refused — counted, and said out loud in one line +//! that goes into the file and is never offered back to the queue. +//! +//! What it cannot do at all it says once, in the boot's own log. A refusal from +//! the peer is final — netd's `ERR_CONNECTION_REFUSED` means the peer will keep +//! refusing — so the attempt stops there; everything else is this machine not +//! being ready yet, so it is retried until [`OPEN_BOUND`] and then said once. + +use std::sync::{Arc, Condvar, Mutex}; +use std::time::{Duration, Instant}; + +use toyos::net::{self, NetError, TcpConnection}; +use toyos_logstream::{Backlog, Due}; + +/// What netd is given for one connect. +/// +/// It bounds netd's own attempt, not this one: the retry below is what covers a +/// machine whose network is not up yet, and a long timeout here would only make +/// each retry coarser. +const CONNECT_TIMEOUT_MS: u32 = 2_000; + +/// Between two attempts. Short enough that a stream is live within a round of +/// netd becoming ready, long enough that a boot with no network is not spending +/// a core on saying so. +const RETRY_EVERY: Duration = Duration::from_millis(250); + +/// How long this machine has to become able to open the connection at all. +/// +/// netd has to claim the function, bring the link up and finish DHCP before the +/// first SYN can go out, and none of that is bounded by anything `logd` knows. +/// What this catches is a machine that will never have a network, and on that +/// machine the cost of being wrong is one line in the log, late. +const OPEN_BOUND: Duration = Duration::from_secs(30); + +/// How often a run of drops may put a line in the log. +/// +/// **A run of loss is one fact, however many records it covers.** A stream that +/// cannot open at all refuses every line for the whole boot, and a report per +/// round of `logd`'s loop would be the log talking about itself instead of about +/// the machine. Each line carries the boot's running total, so the last one is +/// the whole answer. +const DROP_REPORT_EVERY: Duration = Duration::from_secs(1); + +/// The stream, from `logd`'s side: somewhere to put the lines the file has +/// taken, and what the stream owes the log in return. +pub struct Stream { + shared: Arc, + /// When a run of drops last put a line in the log. + said_at: Mutex>, +} + +struct Shared { + backlog: Mutex, + /// Woken by [`Stream::round`]; waited on by the writer. + ready: Condvar, + /// What the stream could not do, waiting to be written into the log. Said + /// once per episode and then taken. + trouble: Mutex>, +} + +impl Stream { + /// The stream this boot was told to open, or `None` when it was told to + /// open none. + /// + /// **A boot that asked for no stream says nothing**: there is no failure to + /// report, and a line in every ordinary boot's log about a feature nobody + /// asked for is noise. A boot that asked for one with an address that is not + /// one says so, because that is a failure. + pub fn start(said: Option<&str>) -> Option { + let value = said?; + let shared = Arc::new(Shared { + backlog: Mutex::new(Backlog::new()), + ready: Condvar::new(), + trouble: Mutex::new(None), + }); + match toyos_logstream::endpoint(value) { + Ok((addr, port)) => { + let theirs = Arc::clone(&shared); + // Named, because a backtrace out of a blocked write should say + // which of `logd`'s two jobs was blocked. + std::thread::Builder::new() + .name("log-stream".into()) + .spawn(move || run(&theirs, addr, port)) + .expect("logd: the log stream's writer could not be started"); + } + Err(why) => { + say(&shared, format!( + "logd: {}{value} is not an address ({}) - this boot's log is on /log only", + toyos_logstream::PARAM, + why.as_str() + )); + } + } + Some(Self { shared, said_at: Mutex::new(None) }) + } + + /// Offer the lines the file has just taken, and answer what the stream owes + /// this boot's log: a failure it has not reported, or the count of what a + /// peer slower than this machine cost. + /// + /// **The caller writes those lines to the file and does not offer them + /// back.** A drop report handed to a queue that is refusing is refused too, + /// which owes another report, for the life of the boot; + /// `Backlog::round` is where that rule is kept. + /// + /// It cannot fail and it cannot wait: the queue either takes a line or + /// counts it. The only lock it touches is one the writer never holds across + /// a write. + pub fn round<'a>(&self, wrote: impl IntoIterator) -> Vec { + let mut said_at = self.said_at.lock().expect("a mutex is poisoned"); + let due = match *said_at { + Some(at) if at.elapsed() < DROP_REPORT_EVERY => Due::NotYet, + _ => Due::Now, + }; + let report = { + let mut backlog = + self.shared.backlog.lock().expect("the log stream's queue is poisoned"); + backlog.round(wrote, due) + }; + self.shared.ready.notify_one(); + + let mut owed = Vec::new(); + if let Some(said) = self.shared.trouble.lock().expect("a mutex is poisoned").take() { + owed.push(said); + } + if let Some(said) = report { + *said_at = Some(Instant::now()); + owed.push(said); + } + owed + } +} + +/// Open the connection and write the queue into it, for the life of the +/// process. +fn run(shared: &Shared, addr: [u8; 4], port: u16) { + let conn = match open(shared, addr, port) { + Some(conn) => conn, + None => return, + }; + loop { + let batch = { + let mut backlog = shared.backlog.lock().expect("the log stream's queue is poisoned"); + while backlog.is_empty() { + // The lock is given up here and taken again with something in + // the queue; `round` never waits behind this thread. + backlog = + shared.ready.wait(backlog).expect("the log stream's queue is poisoned"); + } + backlog.drain() + }; + for line in batch { + // **Blocking, and on purpose.** A full pipe is netd holding a + // closed TCP window, which is the listener asking this machine to + // slow down; waiting here is what turns that into a bounded queue + // and a counted drop instead of an unbounded one. + if let Err(why) = write_all(&conn, line.as_bytes()) { + say(shared, format!( + "logd: the log stream to {}.{}.{}.{}:{port} ended ({why}) - this boot's log \ + continues on /log only", + addr[0], addr[1], addr[2], addr[3] + )); + return; + } + } + } +} + +/// The connection, or `None` once this machine has been given long enough to +/// have one. +fn open(shared: &Shared, addr: [u8; 4], port: u16) -> Option { + let began = Instant::now(); + let at = format!("{}.{}.{}.{}:{port}", addr[0], addr[1], addr[2], addr[3]); + loop { + match net::tcp_connect(addr, port, CONNECT_TIMEOUT_MS) { + Ok(conn) => return Some(conn), + // **The peer answered, and its answer is final.** A refused SYN is + // a listener that is not there; retrying it for thirty seconds + // would delay the one line that says so and change nothing. + Err(NetError::ConnectionRefused) => { + say(shared, format!( + "logd: nothing is listening at {at} for this boot's log stream - \ + this boot's log is on /log only" + )); + return None; + } + // The manifest gave this program no `netd`, so there is no network + // to wait for either. + Err(NetError::NetdNotFound) => { + say(shared, format!( + "logd: this machine has no netd to reach {at} through - this boot's log \ + is on /log only" + )); + return None; + } + Err(e) => { + if began.elapsed() >= OPEN_BOUND { + say(shared, format!( + "logd: {at} did not answer in {:?} ({e:?}) - this boot's log is on \ + /log only", + OPEN_BOUND + )); + return None; + } + std::thread::sleep(RETRY_EVERY); + } + } + } +} + +/// One `write` is one `SYS_WRITE` and a pipe may take part of a line; a line +/// that arrived in halves would be two lines on the listener's side. +/// +/// The error is carried out rather than collapsed: a stream that stopped is one +/// line in the log, and the reason is the whole of what that line is worth. +fn write_all(conn: &TcpConnection, mut bytes: &[u8]) -> Result<(), String> { + while !bytes.is_empty() { + match conn.tx.write(bytes) { + Ok(0) => return Err("netd took none of it".to_string()), + Ok(n) => bytes = &bytes[n..], + Err(e) => return Err(e.to_string()), + } + } + Ok(()) +} + +/// Leave one line for the log. The writer says at most one thing in its life — +/// it either fails to open or fails to write, and returns either way — and a +/// stream that never started one says its refusal from `start`. +fn say(shared: &Shared, line: String) { + *shared.trouble.lock().expect("a mutex is poisoned") = Some(line); +} diff --git a/userland/netd/src/main.rs b/userland/netd/src/main.rs index bde2a5ca80..d7115e6af8 100644 --- a/userland/netd/src/main.rs +++ b/userland/netd/src/main.rs @@ -1076,14 +1076,34 @@ impl NetDaemon { // forgeable closed flags. while socket.can_send() { if let Some(ref pipe) = conn.tx_read { + // **No more is taken out of the pipe than the socket will + // take from us.** `send_slice` answers how many bytes it + // enqueued and takes fewer when the send buffer is short of + // room; bytes read past that are gone, and the peer's stream + // is short in the middle with nothing saying so. The pipe is + // where the rest belongs until there is room. let mut buf = [0u8; 4096]; - match toyos_abi::syscall::read_nonblock(pipe.as_handle(), &mut buf) { + let want = (socket.send_capacity() - socket.send_queue()).min(buf.len()); + // A zero-length read answers `Ok(0)`, which the arm below + // reads as the client hanging up; asked for no bytes, this + // loop has nothing to do instead. + if want == 0 { + break; + } + match toyos_abi::syscall::read_nonblock(pipe.as_handle(), &mut buf[..want]) { Ok(0) => { socket.close(); conn.close_tx(); break; } - Ok(n) => { let _ = socket.send_slice(&buf[..n]); } + Ok(n) => { + // Both refusals are bytes the pipe has already given + // up, so neither may be swallowed here of all places. + let sent = socket.send_slice(&buf[..n]).unwrap_or_else(|e| { + panic!("netd: a socket that could send refused {n} byte(s): {e:?}") + }); + assert_eq!(sent, n, "netd: the send buffer took {sent} of {n} byte(s) it had room for"); + } _ => break, } } else {