perf: a bit faster + restore NaN behavior if absent)

Signed-off-by: Tuan Anh Tran <me@tuananh.org>

Author: tuananh

benchmark/transform.js

@@ -1,7 +1,7 @@
 'use strict'
 
 const fs = require('fs')
-const { transform } = require('..')
+const { transform, destroy } = require('..')
 const { XMLParser } = require('fast-xml-parser')
 const xml2js = require('xml2js')
 const xmljs = require('xml-js')
@@ -64,6 +64,7 @@ const template = {
   })
 
   await run()
+  await destroy()
 })().catch((err) => {
   console.error(err)
   process.exitCode = 1

dist/camaro.js

@@ -8,35 +8,39 @@ if (NODE_MAJOR_VERSION < 12 || process.env.CAMARO_FORCE_SINGLE_THREAD === 'true'
     pool = {
         run(task, opts) {
             void opts
-            return workerFn(task)
+            return workerFn(task).then(parseCamaroJson)
         },
+        destroy() {},
     }
 } else {
-    const WorkerPool = require('piscina')
-    const piscina = new WorkerPool({ filename: resolve(__dirname, 'worker.js') })
+    const { LeanPool } = require('./lean-pool')
+    const { parseCamaroJson } = require('./json-parse')
+    const leanPool = new LeanPool(resolve(__dirname, 'pool-worker.js'))
     pool = {
         run(task, opts) {
-            if (opts && opts.transferList && opts.transferList.length > 0) {
-                return piscina.run(task, { transferList: opts.transferList })
-            }
-            return piscina.run(task)
+            return leanPool.run(task, opts).then(parseCamaroJson)
+        },
+        destroy() {
+            return leanPool.destroy()
         },
     }
 }
 
 /** Wasm Embind Utf-16 string → std::string is slower than Utf-8 bytes + malloc; reuse the Utf-8 path. */
 const textEncoderUtf8 =
     typeof TextEncoder !== 'undefined' ? new TextEncoder() : null
+let xmlUtf8Scratch = null
 
 function utf8BytesFromJsString(xml) {
-    return textEncoderUtf8
-        ? textEncoderUtf8.encode(xml)
-        : Buffer.from(xml, 'utf8')
-}
-
-function canTransferUnderlyingBuffer(view) {
-    if (!(view instanceof Uint8Array) || view.byteLength === 0) return false
-    return view.byteOffset === 0 && view.byteLength === view.buffer.byteLength
+    if (!textEncoderUtf8) {
+        return Buffer.from(xml, 'utf8')
+    }
+    const worstCase = xml.length * 3
+    if (!xmlUtf8Scratch || xmlUtf8Scratch.length < worstCase) {
+        xmlUtf8Scratch = new Uint8Array(Math.max(worstCase, 65536))
+    }
+    const { written } = textEncoderUtf8.encodeInto(xml, xmlUtf8Scratch)
+    return xmlUtf8Scratch.subarray(0, written)
 }
 
 /**
@@ -45,16 +49,10 @@ function canTransferUnderlyingBuffer(view) {
  */
 function xmlPayloadForWorkerThread(xml) {
     if (typeof xml === 'string') {
-        const u8 = utf8BytesFromJsString(xml)
-        if (
-            NODE_MAJOR_VERSION >= 12 &&
-            process.env.CAMARO_FORCE_SINGLE_THREAD !== 'true' &&
-            canTransferUnderlyingBuffer(u8)
-        ) {
-            return { xmlWire: u8, poolOpts: { transferList: [u8.buffer] } }
-        }
-        return { xmlWire: u8 }
+        // Reuse scratch + structured clone; transfer detaches and forces re-allocation.
+        return { xmlWire: utf8BytesFromJsString(xml) }
     }
+    // User-supplied binaries: structured clone only (transfer would detach caller's buffer).
     return { xmlWire: xml }
 }
 
@@ -98,6 +96,17 @@ function isEmptyObject(obj) {
     return Object.entries(obj).length === 0 && obj.constructor === Object
 }
 
+const templateStringCache = new WeakMap()
+
+function templateString(template) {
+    let cached = templateStringCache.get(template)
+    if (cached === undefined) {
+        cached = JSON.stringify(template)
+        templateStringCache.set(template, cached)
+    }
+    return cached
+}
+
 /**
  * convert xml to json base on the template object
  * @param {string|Buffer|Uint8Array|ArrayBuffer} xml xml as UTF-8 string or raw bytes
@@ -115,7 +124,7 @@ function transform(xml, template) {
     return dispatchPool(
         {
             fn: 'transform',
-            args: [payload.xmlWire, JSON.stringify(template)],
+            args: [payload.xmlWire, templateString(template)],
         },
         payload.poolOpts,
     )

dist/camaro.wasm

@@ -0,0 +1,15 @@
+'use strict'
+
+function parseCamaroJson(payload) {
+    if (typeof payload === 'string') {
+        const first = payload[0]
+        if (first === '{' || first === '[') return JSON.parse(payload)
+        return payload
+    }
+    if (payload && typeof payload === 'object' && typeof payload.json === 'string') {
+        return JSON.parse(payload.json, (_, v) => (v === '__camaro_nan__' ? NaN : v))
+    }
+    return payload
+}
+
+module.exports = { parseCamaroJson }

index.js

@@ -0,0 +1,111 @@
+'use strict'
+
+const { Worker } = require('worker_threads')
+
+function idleTimeoutMs() {
+    const raw = process.env.CAMARO_IDLE_TIMEOUT
+    if (raw === 'Infinity') return Infinity
+    if (raw !== undefined && raw !== '') {
+        const n = Number(raw)
+        if (Number.isFinite(n) && n >= 0) return n
+    }
+    // Match piscina default: shut down idle workers immediately so scripts can exit.
+    return 0
+}
+
+class LeanWorker {
+    constructor(filename, idleTimeout) {
+        this.filename = filename
+        this.idleTimeout = idleTimeout
+        this.worker = null
+        this.seq = 0
+        this.pending = new Map()
+        this.idleTimer = null
+    }
+
+    ensureWorker() {
+        if (this.worker) return
+        const worker = new Worker(this.filename)
+        worker.unref()
+        worker.on('message', ({ id, result, error }) => {
+            const p = this.pending.get(id)
+            this.pending.delete(id)
+            if (!p) return
+            if (error) p.reject(new Error(error))
+            else p.resolve(result)
+            if (this.pending.size === 0) this.scheduleIdleShutdown()
+        })
+        worker.on('error', (err) => {
+            for (const p of this.pending.values()) p.reject(err)
+            this.pending.clear()
+            this.terminate()
+        })
+        worker.on('exit', () => {
+            if (this.worker === worker) this.worker = null
+        })
+        this.worker = worker
+    }
+
+    clearIdleTimer() {
+        if (this.idleTimer !== null) {
+            clearTimeout(this.idleTimer)
+            this.idleTimer = null
+        }
+    }
+
+    scheduleIdleShutdown() {
+        this.clearIdleTimer()
+        if (this.idleTimeout === Infinity || !this.worker) return
+        this.idleTimer = setTimeout(() => {
+            this.idleTimer = null
+            if (this.pending.size === 0) this.terminate()
+        }, this.idleTimeout)
+        this.idleTimer.unref()
+    }
+
+    run(task, opts = {}) {
+        this.ensureWorker()
+        this.clearIdleTimer()
+        const id = ++this.seq
+        return new Promise((resolve, reject) => {
+            this.pending.set(id, { resolve, reject })
+            this.worker.postMessage({ id, task }, opts.transferList || [])
+        })
+    }
+
+    terminate() {
+        this.clearIdleTimer()
+        if (!this.worker) return Promise.resolve()
+        const w = this.worker
+        this.worker = null
+        return w.terminate()
+    }
+
+    destroy() {
+        return this.terminate()
+    }
+}
+
+class LeanPool {
+    constructor(filename, size) {
+        const envSize = Number(process.env.CAMARO_POOL_SIZE || 0)
+        const poolSize = size || (envSize > 0 ? envSize : 1)
+        const idleTimeout = idleTimeoutMs()
+        this.workers = Array.from(
+            { length: poolSize },
+            () => new LeanWorker(filename, idleTimeout),
+        )
+        this.next = 0
+    }
+
+    run(task, opts) {
+        const worker = this.workers[this.next++ % this.workers.length]
+        return worker.run(task, opts)
+    }
+
+    destroy() {
+        return Promise.all(this.workers.map((w) => w.destroy()))
+    }
+}
+
+module.exports = { LeanPool }

json-parse.js

@@ -69,7 +69,5 @@
         "xml-js": "^1.6.11",
         "xml2js": "^0.6.0"
     },
-    "dependencies": {
-        "piscina": "^5.2.0"
-    }
+    "dependencies": {}
 }

lean-pool.js

@@ -0,0 +1,16 @@
+'use strict'
+
+const { parentPort } = require('worker_threads')
+const workerFn = require('./worker')
+
+parentPort.on('message', async ({ id, task }) => {
+    try {
+        const result = await workerFn(task)
+        parentPort.postMessage({ id, result })
+    } catch (err) {
+        parentPort.postMessage({
+            id,
+            error: err && err.message ? err.message : String(err),
+        })
+    }
+})