From 1b432cb6c66f0f5a5238bb76d5042240334b7586 Mon Sep 17 00:00:00 2001 From: seb Date: Fri, 30 Oct 2026 05:50:46 +0100 Subject: [PATCH] u --- src/api/index.js | 2 + .../companyImport/articleImageImporter.js | 34 ++++---- src/services/companyImport/articleImporter.js | 37 +++++---- src/services/companyImport/imageHashWorker.js | 36 +++++++++ src/services/companyImport/imageHasher.js | 79 ++++++++++++++----- 5 files changed, 140 insertions(+), 48 deletions(-) create mode 100644 src/services/companyImport/imageHashWorker.js diff --git a/src/api/index.js b/src/api/index.js index d8872e6..86b73e5 100644 --- a/src/api/index.js +++ b/src/api/index.js @@ -13,6 +13,7 @@ import { ensureConnection } from '../db/store.js'; import { closeAll as closeAllSse } from './sse.js'; import { closeAllManagers } from '../services/conversationManager.js'; import { shutdownImageCropWorkers } from '../services/imageCrop.js'; +import { shutdownImageHashWorkers } from '../services/companyImport/imageHasher.js'; import { shutdownJavascriptSandboxWorkers } from '../services/javascriptSandbox.js'; import { getOrCreateCert } from '../util/cert.js'; import { createLogger } from '../util/logger.js'; @@ -279,6 +280,7 @@ export async function stopApi() { // Pooled image-crop workers are live threads and must be terminated too, or the // process stays alive after the last request has been served. await shutdownImageCropWorkers(); + await shutdownImageHashWorkers(); // Same for the pooled sandbox threads that run `javascript_execute` / applets. await shutdownJavascriptSandboxWorkers(); await stopBackupService(); diff --git a/src/services/companyImport/articleImageImporter.js b/src/services/companyImport/articleImageImporter.js index f01d082..eadbc5d 100644 --- a/src/services/companyImport/articleImageImporter.js +++ b/src/services/companyImport/articleImageImporter.js @@ -1,12 +1,13 @@ // MARK: - Imports // #region Imports import fs from 'fs'; -import { computeSkeleHash } from './imageHasher.js'; import { createLogger } from '../../util/logger.js'; // #endregion const log = createLogger('import'); +const IMAGE_CONCURRENCY = 4; + // MARK: - Article Image Importer // #region Article Image Importer @@ -28,20 +29,25 @@ export async function importArticleImages({ }) { let totalImagesImported = 0; - for (const task of articleImageTasks) { - try { - const imgBuf = fs.readFileSync(task.fullPath); - await store.article.addImage(googleAuthId, task.sku, imgBuf); - totalImagesImported++; - - const skele = await computeSkeleHash(imgBuf); - progressTracker.emitSkeleHash(skele?.base62 || null); - } catch (err) { - log.warn(`Article image add failed for ${task.sku} (${task.relPath}):`, err.message); + // addImage decodes/encodes in the store's worker pool before its (serialized) DB transaction, so + // several in flight overlap the CPU work. The store also returns the skelehash — no second decode. + let next = 0; + const runNext = async () => { + while (next < articleImageTasks.length) { + const task = articleImageTasks[next++]; + try { + const imgBuf = await fs.promises.readFile(task.fullPath); + const added = await store.article.addImage(googleAuthId, task.sku, imgBuf); + totalImagesImported++; + progressTracker.emitSkeleHash(added?.skelehash || null); + } catch (err) { + log.warn(`Article image add failed for ${task.sku} (${task.relPath}):`, err.message); + } + progressTracker.counts.imagesdone++; + progressTracker.emitProgress(); } - progressTracker.counts.imagesdone++; - progressTracker.emitProgress(); - } + }; + await Promise.all(Array.from({ length: Math.min(IMAGE_CONCURRENCY, articleImageTasks.length) }, runNext)); progressTracker.flushSkeleHashes(); progressTracker.emitProgress(true); diff --git a/src/services/companyImport/articleImporter.js b/src/services/companyImport/articleImporter.js index f9e5058..771380f 100644 --- a/src/services/companyImport/articleImporter.js +++ b/src/services/companyImport/articleImporter.js @@ -4,7 +4,6 @@ import fs from 'fs'; import { cleanValue, parseGermanFloat, parseSuppliers } from './csvParser.js'; import { setupVariantAttributes } from './variantImporter.js'; import { setupPfandArticles, linkArticlePfand, extractPfandValue } from './pfandImporter.js'; -import { computeSkeleHash } from './imageHasher.js'; import { createLogger } from '../../util/logger.js'; // #endregion @@ -154,6 +153,9 @@ export async function importArticles({ let pfandOptionsLinked = 0; let suppliersLinked = 0; let articlePicturesImported = 0; + const IMAGE_CONCURRENCY = 4; + /** @type {Set>} */ + const inFlightImageJobs = new Set(); // 6. Unified single loop over all articles for (const a of sortedArticles) { @@ -452,26 +454,33 @@ export async function importArticles({ // --- Pictures handling for this article --- const imagesForThisArt = articleImagesBySku.get(sku); if (imagesForThisArt && imagesForThisArt.length > 0) { - for (const imgTask of imagesForThisArt) { - try { - const imgBuf = fs.readFileSync(imgTask.fullPath); - await store.article.addImage(googleAuthId, sku, imgBuf); - articlePicturesImported++; - - const skele = await computeSkeleHash(imgBuf); - progressTracker.emitSkeleHash(skele?.base62 || null); - } catch (imgErr) { - log.warn(`Article image add failed for ${sku} (${imgTask.relPath}):`, imgErr.message); + // Don't await: addImage's decode/encode runs in the store's worker pool, so keep creating the + // next articles while up to IMAGE_CONCURRENCY images are processed. Images of one article stay + // sequential (sort order). Awaiting happens only when the in-flight limit is reached. + const job = (async () => { + for (const imgTask of imagesForThisArt) { + try { + const imgBuf = await fs.promises.readFile(imgTask.fullPath); + const added = await store.article.addImage(googleAuthId, sku, imgBuf); + articlePicturesImported++; + progressTracker.emitSkeleHash(added?.skelehash || null); + } catch (imgErr) { + log.warn(`Article image add failed for ${sku} (${imgTask.relPath}):`, imgErr.message); + } + progressTracker.counts.imagesdone++; + progressTracker.emitProgress(); } - progressTracker.counts.imagesdone++; - progressTracker.emitProgress(); - } + })(); + inFlightImageJobs.add(job); + job.finally(() => inFlightImageJobs.delete(job)); + if (inFlightImageJobs.size >= IMAGE_CONCURRENCY) await Promise.race(inFlightImageJobs); } progressTracker.counts.articlesdone++; progressTracker.emitProgress(); } + await Promise.all(inFlightImageJobs); progressTracker.flushSkeleHashes(); progressTracker.emitProgress(true); log.info(`Articles complete: New: ${articlesImported}, Updated: ${articlesUpdated}, Variants: ${variantsLinked}, Pfand: ${pfandOptionsLinked}, Suppliers: ${suppliersLinked}, Pictures: ${articlePicturesImported}`); diff --git a/src/services/companyImport/imageHashWorker.js b/src/services/companyImport/imageHashWorker.js new file mode 100644 index 0000000..8892a28 --- /dev/null +++ b/src/services/companyImport/imageHashWorker.js @@ -0,0 +1,36 @@ +/** + * Worker-thread entry for SkeleHash computation (decode + resize + encodeRGBA). + * Protocol: parent sends `{ id, buffer }` (ArrayBuffer, transferred); worker replies + * `{ id, ok: true, result }` or `{ id, ok: false, error }`. + */ +import { parentPort } from 'worker_threads'; +import { optimizeImage } from 'wasm-image-optimization'; +import { encodeRGBA } from 'skelehash-wasm'; + +parentPort.postMessage({ ready: true }); + +parentPort.on('message', async ({ id, buffer }) => { + try { + const raw = await optimizeImage({ + image: new Uint8Array(buffer), + width: 200, + height: 200, + fit: 'contain', + format: 'raw', + }); + const result = await encodeRGBA(raw.width, raw.height, raw.data); + if (!result?.base62) throw new Error('encodeRGBA failed to produce a base62 skelehash'); + parentPort.postMessage({ + id, + ok: true, + result: { + base62: result.base62, + width: raw.width, + height: raw.height, + hashLength: result.hash?.length ?? 0, + }, + }); + } catch (err) { + parentPort.postMessage({ id, ok: false, error: err?.message || String(err) }); + } +}); diff --git a/src/services/companyImport/imageHasher.js b/src/services/companyImport/imageHasher.js index ba07cb7..4f9576a 100644 --- a/src/services/companyImport/imageHasher.js +++ b/src/services/companyImport/imageHasher.js @@ -1,17 +1,69 @@ // MARK: - Imports // #region Imports -import { optimizeImage } from 'wasm-image-optimization'; -import { encodeRGBA } from 'skelehash-wasm'; +import { Worker } from 'worker_threads'; +import { fileURLToPath } from 'url'; +import os from 'os'; import { createLogger } from '../../util/logger.js'; // #endregion const log = createLogger('import'); +// `wasm-image-optimization/workers` (worker-lib) cannot be loaded under native Node ESM +// (extensionless `./pool` import -> ERR_MODULE_NOT_FOUND), so use a plain worker_threads pool. +const WORKER_PATH = fileURLToPath(new URL('./imageHashWorker.js', import.meta.url)); +const POOL_SIZE = Math.max(1, Math.min(4, (os.cpus?.()?.length || 2) - 1)); + +/** @type {Array<{ worker: Worker, busy: boolean, pending: Map }>} */ +const pool = []; +const queue = []; +let jobSeq = 0; + +function spawn() { + const entry = { worker: new Worker(WORKER_PATH), busy: false, pending: new Map(), ready: false }; + entry.worker.on('message', (msg) => { + if (msg.ready) { entry.ready = true; drain(); return; } + const p = entry.pending.get(msg.id); + if (!p) return; + entry.pending.delete(msg.id); + entry.busy = false; + entry.worker.unref(); + if (msg.ok) p.resolve(msg.result); else p.reject(new Error(msg.error)); + drain(); + }); + entry.worker.on('error', (err) => { + for (const p of entry.pending.values()) p.reject(err); + entry.pending.clear(); + pool.splice(pool.indexOf(entry), 1); + }); + entry.worker.unref(); + pool.push(entry); +} + +function drain() { + while (queue.length) { + const entry = pool.find((e) => e.ready && !e.busy); + if (!entry) { + if (pool.length < POOL_SIZE) spawn(); + return; + } + const job = queue.shift(); + entry.busy = true; + entry.worker.ref(); // keep the process alive while a job is in flight + entry.pending.set(job.id, job); + entry.worker.postMessage({ id: job.id, buffer: job.buffer }, [job.buffer]); + } +} + +/** Terminates the hashing workers (call on shutdown). */ +export async function shutdownImageHashWorkers() { + await Promise.all(pool.splice(0).map((e) => e.worker.terminate())); +} + // MARK: - SkeleHash Computation // #region SkeleHash Computation /** - * Computes SkeleHash base62 hash for an image buffer or base64 string. + * Computes SkeleHash base62 hash for an image buffer or base64 string, off the main thread. * * @param {Buffer|string} buffer * @returns {Promise<{base62: string, width: number, height: number, hashLength: number}|null>} @@ -23,24 +75,11 @@ export async function computeSkeleHash(buffer) { const base64Data = input.includes(',') ? input.split(',')[1] : input; input = Buffer.from(base64Data, 'base64'); } - const raw = await optimizeImage({ - image: input, - width: 200, - height: 200, - fit: 'contain', - format: 'raw', + const copy = Uint8Array.from(input).buffer; // standalone ArrayBuffer, safe to transfer + return new Promise((resolve, reject) => { + queue.push({ id: ++jobSeq, buffer: copy, resolve, reject }); + drain(); }); - - const result = await encodeRGBA(raw.width, raw.height, raw.data); - if (!result?.base62) { - throw new Error('encodeRGBA failed to produce a base62 skelehash'); - } - return { - base62: result.base62, - width: raw.width, - height: raw.height, - hashLength: result.hash?.length ?? 0, - }; } // #endregion