diff --git a/scripts/backup-helper/commands/commit.mjs b/scripts/backup-helper/commands/commit.mjs index 8874082..5313df2 100644 --- a/scripts/backup-helper/commands/commit.mjs +++ b/scripts/backup-helper/commands/commit.mjs @@ -5,6 +5,9 @@ * provider machine and returns ready-to-commit `pieceCid`s. */ +import fs from 'node:fs/promises' +import path from 'node:path' + import { parse as parsePieceCid } from '@filoz/synapse-core/piece' import { fromSecp256k1 } from '@filoz/synapse-core/session-key' import { addPieces, createDataSet, waitForAddPieces, waitForCreateDataSet } from '@filoz/synapse-core/sp' @@ -195,9 +198,12 @@ function extractParkingJson(stdout) { * @returns {Promise} */ async function runParkingBinary(dir, target) { - let result + // curio toolbox import-pieces now writes its JSON result to a --result file + // (commit 7c8297ca "always write in json file"), not stdout. Local patch + // pending Natalie shipping the same fix upstream to migration-support-scripts. + const resultPath = path.join(dir, '.parking-result.json') try { - result = await execa({ + await execa({ env: { LANG: 'en_US.UTF-8', GOLOG_LOG_LEVEL: 'error', @@ -211,17 +217,27 @@ async function runParkingBinary(dir, target) { target, '--batch-size', String(PARKING_BATCH_SIZE), + '--result', + resultPath, ]) } catch (err) { - const message = err?.stderr || err?.stdout || err?.message || String(err) + // curio writes the result file even on failure, with an `error` field set + let structured + try { + structured = JSON.parse(await fs.readFile(resultPath, 'utf8')) + } catch {} + await fs.rm(resultPath, { force: true }) + const message = structured?.error || err?.stderr || err?.stdout || err?.message || String(err) throw new Error(`commit: parking command failed: ${message}`) } let parsed try { - parsed = JSON.parse(extractParkingJson(result.stdout || '')) + parsed = JSON.parse(await fs.readFile(resultPath, 'utf8')) } catch (err) { throw new Error(`commit: parking command returned invalid JSON: ${err?.message || err}`) + } finally { + await fs.rm(resultPath, { force: true }) } return /** @type {ParkingResult} */ (parkingResultSchema.parse(parsed)) diff --git a/scripts/backup-helper/commands/prepare.mjs b/scripts/backup-helper/commands/prepare.mjs index 9c7b9bc..e2cb2df 100644 --- a/scripts/backup-helper/commands/prepare.mjs +++ b/scripts/backup-helper/commands/prepare.mjs @@ -7,10 +7,10 @@ * CAR bytes and persisted back into tracking.db. */ -import { createReadStream } from 'node:fs' import fs from 'node:fs/promises' +import { fileURLToPath } from 'node:url' +import { Worker } from 'node:worker_threads' -import { calculateFromIterable } from '@filoz/synapse-core/piece' import pMap from 'p-map' import { pathExists, renderProgressLine } from '../../utils.js' @@ -28,15 +28,83 @@ const PREPARE_BATCH_SIZE = 1_000 */ /** - * @param {string} carPath + * Pool of worker threads that each compute a piece CID from a CAR file, so the + * CPU-bound (pure-JS, single-threaded) CommP hashing runs in parallel across + * cores. Each worker calls the same `@filoz/synapse-core/piece` hash as before, + * so output piece CIDs are identical — only throughput changes. DB writes and + * file renames stay on the main thread (sqlite is not shared with workers). */ -async function calculateLocalPieceCid(carPath) { - const stream = createReadStream(carPath) - try { - const pieceCid = await calculateFromIterable(stream) - return pieceCid.toString() - } finally { - stream.destroy() +class PieceCidWorkerPool { + /** @param {number} size */ + constructor(size) { + this.workerPath = fileURLToPath(new URL('../lib/piece-cid-worker.mjs', import.meta.url)) + /** @type {import('node:worker_threads').Worker[]} */ + this.workers = [] + /** @type {import('node:worker_threads').Worker[]} */ + this.idle = [] + /** @type {Array<{carPath: string, resolve: (v: string) => void, reject: (e: Error) => void}>} */ + this.queue = [] + /** @type {Map void, reject: (e: Error) => void}>} */ + this.busy = new Map() + for (let i = 0; i < Math.max(1, size); i++) { + const worker = new Worker(this.workerPath) + worker.on('message', (msg) => this._onMessage(worker, msg)) + worker.on('error', (err) => this._onError(worker, err)) + this.workers.push(worker) + this.idle.push(worker) + } + } + + /** + * @param {string} carPath + * @returns {Promise} + */ + compute(carPath) { + return new Promise((resolve, reject) => { + const worker = this.idle.pop() + if (worker) this._assign(worker, { carPath, resolve, reject }) + else this.queue.push({ carPath, resolve, reject }) + }) + } + + /** + * @param {import('node:worker_threads').Worker} worker + * @param {{carPath: string, resolve: (v: string) => void, reject: (e: Error) => void}} job + */ + _assign(worker, job) { + this.busy.set(worker, { resolve: job.resolve, reject: job.reject }) + worker.postMessage({ carPath: job.carPath }) + } + + /** + * @param {import('node:worker_threads').Worker} worker + * @param {{pieceCid?: string, error?: string}} msg + */ + _onMessage(worker, msg) { + const job = this.busy.get(worker) + this.busy.delete(worker) + if (job) { + if (msg.error) job.reject(new Error(msg.error)) + else job.resolve(/** @type {string} */ (msg.pieceCid)) + } + const next = this.queue.shift() + if (next) this._assign(worker, next) + else this.idle.push(worker) + } + + /** + * @param {import('node:worker_threads').Worker} worker + * @param {Error} err + */ + _onError(worker, err) { + const job = this.busy.get(worker) + this.busy.delete(worker) + if (job) job.reject(err) + // a crashed worker leaves the pool; remaining workers continue + } + + async close() { + await Promise.all(this.workers.map((worker) => worker.terminate())) } } @@ -127,6 +195,7 @@ function renderPrepareProgress(summary) { export async function runPrepare({ dir, concurrency }) { const tracking = openTrackingDb(dir) const workerConcurrency = concurrency ?? DEFAULT_PREPARE_CONCURRENCY + const pool = new PieceCidWorkerPool(workerConcurrency) try { const total = tracking.getDownloadStats().complete @@ -171,7 +240,7 @@ export async function runPrepare({ dir, concurrency }) { let pieceCid = candidate.pieceCid if (!pieceCid) { - pieceCid = await calculateLocalPieceCid(workItem.carPath) + pieceCid = await pool.compute(workItem.carPath) tracking.setPieceCid(workItem.shardCid, pieceCid) computedPieceCid = true } else { @@ -209,6 +278,7 @@ export async function runPrepare({ dir, concurrency }) { `prepare: done. total=${summary.total} done=${summary.done}/${summary.total} computed=${summary.computed} failed=${summary.failed}`, ) } finally { + await pool.close() tracking.close() } } diff --git a/scripts/backup-helper/lib/piece-cid-worker.mjs b/scripts/backup-helper/lib/piece-cid-worker.mjs new file mode 100644 index 0000000..8fd3e36 --- /dev/null +++ b/scripts/backup-helper/lib/piece-cid-worker.mjs @@ -0,0 +1,29 @@ +/** + * Worker thread for `backup-helper prepare`: computes a piece CID from a local + * CAR file using the same `@filoz/synapse-core/piece` hash as the main thread, + * so the CPU-bound (pure-JS) CommP hashing can run in parallel across cores. + * + * Local addition (not upstream) — parallelises prepare's hashing. The hash + * itself is unchanged, so output piece CIDs are identical to the single-thread + * path; only the throughput differs. + */ +import { createReadStream } from 'node:fs' +import { parentPort } from 'node:worker_threads' + +import { calculateFromIterable } from '@filoz/synapse-core/piece' + +if (!parentPort) { + throw new Error('piece-cid-worker.mjs must be run as a worker thread') +} + +parentPort.on('message', async ({ carPath }) => { + const stream = createReadStream(carPath) + try { + const pieceCid = await calculateFromIterable(stream) + parentPort.postMessage({ pieceCid: pieceCid.toString() }) + } catch (err) { + parentPort.postMessage({ error: String(err?.message || err) }) + } finally { + stream.destroy() + } +})