feat(benchmark): authoritative in-test sysbench numbers + results strip (#1082)

Phase 2 of the live benchmark run experience. Streams sysbench
--report-interval=1 interim lines over a Docker attach so the run view
shows authoritative in-test CPU events/sec and disk read/write MiB/s
(overlaying the Phase 1 host-proxy disk numbers), and fills a
"results so far" strip as each stage completes.

- _runSysbenchCommandStreaming: attaches to the container output for
  live onLine callbacks, but returns the authoritative output via
  container.logs() after exit -- byte-identical to _runSysbenchCommand,
  so the SCORED numbers are unchanged (attach 'data' can flush after
  container.wait() resolves, which would truncate the final report).
- CPU + disk stages stream interim eps: / reads:/writes: MiB/s into
  setStageMetric; memory stays non-streaming (too fast to sample).
- _emitPartialResult broadcasts each finished stage's raw result on the
  progress channel; useBenchmarkRun accumulates them; ResultsSoFar
  renders the chip strip.
- Frontend: live CPU ev/s readout+sparkline, disk hero switches to
  "Benchmark throughput" when in-test numbers arrive.

Browser-validated on NOMAD3 (System-Only): CPU 6536 ev/s live, disk
10399 MB/s benchmark throughput, results strip, final score 68.3.

Part of #1082 (tracker stays open for Phase 3).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Sherwood 2026-07-11 20:22:58 -07:00
parent 3bca6552f0
commit 965474d0e4
4 changed files with 292 additions and 25 deletions

View File

@ -22,6 +22,7 @@ import type {
RepositorySubmitResponse,
RepositoryStats,
BenchmarkStageDescriptor,
BenchmarkPartialResult,
} from '../../types/benchmark.js'
import { randomUUID, createHmac } from 'node:crypto'
import { DockerService } from './docker_service.js'
@ -409,6 +410,14 @@ export class BenchmarkService {
if (includeAI && (type === 'full' || type === 'ai')) {
try {
aiScores = await this._runAIBenchmark()
if (aiScores.ai_tokens_per_second !== undefined && aiScores.ai_tokens_per_second !== null) {
this._emitPartialResult({
status: 'running_ai',
label: 'AI',
value: Math.round(aiScores.ai_tokens_per_second),
unit: 'tok/s',
})
}
} catch (error) {
// For AI-only benchmarks, failing is fatal - don't save useless results with all zeros
if (type === 'ai') {
@ -471,17 +480,41 @@ export class BenchmarkService {
// Run CPU benchmark
this._updateStatus('running_cpu', 'Running CPU benchmark...')
const cpuResult = await this._runSysbenchCpu()
this._emitPartialResult({
status: 'running_cpu',
label: 'CPU',
value: Math.round(cpuResult.events_per_second),
unit: 'ev/s',
})
// Run memory benchmark
this._updateStatus('running_memory', 'Running memory benchmark...')
const memoryResult = await this._runSysbenchMemory()
this._emitPartialResult({
status: 'running_memory',
label: 'Memory',
value: Math.round(memoryResult.operations_per_second),
unit: 'ops/s',
})
// Run disk benchmarks
this._updateStatus('running_disk_read', 'Running disk read benchmark...')
const diskReadResult = await this._runSysbenchDiskRead()
this._emitPartialResult({
status: 'running_disk_read',
label: 'Read',
value: Math.round(diskReadResult.read_mb_per_sec),
unit: 'MB/s',
})
this._updateStatus('running_disk_write', 'Running disk write benchmark...')
const diskWriteResult = await this._runSysbenchDiskWrite()
this._emitPartialResult({
status: 'running_disk_write',
label: 'Write',
value: Math.round(diskWriteResult.write_mb_per_sec),
unit: 'MB/s',
})
// Normalize scores to 0-100 scale
return {
@ -708,14 +741,25 @@ export class BenchmarkService {
* Run sysbench CPU benchmark
*/
private async _runSysbenchCpu(): Promise<SysbenchCpuResult> {
const output = await this._runSysbenchCommand([
'sysbench',
'cpu',
'--cpu-max-prime=20000',
'--threads=4',
'--time=30',
'run',
])
// --report-interval=1 emits interim lines like:
// [ 1s ] thds: 4 eps: 7589.48 lat (ms,95%): 0.60
// They drive live telemetry only; the SCORED numbers still come from the
// final report parsed below, exactly as before.
const output = await this._runSysbenchCommandStreaming(
[
'sysbench',
'cpu',
'--cpu-max-prime=20000',
'--threads=4',
'--time=30',
'--report-interval=1',
'run',
],
(line) => {
const m = line.match(/eps:\s*([\d.]+)/)
if (m) this.telemetry?.setStageMetric('events_per_sec', parseFloat(m[1]))
}
)
// Parse output for events per second
const eventsMatch = output.match(/events per second:\s*([\d.]+)/i)
@ -761,13 +805,23 @@ export class BenchmarkService {
private async _runSysbenchDiskRead(): Promise<SysbenchDiskResult> {
// Run prepare, test, and cleanup in a single container
// This is necessary because each container has its own filesystem
const output = await this._runSysbenchCommand([
'sh',
'-c',
'sysbench fileio --file-total-size=1G --file-num=4 prepare && ' +
'sysbench fileio --file-total-size=1G --file-num=4 --file-test-mode=seqrd --time=30 run && ' +
'sysbench fileio --file-total-size=1G --file-num=4 cleanup',
])
// --report-interval=1 (run step only) emits interim lines like:
// [ 1s ] reads: 216.75 MiB/s writes: 0.00 MiB/s fsyncs: 551.89/s ...
// They drive live telemetry only; the SCORED numbers still come from the
// final report parsed below, exactly as before.
const output = await this._runSysbenchCommandStreaming(
[
'sh',
'-c',
'sysbench fileio --file-total-size=1G --file-num=4 prepare && ' +
'sysbench fileio --file-total-size=1G --file-num=4 --file-test-mode=seqrd --time=30 --report-interval=1 run && ' +
'sysbench fileio --file-total-size=1G --file-num=4 cleanup',
],
(line) => {
const m = line.match(/reads:\s*([\d.]+)\s*MiB\/s/)
if (m) this.telemetry?.setStageMetric('mib_s', parseFloat(m[1]))
}
)
// Parse output - look for the Throughput section
const readMatch = output.match(/read,\s*MiB\/s:\s*([\d.]+)/i)
@ -790,13 +844,21 @@ export class BenchmarkService {
private async _runSysbenchDiskWrite(): Promise<SysbenchDiskResult> {
// Run prepare, test, and cleanup in a single container
// This is necessary because each container has its own filesystem
const output = await this._runSysbenchCommand([
'sh',
'-c',
'sysbench fileio --file-total-size=1G --file-num=4 prepare && ' +
'sysbench fileio --file-total-size=1G --file-num=4 --file-test-mode=seqwr --time=30 run && ' +
'sysbench fileio --file-total-size=1G --file-num=4 cleanup',
])
// --report-interval=1 (run step only) drives live telemetry; the SCORED
// numbers still come from the final report parsed below, exactly as before.
const output = await this._runSysbenchCommandStreaming(
[
'sh',
'-c',
'sysbench fileio --file-total-size=1G --file-num=4 prepare && ' +
'sysbench fileio --file-total-size=1G --file-num=4 --file-test-mode=seqwr --time=30 --report-interval=1 run && ' +
'sysbench fileio --file-total-size=1G --file-num=4 cleanup',
],
(line) => {
const m = line.match(/writes:\s*([\d.]+)\s*MiB\/s/)
if (m) this.telemetry?.setStageMetric('mib_s', parseFloat(m[1]))
}
)
// Parse output - look for the Throughput section
const writeMatch = output.match(/written,\s*MiB\/s:\s*([\d.]+)/i)
@ -870,6 +932,125 @@ export class BenchmarkService {
}
}
/**
* Like _runSysbenchCommand but attaches to the container's output stream and
* invokes onLine for every complete line as it arrives, so interim
* --report-interval lines can drive live telemetry. Still returns the full
* accumulated stdout so the caller's final-report regexes (the SCORED numbers)
* parse exactly as before. onLine parsing must never throw.
*/
private async _runSysbenchCommandStreaming(
cmd: string[],
onLine: (line: string) => void
): Promise<string> {
let container: Dockerode.Container | null = null
try {
container = await this.dockerService.docker.createContainer({
Image: SYSBENCH_IMAGE,
Cmd: cmd,
name: `${SYSBENCH_CONTAINER_NAME}_${Date.now()}`,
Tty: true, // Important: prevents multiplexed stdout/stderr headers
HostConfig: {
AutoRemove: false, // Don't auto-remove to avoid race condition with output capture
},
})
// Attach BEFORE starting so no early output is missed. With Tty: true the
// attach stream is raw (non-multiplexed), so no demuxing is needed.
const stream = await container.attach({ stream: true, stdout: true, stderr: true })
// The attach stream drives the live onLine callbacks only (best-effort
// telemetry). A missed trailing line here is harmless — it never feeds the
// scored numbers, which come from container.logs() after exit below.
let buffer = ''
stream.on('data', (chunk: Buffer) => {
buffer += chunk.toString('utf8')
let newlineIdx: number
while ((newlineIdx = buffer.indexOf('\n')) >= 0) {
const line = buffer.slice(0, newlineIdx)
buffer = buffer.slice(newlineIdx + 1)
try {
onLine(line)
} catch {
// Live-parse failures must never affect the benchmark itself.
}
}
})
await container.start()
await container.wait()
// Flush any trailing partial line through onLine (best-effort).
if (buffer.length > 0) {
try {
onLine(buffer)
} catch {
// Live-parse failures must never affect the benchmark itself.
}
}
// Authoritative output for the SCORED parse. Fetch the complete captured
// log AFTER exit, exactly like _runSysbenchCommand does, rather than
// trusting the live attach stream — attach 'data' events can flush after
// container.wait() resolves, which would truncate the final report and
// break the scoring regexes. This makes the returned output byte-identical
// to the non-streaming path.
const logs = await container.logs({
stdout: true,
stderr: true,
})
const cleaned = logs
.toString('utf8')
.replace(/[\x00-\x08]/g, '') // Remove control characters
.trim()
// Manually remove the container after capturing output
try {
await container.remove()
} catch (removeError) {
// Log but don't fail if removal fails (container might already be gone)
logger.warn(`Failed to remove sysbench container: ${removeError.message}`)
}
return cleaned
} catch (error) {
// Clean up container on error if it exists
if (container) {
try {
await container.remove({ force: true })
} catch (removeError) {
// Ignore removal errors
}
}
logger.error(`Sysbench command failed: ${error.message}`)
throw new Error(`Sysbench command failed: ${error.message}`)
}
}
/**
* Broadcast a progress frame carrying a just-finished stage's raw result.
* Reuses the current status/progress so the StageRail is undisturbed; the
* partial_result field fills the "results so far" strip in the live UI.
*/
private _emitPartialResult(result: BenchmarkPartialResult) {
const progress: BenchmarkProgress = {
status: this.currentStatus,
progress: this._getProgressPercent(this.currentStatus),
message: this._getStageLabel(this.currentStatus),
current_stage: this._getStageLabel(this.currentStatus),
timestamp: new Date().toISOString(),
stages: this.currentStages,
stage_index: this.currentStages.findIndex((s) => s.status === this.currentStatus),
stage_count: this.currentStages.length,
partial_result: result,
}
transmit.broadcast(BROADCAST_CHANNELS.BENCHMARK_PROGRESS, {
benchmark_id: this.currentBenchmarkId,
...progress,
})
}
/**
* Broadcast benchmark progress update
*/

View File

@ -4,6 +4,7 @@ import StageRail from './StageRail'
import CoreGrid from './CoreGrid'
import Sparkline from './Sparkline'
import LiveReadout from './LiveReadout'
import ResultsSoFar from './ResultsSoFar'
import type { BenchmarkRunHook } from '~/hooks/useBenchmarkRun'
import type { BenchmarkStatus } from '../../../types/benchmark'
@ -39,18 +40,24 @@ function StageHero({ run }: { run: BenchmarkRunHook }) {
if (status === 'running_disk_read' || status === 'running_disk_write') {
const isRead = status === 'running_disk_read'
// Prefer the authoritative in-test sysbench throughput once it's flowing;
// fall back to the coarse host-proxy disk numbers until then.
const hasInTest = run.diskMibs !== null
return (
<div className={heroCard}>
<div className={title}><IconServer className="w-4 h-4" /> {isRead ? 'Disk Read' : 'Disk Write'}</div>
<div className="flex-1 flex flex-col justify-center gap-4">
<LiveReadout
value={isRead ? run.diskReadMbs : run.diskWriteMbs}
value={hasInTest ? run.diskMibs : isRead ? run.diskReadMbs : run.diskWriteMbs}
unit="MB/s"
label="System disk activity"
label={hasInTest ? 'Benchmark throughput' : 'System disk activity'}
size="lg"
/>
<div className="text-desert-olive">
<Sparkline data={isRead ? run.diskReadHistory : run.diskWriteHistory} height={64} />
<Sparkline
data={hasInTest ? run.diskMibsHistory : isRead ? run.diskReadHistory : run.diskWriteHistory}
height={64}
/>
</div>
</div>
</div>
@ -93,6 +100,14 @@ function StageHero({ run }: { run: BenchmarkRunHook }) {
</div>
<div className="flex-1 flex flex-col justify-center">
<CoreGrid loads={run.perCore} />
{status === 'running_cpu' && run.cpuEventsPerSec !== null && (
<div className="mt-4 space-y-2">
<LiveReadout value={run.cpuEventsPerSec} unit="ev/s" label="Events per second" />
<div className="text-desert-olive">
<Sparkline data={run.cpuEventsHistory} height={48} />
</div>
</div>
)}
</div>
</div>
)
@ -162,6 +177,7 @@ export default function BenchmarkRunView({ run }: { run: BenchmarkRunHook }) {
</div>
<VitalsStrip run={run} />
</div>
<ResultsSoFar partials={run.partials} />
</div>
)
}

View File

@ -0,0 +1,35 @@
import { IconCheck } from '@tabler/icons-react'
import type { BenchmarkPartialResult } from '../../../types/benchmark'
/**
* Horizontal strip of chips that fills in each stage's raw result as stages
* complete during a live run. Renders nothing until the first result lands.
*/
export default function ResultsSoFar({ partials }: { partials: BenchmarkPartialResult[] }) {
if (partials.length === 0) return null
return (
<div className="bg-desert-white rounded-lg p-4 border border-desert-stone-light">
<div className="text-xs font-semibold text-desert-stone-dark uppercase tracking-wide mb-3">
Results so far
</div>
<div className="flex flex-wrap gap-2">
{partials.map((p) => (
<div
key={p.status}
className="flex items-center gap-2 rounded-md border border-desert-stone-light bg-desert-white px-3 py-1.5"
>
<IconCheck className="w-4 h-4 text-desert-green" />
<span className="text-xs font-semibold text-desert-stone-dark uppercase tracking-wide">
{p.label}
</span>
<span className="text-sm font-bold font-mono tabular-nums text-desert-green">
{p.value.toLocaleString()}
<span className="text-xs font-semibold text-desert-stone-dark ml-1">{p.unit}</span>
</span>
</div>
))}
</div>
</div>
)
}

View File

@ -6,6 +6,7 @@ import type {
BenchmarkTelemetry,
BenchmarkStatus,
BenchmarkStageDescriptor,
BenchmarkPartialResult,
} from '../../types/benchmark'
// Rolling window length for the sparklines (~1 minute at 1 Hz).
@ -25,6 +26,11 @@ type LiveState = {
aiTokensPerSec: number | null
aiTokHistory: number[]
aiTtftMs: number | null
cpuEventsPerSec: number | null
cpuEventsHistory: number[]
/** Authoritative in-test disk throughput (sysbench interim lines), MiB/s. */
diskMibs: number | null
diskMibsHistory: number[]
}
const EMPTY_LIVE: LiveState = {
@ -39,6 +45,10 @@ const EMPTY_LIVE: LiveState = {
aiTokensPerSec: null,
aiTokHistory: [],
aiTtftMs: null,
cpuEventsPerSec: null,
cpuEventsHistory: [],
diskMibs: null,
diskMibsHistory: [],
}
const pushRing = (arr: number[], v: number) => {
@ -59,10 +69,12 @@ export function useBenchmarkRun(opts?: { onFinished?: (status: 'completed' | 'er
const [progress, setProgress] = useState<BenchmarkProgressWithID | null>(null)
const [live, setLive] = useState<LiveState>(EMPTY_LIVE)
const [partials, setPartials] = useState<BenchmarkPartialResult[]>([])
const reset = useCallback(() => {
setProgress(null)
setLive(EMPTY_LIVE)
setPartials([])
}, [])
useEffect(() => {
@ -70,6 +82,18 @@ export function useBenchmarkRun(opts?: { onFinished?: (status: 'completed' | 'er
BROADCAST_CHANNELS.BENCHMARK_PROGRESS,
(data: BenchmarkProgressWithID) => {
setProgress(data)
if (data.partial_result) {
const incoming = data.partial_result
setPartials((prev) => {
const idx = prev.findIndex((p) => p.status === incoming.status)
if (idx >= 0) {
const next = prev.slice()
next[idx] = incoming
return next
}
return [...prev, incoming]
})
}
if (data.status === 'completed' || data.status === 'error') {
onFinishedRef.current?.(data.status, data.message)
}
@ -81,6 +105,8 @@ export function useBenchmarkRun(opts?: { onFinished?: (status: 'completed' | 'er
(data: BenchmarkTelemetry) => {
setLive((prev) => {
const isTok = data.stage_metric?.kind === 'tokens_per_sec'
const isEps = data.stage_metric?.kind === 'events_per_sec'
const isMib = data.stage_metric?.kind === 'mib_s'
return {
perCore: data.cpu.per_core,
cpuOverall: data.cpu.overall,
@ -93,6 +119,14 @@ export function useBenchmarkRun(opts?: { onFinished?: (status: 'completed' | 'er
aiTokensPerSec: isTok ? data.stage_metric!.value : prev.aiTokensPerSec,
aiTokHistory: isTok ? pushRing(prev.aiTokHistory, data.stage_metric!.value) : prev.aiTokHistory,
aiTtftMs: isTok && data.stage_metric!.ttft_ms !== undefined ? data.stage_metric!.ttft_ms : prev.aiTtftMs,
cpuEventsPerSec: isEps ? data.stage_metric!.value : prev.cpuEventsPerSec,
cpuEventsHistory: isEps
? pushRing(prev.cpuEventsHistory, data.stage_metric!.value)
: prev.cpuEventsHistory,
diskMibs: isMib ? data.stage_metric!.value : prev.diskMibs,
diskMibsHistory: isMib
? pushRing(prev.diskMibsHistory, data.stage_metric!.value)
: prev.diskMibsHistory,
}
})
}
@ -117,6 +151,7 @@ export function useBenchmarkRun(opts?: { onFinished?: (status: 'completed' | 'er
message: progress?.message ?? '',
progressPercent: progress?.progress ?? 0,
...live,
partials,
reset,
}
}