Run in chunks

Author: cte

benchmark/apps/cli/package.json

@@ -11,6 +11,7 @@
 	"dependencies": {
 		"@benchmark/db": "workspace:^",
 		"@benchmark/ipc": "workspace:^",
+		"@benchmark/lib": "workspace:^",
 		"@benchmark/types": "workspace:^",
 		"execa": "^9.5.2",
 		"gluegun": "^5.1.2",

benchmark/apps/cli/src/index.ts

@@ -27,6 +27,7 @@ import {
 	updateTask,
 	createTaskMetrics,
 } from "@benchmark/db"
+import { inChunksOf } from "@benchmark/lib"
 import { IpcServer, IpcClient } from "@benchmark/ipc"
 
 import { __dirname, extensionDevelopmentPath, exercisesPath } from "./paths.js"
@@ -90,54 +91,30 @@ const run = async (toolbox: GluegunToolbox) => {
 		throw new Error("No tasks found.")
 	}
 
-	let currentTask = tasks[0]
-
 	const server = new IpcServer(run.socketPath, () => {})
 	server.listen()
 
 	server.on(IpcMessageType.Connect, (clientId) => {
 		server.send(clientId, {
 			type: IpcMessageType.TaskEvent,
 			origin: IpcOrigin.Server,
-			data: { eventName: RooCodeEventName.Connect, taskId: currentTask.id },
+			// TODO: Broacast the set of running tasks.
+			data: { eventName: RooCodeEventName.Connect, taskId: -1 },
 		})
 	})
 
-	for (const task of tasks) {
-		currentTask = task
-
-		await runExercise({ run, task, server })
-
-		const cmd = testCommands[task.language]
-		const exercisePath = path.resolve(exercisesPath, task.language, task.exercise)
-		const cwd = cmd.cwd ? path.resolve(exercisePath, cmd.cwd) : exercisePath
-		const commands = cmd.commands.map((cs) => parseCommandString(cs))
-
-		let passed = true
+	const chunks = inChunksOf(tasks, 2)
 
-		for (const command of commands) {
-			const controller = new AbortController()
-			const cancelSignal = controller.signal
-			const timeout = setTimeout(() => controller.abort(), cmd.timeout ?? 15_000)
+	for (const chunk of chunks) {
+		await Promise.all(
+			chunk.map(async (task) => {
+				const runSucceeded = await runExercise({ run, task, server })
+				const passed = runSucceeded ? await runUnitTest({ task }) : false
+				await updateTask(task.id, { passed })
+			}),
+		)
 
-			try {
-				const result = await execa({ cwd, shell: true, reject: false, cancelSignal })`${command}`
-				// console.log('[cli#run] execa result =', { ...result, cwd, command })
-
-				clearTimeout(timeout)
-
-				if (result.failed) {
-					passed = false
-					break
-				}
-			} catch (error) {
-				console.log("[cli#run] execa error =", error)
-				passed = false
-				break
-			}
-		}
-
-		await updateTask(task.id, { passed })
+		break
 	}
 
 	const result = await finishRun(run.id)
@@ -155,8 +132,8 @@ const runExercise = async ({ run, task, server }: { run: Run; task: Task; server
 	const prompt = fs.readFileSync(path.resolve(exercisesPath, `prompts/${language}.md`), "utf-8")
 
 	const dirname = path.dirname(run.socketPath)
-	const basename = path.basename(run.socketPath, ".sock")
-	const taskSocketPath = path.resolve(dirname, `${dirname}/${basename}-${task.id}.sock`)
+	const taskSocketPath = path.resolve(dirname, `${dirname}/task-${task.id}.sock`)
+
 	await execa({
 		env: { ROO_CODE_IPC_SOCKET_PATH: taskSocketPath },
 	})`code -n ${path.resolve(exercisesPath, language, exercise)}`
@@ -179,7 +156,7 @@ const runExercise = async ({ run, task, server }: { run: Run; task: Task; server
 	let isTaskFinished = false
 
 	client.on(IpcMessageType.Disconnect, () => {
-		console.log("disconnect")
+		console.log(`[cli#runExercise | ${language} / ${exercise}] disconnect`)
 		isTaskFinished = true
 	})
 
@@ -202,7 +179,7 @@ const runExercise = async ({ run, task, server }: { run: Run; task: Task; server
 		})
 
 		if (!ignoreEvents.includes(eventName)) {
-			console.log(`[cli#runExercise] taskEvent -> ${eventName}`)
+			console.log(`[cli#runExercise | ${language} / ${exercise}] taskEvent -> ${eventName}`)
 		}
 
 		// if (eventName === RooCodeEventName.Message) {
@@ -215,6 +192,7 @@ const runExercise = async ({ run, task, server }: { run: Run; task: Task; server
 
 		if (eventName === RooCodeEventName.TaskStarted) {
 			taskStartedAt = Date.now()
+			await updateTask(task.id, { startedAt: new Date() })
 		}
 
 		if (eventName === RooCodeEventName.TaskCompleted) {
@@ -233,7 +211,7 @@ const runExercise = async ({ run, task, server }: { run: Run; task: Task; server
 				cacheReads: totalCacheReads ?? 0,
 			})
 
-			await updateTask(task.id, { taskMetricsId: taskMetrics.id })
+			await updateTask(task.id, { taskMetricsId: taskMetrics.id, finishedAt: new Date() })
 			isTaskFinished = true
 		}
 
@@ -242,7 +220,7 @@ const runExercise = async ({ run, task, server }: { run: Run; task: Task; server
 		}
 	})
 
-	console.log(`[cli#runExercise] StartNewTask -> ${language} / ${exercise}`)
+	console.log(`[cli#runExercise | ${language} / ${exercise}] StartNewTask (${taskSocketPath})`)
 
 	client.sendMessage({
 		type: IpcMessageType.TaskCommand,
@@ -273,6 +251,39 @@ const runExercise = async ({ run, task, server }: { run: Run; task: Task; server
 	}
 }
 
+const runUnitTest = async ({ task }: { task: Task }) => {
+	const cmd = testCommands[task.language]
+	const exercisePath = path.resolve(exercisesPath, task.language, task.exercise)
+	const cwd = cmd.cwd ? path.resolve(exercisePath, cmd.cwd) : exercisePath
+	const commands = cmd.commands.map((cs) => parseCommandString(cs))
+
+	let passed = true
+
+	for (const command of commands) {
+		const controller = new AbortController()
+		const cancelSignal = controller.signal
+		const timeout = setTimeout(() => controller.abort(), cmd.timeout ?? 15_000)
+
+		try {
+			const result = await execa({ cwd, shell: true, reject: false, cancelSignal })`${command}`
+			// console.log('[cli#run] execa result =', { ...result, cwd, command })
+
+			clearTimeout(timeout)
+
+			if (result.failed) {
+				passed = false
+				break
+			}
+		} catch (error) {
+			console.log("[cli#run] execa error =", error)
+			passed = false
+			break
+		}
+	}
+
+	return passed
+}
+
 const askLanguage = async (prompt: GluegunPrompt) => {
 	const { language } = await prompt.ask<{ language: ExerciseLanguage }>({
 		type: "select",

benchmark/package.json

@@ -5,6 +5,7 @@
 	"scripts": {
 		"lint": "turbo lint --log-order grouped --output-logs new-only",
 		"check-types": "turbo check-types --log-order grouped --output-logs new-only",
+		"test": "turbo test --log-order grouped --output-logs new-only",
 		"format": "turbo format --log-order grouped --output-logs new-only",
 		"build": "turbo build --log-order grouped --output-logs new-only",
 		"web": "turbo dev --filter @benchmark/web --output-logs new-only --ui tui",

benchmark/packages/ipc/src/index.ts

@@ -20,23 +20,24 @@ export type IpcClientEvents = {
 
 export class IpcClient extends EventEmitter<IpcClientEvents> {
 	private readonly _socketPath: string
+	private readonly _id: string
 	private readonly _log: (...args: unknown[]) => void
-
 	private _isConnected = false
 	private _clientId?: string
 
 	constructor(socketPath: string, log = console.log) {
 		super()
 
 		this._socketPath = socketPath
+		this._id = `benchmark-${crypto.randomBytes(6).toString("hex")}`
 		this._log = log
 
 		ipc.config.silent = true
 
-		ipc.connectTo("benchmarkServer", this.socketPath, () => {
-			ipc.of.benchmarkServer?.on("connect", () => this.onConnect())
-			ipc.of.benchmarkServer?.on("disconnect", () => this.onDisconnect())
-			ipc.of.benchmarkServer?.on("message", (data) => this.onMessage(data))
+		ipc.connectTo(this._id, this.socketPath, () => {
+			ipc.of[this._id]?.on("connect", () => this.onConnect())
+			ipc.of[this._id]?.on("disconnect", () => this.onDisconnect())
+			ipc.of[this._id]?.on("message", (data) => this.onMessage(data))
 		})
 	}
 
@@ -93,12 +94,12 @@ export class IpcClient extends EventEmitter<IpcClientEvents> {
 	}
 
 	public sendMessage(message: IpcMessage) {
-		ipc.of.benchmarkServer?.emit("message", message)
+		ipc.of[this._id]?.emit("message", message)
 	}
 
 	public disconnect() {
 		try {
-			ipc.disconnect("benchmarkServer")
+			ipc.disconnect(this._id)
 			// @TODO: Should we set _disconnect here?
 		} catch (error) {
 			this.log("[client#disconnect] error disconnecting", error)

benchmark/packages/lib/eslint.config.mjs

@@ -0,0 +1,4 @@
+import { config } from "@benchmark/eslint-config/base"
+
+/** @type {import("eslint").Linter.Config} */
+export default [...config]

benchmark/packages/lib/package.json

@@ -0,0 +1,17 @@
+{
+	"name": "@benchmark/lib",
+	"private": true,
+	"type": "module",
+	"exports": "./src/index.ts",
+	"scripts": {
+		"lint": "eslint src --ext ts --max-warnings=0",
+		"check-types": "tsc --noEmit",
+		"test": "vitest --globals --run",
+		"format": "prettier --write src"
+	},
+	"devDependencies": {
+		"@benchmark/eslint-config": "workspace:^",
+		"@benchmark/typescript-config": "workspace:^",
+		"vitest": "^3.0.9"
+	}
+}

benchmark/packages/lib/src/__tests__/in-chunks-of.test.ts

@@ -0,0 +1,18 @@
+// npx vitest run src/__tests__/in-chunks-of.test.ts
+
+import { inChunksOf } from "../in-chunks-of.js"
+
+describe("inChunksOf", () => {
+	it("should return an array of arrays", () => {
+		const result = inChunksOf([1, 2, 3, 4, 5])
+		expect(result).toEqual([[1, 2], [3, 4], [5]])
+	})
+
+	it("should return an array of arrays with a custom chunk size", () => {
+		const result = inChunksOf([1, 2, 3, 4, 5], 3)
+		expect(result).toEqual([
+			[1, 2, 3],
+			[4, 5],
+		])
+	})
+})

benchmark/packages/lib/src/in-chunks-of.ts

@@ -0,0 +1,14 @@
+export function inChunksOf<T>(ary: T[], perChunk = 2) {
+	const result = ary.reduce((collect, item, index) => {
+		const chunkIndex = Math.floor(index / perChunk)
+
+		if (!collect[chunkIndex]) {
+			collect[chunkIndex] = []
+		}
+
+		collect[chunkIndex].push(item)
+		return collect
+	}, [] as T[][])
+
+	return result
+}

benchmark/packages/lib/src/index.ts

@@ -0,0 +1 @@
+export * from "./in-chunks-of.js"

benchmark/packages/lib/tsconfig.json

@@ -0,0 +1,8 @@
+{
+	"extends": "@benchmark/typescript-config/base.json",
+	"compilerOptions": {
+		"types": ["vitest/globals"]
+	},
+	"include": ["src"],
+	"exclude": ["node_modules"]
+}