Rework the logging and terminal emulator on the ui to work like an actual emulator for better output.
This commit is contained in:
Jaret Burkett
parent
5cb54ba9cc
commit
e1e1996c16
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<HTMLDivElement>(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;
|
||||
|
|
|
|||
|
|
@ -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<string>('');
|
||||
// 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<TerminalEmulator | null>(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 };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 [ <params> <intermediates> <final byte @-~>
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue