tom/DESCracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance MasterSearch with worker calibration and progress tracking

Browse Source 
- 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.
This commit is contained in:
Tom Butcher 2026-09-19 22:43:40 +01:00
parent e0c866a542
commit da56f5dcfd
2 changed files with 265 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

272
src/search/orchestrator.ts
View File

@ -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<string, (rate: number) => 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<HitRecord[]> {
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<string, TrackedWorker>();
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<void> {
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<void> {
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<void> {
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<void> {
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<void> {
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}`

15
src/ui/mainWindow.ts
View File

@ -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),