From e1e1996c16520694a3412de7b5e438d3808096f2 Mon Sep 17 00:00:00 2001 From: Jaret Burkett Date: Thu, 16 Jul 2026 08:50:40 -0600 Subject: [PATCH] Rework the logging and terminal emulator on the ui to work like an actual emulator for better output. --- toolkit/print.py | 83 ++++++++- ui/src/app/api/jobs/[jobID]/log/route.ts | 16 +- ui/src/components/CaptionMonitor.tsx | 4 +- ui/src/components/JobOverview.tsx | 10 +- ui/src/hooks/useJobLog.tsx | 25 +-- ui/src/utils/terminalEmulator.ts | 226 +++++++++++++++++++++++ 6 files changed, 327 insertions(+), 37 deletions(-) create mode 100644 ui/src/utils/terminalEmulator.ts diff --git a/toolkit/print.py b/toolkit/print.py index 4d26d243..fbbc82e2 100644 --- a/toolkit/print.py +++ b/toolkit/print.py @@ -1,5 +1,6 @@ import sys import os +import time from toolkit.accelerator import get_accelerator @@ -8,20 +9,81 @@ def print_acc(*args, **kwargs): print(*args, **kwargs) +# Progress bars (tqdm etc.) refresh many times a second with \r / cursor-up +# rewrites. The terminal gets every refresh untouched, but writing them all to +# the log file makes it enormous, so transient refreshes are buffered and +# written at most once per interval — each new refresh of the same line(s) +# replaces the buffered one. Real content (anything with actual text and a +# newline) always writes through immediately, preceded by any buffered refresh +# to preserve stream order. +TRANSIENT_WRITE_INTERVAL = 1.0 +# Safety valve: an unterminated refresh stream writes through past this size. +MAX_TRANSIENT_BUFFER = 65536 + + class Logger: - def __init__(self, filename): - self.terminal = sys.stdout - self.log = open(filename, 'a') + def __init__(self, terminal, log_file): + self.terminal = terminal + self.log = log_file + # Last completed refresh cycle (a new cycle replaces the previous one). + self._cycle = '' + # In-progress refresh: a lone \r line rewrite, or a multi-write + # cursor-movement cycle (nested bars) still being assembled. + self._tail = '' + self._tail_is_lone_refresh = False + self._last_transient_write = 0.0 def write(self, message): self.terminal.write(message) - self.log.write(message) - self.log.flush() # Make sure it's written immediately + self._write_log(message) + + def _write_log(self, message): + has_move = '\r' in message or '\x1b[A' in message + has_newline = '\n' in message + if has_move and not has_newline: + if message.startswith('\r') and '\x1b[A' not in message and (not self._tail or self._tail_is_lone_refresh): + # Single-line rewrite (plain tqdm bar) — replaces the previous one. + self._tail = message + self._tail_is_lone_refresh = True + else: + # Part of a multi-write cursor-movement cycle (nested bars). + self._tail += message + self._tail_is_lone_refresh = False + elif self._tail and has_newline and message.strip('\r\n') == '': + # Pure newline movement closes a multi-write cycle; the completed + # cycle replaces the previously buffered one. + self._cycle = self._tail + message + self._tail = '' + self._tail_is_lone_refresh = False + else: + # Real content — write any buffered refresh first to keep order. + self._flush_transient() + self.log.write(message) + self.log.flush() + self._last_transient_write = time.monotonic() + return + now = time.monotonic() + # Only flush between cycles (or on a lone \r rewrite, which leaves the + # cursor on the same row) — flushing mid-cycle would leave the file's + # cursor moved up and misalign everything written after. + mid_cycle = self._tail and not self._tail_is_lone_refresh + if (now - self._last_transient_write >= TRANSIENT_WRITE_INTERVAL and not mid_cycle) or len( + self._tail) > MAX_TRANSIENT_BUFFER: + self._flush_transient() + self._last_transient_write = now + + def _flush_transient(self): + if self._cycle or self._tail: + self.log.write(self._cycle + self._tail) + self.log.flush() + self._cycle = '' + self._tail = '' + self._tail_is_lone_refresh = False def flush(self): self.terminal.flush() self.log.flush() - + def isatty(self): return self.terminal.isatty() @@ -30,5 +92,10 @@ def setup_log_to_file(filename): if get_accelerator().is_local_main_process: if not os.path.exists(os.path.dirname(filename)): os.makedirs(os.path.dirname(filename)) - sys.stdout = Logger(filename) - sys.stderr = Logger(filename) + # Capture the real streams before replacing them — wrapping the + # already-replaced sys.stdout as the stderr Logger's "terminal" would + # double-write every stderr message to the file. Both wrappers share a + # single file handle. + log_file = open(filename, 'a') + sys.stdout = Logger(sys.stdout, log_file) + sys.stderr = Logger(sys.stderr, log_file) diff --git a/ui/src/app/api/jobs/[jobID]/log/route.ts b/ui/src/app/api/jobs/[jobID]/log/route.ts index 4748524e..c3531662 100644 --- a/ui/src/app/api/jobs/[jobID]/log/route.ts +++ b/ui/src/app/api/jobs/[jobID]/log/route.ts @@ -28,7 +28,12 @@ export async function GET(request: NextRequest, { params }: { params: { jobID: s return NextResponse.json({ log: '', offset: 0, reset: true }); } - const MAX_LINES = 2000; + // Cap on the initial payload. The client renders the log through a terminal + // emulator that collapses \r/cursor-movement rewrites (progress bars), so a + // small newline-counted tail would be mostly bar churn that collapses to a + // few rendered lines — send a generous byte tail instead and let the + // emulator (which caps its own scrollback) do the trimming. + const MAX_TAIL_BYTES = 5 * 1024 * 1024; // Client sends the byte offset it has already consumed so we only return new // content. `offset` omitted (or NaN) => initial load / full tail. const offsetParam = request.nextUrl.searchParams.get('offset'); @@ -52,9 +57,8 @@ export async function GET(request: NextRequest, { params }: { params: { jobID: s const fh = await fs.promises.open(logPath, 'r'); try { if (isReset) { - // Read only the tail of the file to avoid loading huge logs into memory. - // Assume an average line length so we grab enough bytes to cover MAX_LINES. - const start = Math.max(0, size - MAX_LINES * 512); + // Read only the tail of very large files to bound memory/payload. + const start = Math.max(0, size - MAX_TAIL_BYTES); let log = await readRange(fh, start, size); // Drop a partial first line if we started mid-file. if (start > 0) { @@ -63,10 +67,6 @@ export async function GET(request: NextRequest, { params }: { params: { jobID: s log = log.slice(newlineIdx + 1); } } - const lines = log.split('\n'); - if (lines.length > MAX_LINES) { - log = lines.slice(-MAX_LINES).join('\n'); - } return NextResponse.json({ log, offset: size, reset: true }); } // Incremental: return only the bytes appended since the last offset. diff --git a/ui/src/components/CaptionMonitor.tsx b/ui/src/components/CaptionMonitor.tsx index d2e1e021..e7be1526 100644 --- a/ui/src/components/CaptionMonitor.tsx +++ b/ui/src/components/CaptionMonitor.tsx @@ -45,8 +45,8 @@ export default function CaptionMonitor({ datasetPath, onHeightChange }: CaptionM const [isScrolledToBottom, setIsScrolledToBottom] = useState(true); const logLines: string[] = useMemo(() => { - let splits: string[] = log.split(/\n|\r\n/); - splits = splits.map(line => line.split(/\r/).pop()) as string[]; + // Log is already terminal-rendered by useJobLog — one entry per line. + let splits: string[] = log.split('\n'); const maxLines = 1000; if (splits.length > maxLines) { splits = splits.slice(splits.length - maxLines); diff --git a/ui/src/components/JobOverview.tsx b/ui/src/components/JobOverview.tsx index 37df5baf..dcd831dd 100644 --- a/ui/src/components/JobOverview.tsx +++ b/ui/src/components/JobOverview.tsx @@ -20,7 +20,7 @@ export default function JobOverview({ job }: JobOverviewProps) { } return job.gpu_ids.split(',').map(id => parseInt(id)); }, [job.gpu_ids]); - const { log, setLog, status: statusLog, refresh: refreshLog } = useJobLog(job.id, 2000); + const { log, status: statusLog, refresh: refreshLog } = useJobLog(job.id, 2000); const logRef = useRef(null); // Track whether we should auto-scroll to bottom const [isScrolledToBottom, setIsScrolledToBottom] = useState(true); @@ -32,12 +32,8 @@ export default function JobOverview({ job }: JobOverviewProps) { const isStopping = job.stop && job.status === 'running'; const logLines: string[] = useMemo(() => { - // split at line breaks on \n or \r\n but not \r - let splits: string[] = log.split(/\n|\r\n/); - - splits = splits.map(line => { - return line.split(/\r/).pop(); - }) as string[]; + // Log is already terminal-rendered by useJobLog — one entry per line. + let splits: string[] = log.split('\n'); // only return last 100 lines max const maxLines = 1000; diff --git a/ui/src/hooks/useJobLog.tsx b/ui/src/hooks/useJobLog.tsx index 833871a4..001e4dc4 100644 --- a/ui/src/hooks/useJobLog.tsx +++ b/ui/src/hooks/useJobLog.tsx @@ -2,20 +2,21 @@ import { useEffect, useState, useRef } from 'react'; import { apiClient } from '@/utils/api'; +import { TerminalEmulator } from '@/utils/terminalEmulator'; interface FileObject { path: string; size: number; } -const clean = (text: string): string => { - // remove \x1B[A\x1B[A - text = text.replace(/\x1B\[A/g, ''); - return text; -}; - export default function useJobLog(jobID: string, reloadInterval: null | number = null) { const [log, setLog] = useState(''); + // Emulates a terminal over the raw log stream so carriage returns, cursor + // movement, and erase sequences collapse lines like a real terminal. + const terminalRef = useRef(null); + if (terminalRef.current === null) { + terminalRef.current = new TerminalEmulator(); + } const didInitialLoadRef = useRef(false); // Byte offset into the log file that we've already consumed. Sent to the // server so it only returns newly appended content. @@ -42,14 +43,13 @@ export default function useJobLog(jobID: string, reloadInterval: null | number = .then(res => res.data) .then(data => { offsetRef.current = data.offset ?? null; - const cleanLog = clean(data.log ?? ''); + const terminal = terminalRef.current!; if (data.reset) { // Log was reset/truncated (or initial load) — replace everything. - setLog(cleanLog); - } else if (cleanLog) { - // Incremental — append only the new content. - setLog(prev => prev + cleanLog); + terminal.reset(); } + terminal.write(data.log ?? ''); + setLog(terminal.toString()); setStatus('success'); didInitialLoadRef.current = true; }) @@ -67,6 +67,7 @@ export default function useJobLog(jobID: string, reloadInterval: null | number = offsetRef.current = null; didInitialLoadRef.current = false; inFlightRef.current = false; + terminalRef.current?.reset(); setLog(''); refresh(); @@ -81,5 +82,5 @@ export default function useJobLog(jobID: string, reloadInterval: null | number = } }, [jobID]); - return { log, setLog, status, refresh }; + return { log, status, refresh }; } diff --git a/ui/src/utils/terminalEmulator.ts b/ui/src/utils/terminalEmulator.ts new file mode 100644 index 00000000..63ff378b --- /dev/null +++ b/ui/src/utils/terminalEmulator.ts @@ -0,0 +1,226 @@ +// Minimal terminal emulator for rendering job logs the way a real terminal +// would. Handles the control characters and ANSI escape sequences that show up +// in training logs on Linux, macOS, and Windows: +// \r carriage return — cursor to column 0, output overwrites in place +// \n newline — next line (also covers \r\n) +// \b backspace — cursor left one column +// \t tab — advance to next 8-column tab stop +// ESC[A / ESC[B / ESC[C / ESC[D ... cursor movement (tqdm nested bars) +// ESC[K erase in line, ESC[J erase in display +// ESC[m colors/styles — stripped (the UI renders plain text) +// OSC sequences (window title etc.) — stripped +// It is stateful and incremental: feed it chunks as they arrive and it keeps +// cursor position and partial escape sequences across chunk boundaries. + +// Cap the scrollback so multi-day jobs don't grow memory without bound. +const MAX_LINES = 5000; +// If an escape sequence is never terminated, stop buffering it after this. +const MAX_PENDING = 4096; + +export class TerminalEmulator { + private lines: string[] = ['']; + private row = 0; + private col = 0; + // Incomplete escape sequence carried over to the next write(). + private pending = ''; + + reset(): void { + this.lines = ['']; + this.row = 0; + this.col = 0; + this.pending = ''; + } + + write(chunk: string): void { + const text = this.pending + chunk; + this.pending = ''; + const n = text.length; + let i = 0; + while (i < n) { + const code = text.charCodeAt(i); + if (code === 0x1b) { + const consumed = this.consumeEscape(text, i); + if (consumed === 0) { + // Sequence is split across chunks — save it for the next write. + if (n - i <= MAX_PENDING) { + this.pending = text.slice(i); + } + return; + } + i += consumed; + } else if (code === 0x0d) { + // \r + this.col = 0; + i++; + } else if (code === 0x0a) { + // \n + this.newline(); + i++; + } else if (code === 0x08) { + // \b + if (this.col > 0) this.col--; + i++; + } else if (code === 0x09) { + // \t + this.putText(' '.repeat(8 - (this.col % 8))); + i++; + } else if (code < 0x20 || code === 0x7f) { + // Other control chars (BEL, etc.) — ignore. + i++; + } else { + // Bulk-copy a run of printable characters. + let j = i + 1; + while (j < n) { + const c = text.charCodeAt(j); + if (c < 0x20 || c === 0x7f) break; + j++; + } + this.putText(text.slice(i, j)); + i = j; + } + } + } + + /** The rendered screen/scrollback as plain text lines. */ + toLines(): string[] { + return this.lines.slice(); + } + + toString(): string { + return this.lines.join('\n'); + } + + // Overwrite text at the cursor, padding with spaces if the cursor sits past + // the end of the line (just like a real terminal cell grid). + private putText(s: string): void { + let line = this.lines[this.row]; + if (line.length < this.col) { + line = line.padEnd(this.col); + } + this.lines[this.row] = line.slice(0, this.col) + s + line.slice(this.col + s.length); + this.col += s.length; + } + + private newline(): void { + this.row++; + this.col = 0; + this.ensureRow(); + const excess = this.lines.length - MAX_LINES; + if (excess > 0) { + this.lines.splice(0, excess); + this.row = Math.max(0, this.row - excess); + } + } + + private ensureRow(): void { + while (this.lines.length <= this.row) { + this.lines.push(''); + } + } + + // Consume one escape sequence starting at text[i] (which is ESC). Returns + // the number of characters consumed, or 0 if the sequence is incomplete. + private consumeEscape(text: string, i: number): number { + const n = text.length; + if (i + 1 >= n) return 0; + const kind = text[i + 1]; + + if (kind === '[') { + // CSI: ESC [ + let j = i + 2; + while (j < n && /[0-9;?]/.test(text[j])) j++; + while (j < n && text[j] >= ' ' && text[j] <= '/') j++; + if (j >= n) return 0; + const final = text[j]; + if (final >= '@' && final <= '~') { + this.applyCsi(text.slice(i + 2, j), final); + return j - i + 1; + } + // Malformed — drop the ESC [ and let the rest render as text. + return 2; + } + + if (kind === ']') { + // OSC: ESC ] ... terminated by BEL or ST (ESC \) + for (let j = i + 2; j < n; j++) { + if (text.charCodeAt(j) === 0x07) return j - i + 1; + if (text.charCodeAt(j) === 0x1b) { + if (j + 1 >= n) return 0; + if (text[j + 1] === '\\') return j - i + 2; + } + } + return 0; + } + + if (kind === 'M') { + // Reverse index — cursor up one line. + this.row = Math.max(0, this.row - 1); + return 2; + } + + // Any other two-character escape (ESC 7, ESC 8, ESC c, ...) — ignore. + return 2; + } + + private applyCsi(params: string, final: string): void { + const args = params.replace(/^\?/, '').split(';').map(p => parseInt(p, 10)); + const arg = (idx: number, fallback: number): number => { + const v = args[idx]; + return Number.isNaN(v) || v === undefined ? fallback : v; + }; + + switch (final) { + case 'A': // cursor up + case 'F': // cursor to start of previous line + this.row = Math.max(0, this.row - arg(0, 1)); + if (final === 'F') this.col = 0; + break; + case 'B': // cursor down + case 'e': + case 'E': // cursor to start of next line + this.row += arg(0, 1); + this.ensureRow(); + if (final === 'E') this.col = 0; + break; + case 'C': // cursor forward + case 'a': + this.col += arg(0, 1); + break; + case 'D': // cursor back + this.col = Math.max(0, this.col - arg(0, 1)); + break; + case 'G': // cursor to absolute column (1-based) + case '`': + this.col = Math.max(0, arg(0, 1) - 1); + break; + case 'K': { + // Erase in line: 0 = cursor to end, 1 = start to cursor, 2 = all + const mode = arg(0, 0); + const line = this.lines[this.row]; + if (mode === 0) { + this.lines[this.row] = line.slice(0, this.col); + } else if (mode === 1) { + this.lines[this.row] = ' '.repeat(Math.min(this.col + 1, line.length)) + line.slice(this.col + 1); + } else if (mode === 2) { + this.lines[this.row] = ''; + } + break; + } + case 'J': { + // Erase in display: 0 = cursor to end, 2/3 = everything + const mode = arg(0, 0); + if (mode === 0) { + this.lines[this.row] = this.lines[this.row].slice(0, this.col); + this.lines.length = this.row + 1; + } else if (mode === 2 || mode === 3) { + this.reset(); + } + break; + } + // Colors ('m'), cursor show/hide ('h'/'l'), positioning we can't map to + // scrollback ('H'/'f'/'d'), and anything else — ignore. + default: + break; + } + } +}