Merge pull request #8 from BravoNatalie/review/tippyflitsuk-parallel-prepare

perf(backup-helper): compute prepare CommP across a worker_threads pool

Author: BravoNatalie

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,154 @@ 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<import('node:worker_threads').Worker, {resolve: (v: string) => void, reject: (e: Error) => void}>} */
+    this.busy = new Map()
+    /** @type {Error | null} */
+    this.fatalError = null
+    this.closing = false
+    /** @type {Set<import('node:worker_threads').Worker>} */
+    this.failedWorkers = new Set()
+    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))
+      worker.on('exit', (code) => this._onExit(worker, code))
+      this.workers.push(worker)
+      this.idle.push(worker)
+    }
+  }
+
+  /**
+   * @param {string} carPath
+   * @returns {Promise<string>}
+   */
+  compute(carPath) {
+    return new Promise((resolve, reject) => {
+      if (this.fatalError) {
+        reject(this.fatalError)
+        return
+      }
+
+      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) {
+    this._handleWorkerFailure(worker, err, `prepare: piece CID worker error: ${err?.message || err}`)
+  }
+
+  /**
+   * @param {import('node:worker_threads').Worker} worker
+   * @param {number} code
+   */
+  _onExit(worker, code) {
+    if (this.closing) return
+    if (code !== 0) {
+      const err = new Error(`piece CID worker exited with code ${code}`)
+      this._handleWorkerFailure(worker, err, `prepare: ${err.message}`)
+      return
+    }
+
+    if (this.busy.has(worker)) {
+      const err = new Error('piece CID worker exited unexpectedly while processing a job')
+      this._handleWorkerFailure(worker, err, `prepare: ${err.message}`)
+    }
+  }
+
+  /**
+   * @param {import('node:worker_threads').Worker} worker
+   */
+  _removeWorker(worker) {
+    this.workers = this.workers.filter((candidate) => candidate !== worker)
+    this.idle = this.idle.filter((candidate) => candidate !== worker)
+  }
+
+  /**
+   * @param {import('node:worker_threads').Worker} worker
+   * @param {Error} err
+   * @param {string} logMessage
+   */
+  _handleWorkerFailure(worker, err, logMessage) {
+    if (this.failedWorkers.has(worker)) return
+    this.failedWorkers.add(worker)
+
+    const job = this.busy.get(worker)
+    this.busy.delete(worker)
+    if (job) job.reject(err)
+    this._removeWorker(worker)
+    console.error(logMessage)
+    this._drainOrFailQueue()
+  }
+
+  _drainOrFailQueue() {
+    while (this.idle.length > 0 && this.queue.length > 0) {
+      const worker = this.idle.pop()
+      const next = this.queue.shift()
+      if (!worker || !next) break
+      this._assign(worker, next)
+    }
+
+    if (this.workers.length > 0) return
+
+    this.fatalError = new Error('all piece CID workers failed')
+    console.error(`prepare: ${this.fatalError.message}`)
+    while (this.queue.length > 0) {
+      const next = this.queue.shift()
+      if (!next) break
+      next.reject(this.fatalError)
+    }
+  }
+
+  async close() {
+    this.closing = true
+    await Promise.all(this.workers.map((worker) => worker.terminate()))
   }
 }
 
@@ -127,6 +266,8 @@ function renderPrepareProgress(summary) {
 export async function runPrepare({ dir, concurrency }) {
   const tracking = openTrackingDb(dir)
   const workerConcurrency = concurrency ?? DEFAULT_PREPARE_CONCURRENCY
+  /** @type {PieceCidWorkerPool | null} */
+  let pool = null
 
   try {
     const total = tracking.getDownloadStats().complete
@@ -135,6 +276,8 @@ export async function runPrepare({ dir, concurrency }) {
       return
     }
 
+    pool = new PieceCidWorkerPool(workerConcurrency)
+
     const summary = {
       total,
       done: 0,
@@ -171,7 +314,8 @@ export async function runPrepare({ dir, concurrency }) {
 
             let pieceCid = candidate.pieceCid
             if (!pieceCid) {
-              pieceCid = await calculateLocalPieceCid(workItem.carPath)
+              if (!pool) throw new Error('prepare: piece CID worker pool was not initialized')
+              pieceCid = await pool.compute(workItem.carPath)
               tracking.setPieceCid(workItem.shardCid, pieceCid)
               computedPieceCid = true
             } else {
@@ -211,6 +355,7 @@ export async function runPrepare({ dir, concurrency }) {
       `prepare: done. total=${summary.total} done=${summary.done}/${summary.total} computed=${summary.computed} failed=${summary.failed}`,
     )
   } finally {
+    if (pool) await pool.close()
     tracking.close()
   }
 }

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()
+  }
+})