diff --git a/.gitignore b/.gitignore index a6f4b109..cfc91851 100644 --- a/.gitignore +++ b/.gitignore @@ -190,4 +190,5 @@ aitk_db.db-shm /notes.md /data .claude -original_repo \ No newline at end of file +original_repo +.next \ No newline at end of file diff --git a/ui/src/app/api/monitor/route.ts b/ui/src/app/api/monitor/route.ts new file mode 100644 index 00000000..baa877d3 --- /dev/null +++ b/ui/src/app/api/monitor/route.ts @@ -0,0 +1,60 @@ +import { startMonitor } from '@/server/monitor'; +import { MonitorSample } from '@/types'; + +// SSE stream of system stats. On connect: an `init` event with the 2-minute +// rolling history plus the latest full sample; then a `sample` event every +// MONITOR_TICK_MS. Auth is the normal middleware bearer check, so the client +// uses fetch (EventSource can't send headers). +export const dynamic = 'force-dynamic'; + +export async function GET(request: Request) { + const monitor = startMonitor(); + const encoder = new TextEncoder(); + let unsubscribe: (() => void) | null = null; + + const stream = new ReadableStream({ + start(controller) { + const send = (event: string, data: unknown) => { + controller.enqueue(encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)); + }; + const cleanup = () => { + unsubscribe?.(); + unsubscribe = null; + try { + controller.close(); + } catch { + // already closed + } + }; + try { + send('init', monitor.getInit()); + } catch { + cleanup(); + return; + } + unsubscribe = monitor.subscribe((sample: MonitorSample) => { + try { + send('sample', sample); + } catch { + // Client is gone but abort hasn't fired yet + cleanup(); + } + }); + request.signal.addEventListener('abort', cleanup); + }, + cancel() { + unsubscribe?.(); + unsubscribe = null; + }, + }); + + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + // Belt-and-braces against any buffering proxy in front of us + 'X-Accel-Buffering': 'no', + }, + }); +} diff --git a/ui/src/components/GPUMonitor.tsx b/ui/src/components/GPUMonitor.tsx index e28a1ee3..7e83af34 100644 --- a/ui/src/components/GPUMonitor.tsx +++ b/ui/src/components/GPUMonitor.tsx @@ -1,43 +1,13 @@ -import React, { useState, useRef, useMemo } from 'react'; -import { GPUApiResponse } from '@/types'; +import React, { useMemo } from 'react'; import Loading from '@/components/Loading'; import GPUWidget from '@/components/GPUWidget'; -import { apiClient } from '@/utils/api'; -import usePollLoop from '@/hooks/usePollLoop'; +import useMonitorStream from '@/hooks/useMonitorStream'; const GpuMonitor: React.FC = () => { - const [gpuData, setGpuData] = useState(null); - const [loading, setLoading] = useState(true); - const [error, setError] = useState(null); - const [lastUpdated, setLastUpdated] = useState(null); - const isFetchingGpuRef = useRef(false); - - const fetchGpuInfo = () => { - if (isFetchingGpuRef.current) { - return; - } - setLoading(true); - isFetchingGpuRef.current = true; - return apiClient - .get('/api/gpu') - .then(res => res.data) - .then(data => { - setGpuData(data); - setLastUpdated(new Date()); - setError(null); - }) - .catch(err => { - setError(`Failed to fetch GPU data: ${err instanceof Error ? err.message : String(err)}`); - }) - .finally(() => { - isFetchingGpuRef.current = false; - setLoading(false); - }); - }; - - // Fetch every second, but only schedule the next fetch after the current - // one finishes so slow responses can't stack requests - usePollLoop(fetchGpuInfo, 1000); + // Live samples arrive every MONITOR_TICK_MS over the shared /api/monitor SSE stream + const { gpu: gpuData, lastUpdated } = useMonitorStream(); + const loading = gpuData === null; + const error = null; const getGridClasses = (gpuCount: number): string => { switch (gpuCount) { diff --git a/ui/src/hooks/useCPUInfo.tsx b/ui/src/hooks/useCPUInfo.tsx index 7563f448..ee55d895 100644 --- a/ui/src/hooks/useCPUInfo.tsx +++ b/ui/src/hooks/useCPUInfo.tsx @@ -1,30 +1,24 @@ 'use client'; -import { CpuInfo } from '@/types'; -import { useState } from 'react'; -import { apiClient } from '@/utils/api'; -import usePollLoop from '@/hooks/usePollLoop'; +import useMonitorStream from '@/hooks/useMonitorStream'; +/** + * CPU stats from the shared /api/monitor SSE stream. Data arrives live every + * MONITOR_TICK_MS, so `reloadInterval` is accepted only for call-site + * compatibility with the old polling implementation and is ignored. + */ export default function useCPUInfo(reloadInterval: null | number = null) { - const [cpuInfo, setCpuInfo] = useState(null); - const [isCPUInfoLoaded, setIsLoaded] = useState(false); - const [status, setStatus] = useState<'idle' | 'loading' | 'success' | 'error'>('idle'); + void reloadInterval; + const { cpu, connected } = useMonitorStream(); - const fetchCpuInfo = async () => { - setStatus('loading'); - try { - const data: CpuInfo = await apiClient.get('/api/cpu').then(res => res.data); - setCpuInfo(data); - setStatus('success'); - } catch (err) { - console.error(`Failed to fetch CPU data: ${err instanceof Error ? err.message : String(err)}`); - setStatus('error'); - } finally { - setIsLoaded(true); - } + const isCPUInfoLoaded = cpu !== null; + const status: 'idle' | 'loading' | 'success' | 'error' = cpu !== null ? 'success' : connected ? 'loading' : 'idle'; + + return { + cpuInfo: cpu, + isCPUInfoLoaded, + status, + // The stream is always live; nothing to refresh manually. + refreshCpuInfo: async () => {}, }; - - usePollLoop(fetchCpuInfo, reloadInterval); - - return { cpuInfo, isCPUInfoLoaded, status, refreshCpuInfo: fetchCpuInfo }; } diff --git a/ui/src/hooks/useGPUInfo.tsx b/ui/src/hooks/useGPUInfo.tsx index 4cb69a56..214f10e5 100644 --- a/ui/src/hooks/useGPUInfo.tsx +++ b/ui/src/hooks/useGPUInfo.tsx @@ -1,44 +1,39 @@ 'use client'; -import { GPUApiResponse, GpuInfo } from '@/types'; -import { useEffect, useState } from 'react'; -import { apiClient } from '@/utils/api'; -import usePollLoop from '@/hooks/usePollLoop'; +import { GpuInfo } from '@/types'; +import { useMemo } from 'react'; +import useMonitorStream from '@/hooks/useMonitorStream'; +/** + * GPU stats from the shared /api/monitor SSE stream. Data arrives live every + * MONITOR_TICK_MS, so `reloadInterval` is accepted only for call-site + * compatibility with the old polling implementation and is ignored. + */ export default function useGPUInfo(gpuIds: null | number[] = null, reloadInterval: null | number = null) { - const [gpuList, setGpuList] = useState([]); - const [isGPUInfoLoaded, setIsLoaded] = useState(false); - const [status, setStatus] = useState<'idle' | 'loading' | 'success' | 'error'>('idle'); + void reloadInterval; + const { gpu, connected } = useMonitorStream(); - const fetchGpuInfo = async () => { - setStatus('loading'); - try { - const data: GPUApiResponse = await apiClient.get('/api/gpu').then(res => res.data); - let gpus = data.gpus.sort((a, b) => a.index - b.index); - if (gpuIds) { - gpus = gpus.filter(gpu => gpuIds.includes(gpu.index)); - } - setGpuList(gpus); - setStatus('success'); - } catch (err) { - console.error(`Failed to fetch GPU data: ${err instanceof Error ? err.message : String(err)}`); - setStatus('error'); - } finally { - setIsLoaded(true); + // Key on contents, not identity — call sites often build gpuIds inline. + const gpuIdsKey = gpuIds ? gpuIds.join(',') : null; + const gpuList: GpuInfo[] = useMemo(() => { + if (!gpu) return []; + let gpus = [...gpu.gpus].sort((a, b) => a.index - b.index); + if (gpuIdsKey !== null) { + const ids = gpuIdsKey === '' ? [] : gpuIdsKey.split(',').map(Number); + gpus = gpus.filter(g => ids.includes(g.index)); } + return gpus; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [gpu, gpuIdsKey]); + + const isGPUInfoLoaded = gpu !== null; + const status: 'idle' | 'loading' | 'success' | 'error' = gpu !== null ? 'success' : connected ? 'loading' : 'idle'; + + return { + gpuList, + isGPUInfoLoaded, + status, + // The stream is always live; nothing to refresh manually. + refreshGpuInfo: async () => {}, }; - - usePollLoop(fetchGpuInfo, reloadInterval, [gpuIds]); - - // If the initial fetch failed (e.g. nvidia-smi was transiently busy on startup), - // retry automatically so users don't have to reload the page to see their GPUs. - useEffect(() => { - if (status !== 'error') return; - const retry = setTimeout(() => { - fetchGpuInfo(); - }, 3000); - return () => clearTimeout(retry); - }, [status]); - - return { gpuList, setGpuList, isGPUInfoLoaded, status, refreshGpuInfo: fetchGpuInfo }; } diff --git a/ui/src/hooks/useMonitorStream.tsx b/ui/src/hooks/useMonitorStream.tsx new file mode 100644 index 00000000..271454b9 --- /dev/null +++ b/ui/src/hooks/useMonitorStream.tsx @@ -0,0 +1,156 @@ +'use client'; + +import { useEffect, useState } from 'react'; +import { CpuInfo, GPUApiResponse, MonitorHistoryPoint, MonitorInit, MonitorSample } from '@/types'; +import { isAuthorizedState } from '@/utils/api'; +import { historyPointFromSample, MONITOR_HISTORY_LENGTH } from '@/utils/monitorSample'; + +export interface MonitorStreamState { + connected: boolean; + lastUpdated: Date | null; + cpu: CpuInfo | null; + gpu: GPUApiResponse | null; + history: MonitorHistoryPoint[]; +} + +/** + * One shared SSE connection to /api/monitor for the whole app, no matter how + * many components subscribe. Connects while at least one hook instance is + * mounted, reconnects automatically, and keeps the 2-minute rolling history + * (seeded by the server's backlog on connect) up to date from live samples. + * + * Native EventSource can't send the Authorization header the middleware + * expects, so this reads the SSE stream through fetch. + */ +let state: MonitorStreamState = { + connected: false, + lastUpdated: null, + cpu: null, + gpu: null, + history: [], +}; +const listeners = new Set<(s: MonitorStreamState) => void>(); +let refCount = 0; +let running = false; +let abortController: AbortController | null = null; + +function emit(next: MonitorStreamState) { + state = next; + for (const listener of listeners) { + listener(state); + } +} + +function handleEventBlock(block: string) { + let event = 'message'; + const dataLines: string[] = []; + for (const line of block.split('\n')) { + if (line.startsWith('event:')) { + event = line.slice('event:'.length).trim(); + } else if (line.startsWith('data:')) { + dataLines.push(line.slice('data:'.length).trimStart()); + } + } + if (dataLines.length === 0) return; + + if (event === 'init') { + const init: MonitorInit = JSON.parse(dataLines.join('\n')); + emit({ + connected: true, + lastUpdated: new Date(init.t), + cpu: init.cpu, + gpu: init.gpu, + history: init.history, + }); + } else if (event === 'sample') { + const sample: MonitorSample = JSON.parse(dataLines.join('\n')); + const history = [...state.history, historyPointFromSample(sample)]; + if (history.length > MONITOR_HISTORY_LENGTH) { + history.splice(0, history.length - MONITOR_HISTORY_LENGTH); + } + emit({ + connected: true, + lastUpdated: new Date(sample.t), + cpu: sample.cpu, + gpu: sample.gpu, + history, + }); + } +} + +async function runLoop() { + if (running) return; + running = true; + try { + while (refCount > 0) { + abortController = new AbortController(); + try { + const headers: Record = { Accept: 'text/event-stream' }; + const token = localStorage.getItem('AI_TOOLKIT_AUTH'); + if (token) { + headers['Authorization'] = `Bearer ${token}`; + } + const res = await fetch('/api/monitor', { + headers, + cache: 'no-store', + signal: abortController.signal, + }); + if (res.status === 401) { + // Mirror the axios interceptor's behavior + localStorage.removeItem('AI_TOOLKIT_AUTH'); + isAuthorizedState.set(false); + throw new Error('unauthorized'); + } + if (!res.ok || !res.body) { + throw new Error(`monitor stream returned HTTP ${res.status}`); + } + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + let sep; + while ((sep = buffer.indexOf('\n\n')) !== -1) { + const block = buffer.slice(0, sep); + buffer = buffer.slice(sep + 2); + handleEventBlock(block); + } + } + } catch (err) { + if (!abortController.signal.aborted) { + console.error(`Monitor stream error: ${err instanceof Error ? err.message : String(err)}`); + } + } + if (state.connected) { + emit({ ...state, connected: false }); + } + if (refCount <= 0) break; + await new Promise(r => setTimeout(r, 2000)); + } + } finally { + running = false; + } +} + +export default function useMonitorStream(): MonitorStreamState { + const [snapshot, setSnapshot] = useState(state); + + useEffect(() => { + const listener = (s: MonitorStreamState) => setSnapshot(s); + listeners.add(listener); + refCount++; + setSnapshot(state); + runLoop(); + return () => { + listeners.delete(listener); + refCount--; + if (refCount <= 0) { + abortController?.abort(); + } + }; + }, []); + + return snapshot; +} diff --git a/ui/src/instrumentation.ts b/ui/src/instrumentation.ts new file mode 100644 index 00000000..147c92a2 --- /dev/null +++ b/ui/src/instrumentation.ts @@ -0,0 +1,11 @@ +/** + * Next.js instrumentation hook — runs once when the server process starts + * (not during build). Starts the always-on system monitor so the rolling + * stats history is already populated when the first client connects. + */ +export async function register() { + if (process.env.NEXT_RUNTIME === 'nodejs') { + const { startMonitor } = await import('@/server/monitor'); + startMonitor(); + } +} diff --git a/ui/src/server/monitor.ts b/ui/src/server/monitor.ts new file mode 100644 index 00000000..32da712a --- /dev/null +++ b/ui/src/server/monitor.ts @@ -0,0 +1,470 @@ +import { spawn, execFile, ChildProcess } from 'child_process'; +import { promisify } from 'util'; +import os from 'os'; +import si from 'systeminformation'; +import { loadMacstats } from '@/server/macstats'; +import { CpuInfo, GpuInfo, GPUApiResponse, MonitorHistoryPoint, MonitorInit, MonitorSample } from '@/types'; +import { historyPointFromSample, MONITOR_HISTORY_LENGTH, MONITOR_TICK_MS } from '@/utils/monitorSample'; + +const execFileAsync = promisify(execFile); + +/** + * Always-on system monitor. Samples CPU + GPU every MONITOR_TICK_MS, keeps a + * 2-minute rolling history (load + memory only), and pushes every full sample + * to subscribers (the SSE route at /api/monitor). + * + * GPU stats come from a single resident `nvidia-smi ... -lms` child process — + * NVML stays initialized between samples, which is what made the old + * spawn-per-request /api/gpu route slow. If loop mode never produces output + * (or nvidia-smi keeps hanging), we fall back to a one-shot spawn per tick, + * which matches the old route's behavior exactly. + */ + +const NV_QUERY_ARGS = [ + '--query-gpu=index,name,driver_version,temperature.gpu,utilization.gpu,utilization.memory,memory.total,memory.free,memory.used,power.draw,power.limit,clocks.current.graphics,clocks.current.memory,fan.speed', + '--format=csv,noheader,nounits', +]; +const NV_ENV = { ...process.env, CUDA_DEVICE_ORDER: 'PCI_BUS_ID' }; +// No stdout line for this long means the loop child is buffering or hung. +const NV_WATCHDOG_MS = 15_000; +// A line-less gap this long after the last line closes out a batch (all +// lines of one iteration arrive together; iterations are MONITOR_TICK_MS +// apart, so this can never bleed into the next batch). +const NV_BATCH_FLUSH_MS = 100; + +function parseGpuLine(line: string): GpuInfo | null { + const [ + index, + name, + driverVersion, + temperature, + gpuUtil, + memoryUtil, + memoryTotal, + memoryFree, + memoryUsed, + powerDraw, + powerLimit, + clockGraphics, + clockMemory, + fanSpeed, + ] = line.split(', ').map(item => item.trim()); + if (isNaN(parseInt(index))) return null; + return { + index: parseInt(index), + name, + driverVersion, + temperature: parseInt(temperature), + utilization: { + gpu: parseInt(gpuUtil), + memory: parseInt(memoryUtil), + }, + memory: { + total: parseInt(memoryTotal), + free: parseInt(memoryFree), + used: parseInt(memoryUsed), + }, + power: { + draw: parseFloat(powerDraw), + limit: parseFloat(powerLimit), + }, + clocks: { + graphics: parseInt(clockGraphics), + memory: parseInt(clockMemory), + }, + fan: { + speed: parseInt(fanSpeed) || 0, // Some GPUs might not report fan speed, default to 0 + }, + }; +} + +type Subscriber = (sample: MonitorSample) => void; + +class SystemMonitor { + private isMac = os.platform() === 'darwin'; + private history: MonitorHistoryPoint[] = []; + private subscribers = new Set(); + private latestCpu: CpuInfo | null = null; + private latestGpu: GPUApiResponse = { hasNvidiaSmi: false, isMac: this.isMac, gpus: [] }; + private macGpuName = 'Apple GPU'; + private nvChild: ChildProcess | null = null; + private nvBatch: GpuInfo[] = []; + private nvStdoutBuffer = ''; + private nvFlushTimer: NodeJS.Timeout | null = null; + private nvUnavailable = false; + private nvReaped = false; + private nvEverGotLine = false; + private nvOneShotMode = false; + private nvOneShotInFlight = false; + private lastNvLineAt = 0; + private tickInFlight = false; + private lastCpuTemp = 0; + private cpuTempInFlight = false; + + start(): void { + if (this.isMac) { + this.initMacGpuName(); + } else { + this.startNvLoop(); + } + // Never leave a resident nvidia-smi behind. + process.once('exit', () => { + try { + this.nvChild?.kill('SIGKILL'); + } catch { + // already gone + } + }); + // Fixed cadence; a tick that overruns the interval (slow temperature + // read, hung nvidia-smi one-shot) just skips beats instead of stacking. + this.tick(); + setInterval(() => this.tick(), MONITOR_TICK_MS); + } + + subscribe(fn: Subscriber): () => void { + this.subscribers.add(fn); + return () => this.subscribers.delete(fn); + } + + getInit(): MonitorInit { + return { + t: Date.now(), + cpu: this.latestCpu, + gpu: this.latestGpu, + history: [...this.history], + }; + } + + // ------------------------------------------------------------------------- + // Tick loop + // ------------------------------------------------------------------------- + private async tick(): Promise { + if (this.tickInFlight) return; + this.tickInFlight = true; + try { + await this.doTick(); + } finally { + this.tickInFlight = false; + } + } + + private async doTick(): Promise { + const t = Date.now(); + try { + this.latestCpu = await this.sampleCpu(); + } catch (error) { + console.error('Monitor: CPU sample failed:', error); + } + try { + if (this.isMac) { + this.latestGpu = this.sampleMacGpu(); + } else if (this.nvOneShotMode) { + await this.sampleNvOneShot(); + } else { + this.nvWatchdog(); + } + } catch (error) { + console.error('Monitor: GPU sample failed:', error); + } + + const sample: MonitorSample = { t, cpu: this.latestCpu, gpu: this.latestGpu }; + this.history.push(historyPointFromSample(sample)); + if (this.history.length > MONITOR_HISTORY_LENGTH) { + this.history.splice(0, this.history.length - MONITOR_HISTORY_LENGTH); + } + for (const subscriber of this.subscribers) { + try { + subscriber(sample); + } catch (error) { + console.error('Monitor: subscriber failed:', error); + } + } + } + + // ------------------------------------------------------------------------- + // CPU (mirrors /api/cpu exactly) + // ------------------------------------------------------------------------- + private async sampleCpu(): Promise { + const cpuInfoRaw = await si.cpu(); // static info, cached by systeminformation + + if (this.isMac) { + try { + const ms = loadMacstats(); + if (!ms) throw new Error('macstats unavailable'); + const ramData = ms.getRAMUsageSync(); + const cpuData = ms.getCpuDataSync(); + return { + name: `${cpuInfoRaw.manufacturer} ${cpuInfoRaw.brand}`, + cores: cpuInfoRaw.cores, + temperature: cpuData.temperature || 0, + totalMemory: ramData.total / (1024 * 1024), + availableMemory: ramData.free / (1024 * 1024), + freeMemory: ramData.free / (1024 * 1024), + currentLoad: (await si.currentLoad()).currentLoad || 0, + }; + } catch { + // Fallback to systeminformation if macstats fails + } + } + + // The temperature read can take >1s on some machines and would make the + // tick skip beats, so it refreshes concurrently and we use the cached + // value (at most a tick or two old). + this.refreshCpuTemp(); + const [memoryData, load] = await Promise.all([si.mem(), si.currentLoad()]); + return { + name: `${cpuInfoRaw.manufacturer} ${cpuInfoRaw.brand}`, + cores: cpuInfoRaw.cores, + temperature: this.lastCpuTemp, + totalMemory: memoryData.total / (1024 * 1024), + availableMemory: memoryData.available / (1024 * 1024), + freeMemory: memoryData.free / (1024 * 1024), + currentLoad: load.currentLoad || 0, + }; + } + + private refreshCpuTemp(): void { + if (this.cpuTempInFlight) return; + this.cpuTempInFlight = true; + si.cpuTemperature() + .then(t => { + this.lastCpuTemp = t.main || 0; + }) + .catch(() => { + // keep the previous value + }) + .finally(() => { + this.cpuTempInFlight = false; + }); + } + + // ------------------------------------------------------------------------- + // Mac GPU (mirrors /api/gpu's mac path) + // ------------------------------------------------------------------------- + private initMacGpuName(): void { + execFileAsync('sh', ['-c', 'system_profiler SPDisplaysDataType 2>/dev/null | grep -E "Chipset Model|Total Number of Cores"'], { + timeout: 5000, + }) + .then(({ stdout }) => { + const nameMatch = stdout.match(/Chipset Model:\s*(.+)/); + const coresMatch = stdout.match(/Total Number of Cores:\s*(\d+)/); + if (nameMatch) { + this.macGpuName = nameMatch[1].trim(); + if (coresMatch) { + this.macGpuName += ` GPU (${coresMatch[1]} cores)`; + } + } + }) + .catch(() => { + // fallback to generic name + }); + } + + private sampleMacGpu(): GPUApiResponse { + let temperature = 0; + let gpuLoad = 0; + let fanSpeed = 0; + let powerDraw = 0; + let memUsed = 0; + let memTotal = os.totalmem() / (1024 * 1024); + + const ms = loadMacstats(); + if (ms) { + try { + const gpuData = ms.getGpuDataSync(); + temperature = gpuData.temperature || 0; + gpuLoad = gpuData.usage || 0; + } catch { + // ignore + } + try { + const fanData = ms.getFanDataSync(); + const fanKeys = Object.keys(fanData); + if (fanKeys.length > 0) { + fanSpeed = fanData[fanKeys[0]].rpm || 0; + } + } catch { + // ignore + } + try { + const powerData = ms.getPowerDataSync(); + powerDraw = powerData.gpu || 0; + } catch { + // ignore + } + try { + const ramData = ms.getRAMUsageSync(); + memUsed = ramData.used / (1024 * 1024); + memTotal = ramData.total / (1024 * 1024); + } catch { + // ignore + } + } + + return { + hasNvidiaSmi: false, + isMac: true, + gpus: [ + { + index: 0, + name: this.macGpuName, + driverVersion: 'macOS', + temperature: Math.round(temperature), + utilization: { + gpu: gpuLoad, + memory: memTotal > 0 ? Math.round((memUsed / memTotal) * 100) : 0, + }, + memory: { + total: Math.round(memTotal), + free: Math.round(memTotal - memUsed), + used: Math.round(memUsed), + }, + power: { draw: powerDraw, limit: 0 }, + clocks: { graphics: 0, memory: 0 }, + fan: { speed: fanSpeed }, + }, + ], + }; + } + + // ------------------------------------------------------------------------- + // NVIDIA loop-mode child + // ------------------------------------------------------------------------- + private startNvLoop(): void { + if (this.nvUnavailable || this.nvOneShotMode || this.nvChild) return; + + // A hard-killed server (SIGKILL never runs the exit hook) can orphan the + // resident loop child. Our exact query string only ever appears in + // children we spawned, so reap any stray once at boot — after it + // completes, to not race the kill against our own fresh child. + if (!this.nvReaped && process.platform !== 'win32') { + this.nvReaped = true; + // No loop flag in the pattern so strays from any past cadence match + execFile('pkill', ['-9', '-f', `nvidia-smi ${NV_QUERY_ARGS.join(' ')}`], () => this.startNvLoop()); + return; + } + this.nvReaped = true; + + let child: ChildProcess; + try { + child = spawn('nvidia-smi', [...NV_QUERY_ARGS, '-lms', String(MONITOR_TICK_MS)], { + env: NV_ENV, + stdio: ['ignore', 'pipe', 'pipe'], + }); + } catch { + this.markNvUnavailable(); + return; + } + this.nvChild = child; + this.nvStdoutBuffer = ''; + this.nvBatch = []; + this.lastNvLineAt = Date.now(); + + child.stdout!.on('data', (chunk: Buffer) => this.onNvData(chunk.toString())); + child.stderr!.on('data', () => { + // nvidia-smi warnings are not actionable here + }); + child.on('error', (err: NodeJS.ErrnoException) => { + if (err.code === 'ENOENT') { + this.markNvUnavailable(); + } + }); + child.on('exit', () => { + if (this.nvChild !== child) return; + this.nvChild = null; + if (this.nvFlushTimer) { + clearTimeout(this.nvFlushTimer); + this.nvFlushTimer = null; + } + if (!this.nvUnavailable && !this.nvOneShotMode) { + setTimeout(() => this.startNvLoop(), 5000); + } + }); + } + + private onNvData(text: string): void { + this.nvStdoutBuffer += text; + const lines = this.nvStdoutBuffer.split('\n'); + this.nvStdoutBuffer = lines.pop() || ''; + for (const line of lines) { + if (!line.trim()) continue; + const gpu = parseGpuLine(line); + if (!gpu) continue; + this.nvEverGotLine = true; + this.lastNvLineAt = Date.now(); + this.nvBatch.push(gpu); + } + if (this.nvFlushTimer) clearTimeout(this.nvFlushTimer); + this.nvFlushTimer = setTimeout(() => this.flushNvBatch(), NV_BATCH_FLUSH_MS); + } + + private flushNvBatch(): void { + this.nvFlushTimer = null; + if (this.nvBatch.length === 0) return; + this.latestGpu = { + hasNvidiaSmi: true, + isMac: false, + gpus: this.nvBatch.sort((a, b) => a.index - b.index), + }; + this.nvBatch = []; + } + + private nvWatchdog(): void { + if (this.nvUnavailable || !this.nvChild) return; + if (Date.now() - this.lastNvLineAt <= NV_WATCHDOG_MS) return; + console.warn('Monitor: no output from nvidia-smi loop, restarting it'); + // Loop mode that never produced a single line isn't going to start — + // switch to a one-shot spawn per tick instead of kill/respawn forever. + if (!this.nvEverGotLine) { + this.nvOneShotMode = true; + } + this.nvChild.kill('SIGKILL'); + } + + private async sampleNvOneShot(): Promise { + if (this.nvUnavailable || this.nvOneShotInFlight) return; + this.nvOneShotInFlight = true; + try { + const { stdout } = await execFileAsync('nvidia-smi', NV_QUERY_ARGS, { env: NV_ENV }); + const gpus = stdout + .trim() + .split('\n') + .map(parseGpuLine) + .filter((gpu): gpu is GpuInfo => gpu !== null) + .sort((a, b) => a.index - b.index); + this.latestGpu = { hasNvidiaSmi: true, isMac: false, gpus }; + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') { + this.markNvUnavailable(); + } else { + console.error('Monitor: one-shot nvidia-smi failed:', err); + } + } finally { + this.nvOneShotInFlight = false; + } + } + + private markNvUnavailable(): void { + this.nvUnavailable = true; + this.latestGpu = { + hasNvidiaSmi: false, + isMac: false, + gpus: [], + error: 'nvidia-smi not found or not accessible', + }; + } +} + +/** + * Idempotent starter. Guarded on globalThis so dev-mode module reloads never + * stack a second sampler (and a second resident nvidia-smi) in the same + * process. + */ +export function startMonitor(): SystemMonitor { + const g = globalThis as unknown as { __aiToolkitSystemMonitor?: SystemMonitor }; + if (!g.__aiToolkitSystemMonitor) { + g.__aiToolkitSystemMonitor = new SystemMonitor(); + g.__aiToolkitSystemMonitor.start(); + } + return g.__aiToolkitSystemMonitor; +} diff --git a/ui/src/types.ts b/ui/src/types.ts index f1de3fa2..53cca389 100644 --- a/ui/src/types.ts +++ b/ui/src/types.ts @@ -56,6 +56,29 @@ export interface GPUApiResponse { error?: string; } +/** + * System monitor stream (SSE at /api/monitor) + */ + +// Rolling history only logs load + memory; everything else (temps, fans, +// power, clocks) is instantaneous-only via MonitorSample. +export interface MonitorHistoryPoint { + t: number; // epoch ms + cpu: { load: number; memUsedMb: number }; + // one entry per GPU, same order as MonitorSample.gpu.gpus (sorted by index) + gpus: { load: number; memUsedMb: number }[]; +} + +export interface MonitorSample { + t: number; + cpu: CpuInfo | null; + gpu: GPUApiResponse; +} + +export interface MonitorInit extends MonitorSample { + history: MonitorHistoryPoint[]; +} + /** * Training configuration */ diff --git a/ui/src/utils/monitorSample.ts b/ui/src/utils/monitorSample.ts new file mode 100644 index 00000000..04e6ede0 --- /dev/null +++ b/ui/src/utils/monitorSample.ts @@ -0,0 +1,23 @@ +import { MonitorHistoryPoint, MonitorSample } from '@/types'; + +export const MONITOR_TICK_MS = 500; +export const MONITOR_HISTORY_LENGTH = 120_000 / MONITOR_TICK_MS; // 2 minutes of samples + +/** + * Reduce a full sample to the slim point kept in the rolling history: + * load + memory only. Used by both the server (to build the backlog sent on + * connect) and the client (to extend that backlog from live samples). + */ +export function historyPointFromSample(sample: MonitorSample): MonitorHistoryPoint { + return { + t: sample.t, + cpu: { + load: sample.cpu?.currentLoad ?? 0, + memUsedMb: sample.cpu ? sample.cpu.totalMemory - sample.cpu.availableMemory : 0, + }, + gpus: sample.gpu.gpus.map(gpu => ({ + load: gpu.utilization.gpu, + memUsedMb: gpu.memory.used, + })), + }; +}