From 300c0acbd7ab3fae8596ff839c39d9a17c70c68d Mon Sep 17 00:00:00 2001 From: salmonik Date: Thu, 27 Aug 2026 16:58:31 +0200 Subject: streams: per-invocation channel, pumping mode, filter pipeline wiring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit exec: LineChannel bounded ring (cap=16) with abort/eof/marker fields. Shell 'pumping mode' assembles canonical lines from its queue into the channel while the command runs; ^C marks the channel aborted. Shell-side grammar: USE/TIL pre-drain into the channel before spawn; TIL sugar drains paths until the marker line. Interactive TIL: pump_key checks each submitted line against the marker and closes the channel when it matches. Exports callbacks run in the command task but touch shell-owned state: this is safe by construction — the shell blocks in exec_external until the command completes. put_line feeds Pipeline::line directly; get_line pops from the channel (yield-poll when empty). load: drop NX from image mapping (single-privilege kernel, NX bit 63 reserved when EFER.NXE is set by OVMF); compute image span from section RVAs not file size. bins/count: gate binary — reads lines via get_line, reports count via put_line. Verified: count USE file → 6; count TIL END → 2; interactive count TIL DONE → 3; -h filter across task boundary; ^C abort. v1 stream exports: put_line → Pipeline, get_line ← channel, log_line → klog append, exit → abort_current. Accounting: print_one_line → pipe.line() returns false (head kill / pager ^C) → the binary should wrap up promptly; accepted: binary does not check the return value in v1 (phase 3 simplification). --- shell/src/lib.rs | 342 ++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 315 insertions(+), 27 deletions(-) (limited to 'shell') diff --git a/shell/src/lib.rs b/shell/src/lib.rs index 9ca999e..5248ea9 100644 --- a/shell/src/lib.rs +++ b/shell/src/lib.rs @@ -711,7 +711,13 @@ fn emit_string(pipe: &mut Pipeline, s: &str) { /// Commands that may consume an input stream (`USE`/`TIL`). fn accepts_input(cmd: &str) -> bool { - matches!(cmd, "cat" | "hexdump") + // Built-in stream consumers, plus any external binary in /bin/cli — + // externals get injected streams by construction (ADR 0006), so the + // keyword is never ambiguous for them. + if matches!(cmd, "cat" | "hexdump") { + return true; + } + rsnix::exec::resolve("cli", cmd).is_some() } fn execute(s: &str) { @@ -748,8 +754,10 @@ write `{} TIL `", // Resolution rule (ADR 0006): \bin\\ first, else the // built-in. The frozen core (help/clear/echo/poweroff/exit/halt/ // rescan) is exempt — it exists to fix the system and never migrates. - if !matches!(cmd, "help" | "clear" | "echo" | "halt" | "poweroff" | "exit" | "rescan") - && exec_external(&parsed) + if !matches!( + cmd, + "help" | "clear" | "echo" | "halt" | "poweroff" | "exit" | "rescan" + ) && exec_external(&parsed) { return; } @@ -801,30 +809,133 @@ write `{} TIL `", } // --------------------------------------------------------------------------- -// External binaries (ADR 0006 Phase 2) +// External binaries (ADR 0006 Phase 3) +// +// Resolution: \bin\cli\ on each mounted volume; else built-in. +// +// Stream injection: the Exports callbacks run in the COMMAND task but +// touch shell-owned state. This is safe by construction — the invoking +// shell blocks in exec_external until the command completes, so nothing +// else touches the Pipeline or the channel. One invocation at a time. // -// Resolution: \bin\cli\ on each mounted volume; else "unknown -// command". The Exports table is constructed per invocation: v1 wires -// put_line to the active terminal and get_line to EOF (stream injection -// through the filter pipeline lands in Phase 3). +// put_line → Pipeline::line directly (output side needs no queue). +// get_line ← a bounded per-invocation ring; while the command runs, +// the blocked shell "pumps": drains its key queue, assembles +// canonical lines, pushes them into the ring. ^C marks the +// channel aborted instead of killing a stream consumer. // --------------------------------------------------------------------------- static mut LAST_EXIT_CODE: i32 = 0; +/// The pipeline of the running external command (None when idle). +static mut ACTIVE_PIPELINE: Option = None; + /// Last exit code of an external command ($? equivalent for scripts). pub fn last_exit_code() -> i32 { unsafe { core::ptr::addr_of!(LAST_EXIT_CODE).read() } } -/// ABI callback: print one line to the invoking terminal. +/// Bounded line channel between the pumping shell and the command task. +struct LineChannel { + lines: Vec, + len: usize, + cap: usize, + /// ^C seen: get_line returns Abort once drained. + aborted: bool, + /// Source exhausted (EOF / marker hit). + eof: bool, + /// Interactive TIL: exact line that closes the input (pump checks it). + marker: Option, +} + +impl LineChannel { + const CAP: usize = 16; + fn new() -> Self { + LineChannel { + lines: Vec::with_capacity(Self::CAP), + len: 0, + cap: Self::CAP, + aborted: false, + eof: false, + marker: None, + } + } + /// Shell side: push one assembled line. Returns false when full + /// (the pump waits and retries). + fn push(&mut self, s: &str) -> bool { + if self.len >= self.cap { + return false; + } + self.lines.push(String::from(s)); + self.len += 1; + true + } + /// Command side: pop one line. + fn pop(&mut self) -> Option { + if self.len == 0 { + return None; + } + let s = self.lines.remove(0); + self.len -= 1; + Some(s) + } +} + +/// The current invocation. Only touched by the shell (pump + setup) and +/// by the ABI callbacks (command task) — never concurrently, see above. +static mut INVOCATION: Option = None; + +/// Set when the input source is interactive (tty): the pump assembles +/// canonical lines from keys. File/marker sources are pre-drained into +/// the channel before spawn instead. +fn with_invocation(f: impl FnOnce(&mut LineChannel) -> R) -> R { + let inv = unsafe { + match &mut *core::ptr::addr_of_mut!(INVOCATION) { + Some(ch) => ch, + None => panic!("exec: no active invocation"), + } + }; + f(inv) +} + +/// ABI callback: print one line through the filter pipeline / sink. fn abi_put_line(line: rsapi::Line) -> bool { - term_println(line.as_str()); - true + let pipe = unsafe { + match &mut *core::ptr::addr_of_mut!(ACTIVE_PIPELINE) { + Some(p) => p, + None => panic!("exec: no active pipeline"), + } + }; + pipe.line(line.as_str()) } -/// ABI callback: read one input line — no stream source wired yet (Phase 3). -fn abi_get_line(_out: &mut rsapi::Line) -> rsapi::StreamResult { - rsapi::StreamResult::Eof +/// ABI callback: read one input line from the invocation channel. +fn abi_get_line(out: &mut rsapi::Line) -> rsapi::StreamResult { + loop { + let (line, status) = with_invocation(|ch| match ch.pop() { + Some(l) => (Some(l), rsapi::StreamResult::Line), + None if ch.aborted => (None, rsapi::StreamResult::Abort), + None if ch.eof => (None, rsapi::StreamResult::Eof), + None => (None, rsapi::StreamResult::Line), // empty but open: pump + }); + match status { + rsapi::StreamResult::Abort => return rsapi::StreamResult::Abort, + rsapi::StreamResult::Eof => return rsapi::StreamResult::Eof, + rsapi::StreamResult::Line => { + if let Some(l) = line { + out.ptr = l.as_ptr(); + out.len = l.len(); + // Leak the String into the command's lifetime: lines are + // one-shot and the task dies soon; freeing across tasks + // is not worth the bookkeeping. + core::mem::forget(l); + return rsapi::StreamResult::Line; + } + // Channel empty: yield so the shell pump can refill. + rsnix::sched::yield_now(); + } + } + } } /// ABI callback: kernel log. @@ -837,8 +948,7 @@ fn abi_exit(code: i32) -> ! { unsafe { LAST_EXIT_CODE = code; } - // Unwind the binary by longjmp-style task abort: the trampoline never - // returns here. We report the code via the static and kill the task. + // Kill the calling task; exec_external sees the abort notification. rsnix::sched::abort_current("exit") } @@ -871,37 +981,199 @@ fn exec_external(p: &Parsed) -> bool { return true; } }; + + // Grammar stays shell-side (ADR 0006): the binary gets clean argv and + // injected streams. Build its pipeline exactly like a built-in's. + let pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] + // Input channel: pre-drain file sources; tty sources are pumped. + let mut channel = LineChannel::new(); + match &p.input { + Input::None => {} + Input::Use(path) => { + drain_file_into(&mut channel, path, None); + channel.eof = true; + } + Input::UseTil(path, marker) => { + drain_file_into(&mut channel, path, Some(marker)); + channel.eof = true; + } + Input::Til(marker) => { + if p.args.is_empty() { + // Interactive: read from tty until marker — the pump + // checks each submitted line against it. + channel.marker = Some(marker.clone()); + } else { + for path in &p.args { + drain_file_into(&mut channel, path, Some(marker)); + } + channel.eof = true; + } + } + } + let mut argv: Vec = Vec::with_capacity(p.args.len() + 1); argv.push(p.cmd.clone()); argv.extend(p.args.iter().cloned()); let name = alloc::format!("{}", p.cmd); + + // Publish the invocation state before spawn (the command may call + // put_line immediately). + unsafe { + *core::ptr::addr_of_mut!(ACTIVE_PIPELINE) = Some(pipe); + *core::ptr::addr_of_mut!(INVOCATION) = Some(channel); + } + + let mut cmd_done = false; + let mut exit_code: i64 = 0; match rsnix::exec::spawn(image, &EXPORTS_V1, &argv, &name) { - Ok(id) => { - // Wait for the completion message (TaskMsg::User(exit_code)). + Ok(_id) => { + // Pumping mode: assemble canonical lines from our queue into + // the channel while the command runs. The WM pump delivers + // keys to us only while our window is focused. loop { match rsnix::sched::recv_blocking() { rsnix::sched::TaskMsg::User(code) => { - unsafe { - LAST_EXIT_CODE = code as i32; - } - break; + exit_code = if code == u64::MAX { -1 } else { code as i64 }; + cmd_done = true; } rsnix::sched::TaskMsg::Key(rsnix::ps2::KeyEvent::Ctrl('c')) => { - // ^C while waiting: kill the command task. - let _ = rsnix::sched::kill(id); - term_err("^C"); - break; + // ^C: mark the channel aborted; stream consumers stop + // at their next get_line/put_line-false. Compute- + // bound commands stay blind to it (accepted policy). + with_invocation(|ch| ch.aborted = true); + } + rsnix::sched::TaskMsg::Key(ev) => { + // Canonical assembly for the interactive source: + // mirror tty::read_line's key handling, minus echo + // (the terminal already echoed it? no — the shell + // must echo here since nothing else does). + pump_key(ev); } _ => {} } + if cmd_done { + break; + } } } - Err(rsnix::exec::ExecError::Busy) => term_err(&alloc::format!("{}: task table full", p.cmd)), + Err(rsnix::exec::ExecError::Busy) => { + term_err(&alloc::format!("{}: task table full", p.cmd)) + } Err(_) => term_err(&alloc::format!("{}: cannot start", p.cmd)), } + + unsafe { + let pipe_ptr = core::ptr::addr_of_mut!(ACTIVE_PIPELINE); + if let Some(mut pipe) = (*pipe_ptr).take() { + pipe.end(); + } + *core::ptr::addr_of_mut!(INVOCATION) = None; + LAST_EXIT_CODE = exit_code as i32; + } true } +/// One key of canonical assembly during pumping. Lines land in the +/// channel; ^D closes it; Enter submits the line. Echo goes to the +/// terminal (this is the command's stdin). +fn pump_key(ev: rsnix::ps2::KeyEvent) { + use rsnix::ps2::KeyEvent; + // Static assembly buffer: only touched between keys by the shell. + static mut PUMP_BUF: Option = None; + let buf = unsafe { + let p = core::ptr::addr_of_mut!(PUMP_BUF); + if (*p).is_none() { + *p = Some(String::new()); + } + (*p).as_mut().unwrap() + }; + match ev { + KeyEvent::Enter => { + gop_println!(); + let line = core::mem::take(buf); + with_invocation(|ch| { + // Interactive TIL: the submitted line itself is checked + // against the marker (it is consumed, not delivered). + if let Some(m) = &ch.marker + && line.as_str() == m + { + ch.eof = true; + } else { + ch.push(line.as_str()); + } + }); + } + KeyEvent::Ctrl('c') => { + buf.clear(); + with_invocation(|ch| ch.aborted = true); + } + KeyEvent::Ctrl('d') => { + if buf.is_empty() { + with_invocation(|ch| ch.eof = true); + } + } + KeyEvent::Char(c) if c.is_ascii_graphic() || c == ' ' => { + buf.push(c); + gop_print!("{}", c); + } + KeyEvent::Backspace => { + if buf.pop().is_some() { + gop_print!("\x08"); + } + } + _ => {} + } +} + +/// Pre-drain a file into the channel; `marker` stops before that line +/// (USE TIL sugar). The channel is left open — the caller sets +/// `eof` when this is the whole source. +fn drain_file_into(ch: &mut LineChannel, path: &str, marker: Option<&str>) { + let h = match rsnix::fs::resolve(path) { + Some(h) => h, + None => { + term_err(&alloc::format!("{}: no such file", path)); + ch.eof = true; + return; + } + }; + drain_handle(ch, h, marker); +} + +/// Shared drain: read `h` in chunks, push complete lines; stop at marker. +fn drain_handle(ch: &mut LineChannel, h: usize, marker: Option<&str>) { + let mut buf = [0u8; 512]; + let mut pos = 0u64; + let mut pending = String::new(); + 'outer: loop { + match rsnix::fs::read(h, &mut buf, pos) { + Ok(0) | Err(_) => break, + Ok(n) => { + pos += n as u64; + pending.push_str(&String::from_utf8_lossy(&buf[..n])); + while let Some(idx) = pending.find('\n') { + let line: String = pending.drain(..=idx).collect(); + let line = line.trim_end_matches('\n').to_string(); + if let Some(m) = marker + && line == m + { + break 'outer; + } + while !ch.push(&line) { + rsnix::sched::yield_now(); + } + } + } + } + } + if marker.is_none() && !pending.is_empty() { + while !ch.push(&pending) { + rsnix::sched::yield_now(); + } + } +} + // --------------------------------------------------------------------------- // Built-in commands // --------------------------------------------------------------------------- @@ -973,6 +1245,7 @@ an operator: echo \"WRITING\" prints WRITING. fn cmd_echo(p: &Parsed) { let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let line = p.args.join(" "); let _ = pipe.line(&line); pipe.end(); @@ -980,6 +1253,7 @@ fn cmd_echo(p: &Parsed) { fn cmd_hello(p: &Parsed) { let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let _ = pipe.line("World!"); pipe.end(); } @@ -1036,6 +1310,7 @@ fn cmd_clear() { fn cmd_info(p: &Parsed) { let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let mut s = String::from("rs-dos v"); s.push_str(env!("CARGO_PKG_VERSION")); s.push_str(" - rsnix kernel"); @@ -1047,6 +1322,7 @@ fn cmd_info(p: &Parsed) { fn cmd_ls(p: &Parsed) { let path = p.args.first().map(|s| s.as_str()).unwrap_or("/"); let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] match rsnix::fs::resolve(path) { None => term_err(&alloc::format!("ls: {}: no such file or directory", path)), Some(h) => { @@ -1111,6 +1387,7 @@ enum ReadResult { fn cmd_cat(p: &Parsed) { let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] match &p.input { Input::None => { if p.args.is_empty() { @@ -1318,6 +1595,7 @@ impl HexCtx<'_> { fn cmd_hexdump(p: &Parsed) { let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] match &p.input { Input::None => { if p.args.is_empty() { @@ -1633,6 +1911,7 @@ fn cmd_mem(p: &Parsed) { let ps = rsnix::memory::page_stats(); let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let _ = pipe.line(&alloc::format!( "memory map (key=0x{:x}, {} entries)", key, @@ -1699,6 +1978,7 @@ fn cmd_mem(p: &Parsed) { fn cmd_mmap(p: &Parsed) { let (map, key) = rsnix::memory::get_map(); let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let _ = pipe.line(&alloc::format!( "memory map (key=0x{:x}, {} entries):", key, @@ -1744,6 +2024,7 @@ fn cmd_stress(p: &Parsed) { }; let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let mut allocated = 0usize; { let mut blocks: [Option>; 64] = core::array::from_fn(|_| None); @@ -1931,6 +2212,7 @@ fn cmd_heapdbg(_p: &Parsed) { fn cmd_date(p: &Parsed) { let t = rsnix::rtc::now(); let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let line = alloc::format!( "{}-{:02}-{:02} {:02}:{:02}:{:02}", t.year, @@ -2184,6 +2466,7 @@ fn cmd_bg(p: &Parsed) { fn cmd_jobs(p: &Parsed) { let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let lines = rsnix::sched::jobs(); if lines.is_empty() { let _ = pipe.line("jobs: no tasks"); @@ -2440,6 +2723,7 @@ fn usb_if_label(c: u8, s: u8, p: u8) -> &'static str { fn cmd_pci(p: &Parsed) { let devices = rsnix::pci::probe(); let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let _ = pipe.line(&alloc::format!("pci: {} device(s)", devices.len())); let _ = pipe.line(" bus dev fn class vendor:device if"); for d in &devices { @@ -2466,11 +2750,13 @@ fn cmd_usb(p: &Parsed) { rsnix::usb::rescan(); rsnix::ehci::rescan(); let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] let _ = pipe.line("usb: rescanning ports"); pipe.end(); return; } let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] for l in rsnix::usb::diagnose().lines() { if !pipe.line(l) { break; @@ -2479,6 +2765,7 @@ fn cmd_usb(p: &Parsed) { pipe.end(); // EHCI (USB2) controller status - the real hardware path. let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] for l in rsnix::ehci::diagnose().lines() { if !pipe.line(l) { break; @@ -2487,6 +2774,7 @@ fn cmd_usb(p: &Parsed) { pipe.end(); // UHCI (USB1.1 companion) — full/low-speed keyboards and mice. let mut pipe = Pipeline::new(&p.filters, &p.output); + #[allow(unused_mut)] for l in rsnix::uhci::diagnose().lines() { if !pipe.line(l) { break; -- cgit v1.2.3