From 1de3e21824afc88cf466508378154de25d04d00d Mon Sep 17 00:00:00 2001 From: TippyFlits Date: Thu, 4 Jun 2026 20:14:39 +0100 Subject: [PATCH 1/2] perf(backup-helper): compute prepare CommP across a worker_threads pool prepare computes a piece CID for every shard missing one via the pure-JS @filoz/synapse-core/piece hash. That hash is CPU-bound and single-threaded, so on a multi-core node prepare pins one core and leaves the rest idle. Move the hashing into a worker_threads pool sized to --concurrency. Each worker calls the same calculateFromIterable, so output piece CIDs are byte-identical to the single-thread path; only throughput changes. DB writes and shard->piece renames stay on the main thread (sqlite is not shared with workers). Measured on a 64-core machine (~10 MB avg CARs): ~20 MB/s at -c 8, ~69 MB/s at -c 24, ~84 MB/s at -c 32 (knee ~32 on that box). No change to default concurrency or any existing flag. --- scripts/backup-helper/commands/prepare.mjs | 92 ++++++++++++++++--- .../backup-helper/lib/piece-cid-worker.mjs | 29 ++++++ 2 files changed, 110 insertions(+), 11 deletions(-) create mode 100644 scripts/backup-helper/lib/piece-cid-worker.mjs 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() + } +}) From bc3181bf7efa338843e995a0b1b33a446670af79 Mon Sep 17 00:00:00 2001 From: TippyFlits Date: Thu, 4 Jun 2026 20:14:39 +0100 Subject: [PATCH 2/2] fix(backup-helper): read curio import-pieces result from --result file curio toolbox import-pieces moved its JSON result off stdout into a file passed via a now-required --result flag, and writes that file even on non-zero exit. Against a curio build with that change, runParkingBinary fails with 'ERROR: result is required'. Pass --result /.parking-result.json, read the JSON from there on both success and failure (surfacing curio's structured error field), and remove the file afterward. Tracks filecoin-project/curio#1264 (commit 7c8297ca). Note that #1264 is currently a draft; if the flag contract shifts before it merges this will need a follow-up. --- scripts/backup-helper/commands/commit.mjs | 24 +++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) 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))