From da56f5dcfd1d2810b61a1ebdb6330a6a608722f3 Mon Sep 17 00:00:00 2001 From: Tom Butcher Date: Sat, 19 Sep 2026 22:43:40 +0100 Subject: [PATCH] Enhance MasterSearch with worker calibration and progress tracking - Added calibration functionality for workers to improve performance measurement. - Introduced new properties in ProgressSnapshot and TrackedWorker interfaces to support calibration state and benchmark rates. - Updated MasterSearch class to manage calibration process, including resetting worker samples and handling remote calibration. - Modified UI to reflect calibration status during progress updates, enhancing user feedback during operations. --- src/search/orchestrator.ts | 272 ++++++++++++++++++++++++++++++++++--- src/ui/mainWindow.ts | 15 +- 2 files changed, 265 insertions(+), 22 deletions(-) diff --git a/src/search/orchestrator.ts b/src/search/orchestrator.ts index 2e151dc..45367e7 100644 --- a/src/search/orchestrator.ts +++ b/src/search/orchestrator.ts @@ -43,6 +43,7 @@ export interface ProgressSnapshot { etaSec: number | null; hits: number; workerLabel?: string; + calibrating?: boolean; } export interface SearchEvents { @@ -166,6 +167,7 @@ interface TrackedWorker { keysDone: bigint; activeMs: number; busyStarted: number | null; + benchRate: number; } function beginWork(worker: TrackedWorker): void { @@ -224,6 +226,8 @@ export class MasterSearch { private dispatch: (() => void) | null = null; private feedWorker: ((worker: TrackedWorker) => boolean) | null = null; private stageAlloc: BlockAllocator | null = null; + private calibrating = false; + private calibPending = new Map void>(); constructor( private readonly server: MasterServer | null, @@ -238,12 +242,17 @@ export class MasterSearch { } this.kills.clear(); this.server?.cancelAll(); + const pending = [...this.calibPending.values()]; + this.calibPending.clear(); + for (const finish of pending) { + finish(0); + } } async run(): Promise { this.cancel = false; this.hits = []; - this.started = Date.now(); + this.started = 0; if (this.params.devices.length === 0 && !(this.server && this.server.listWorkers().length > 0)) { throw new Error("Select at least one local device, or connect a slave that exposes workers."); } @@ -268,12 +277,19 @@ export class MasterSearch { this.bindServer(data, firstCt); try { - for (const job of jobs) { - if (this.cancel) { - break; + if (jobs[0]) { + await this.calibrateWorkers(jobs[0]); + } + if (!this.cancel) { + this.started = Date.now(); + this.events.onLog?.("Search timer started."); + for (const job of jobs) { + if (this.cancel) { + break; + } + await this.runStage(job, data, firstCt); + this.overallPrior += job.space; } - await this.runStage(job, data, firstCt); - this.overallPrior += job.space; } } finally { for (const off of this.unsub) { @@ -284,7 +300,7 @@ export class MasterSearch { this.feedWorker = null; } - const elapsed = (Date.now() - this.started) / 1000; + const elapsed = this.started > 0 ? (Date.now() - this.started) / 1000 : 0; writeResults(this.params.outDir, elapsed, this.hits); this.events.onDone?.(this.hits, this.cancel); return this.hits; @@ -294,6 +310,19 @@ export class MasterSearch { return this.params.blockSizeFor(worker.id, worker.deviceType); } + private testBlockCount(worker: TrackedWorker): bigint { + const size = this.sizeFor(worker); + return size < 1n ? 1n : size; + } + + private priorityRate(worker: TrackedWorker): number { + return worker.benchRate > 0 ? worker.benchRate : averageRate(worker); + } + + private rankWorkers(workers: TrackedWorker[]): TrackedWorker[] { + return [...workers].sort((a, b) => this.priorityRate(b) - this.priorityRate(a)); + } + private rebuildRoster(): void { const next = new Map(); for (const device of this.params.devices) { @@ -315,6 +344,7 @@ export class MasterSearch { keysDone: prev?.keysDone ?? 0n, activeMs: prev?.activeMs ?? 0, busyStarted: prev?.busyStarted ?? null, + benchRate: prev?.benchRate ?? 0, }); } if (this.server) { @@ -338,6 +368,7 @@ export class MasterSearch { keysDone: prev?.keysDone ?? 0n, activeMs: prev?.activeMs ?? 0, busyStarted: prev?.busyStarted ?? null, + benchRate: prev?.benchRate ?? 0, }); } } @@ -359,10 +390,17 @@ export class MasterSearch { }; this.server.events.onHit = (slave, jobId, index, keyHex, plainHex) => { prev.onHit?.(slave, jobId, index, keyHex, plainHex); + if (this.calibrating) { + return; + } this.recordHit(keyHex, plainHex, `slave_${slave.hostname}`, data, firstCt); }; this.server.events.onComplete = (slave, worker, jobId) => { prev.onComplete?.(slave, worker, jobId); + if (this.calibrating) { + this.finishRemoteCalibration(remoteWorkerId(slave.id, worker.device.key), jobId); + return; + } this.stageAlloc?.complete(jobId); this.syncStageCompleted(); const tracked = this.roster.get(remoteWorkerId(slave.id, worker.device.key)); @@ -383,6 +421,15 @@ export class MasterSearch { }; this.server.events.onDropped = (slave, jobId) => { prev.onDropped?.(slave, jobId); + if (this.calibrating) { + if (jobId) { + const tracked = [...this.roster.values()].find((w) => w.jobId === jobId); + this.finishRemoteCalibration(tracked?.id ?? "", jobId, 0); + } + this.rebuildRoster(); + this.emitProgress(); + return; + } if (jobId) { this.stageAlloc?.fail(jobId); this.syncStageCompleted(); @@ -418,6 +465,191 @@ export class MasterSearch { }); } + private resetWorkerSample(worker: TrackedWorker): void { + if (worker.busyStarted != null) { + endWork(worker, 0n); + } + worker.completedBlocks = 0; + worker.keysDone = 0n; + worker.activeMs = 0; + worker.busyStarted = null; + worker.busy = false; + worker.done = 0n; + worker.count = 0n; + worker.jobId = null; + worker.rate = 0; + } + + private finishRemoteCalibration(workerId: string, jobId: string, forcedRate?: number): void { + const pending = this.calibPending.get(jobId); + if (!pending) { + return; + } + this.calibPending.delete(jobId); + const tracked = workerId ? this.roster.get(workerId) : undefined; + let rate = forcedRate ?? 0; + if (tracked && forcedRate == null) { + endWork(tracked, tracked.count); + rate = averageRate(tracked) || tracked.rate; + } else if (tracked && tracked.busyStarted != null) { + endWork(tracked, 0n); + } + if (tracked) { + tracked.benchRate = rate; + tracked.jobId = null; + tracked.busy = false; + tracked.rate = 0; + tracked.done = 0n; + tracked.count = 0n; + } + pending(rate); + this.emitProgress(); + } + + private async calibrateWorkers(probe: StageJob): Promise { + this.rebuildRoster(); + const workers = [...this.roster.values()]; + if (workers.length === 0) { + return; + } + this.calibrating = true; + this.events.onLog?.( + `Calibrating ${workers.length} worker(s) with one test block each (excluded from elapsed time)…`, + ); + this.emitProgress(); + try { + await Promise.all(workers.map((worker) => this.calibrateWorker(worker, probe))); + } finally { + for (const worker of this.roster.values()) { + this.resetWorkerSample(worker); + } + this.calibPending.clear(); + this.calibrating = false; + } + if (this.cancel) { + return; + } + const ranked = this.rankWorkers([...this.roster.values()]); + for (const worker of ranked) { + this.events.onLog?.( + ` ${worker.host} / ${worker.device}: ${worker.benchRate > 0 ? formatRate(worker.benchRate) : "no sample"}`, + ); + } + this.emitProgress(); + } + + private calibrateWorker(worker: TrackedWorker, job: StageJob): Promise { + if (this.cancel) { + return Promise.resolve(); + } + const count = this.testBlockCount(worker); + if (worker.slaveId && this.server) { + return this.calibrateRemote(worker, job, count); + } + const device = this.params.devices.find((d) => d.key === worker.deviceKey); + if (!device) { + worker.benchRate = 0; + return Promise.resolve(); + } + return this.calibrateLocal(device, worker, job, count); + } + + private calibrateRemote(worker: TrackedWorker, job: StageJob, count: bigint): Promise { + return new Promise((resolve) => { + const jobId = randomUUID(); + this.calibPending.set(jobId, (rate) => { + worker.benchRate = rate; + resolve(); + }); + worker.jobId = jobId; + worker.busy = true; + worker.done = 0n; + worker.count = count; + worker.rate = 0; + const ok = this.server!.assign(worker.slaveId!, worker.deviceKey, { + jobId, + deviceKey: worker.deviceKey, + start: "0", + count: count.toString(), + keyLen: job.length, + charset: this.params.charset, + padByte: job.pad, + target: PAD_CT.toString("hex"), + fills: FILL_HEX, + cpuWorkers: worker.deviceType === "cpu" ? this.params.cpuWorkersFor(worker.id) : undefined, + }); + if (!ok) { + this.finishRemoteCalibration(worker.id, jobId, 0); + return; + } + beginWork(worker); + this.emitProgress(); + }); + } + + private async calibrateLocal( + device: ComputeDevice, + worker: TrackedWorker, + job: StageJob, + count: bigint, + ): Promise { + const jobId = randomUUID(); + worker.jobId = jobId; + worker.busy = true; + worker.done = 0n; + worker.count = count; + worker.rate = 0; + beginWork(worker); + this.emitProgress(); + const handle = runBrute( + { + device, + keyLen: job.length, + charset: this.params.charset, + padByte: job.pad, + start: 0n, + count, + batch: device.type === "cpu" ? 200_000 : this.params.gpuBatch, + workers: this.params.cpuWorkersFor(worker.id), + }, + { + onProgress: (done, total, rate) => { + const live = this.roster.get(worker.id); + if (live) { + this.applyProgress(live, jobId, done, total, rate); + } + this.emitProgress(device.name); + }, + onHit: () => undefined, + onLog: (line) => this.events.onLog?.(line), + }, + ); + this.kills.add(handle.kill); + try { + await handle.promise; + if (!this.cancel) { + endWork(worker, count); + worker.benchRate = averageRate(worker) || worker.rate; + } + } catch (err: unknown) { + this.events.onLog?.( + `${device.name} calibration: ${err instanceof Error ? err.message : String(err)}`, + ); + worker.benchRate = 0; + } finally { + this.kills.delete(handle.kill); + if (worker.busyStarted != null) { + endWork(worker, 0n); + } + worker.jobId = null; + worker.busy = false; + worker.rate = 0; + worker.done = 0n; + worker.count = 0n; + this.emitProgress(); + } + } + private async runStage(job: StageJob, data: Buffer, firstCt: Buffer): Promise { this.stage = job; this.stageTotal = job.space; @@ -460,7 +692,7 @@ export class MasterSearch { } } } - return idle; + return this.rankWorkers(idle); }; const feedWorker = (worker: TrackedWorker): boolean => { @@ -702,11 +934,13 @@ export class MasterSearch { for (const w of this.roster.values()) { const liveJob = w.busy && w.jobId != null && (!this.stageAlloc || this.stageAlloc.has(w.jobId)); if (liveJob) { - inflight += this.liveDone(w); + if (!this.calibrating) { + inflight += this.liveDone(w); + } rateSum += w.rate; } const pct = w.busy && w.count > 0n ? Number((w.done * 1000n) / w.count) / 10 : 0; - const avg = averageRate(w); + const avg = averageRate(w) || w.benchRate; rows.push({ id: w.id, host: w.host, @@ -735,25 +969,26 @@ export class MasterSearch { } else { this.stagePeak = stageDone; } - const overallDone = this.overallPrior + stageDone; - const elapsed = Math.max((Date.now() - this.started) / 1000, 1e-6); - const rate = rateSum > 0 ? rateSum : Number(overallDone) / elapsed; + const overallDone = this.calibrating ? 0n : this.overallPrior + stageDone; + const elapsed = this.started > 0 ? Math.max((Date.now() - this.started) / 1000, 1e-6) : 0; + const rate = rateSum > 0 ? rateSum : elapsed > 0 ? Number(overallDone) / elapsed : 0; const remain = this.overallTotal > overallDone ? this.overallTotal - overallDone : 0n; const snap: ProgressSnapshot = { length: this.stage?.length ?? 0, pad: this.stage?.pad ?? 0, cipher: "des", - stageDone, + stageDone: this.calibrating ? 0n : stageDone, stageTotal: this.stageTotal, - stagePct: this.stageTotal > 0n ? Number((stageDone * 1000n) / this.stageTotal) / 10 : 0, + stagePct: this.calibrating || this.stageTotal <= 0n ? 0 : Number((stageDone * 1000n) / this.stageTotal) / 10, overallDone, overallTotal: this.overallTotal, - overallPct: Number((overallDone * 1000n) / this.overallTotal) / 10, + overallPct: this.calibrating ? 0 : Number((overallDone * 1000n) / this.overallTotal) / 10, rate, elapsed, - etaSec: rate > 0 ? Number(remain) / rate : null, + etaSec: this.calibrating || rate <= 0 ? null : Number(remain) / rate, hits: this.hits.length, workerLabel, + calibrating: this.calibrating, }; this.events.onProgress?.(snap); this.events.onWorkers?.(rows); @@ -761,6 +996,9 @@ export class MasterSearch { } export function describeProgress(snap: ProgressSnapshot): string { + if (snap.calibrating) { + return `${formatRate(snap.rate)} calibrating elapsed -- ETA -- hits ${snap.hits}`; + } return ( `${formatRate(snap.rate)} elapsed ${formatEta(snap.elapsed)} ` + `ETA ${formatEta(snap.etaSec)} hits ${snap.hits}` diff --git a/src/ui/mainWindow.ts b/src/ui/mainWindow.ts index b26a63c..386ddde 100644 --- a/src/ui/mainWindow.ts +++ b/src/ui/mainWindow.ts @@ -1126,11 +1126,16 @@ export function createMainWindow(): QMainWindow { onProgress: (snap) => { stageMeter.setValue(snap.stagePct * 10); overallMeter.setValue(snap.overallPct * 10); - stageLabel.setText( - `Current · DES len=${snap.length} pad=0x${snap.pad.toString(16).padStart(2, "0")} ` + - `${snap.stageDone.toString()} / ${snap.stageTotal.toString()}`, - ); - overallLabel.setText(`Overall · ${snap.overallDone.toString()} / ${snap.overallTotal.toString()}`); + if (snap.calibrating) { + stageLabel.setText("Current · calibrating workers"); + overallLabel.setText("Overall · waiting for rate samples"); + } else { + stageLabel.setText( + `Current · DES len=${snap.length} pad=0x${snap.pad.toString(16).padStart(2, "0")} ` + + `${snap.stageDone.toString()} / ${snap.stageTotal.toString()}`, + ); + overallLabel.setText(`Overall · ${snap.overallDone.toString()} / ${snap.overallTotal.toString()}`); + } stats.setText(describeProgress(snap)); }, onWorkers: (rows) => paintWorkers(rows),