perf: reduce single transform latency

Avoid worker round-trips for single in-flight transforms while retaining the pool for concurrent requests.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: tuananh

index.js

@@ -19,13 +19,42 @@ if (forceMainThread) {
     }
 } else {
     const { LeanPool } = require('./lean-pool')
+    const workerFn = require('./worker')
     const leanPool = new LeanPool(resolve(__dirname, 'pool-worker.js'), {
         maxThreads: poolSize > 0 ? poolSize : undefined,
         onRecycleXml: recycleXmlUtf8Buffer,
         useSab: useSabIpc,
     })
+    let directTaskActive = false
+
+    function directTask(task) {
+        if (!task.sab) return task
+        return {
+            ...task,
+            sab: false,
+            recycleXml: false,
+            args: [
+                typeof task.xmlString === 'string'
+                    ? utf8BytesFromJsString(task.xmlString)
+                    : task.args[0],
+                ...task.args.slice(1),
+            ],
+        }
+    }
+
     pool = {
         run(task, opts) {
+            // A single request does not benefit from crossing a worker boundary.
+            // Keep the first in-flight request local, but retain the pool for
+            // concurrent work so multi-core throughput remains available.
+            if (!directTaskActive) {
+                directTaskActive = true
+                return workerFn(directTask(task))
+                    .then(parseCamaroJson)
+                    .finally(() => {
+                        directTaskActive = false
+                    })
+            }
             return leanPool.run(task, opts)
         },
         destroy() {

lean-pool.js

@@ -2,6 +2,7 @@
 
 const os = require('os')
 const { Worker } = require('worker_threads')
+const { parseCamaroJson } = require('./json-parse')
 const {
     sabEnabled,
     createSabChannel,
@@ -10,6 +11,7 @@ const {
     readResultFromSab,
     stripXmlFromTask,
     CTRL,
+    STATE,
     fnCode,
 } = require('./sab-ipc')
 
@@ -42,6 +44,10 @@ class LeanWorker {
         this.idleTimeout = idleTimeout
         this.onRecycleXml = onRecycleXml
         this.useSab = useSab
+        this.useSabWait =
+            useSab &&
+            typeof Atomics.waitAsync === 'function' &&
+            typeof Atomics.notify === 'function'
         this.sabChannel = useSab ? createSabChannel() : null
         this.worker = null
         this.seq = 0
@@ -59,7 +65,7 @@ class LeanWorker {
             this.sabChannel != null ? { workerData: { sab: this.sabChannel } } : undefined
         const worker = new Worker(this.filename, workerOpts)
         worker.unref()
-        worker.on('message', ({ id, result, error, xmlBuf, sab }) => {
+        worker.on('message', ({ id, result, error, xmlBuf, sab, raw }) => {
             if (xmlBuf && this.onRecycleXml) this.onRecycleXml(xmlBuf)
             const p = this.pending.get(id)
             this.pending.delete(id)
@@ -72,6 +78,12 @@ class LeanWorker {
                 } catch (err) {
                     p.reject(err)
                 }
+            } else if (raw) {
+                try {
+                    p.resolve(parseCamaroJson(result))
+                } catch (err) {
+                    p.reject(err)
+                }
             } else {
                 p.resolve(result)
             }
@@ -105,6 +117,35 @@ class LeanWorker {
         this.idleTimer.unref()
     }
 
+    settleSabResult(id) {
+        if (
+            !this.sabChannel ||
+            Atomics.load(this.sabChannel.control, CTRL.STATE) !== STATE.DONE
+        ) {
+            return
+        }
+        const p = this.pending.get(id)
+        this.pending.delete(id)
+        if (!p) return
+        try {
+            p.resolve(readResultFromSab(this.sabChannel))
+        } catch (err) {
+            p.reject(err)
+        }
+        if (this.pending.size === 0) this.scheduleIdleShutdown()
+    }
+
+    waitForSabResult(id) {
+        const channel = this.sabChannel
+        if (!channel) return
+        const waiter = Atomics.waitAsync(channel.control, CTRL.STATE, STATE.BUSY)
+        if (waiter.async) {
+            waiter.value.then(() => this.settleSabResult(id))
+        } else {
+            this.settleSabResult(id)
+        }
+    }
+
     run(task, opts = {}) {
         this.ensureWorker()
         this.clearIdleTimer()
@@ -124,11 +165,16 @@ class LeanWorker {
             }
             return new Promise((resolve, reject) => {
                 this.pending.set(id, { resolve, reject })
+                if (this.useSabWait) {
+                    Atomics.store(this.sabChannel.control, CTRL.STATE, STATE.BUSY)
+                }
                 this.worker.postMessage({
                     id,
                     sab: true,
+                    waitForSab: this.useSabWait,
                     task: stripXmlFromTask(task),
                 })
+                if (this.useSabWait) this.waitForSabResult(id)
             })
         }
 

pool-worker.js

@@ -7,7 +7,7 @@ const { CTRL, STATE, writeResultToSab } = require('./sab-ipc')
 
 const sabChannel = workerData && workerData.sab ? workerData.sab : null
 
-parentPort.on('message', async ({ id, task, sab: useSab }) => {
+parentPort.on('message', async ({ id, task, sab: useSab, waitForSab }) => {
     try {
         if (useSab && sabChannel) {
             const xmlLen = Atomics.load(sabChannel.control, CTRL.XML_LEN)
@@ -18,9 +18,17 @@ parentPort.on('message', async ({ id, task, sab: useSab }) => {
 
             Atomics.store(sabChannel.control, CTRL.STATE, STATE.BUSY)
             const raw = await workerFn(localTask)
-            writeResultToSab(sabChannel, raw)
+            if (waitForSab) {
+                writeResultToSab(sabChannel, raw)
+                Atomics.store(sabChannel.control, CTRL.STATE, STATE.DONE)
+                Atomics.notify(sabChannel.control, CTRL.STATE, 1)
+                return
+            }
             Atomics.store(sabChannel.control, CTRL.STATE, STATE.DONE)
-            parentPort.postMessage({ id, sab: true })
+            // JSON strings are cheaply cloned by V8. Avoiding the explicit
+            // UTF-8 encode into SAB and decode on the parent removes two
+            // copies from the latency-sensitive single-request path.
+            parentPort.postMessage({ id, raw: true, result: raw })
             return
         }
 
@@ -43,6 +51,7 @@ parentPort.on('message', async ({ id, task, sab: useSab }) => {
     } catch (err) {
         if (useSab && sabChannel) {
             Atomics.store(sabChannel.control, CTRL.STATE, STATE.ERROR)
+            Atomics.notify(sabChannel.control, CTRL.STATE, 1)
         }
         parentPort.postMessage({
             id,

test/worker-result-ipc.test.js

@@ -0,0 +1,18 @@
+'use strict'
+
+const t = require('tape')
+const { transform } = require('../')
+
+t.test('worker result IPC preserves JSON and NaN values', async (t) => {
+    const [, result] = await Promise.all([
+        transform('<warmup/>', { value: 'warmup' }),
+        transform('<root><item>one</item><item>two</item></root>', {
+            items: ['root/item', '.'],
+            missing: 'number(root/@missing)',
+        }),
+    ])
+
+    t.deepEqual(result.items, ['one', 'two'], 'parses the worker JSON result')
+    t.ok(Number.isNaN(result.missing), 'restores NaN from the worker result')
+    t.end()
+})