diff --git a/src/util/processAvifImage.js b/src/util/processAvifImage.js index a921849..ff484ed 100644 --- a/src/util/processAvifImage.js +++ b/src/util/processAvifImage.js @@ -1,140 +1,85 @@ -import { optimizeImage } from 'wasm-image-optimization'; -import { rgbaToThumbHash } from 'thumbhash'; -import { encodeRGBA } from 'skelehash-wasm'; +import { Worker } from 'node:worker_threads'; +import { fileURLToPath } from 'node:url'; +import os from 'node:os'; +import processAvifImageCore from './processAvifImageCore.js'; -const FULL_MAX = 1000; -const THUMB_MAX = 200; -const SMALL_THUMB_MAX = 32; +// Image decode/resize/AVIF encode is CPU-bound WASM; running it on the main thread blocks the event +// loop and serializes batch imports. Jobs run in a small worker_threads pool instead. +// (`wasm-image-optimization/workers` is not used: its `worker-lib` dependency cannot be loaded by +// native Node ESM.) Set DASHSTORE_IMAGE_WORKERS=0 to process inline. +const WORKER_PATH = fileURLToPath(new URL('./processAvifImageWorker.js', import.meta.url)); +const configured = process.env.DASHSTORE_IMAGE_WORKERS; +const POOL_SIZE = configured !== undefined + ? Math.max(0, Number.parseInt(configured, 10) || 0) + : Math.max(1, Math.min(4, (os.cpus?.()?.length || 2) - 1)); -function fitInsideDims(srcW, srcH, maxW, maxH) { - if (!srcW || !srcH || (srcW <= maxW && srcH <= maxH)) { - return {}; - } - const scale = Math.min(maxW / srcW, maxH / srcH); - return { - width: Math.max(1, Math.round(srcW * scale)), - height: Math.max(1, Math.round(srcH * scale)), - fit: 'fill', +const pool = []; +const queue = []; +let jobSeq = 0; + +function spawn() { + const entry = { worker: new Worker(WORKER_PATH), ready: false, busy: false, job: null }; + entry.worker.unref(); + entry.worker.on('message', (msg) => { + if (msg.ready) { + entry.ready = true; + drain(); + return; + } + const job = entry.job; + if (!job || job.id !== msg.id) return; + entry.job = null; + entry.busy = false; + entry.worker.unref(); + if (msg.ok) { + const r = msg.result; + r.full.buffer = Buffer.from(r.full.buffer); + r.thumb.buffer = Buffer.from(r.thumb.buffer); + r.smallThumb.buffer = Buffer.from(r.smallThumb.buffer); + job.resolve(r); + } else { + job.reject(new Error(msg.error)); + } + drain(); + }); + const fail = (err) => { + const i = pool.indexOf(entry); + if (i >= 0) pool.splice(i, 1); + if (entry.job) entry.job.reject(err); + entry.job = null; + if (queue.length) drain(); }; + entry.worker.on('error', fail); + entry.worker.on('exit', (code) => fail(new Error(`Image worker exited (code ${code})`))); + pool.push(entry); } -function unpremultiplyRgba(data) { - const out = new Uint8Array(data.length); - for (let i = 0; i < data.length; i += 4) { - const a = data[i + 3]; - if (a === 255 || a === 0) { - out[i] = data[i]; - out[i + 1] = data[i + 1]; - out[i + 2] = data[i + 2]; - out[i + 3] = a; - } else { - const f = 255 / a; - out[i] = Math.min(255, Math.round(data[i] * f)); - out[i + 1] = Math.min(255, Math.round(data[i + 1] * f)); - out[i + 2] = Math.min(255, Math.round(data[i + 2] * f)); - out[i + 3] = a; +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.job = job; + entry.worker.ref(); // keep process alive while the job is in flight + entry.worker.postMessage({ id: job.id, buffer: job.buffer }, [job.buffer]); } - return out; +} + +/** Terminates pooled image workers (optional; they are unref'd while idle). */ +export async function shutdownImageWorkers() { + await Promise.all(pool.splice(0).map((e) => e.worker.terminate())); } export default async function processAvifImage(inputBuffer) { - const input = Buffer.isBuffer(inputBuffer) - ? inputBuffer - : Buffer.from(inputBuffer); - - const meta = await optimizeImage({ image: input, format: 'none' }); - const w = meta.originalWidth ?? 0; - const h = meta.originalHeight ?? 0; - const isAvif = meta.originalFormat === 'avif'; - const useOriginalFull = - isAvif && w > 0 && h > 0 && w <= FULL_MAX && h <= FULL_MAX; - - if (!useOriginalFull) { - console.log( - `processAvifImage: converting full image (format=${meta.originalFormat ?? 'unknown'}, ${w}×${h}, avif=${isAvif})`, - ); - } - - const fullDims = fitInsideDims(w, h, FULL_MAX, FULL_MAX); - const thumbDims = fitInsideDims(w, h, THUMB_MAX, THUMB_MAX); - const rgbaDims = fitInsideDims(w, h, 100, 100); - const smallThumbDims = fitInsideDims(w, h, SMALL_THUMB_MAX, SMALL_THUMB_MAX); - - const [full, thumb, thumbRgba, smallThumb] = await Promise.all([ - useOriginalFull - ? Promise.resolve({ - width: w, - height: h, - data: input, - }) - : optimizeImage({ - image: input, - ...fullDims, - format: 'avif', - speed: 8, - }), - optimizeImage({ - image: input, - ...thumbDims, - format: 'avif', - speed: 8, - }), - optimizeImage({ - image: input, - ...rgbaDims, - format: 'raw', - }), - optimizeImage({ - image: input, - ...smallThumbDims, - format: 'avif', - speed: 8, - }), - ]); - - const rgbaPixels = unpremultiplyRgba(thumbRgba.data); - - const [thumbHashBytes, skele] = await Promise.all([ - rgbaToThumbHash( - thumbRgba.width, - thumbRgba.height, - rgbaPixels, - ), - encodeRGBA( - thumbRgba.width, - thumbRgba.height, - rgbaPixels, - ), - ]); - - if (!skele?.base62) { - throw new Error('encodeRGBA failed to produce a base62 skelehash'); - } - - const fullBuf = Buffer.isBuffer(full.data) ? full.data : Buffer.from(full.data); - const thumbBuf = Buffer.from(thumb.data); - const smallThumbBuf = Buffer.from(smallThumb.data); - - return { - full: { - w: full.width, - h: full.height, - size: fullBuf.length, - buffer: fullBuf, - }, - thumb: { - w: thumb.width, - h: thumb.height, - size: thumbBuf.length, - buffer: thumbBuf, - hash: Buffer.from(thumbHashBytes).toString('base64'), - }, - smallThumb: { - w: smallThumb.width, - h: smallThumb.height, - buffer: smallThumbBuf, - }, - skelehash: skele.base62, - }; + if (POOL_SIZE === 0) return processAvifImageCore(inputBuffer); + const input = Buffer.isBuffer(inputBuffer) ? inputBuffer : Buffer.from(inputBuffer); + const buffer = Uint8Array.from(input).buffer; // standalone ArrayBuffer, safe to transfer + return new Promise((resolve, reject) => { + queue.push({ id: ++jobSeq, buffer, resolve, reject }); + drain(); + }); } diff --git a/src/util/processAvifImageCore.js b/src/util/processAvifImageCore.js new file mode 100644 index 0000000..a921849 --- /dev/null +++ b/src/util/processAvifImageCore.js @@ -0,0 +1,140 @@ +import { optimizeImage } from 'wasm-image-optimization'; +import { rgbaToThumbHash } from 'thumbhash'; +import { encodeRGBA } from 'skelehash-wasm'; + +const FULL_MAX = 1000; +const THUMB_MAX = 200; +const SMALL_THUMB_MAX = 32; + +function fitInsideDims(srcW, srcH, maxW, maxH) { + if (!srcW || !srcH || (srcW <= maxW && srcH <= maxH)) { + return {}; + } + const scale = Math.min(maxW / srcW, maxH / srcH); + return { + width: Math.max(1, Math.round(srcW * scale)), + height: Math.max(1, Math.round(srcH * scale)), + fit: 'fill', + }; +} + +function unpremultiplyRgba(data) { + const out = new Uint8Array(data.length); + for (let i = 0; i < data.length; i += 4) { + const a = data[i + 3]; + if (a === 255 || a === 0) { + out[i] = data[i]; + out[i + 1] = data[i + 1]; + out[i + 2] = data[i + 2]; + out[i + 3] = a; + } else { + const f = 255 / a; + out[i] = Math.min(255, Math.round(data[i] * f)); + out[i + 1] = Math.min(255, Math.round(data[i + 1] * f)); + out[i + 2] = Math.min(255, Math.round(data[i + 2] * f)); + out[i + 3] = a; + } + } + return out; +} + +export default async function processAvifImage(inputBuffer) { + const input = Buffer.isBuffer(inputBuffer) + ? inputBuffer + : Buffer.from(inputBuffer); + + const meta = await optimizeImage({ image: input, format: 'none' }); + const w = meta.originalWidth ?? 0; + const h = meta.originalHeight ?? 0; + const isAvif = meta.originalFormat === 'avif'; + const useOriginalFull = + isAvif && w > 0 && h > 0 && w <= FULL_MAX && h <= FULL_MAX; + + if (!useOriginalFull) { + console.log( + `processAvifImage: converting full image (format=${meta.originalFormat ?? 'unknown'}, ${w}×${h}, avif=${isAvif})`, + ); + } + + const fullDims = fitInsideDims(w, h, FULL_MAX, FULL_MAX); + const thumbDims = fitInsideDims(w, h, THUMB_MAX, THUMB_MAX); + const rgbaDims = fitInsideDims(w, h, 100, 100); + const smallThumbDims = fitInsideDims(w, h, SMALL_THUMB_MAX, SMALL_THUMB_MAX); + + const [full, thumb, thumbRgba, smallThumb] = await Promise.all([ + useOriginalFull + ? Promise.resolve({ + width: w, + height: h, + data: input, + }) + : optimizeImage({ + image: input, + ...fullDims, + format: 'avif', + speed: 8, + }), + optimizeImage({ + image: input, + ...thumbDims, + format: 'avif', + speed: 8, + }), + optimizeImage({ + image: input, + ...rgbaDims, + format: 'raw', + }), + optimizeImage({ + image: input, + ...smallThumbDims, + format: 'avif', + speed: 8, + }), + ]); + + const rgbaPixels = unpremultiplyRgba(thumbRgba.data); + + const [thumbHashBytes, skele] = await Promise.all([ + rgbaToThumbHash( + thumbRgba.width, + thumbRgba.height, + rgbaPixels, + ), + encodeRGBA( + thumbRgba.width, + thumbRgba.height, + rgbaPixels, + ), + ]); + + if (!skele?.base62) { + throw new Error('encodeRGBA failed to produce a base62 skelehash'); + } + + const fullBuf = Buffer.isBuffer(full.data) ? full.data : Buffer.from(full.data); + const thumbBuf = Buffer.from(thumb.data); + const smallThumbBuf = Buffer.from(smallThumb.data); + + return { + full: { + w: full.width, + h: full.height, + size: fullBuf.length, + buffer: fullBuf, + }, + thumb: { + w: thumb.width, + h: thumb.height, + size: thumbBuf.length, + buffer: thumbBuf, + hash: Buffer.from(thumbHashBytes).toString('base64'), + }, + smallThumb: { + w: smallThumb.width, + h: smallThumb.height, + buffer: smallThumbBuf, + }, + skelehash: skele.base62, + }; +} diff --git a/src/util/processAvifImageWorker.js b/src/util/processAvifImageWorker.js new file mode 100644 index 0000000..c9a000e --- /dev/null +++ b/src/util/processAvifImageWorker.js @@ -0,0 +1,31 @@ +// Worker-thread entry for processAvifImage. Protocol: parent sends `{ id, buffer }` (ArrayBuffer, +// transferred); worker replies `{ id, ok: true, result }` (buffers transferred back) or `{ id, ok: false, error }`. +import { parentPort } from 'node:worker_threads'; +import processAvifImageCore from './processAvifImageCore.js'; + +parentPort.postMessage({ ready: true }); + +parentPort.on('message', async ({ id, buffer }) => { + try { + const result = await processAvifImageCore(Buffer.from(buffer)); + const own = (buf) => Uint8Array.from(buf).buffer; + const full = own(result.full.buffer); + const thumb = own(result.thumb.buffer); + const small = own(result.smallThumb.buffer); + parentPort.postMessage( + { + id, + ok: true, + result: { + ...result, + full: { ...result.full, buffer: full }, + thumb: { ...result.thumb, buffer: thumb }, + smallThumb: { ...result.smallThumb, buffer: small }, + }, + }, + [full, thumb, small], + ); + } catch (err) { + parentPort.postMessage({ id, ok: false, error: err?.message || String(err) }); + } +});