1#![cfg(feature = "tui")]
4
5use std::collections::VecDeque;
6use std::io;
7use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
8use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, Ordering};
9use std::sync::{Arc, Mutex, Once};
10use std::thread::ThreadId;
11use std::time::{Duration, Instant};
12
13use ratatui::crossterm::event::{
14 self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode, KeyEventKind, MouseEventKind,
15};
16use ratatui::crossterm::terminal::{
17 disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen,
18};
19use ratatui::crossterm::{execute, ExecutableCommand};
20use ratatui::layout::{Alignment, Constraint, Layout, Rect};
21use ratatui::style::{Color, Modifier, Style, Stylize};
22use ratatui::text::{Line, Span, Text};
23use ratatui::symbols::Marker;
24use ratatui::widgets::{
25 Axis, Block, BorderType, Borders, Chart, Dataset, Gauge, GraphType, Paragraph, Wrap,
26};
27use ratatui::{Frame, Terminal};
28
29use crate::scheduler::SchedStats;
30use crate::telemetry::{push_ring_line, LogRing};
31use crate::Daemon;
32
33static SAVED_STDERR: AtomicI32 = AtomicI32::new(-1);
34static PIPE_W: AtomicI32 = AtomicI32::new(-1);
35static TUI_ACTIVE: AtomicBool = AtomicBool::new(false);
36static TUI_THREAD: Mutex<Option<ThreadId>> = Mutex::new(None);
37static PANIC_HOOK: Once = Once::new();
38static DRAIN_REQUESTED: AtomicU64 = AtomicU64::new(0);
39static DRAIN_SEEN: AtomicU64 = AtomicU64::new(0);
40const DRAIN_MARK: &[u8] = b"\x00superfluid-stderr-drain:";
41const MAX_LINE_BYTES: u64 = 16 * 1024;
42
43pub struct StderrCapture {
44 saved: OwnedFd,
45 pipe_w: Arc<OwnedFd>,
46}
47
48impl StderrCapture {
49 pub fn worker_stderr(&self) -> Arc<OwnedFd> {
50 Arc::clone(&self.pipe_w)
51 }
52
53 pub fn drain(&self, timeout: std::time::Duration) {
54 use std::io::Write;
55 let seq = DRAIN_REQUESTED.fetch_add(1, Ordering::SeqCst) + 1;
56 let mark = format!("\x00superfluid-stderr-drain:{seq}\n");
57 let mut w = std::mem::ManuallyDrop::new(unsafe { std::fs::File::from_raw_fd(self.pipe_w.as_raw_fd()) });
60 if w.write_all(mark.as_bytes()).is_err() {
61 return;
62 }
63 let deadline = Instant::now() + timeout;
64 while DRAIN_SEEN.load(Ordering::SeqCst) < seq && Instant::now() < deadline {
65 std::thread::sleep(std::time::Duration::from_millis(2));
66 }
67 }
68}
69
70pub fn capture_stderr(ring: LogRing) -> io::Result<StderrCapture> {
71 use std::io::Write;
72 unsafe {
76 let saved = libc::fcntl(2, libc::F_DUPFD_CLOEXEC, 3);
77 if saved < 0 {
78 return Err(io::Error::last_os_error());
79 }
80 let saved = OwnedFd::from_raw_fd(saved);
81 let mut fds = [0i32; 2];
82 if libc::pipe(fds.as_mut_ptr()) < 0 {
83 return Err(io::Error::last_os_error());
84 }
85 let (pipe_r, pipe_w) = (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1]));
86 libc::fcntl(pipe_r.as_raw_fd(), libc::F_SETFD, libc::FD_CLOEXEC);
87 libc::fcntl(pipe_w.as_raw_fd(), libc::F_SETFD, libc::FD_CLOEXEC);
88 let mirror = libc::fcntl(saved.as_raw_fd(), libc::F_DUPFD_CLOEXEC, 3);
89 if mirror < 0 {
90 return Err(io::Error::last_os_error());
91 }
92 let mut mirror = std::fs::File::from_raw_fd(mirror);
93 let mirror_always = libc::isatty(saved.as_raw_fd()) == 0;
94 let mut reader = std::fs::File::from(pipe_r);
95 std::thread::Builder::new()
96 .name("superfluid-stderr-ring".into())
97 .spawn(move || {
98 use std::io::Read;
99 let mut emit = |raw: &[u8]| {
100 let (raw, seen) = match raw.windows(DRAIN_MARK.len()).position(|w| w == DRAIN_MARK) {
101 Some(at) => {
102 let seq = std::str::from_utf8(&raw[at + DRAIN_MARK.len()..])
103 .ok()
104 .and_then(|t| t.trim_end().parse::<u64>().ok());
105 (&raw[..at], seq)
106 }
107 None => (raw, None),
108 };
109 for piece in raw.chunks(MAX_LINE_BYTES as usize) {
110 if mirror_always || !TUI_ACTIVE.load(Ordering::SeqCst) {
111 let _ = mirror.write_all(piece);
112 }
113 let line = String::from_utf8_lossy(piece);
114 let line = line.trim_end();
115 if !line.is_empty() {
116 push_ring_line(&ring, line.to_string());
117 }
118 }
119 if let Some(seq) = seen {
120 DRAIN_SEEN.fetch_max(seq, Ordering::SeqCst);
121 }
122 };
123 let mut chunk = vec![0u8; MAX_LINE_BYTES as usize];
124 let mut carry: Vec<u8> = Vec::new();
125 loop {
126 let n = match reader.read(&mut chunk) {
127 Ok(0) | Err(_) => break,
128 Ok(n) => n,
129 };
130 carry.extend_from_slice(&chunk[..n]);
131 while let Some(pos) = carry.iter().position(|&b| b == b'\n') {
132 let line: Vec<u8> = carry.drain(..=pos).collect();
133 emit(&line);
134 }
135 while carry.len() as u64 >= MAX_LINE_BYTES {
136 let piece: Vec<u8> = carry.drain(..MAX_LINE_BYTES as usize).collect();
137 emit(&piece);
138 }
139 }
140 if !carry.is_empty() {
141 emit(&carry);
142 }
143 })?;
144 SAVED_STDERR.store(saved.as_raw_fd(), Ordering::SeqCst);
145 PIPE_W.store(pipe_w.as_raw_fd(), Ordering::SeqCst);
146 PANIC_HOOK.call_once(|| {
147 let prev = std::panic::take_hook();
148 std::panic::set_hook(Box::new(move |info| {
149 let on_tui_thread = TUI_ACTIVE.load(Ordering::SeqCst)
150 && TUI_THREAD.lock().ok().and_then(|g| *g) == Some(std::thread::current().id());
151 if on_tui_thread {
152 restore_terminal();
153 }
154 prev(info);
155 }));
156 });
157 Ok(StderrCapture {
158 saved,
159 pipe_w: Arc::new(pipe_w),
160 })
161 }
162}
163
164impl Drop for StderrCapture {
165 fn drop(&mut self) {
166 if TUI_ACTIVE.load(Ordering::SeqCst) {
167 restore_terminal();
168 }
169 let _ = SAVED_STDERR.compare_exchange(
170 self.saved.as_raw_fd(),
171 -1,
172 Ordering::SeqCst,
173 Ordering::SeqCst,
174 );
175 let _ = PIPE_W.compare_exchange(self.pipe_w.as_raw_fd(), -1, Ordering::SeqCst, Ordering::SeqCst);
176 }
177}
178
179fn restore_terminal() {
180 TUI_ACTIVE.store(false, Ordering::SeqCst);
181 if let Ok(mut g) = TUI_THREAD.lock() {
182 *g = None;
183 }
184 let _ = disable_raw_mode();
185 let _ = execute!(io::stdout(), DisableMouseCapture, LeaveAlternateScreen);
186 let _ = io::stdout().execute(ratatui::crossterm::cursor::Show);
187 let saved = SAVED_STDERR.load(Ordering::SeqCst);
188 if saved >= 0 {
189 unsafe {
191 libc::dup2(saved, 2);
192 }
193 }
194}
195
196pub struct NodeInfo {
197 pub identity: String,
198 pub model: String,
199 pub engine: String,
200 pub max_lanes: u32,
201}
202
203const SPINNER: [&str; 10] = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"];
204const HIST: usize = 160;
205const ACCENT: Color = Color::Cyan;
206
207pub fn run(
208 daemon: Arc<Daemon>,
209 ring: LogRing,
210 node: NodeInfo,
211 capture: Option<&StderrCapture>,
212) -> io::Result<()> {
213 struct RestoreGuard;
214 impl Drop for RestoreGuard {
215 fn drop(&mut self) {
216 restore_terminal();
217 }
218 }
219 let guard = RestoreGuard;
220 enable_raw_mode()?;
221 io::stdout().execute(EnterAlternateScreen)?;
222 io::stdout().execute(EnableMouseCapture)?;
223 let mut term = Terminal::new(ratatui::backend::CrosstermBackend::new(io::stdout()))?;
224 if let Some(c) = capture {
225 if let Ok(mut g) = TUI_THREAD.lock() {
226 *g = Some(std::thread::current().id());
227 }
228 TUI_ACTIVE.store(true, Ordering::SeqCst);
229 unsafe {
231 libc::dup2(c.pipe_w.as_raw_fd(), 2);
232 }
233 }
234
235 let res = event_loop(&mut term, daemon, ring, node);
236
237 drop(guard);
238 res
239}
240
241fn event_loop<B: ratatui::backend::Backend>(
242 term: &mut Terminal<B>,
243 daemon: Arc<Daemon>,
244 ring: LogRing,
245 node: NodeInfo,
246) -> io::Result<()> {
247 let stats = daemon.sched_stats();
248 let start = Instant::now();
249 let mut tput: VecDeque<u64> = VecDeque::with_capacity(HIST);
250 let mut pref: VecDeque<u64> = VecDeque::with_capacity(HIST);
251 let mut kvh: VecDeque<u64> = VecDeque::with_capacity(HIST);
252 let mut samples: VecDeque<(Instant, u64)> = VecDeque::with_capacity(8);
253 samples.push_back((Instant::now(), stats.decode_tokens.load(Ordering::Relaxed)));
254 let mut prev_t = Instant::now();
255 let mut last_work_ticks = stats.work_ticks.load(Ordering::Relaxed);
256 let mut frame = 0usize;
257 let mut paused = false;
258 let mut log_scroll: usize = 0;
259
260 loop {
261 if event::poll(Duration::from_millis(125))? {
262 let ev = event::read()?;
263 if let Event::Mouse(m) = &ev {
264 match m.kind {
265 MouseEventKind::ScrollUp => log_scroll = log_scroll.saturating_add(3),
266 MouseEventKind::ScrollDown => log_scroll = log_scroll.saturating_sub(3),
267 _ => {}
268 }
269 }
270 if let Event::Key(k) = ev {
271 if k.kind == KeyEventKind::Press {
272 if k.code == KeyCode::Char('c')
273 && k.modifiers.contains(event::KeyModifiers::CONTROL)
274 {
275 break;
276 }
277 match k.code {
278 KeyCode::Char('q') | KeyCode::Esc => break,
279 KeyCode::Char('p') => paused = !paused,
280 KeyCode::Up => log_scroll = log_scroll.saturating_add(1),
281 KeyCode::Down => log_scroll = log_scroll.saturating_sub(1),
282 KeyCode::PageUp => log_scroll = log_scroll.saturating_add(10),
283 KeyCode::PageDown => log_scroll = log_scroll.saturating_sub(10),
284 KeyCode::Char('g') => log_scroll = usize::MAX / 2,
285 KeyCode::End => log_scroll = 0,
286 _ => {}
287 }
288 }
289 }
290 }
291
292 let now = Instant::now();
293 let dt = (now - prev_t).as_secs_f64();
294 if !paused && dt >= 0.20 {
295 let decode = stats.decode_tokens.load(Ordering::Relaxed);
296 samples.push_back((now, decode));
297 while samples.len() > 2 && now.duration_since(samples[1].0) >= Duration::from_secs(1) {
298 samples.pop_front();
299 }
300 let (t_old, d_old) = samples[0];
301 let win = now.duration_since(t_old).as_secs_f64().max(1e-3);
302 push(&mut tput, ((decode.saturating_sub(d_old)) as f64 / win) as u64);
303 let work_ticks = stats.work_ticks.load(Ordering::Relaxed);
304 let fresh = work_ticks != last_work_ticks;
305 last_work_ticks = work_ticks;
306 let resident = stats.lanes_active.load(Ordering::Relaxed) > 0;
307 push(&mut pref, if resident || fresh { stats.prefill_rate_live.load(Ordering::Relaxed) } else { 0 });
308 push(&mut kvh, stats.pool_blocks_used.load(Ordering::Relaxed));
309 prev_t = now;
310 }
311 frame = frame.wrapping_add(1);
312
313 term.draw(|f| {
314 draw(f, &node, &stats, start, &tput, &pref, &kvh, frame, paused, &ring, log_scroll)
315 })?;
316 }
317 Ok(())
318}
319
320fn push(q: &mut VecDeque<u64>, v: u64) {
321 if q.len() >= HIST {
322 q.pop_front();
323 }
324 q.push_back(v);
325}
326
327#[allow(clippy::too_many_arguments)]
328fn draw(
329 f: &mut Frame,
330 node: &NodeInfo,
331 s: &SchedStats,
332 start: Instant,
333 tput: &VecDeque<u64>,
334 pref: &VecDeque<u64>,
335 kvh: &VecDeque<u64>,
336 frame: usize,
337 paused: bool,
338 ring: &LogRing,
339 log_scroll: usize,
340) {
341 let g = |a: &std::sync::atomic::AtomicU64| a.load(Ordering::Relaxed);
342 let rows = Layout::vertical([
343 Constraint::Length(2),
344 Constraint::Length(7),
345 Constraint::Length(14),
346 Constraint::Length(5),
347 Constraint::Min(6),
348 Constraint::Length(1),
349 ])
350 .split(f.area());
351
352 let spin = SPINNER[(frame / 2) % SPINNER.len()];
353 let up = start.elapsed().as_secs();
354 let header = Line::from(vec![
355 Span::styled(format!(" {spin} superfluid "), Style::new().fg(Color::Black).bg(ACCENT).bold()),
356 Span::styled(format!(" {} ", node.identity), Style::new().fg(ACCENT).bold()),
357 Span::raw(format!("· {} · engine {} · up {:02}:{:02}:{:02}", node.model, node.engine, up / 3600, (up % 3600) / 60, up % 60)),
358 Span::styled(if paused { " [PAUSED]" } else { "" }, Style::new().fg(Color::Yellow).bold()),
359 ]);
360 f.render_widget(Paragraph::new(vec![header, in_tick_line(s, rows[0].width)]), rows[0]);
361
362 let node_cols = Layout::horizontal([Constraint::Percentage(48), Constraint::Percentage(52)]).split(rows[1]);
363 let gauges = Layout::vertical([Constraint::Length(3), Constraint::Length(3)]).split(node_cols[0]);
364 let lanes = g(&s.lanes_active);
365 let maxl = node.max_lanes as u64;
366 let lr = if maxl > 0 { (lanes as f64 / maxl as f64).min(1.0) } else { 0.0 };
367 f.render_widget(
368 Gauge::default()
369 .block(bordered("lane occupancy"))
370 .gauge_style(Style::new().fg(util_color(lr)).bg(Color::Rgb(28, 28, 32)))
371 .ratio(lr)
372 .label(format!("{lanes}/{maxl} lanes · {}% · {} free", (lr * 100.0) as u64, maxl.saturating_sub(lanes))),
373 gauges[0],
374 );
375 let used = g(&s.pool_blocks_used);
376 let total = g(&s.pool_blocks_total);
377 if total > 0 {
378 let pr = (used as f64 / total as f64).min(1.0);
379 f.render_widget(
380 Gauge::default()
381 .block(bordered("KV pool"))
382 .gauge_style(Style::new().fg(util_color(pr)).bg(Color::Rgb(28, 28, 32)))
383 .ratio(pr)
384 .label(format!("{used}/{total} blk · {}% · {} free", (pr * 100.0) as u64, total - used)),
385 gauges[1],
386 );
387 } else {
388 f.render_widget(
389 Paragraph::new(vec![
390 Line::from(vec![
391 Span::styled("blocks in use ", Style::new().dim()),
392 Span::styled(format!("{used}"), Style::new().fg(ACCENT).bold()),
393 ]),
394 Line::from(Span::styled("capacity n/a (engine)", Style::new().dim().italic())),
395 ])
396 .block(bordered("KV pool")),
397 gauges[1],
398 );
399 }
400 let ttft = s.ttft_last_ms.load(Ordering::Relaxed);
401 let warm = warm_pct(s);
402 let (sa, sp) = (g(&s.spec_accepted), g(&s.spec_proposed));
403 let spec = (100 * sa).checked_div(sp).unwrap_or(0);
404 let dec = g(&s.decode_rate_live);
405 let pre = g(&s.prefill_rate_live);
406 let idle = lanes == 0;
407 let rate = |v: u64, col: Color| -> (String, Color) {
408 if idle {
409 (format!("{v:>6} tok/s last"), Color::DarkGray)
410 } else {
411 (format!("{v:>6} tok/s"), col)
412 }
413 };
414 let (dec_s, dec_c) = rate(dec, ACCENT);
415 let (pre_s, pre_c) = rate(pre, Color::Magenta);
416 let stat_lines = vec![
417 kv("decode", dec_s, dec_c),
418 kv("prefill", pre_s, pre_c),
419 kv("ttft last", format!("{ttft:>6} ms"), ttft_color(ttft)),
420 kv("warm reuse", format!("{warm:>5}%"), Color::Green),
421 kv("spec accept", format!("{spec:>5}%"), Color::Blue),
422 ];
423 f.render_widget(Paragraph::new(stat_lines).block(bordered("throughput")), node_cols[1]);
424
425 let mid = Layout::horizontal([Constraint::Percentage(60), Constraint::Percentage(40)]).split(rows[2]);
426 let left = Layout::vertical([Constraint::Percentage(50), Constraint::Percentage(50)]).split(mid[0]);
427 let right = Layout::vertical([Constraint::Length(6), Constraint::Min(4)]).split(mid[1]);
428 rate_chart(f, left[0], "decode · delivered tok/s · 1 s window", tput, ACCENT, None);
429 rate_chart(f, left[1], "prefill · engine tok/s · per tick", pref, Color::Magenta, None);
430 level_chart(f, right[1], "KV pool · blocks used", kvh, Color::Green);
431 let classes = ["chat", "compl", "fg-agent", "bg-agent"];
432 let qlines: Vec<Line> = classes
433 .iter()
434 .enumerate()
435 .map(|(i, name)| {
436 let d = g(&s.queue_depth[i]);
437 let bar = "▊".repeat((d.min(20)) as usize);
438 Line::from(vec![
439 Span::styled(format!("{name:>8} "), Style::new().dim()),
440 Span::styled(bar, Style::new().fg(if d == 0 { Color::DarkGray } else { Color::Yellow })),
441 Span::raw(format!(" {d}")),
442 ])
443 })
444 .collect();
445 f.render_widget(Paragraph::new(qlines).block(bordered("queue depth / QoS")), right[0]);
446
447 let c = |name: &str, v: u64, col: Color| {
448 vec![
449 Span::styled(format!("{name} "), Style::new().dim()),
450 Span::styled(format!("{v}"), Style::new().fg(col).bold()),
451 Span::raw(" "),
452 ]
453 };
454 let mut cline: Vec<Span> = Vec::new();
455 cline.extend(c("ticks", g(&s.ticks), Color::White));
456 cline.extend(c("parks✓", g(&s.parks_lossless), Color::Green));
457 cline.extend(c("parks~", g(&s.parks_lossy), Color::Yellow));
458 cline.extend(c("resumes", g(&s.resumes), Color::Green));
459 cline.extend(c("preempt", g(&s.preemptions), Color::Magenta));
460 let mut cline2: Vec<Span> = Vec::new();
461 cline2.extend(c("evictions", g(&s.pressure_evictions), Color::Yellow));
462 cline2.extend(c("respawns", g(&s.worker_respawns), Color::Red));
463 cline2.extend(c("cold", g(&s.cold_admissions), Color::Blue));
464 cline2.extend(c("starve-grants", g(&s.starvation_grants), Color::Magenta));
465 cline2.extend(c("os-pressure", g(&s.os_pressure_events), Color::Red));
466 let idle_now = lanes == 0;
467 let mut cline3: Vec<Span> = Vec::new();
468 cline3.extend(c(
469 if idle_now { "last-tick(ms)" } else { "tick(ms)" },
470 g(&s.tick_wall_ms_live),
471 if idle_now { Color::DarkGray } else { Color::White },
472 ));
473 cline3.extend(c("grant(tok/lane)", g(&s.decode_grant_live), Color::Cyan));
474 cline3.extend(c("prefill-budget(tok)", g(&s.prefill_budget_live), Color::Magenta));
475 f.render_widget(
476 Paragraph::new(vec![Line::from(cline), Line::from(cline2), Line::from(cline3)])
477 .block(bordered("counters")),
478 rows[3],
479 );
480
481 render_log(f, rows[4], ring, log_scroll);
482
483 f.render_widget(
484 Paragraph::new(Line::from(vec![Span::styled(
485 " q quit · p pause · ↑↓/PgUp/PgDn/wheel scroll · End follow ",
486 Style::new().dim(),
487 )]))
488 .alignment(Alignment::Center),
489 rows[5],
490 );
491}
492
493fn rate_chart(f: &mut Frame, area: Rect, title: &str, data: &VecDeque<u64>, col: Color, ymax: Option<u64>) {
494 let n = data.len();
495 let now = data.back().copied().unwrap_or(0);
496 let peak = data.iter().copied().max().unwrap_or(0);
497 let top = ymax.unwrap_or_else(|| nice_ceil(peak));
498 let pts: Vec<(f64, f64)> = data
499 .iter()
500 .enumerate()
501 .map(|(i, v)| ((HIST - n + i) as f64, (*v).min(top) as f64))
502 .collect();
503 let ds = Dataset::default()
504 .marker(Marker::Braille)
505 .graph_type(GraphType::Line)
506 .style(Style::new().fg(col))
507 .data(&pts);
508 let mid = top / 2;
509 let chart = Chart::new(vec![ds])
510 .block(bordered(&format!("{title} · now {now} · peak {peak}")))
511 .x_axis(Axis::default().bounds([0.0, (HIST - 1) as f64]))
512 .y_axis(
513 Axis::default()
514 .bounds([0.0, top as f64])
515 .style(Style::new().dim())
516 .labels([
517 Span::styled("0", Style::new().dim()),
518 Span::styled(format!("{mid}"), Style::new().dim()),
519 Span::styled(format!("{top}"), Style::new().dim()),
520 ]),
521 );
522 f.render_widget(chart, area);
523}
524
525fn level_chart(f: &mut Frame, area: Rect, title: &str, data: &VecDeque<u64>, col: Color) {
526 let n = data.len();
527 let now = data.back().copied().unwrap_or(0);
528 let hi = data.iter().copied().max().unwrap_or(0);
529 let lo = data.iter().copied().min().unwrap_or(0);
530 let (bottom, top) = level_band(lo, hi);
531 let pts: Vec<(f64, f64)> = data
532 .iter()
533 .enumerate()
534 .map(|(i, v)| ((HIST - n + i) as f64, (*v).clamp(bottom, top) as f64))
535 .collect();
536 let ds = Dataset::default()
537 .marker(Marker::Braille)
538 .graph_type(GraphType::Line)
539 .style(Style::new().fg(col))
540 .data(&pts);
541 let mid = bottom + (top - bottom) / 2;
542 let chart = Chart::new(vec![ds])
543 .block(bordered(&format!("{title} · now {now} · low {lo} · high {hi}")))
544 .x_axis(Axis::default().bounds([0.0, (HIST - 1) as f64]))
545 .y_axis(
546 Axis::default()
547 .bounds([bottom as f64, top as f64])
548 .style(Style::new().dim())
549 .labels([
550 Span::styled(format!("{bottom}"), Style::new().dim()),
551 Span::styled(format!("{mid}"), Style::new().dim()),
552 Span::styled(format!("{top}"), Style::new().dim()),
553 ]),
554 );
555 f.render_widget(chart, area);
556}
557
558fn level_band(lo: u64, hi: u64) -> (u64, u64) {
559 let span = hi.saturating_sub(lo).max(1);
560 let mut step = 1u64;
561 while step.saturating_mul(10) <= span {
562 step = step.saturating_mul(10);
563 }
564 let step = step.max(10);
565 let bottom = (lo / step) * step;
566 let top = hi.div_ceil(step) * step;
567 let top = if top == bottom { bottom + step } else { top };
568 (bottom.min(lo), top.max(hi))
569}
570
571fn nice_ceil(v: u64) -> u64 {
572 if v <= 10 {
573 return 10;
574 }
575 let mut m = 1u64;
576 while m.saturating_mul(10) <= v {
577 m = m.saturating_mul(10);
578 }
579 for k in [1u64, 2, 5, 10] {
580 if k.saturating_mul(m) >= v {
581 return k.saturating_mul(m);
582 }
583 }
584 10 * m
585}
586
587fn in_tick_line(s: &SchedStats, width: u16) -> Line<'static> {
588 let Some(t) = s.tick_in_flight() else {
589 return Line::from(Span::styled(" between ticks", Style::new().dim()));
590 };
591 let age_ms = crate::scheduler::uptime_ms().saturating_sub(t.since_ms);
592 let age = if age_ms < 1000 || t.target_ms < 1000 {
593 format!("{age_ms}ms")
594 } else {
595 format!("{:.1}s", age_ms as f64 / 1e3)
596 };
597 let target = (t.target_ms as f64).max(1.0);
598 let col = if age_ms as f64 > 3.0 * target {
599 Color::Red
600 } else if age_ms as f64 > 1.25 * target {
601 Color::Yellow
602 } else {
603 Color::Green
604 };
605 let lanes = |n: u64| format!("{n} lane{}", if n == 1 { "" } else { "s" });
606 let mut segs: Vec<String> = Vec::new();
607 if t.prefill_tokens > 0 {
608 segs.push(format!("prefill {} tok/{}", t.prefill_tokens, lanes(t.prefill_lanes)));
609 }
610 if t.decode_lanes > 0 {
611 segs.push(format!("decode {}", lanes(t.decode_lanes)));
612 }
613 if t.admits > 0 {
614 segs.push(format!("admit {}", t.admits));
615 }
616 if t.retires > 0 {
617 segs.push(format!("retire {}", t.retires));
618 }
619 if segs.is_empty() {
620 segs.push("empty plan".into());
621 }
622 let prefix = format!(" in tick {age}: ");
623 let budget = (width as usize).saturating_sub(prefix.chars().count());
624 let mut text = String::new();
625 let mut dropped = false;
626 for (i, seg) in segs.iter().enumerate() {
627 let piece = if i == 0 { seg.clone() } else { format!(" · {seg}") };
628 let tail = if i + 1 < segs.len() { 2 } else { 0 };
629 if text.chars().count() + piece.chars().count() + tail <= budget {
630 text.push_str(&piece);
631 } else {
632 dropped = true;
633 break;
634 }
635 }
636 if dropped {
637 text.push_str(" …");
638 }
639 Line::from(Span::styled(format!("{prefix}{text}"), Style::new().fg(col).bold()))
640}
641
642fn render_log(f: &mut Frame, area: Rect, ring: &LogRing, scroll: usize) {
643 let lines: Vec<String> = ring.lock().map(|r| r.iter().cloned().collect()).unwrap_or_default();
644 let inner_h = area.height.saturating_sub(2) as usize;
645 let total = lines.len();
646 let scroll = scroll.min(total.saturating_sub(inner_h.max(1)));
647 let end = total.saturating_sub(scroll);
648 let begin = end.saturating_sub(inner_h);
649 let view: Vec<Line> = lines[begin..end].iter().map(|l| color_log(l)).collect();
650 let title = if scroll == 0 {
651 "logs (following)".to_string()
652 } else {
653 format!("logs (−{scroll})")
654 };
655 f.render_widget(
656 Paragraph::new(Text::from(view)).block(bordered(&title)).wrap(Wrap { trim: false }),
657 area,
658 );
659}
660
661fn color_log(l: &str) -> Line<'static> {
662 let (col, dim) = if l.contains("ERROR") {
663 (Color::Red, false)
664 } else if l.contains("WARN") {
665 (Color::Yellow, false)
666 } else if l.contains("DEBUG") || l.contains("TRACE") {
667 (Color::DarkGray, true)
668 } else {
669 (Color::Gray, false)
670 };
671 let mut st = Style::new().fg(col);
672 if dim {
673 st = st.add_modifier(Modifier::DIM);
674 }
675 Line::from(Span::styled(l.to_string(), st))
676}
677
678fn bordered(title: &str) -> Block<'static> {
679 Block::default()
680 .borders(Borders::ALL)
681 .border_type(BorderType::Rounded)
682 .border_style(Style::new().fg(Color::Rgb(70, 70, 80)))
683 .title(Span::styled(format!(" {title} "), Style::new().fg(ACCENT).bold()))
684}
685
686fn kv(k: &str, v: String, col: Color) -> Line<'static> {
687 Line::from(vec![
688 Span::styled(format!("{k:>11} "), Style::new().dim()),
689 Span::styled(v, Style::new().fg(col).bold()),
690 ])
691}
692
693fn util_color(r: f64) -> Color {
694 if r > 0.85 {
695 Color::Red
696 } else if r > 0.6 {
697 Color::Yellow
698 } else {
699 Color::Green
700 }
701}
702fn ttft_color(ms: u64) -> Color {
703 if ms > 800 {
704 Color::Red
705 } else if ms > 300 {
706 Color::Yellow
707 } else {
708 Color::Green
709 }
710}
711
712fn warm_pct(s: &SchedStats) -> u64 {
713 let warm = s.warm_prefix_tokens.load(Ordering::Relaxed);
714 let cold = s.prefill_tokens.load(Ordering::Relaxed);
715 let tot = warm + cold;
716 (100 * warm).checked_div(tot).unwrap_or(0)
717}
718
719#[cfg(test)]
720mod chart_tests {
721 use super::level_band;
722
723 #[test]
724 fn level_band_frames_the_window_with_round_headroom() {
725 assert_eq!(level_band(9369, 9800), (9300, 9800));
726 assert_eq!(level_band(9369, 9369), (9360, 9370));
727 assert_eq!(level_band(0, 0), (0, 10));
728 assert_eq!(level_band(3, 7), (0, 10));
729 assert_eq!(level_band(1200, 31000), (0, 40000));
730 for (lo, hi) in [(0, 1), (5, 5), (999, 1001), (12345, 67890)] {
731 let (b, t) = level_band(lo, hi);
732 assert!(b <= lo && t >= hi && t > b, "{lo}..{hi} -> {b}..{t}");
733 }
734 }
735}
736
737#[cfg(test)]
738mod capture_tests {
739 use super::*;
740 use std::io::Write;
741 use std::time::Duration;
742
743 #[test]
744 fn worker_end_lines_reach_the_ring_and_fd2_is_untouched() {
745 let ring = crate::telemetry::new_log_ring();
746 let before = unsafe {
748 let mut st: libc::stat = std::mem::zeroed();
749 assert_eq!(libc::fstat(2, &mut st), 0);
750 (st.st_dev, st.st_ino)
751 };
752 let cap = capture_stderr(ring.clone()).expect("capture");
753 let w = cap.worker_stderr();
754 let mut f = std::fs::File::from(w.try_clone().expect("clone"));
755 f.write_all(b"[capture-test] one\n").unwrap();
756 f.write_all(b"[capture-test] \xff\xfe bad\n").unwrap();
757 let long = vec![b'x'; (MAX_LINE_BYTES as usize) * 2 + 7];
758 f.write_all(&long).unwrap();
759 f.write_all(b"\n[capture-test] two\n[capture-test] three\n[capture-test] four\n").unwrap();
760 cap.drain(Duration::from_secs(5));
761 {
762 let lines = ring.lock().unwrap();
763 for want in ["[capture-test] two", "[capture-test] three", "[capture-test] four"] {
764 assert!(lines.iter().any(|l| l == want), "after drain, missing {want}: {:?}", *lines);
765 }
766 }
767 let deadline = Instant::now() + Duration::from_secs(5);
768 loop {
769 let n = ring.lock().unwrap().iter().filter(|l| l.contains("[capture-test]")).count();
770 if n >= 3 {
771 break;
772 }
773 assert!(Instant::now() < deadline, "ring never saw the lines: {:?}", ring.lock().unwrap());
774 std::thread::sleep(Duration::from_millis(10));
775 }
776 let after = unsafe {
778 let mut st: libc::stat = std::mem::zeroed();
779 assert_eq!(libc::fstat(2, &mut st), 0);
780 (st.st_dev, st.st_ino)
781 };
782 assert_eq!(before, after, "installing the capture must not move fd 2");
783 let lines: Vec<String> = ring.lock().unwrap().iter().cloned().collect();
784 assert!(lines.iter().any(|l| l == "[capture-test] one"), "{lines:?}");
785 assert!(lines.iter().any(|l| l == "[capture-test] two"), "{lines:?}");
786 assert!(lines.iter().any(|l| l.starts_with("[capture-test] ") && l.ends_with(" bad")), "{lines:?}");
787 let xs: usize = lines.iter().filter(|l| l.bytes().all(|b| b == b'x')).map(|l| l.len()).sum();
788 assert_eq!(xs, (MAX_LINE_BYTES as usize) * 2 + 7, "long run published in pieces: {lines:?}");
789 assert!(lines.iter().filter(|l| l.bytes().all(|b| b == b'x')).all(|l| l.len() <= MAX_LINE_BYTES as usize));
790 drop(f);
791 drop(w);
792 drop(cap);
793 }
794}