fix: coalesce tsfn send message (#210)

toyobayashi · toyobayashi · commit b02054aef5e6 · 2026-04-15T18:37:09.000+08:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -68,7 +68,7 @@ jobs:
 
   wasm32-wasip1-threads:
     timeout-minutes: 15
-    name: CI wasm32-wasip1-threads
+    name: CI wasm32-wasip1-threads (libemnapi-mt)
     runs-on: ubuntu-latest
 
     steps:
@@ -93,6 +93,28 @@ jobs:
       run: EMNAPI_TEST_4GB=1 npm run test:wt -w packages/test
       timeout-minutes: 3
 
+  wasm32-wasip1-threads-use-js:
+    timeout-minutes: 15
+    name: CI wasm32-wasip1-threads (libemnapi-basic-mt)
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v6
+    - uses: ./.github/actions/setup-target-ci
+      with:
+        target: 'wasm32-wasip1-threads-js'
+        needs-wasi-sdk: 'true'
+    - name: NPM Build
+      run: npm run build --workspaces --if-present
+    - name: Build target
+      run: EMNAPI_WASI_THREADS_LINK_BASIC_MT=1 npm run rebuild:wt -w packages/test
+    - name: Test target
+      run: npm run test:wt -w packages/test
+      timeout-minutes: 3
+    - name: Test target (4GB)
+      run: EMNAPI_TEST_4GB=1 npm run test:wt -w packages/test
+      timeout-minutes: 3
+
   wasm32-wasip1:
     timeout-minutes: 15
     name: CI wasm32-wasip1
diff --git a/packages/emnapi/script/generate_struct_info.c b/packages/emnapi/script/generate_struct_info.c
@@ -41,6 +41,8 @@ void print_napi_threadsafe_function_info(FILE* f) {
   fprintf(f, "  queue_size = %zu,\n", offsetof(struct napi_threadsafe_function__, queue_size));
   fprintf(f, "  queue = %zu,\n", offsetof(struct napi_threadsafe_function__, queue));
   fprintf(f, "  async = %zu,\n", offsetof(struct napi_threadsafe_function__, async));
+  fprintf(f, "  async_pending = %zu,\n", offsetof(struct napi_threadsafe_function__, async) + offsetof(uv_async_t, pending));
+  fprintf(f, "  async_u_fd = %zu,\n", offsetof(struct napi_threadsafe_function__, async) + offsetof(uv_async_t, u.fd));
   fprintf(f, "  thread_count = %zu,\n", offsetof(struct napi_threadsafe_function__, thread_count));
   fprintf(f, "  state = %zu,\n", offsetof(struct napi_threadsafe_function__, state));
   fprintf(f, "  dispatch_state = %zu,\n", offsetof(struct napi_threadsafe_function__, dispatch_state));
diff --git a/packages/emnapi/src/threadsafe-function.ts b/packages/emnapi/src/threadsafe-function.ts
@@ -30,6 +30,9 @@ export const emnapiTSFN = {
     /* size_t */ queue_size: 0,
     /* bool */ is_some: 0,
     /* void* */ queue: 0,
+    // Reuse uv_async_t storage as JS-side wakeup state: pending event + scheduled drain.
+    async_pending: 0,
+    async_u_fd: 0,
     /* size_t */ thread_count: 0,
     /* int32_t */ state: 0,
     /* atomic_uchar */ dispatch_state: 0,
@@ -54,6 +57,8 @@ export const emnapiTSFN = {
     emnapiTSFN.offset.queue_size = NapiTSFNOffset64.queue_size
     emnapiTSFN.offset.is_some = NapiTSFNOffset64.async_resource_is_some
     emnapiTSFN.offset.queue = NapiTSFNOffset64.queue
+    emnapiTSFN.offset.async_pending = NapiTSFNOffset64.async_pending
+    emnapiTSFN.offset.async_u_fd = NapiTSFNOffset64.async_u_fd
     emnapiTSFN.offset.thread_count = NapiTSFNOffset64.thread_count
     emnapiTSFN.offset.state = NapiTSFNOffset64.state
     emnapiTSFN.offset.dispatch_state = NapiTSFNOffset64.dispatch_state
@@ -76,6 +81,8 @@ export const emnapiTSFN = {
     emnapiTSFN.offset.queue_size = NapiTSFNOffset32.queue_size
     emnapiTSFN.offset.is_some = NapiTSFNOffset32.async_resource_is_some
     emnapiTSFN.offset.queue = NapiTSFNOffset32.queue
+    emnapiTSFN.offset.async_pending = NapiTSFNOffset32.async_pending
+    emnapiTSFN.offset.async_u_fd = NapiTSFNOffset32.async_u_fd
     emnapiTSFN.offset.thread_count = NapiTSFNOffset32.thread_count
     emnapiTSFN.offset.state = NapiTSFNOffset32.state
     emnapiTSFN.offset.dispatch_state = NapiTSFNOffset32.dispatch_state
@@ -113,7 +120,10 @@ export const emnapiTSFN = {
         const type = __emnapi__.type
         const payload = __emnapi__.payload
         if (type === 'tsfn-send') {
-          emnapiTSFN.dispatch(payload.tsfn)
+          const pendng = payload.tsfn + emnapiTSFN.offset.async_pending
+          if (Atomics.load(new Int32Array(wasmMemory.buffer), pendng >>> 2) !== 0) {
+            emnapiTSFN.enqueue(payload.tsfn)
+          }
         }
       }
     }
@@ -708,24 +718,77 @@ export const emnapiTSFN = {
       emnapiTSFN.send(func)
     }
   },
+  enqueue (func: number): void {
+    // `pending` means a worker thread has requested a wakeup that has not
+    // been drained on the main thread yet.
+    const pending = func + emnapiTSFN.offset.async_pending
+    // `scheduled` prevents queueing the same main-thread drain chain more than
+    // once while a previous wakeup is still in flight.
+    const scheduled = func + emnapiTSFN.offset.async_u_fd
+    const i32a = new Int32Array(wasmMemory.buffer)
+    if (Atomics.exchange(i32a, scheduled >>> 2, 1) !== 0) {
+      return
+    }
+
+    // Match uv_async_send-style coalescing in JS: the first turn represents
+    // the wakeup reaching the main thread, and the second turn performs the
+    // actual TSFN drain after nearby Send/Signal calls have had a chance to
+    // collapse into the shared AsyncProgressWorker state.
+    emnapiCtx.feature.setImmediate(() => {
+      if (Atomics.load(i32a, pending >>> 2) === 0) {
+        Atomics.store(i32a, scheduled >>> 2, 0)
+        return
+      }
+
+      emnapiCtx.feature.setImmediate(() => {
+        try {
+          // Consume the coalesced wakeup once, then let dispatch() observe any
+          // queue mutations through dispatch_state like the C implementation.
+          if (Atomics.exchange(i32a, pending >>> 2, 0) === 0) {
+            return
+          }
+          emnapiTSFN.dispatch(func)
+        } finally {
+          // Allow a later wakeup to schedule a new drain chain. If another
+          // worker-thread send raced with this drain, enqueue one more turn.
+          Atomics.store(i32a, scheduled >>> 2, 0)
+          if (Atomics.load(i32a, pending >>> 2) !== 0) {
+            emnapiTSFN.enqueue(func)
+          }
+        }
+      })
+    })
+  },
   send (func: number): void {
     const current_state = Atomics.or(new Uint32Array(wasmMemory.buffer), (func + emnapiTSFN.offset.dispatch_state) >>> 2, 1 << 1)
     if ((current_state & 1) === 1) {
       return
     }
-    if ((typeof ENVIRONMENT_IS_PTHREAD !== 'undefined') && ENVIRONMENT_IS_PTHREAD) {
-      postMessage({
-        __emnapi__: {
-          type: 'tsfn-send',
-          payload: {
-            tsfn: func
+
+    const pendng = func + emnapiTSFN.offset.async_pending
+    // A wakeup is already pending, so this send only needs to leave the queued
+    // work in the TSFN queue and let the existing drain pick it up.
+    if (Atomics.load(new Int32Array(wasmMemory.buffer), pendng >>> 2) !== 0) {
+      return
+    }
+
+    if (Atomics.exchange(new Int32Array(wasmMemory.buffer), pendng >>> 2, 1) === 0) {
+      if ((typeof ENVIRONMENT_IS_PTHREAD !== 'undefined') && ENVIRONMENT_IS_PTHREAD) {
+        // Worker threads only post a wakeup token. Main-thread draining is
+        // serialized by enqueue() once the message is received.
+        postMessage({
+          __emnapi__: {
+            type: 'tsfn-send',
+            payload: {
+              tsfn: func
+            }
           }
-        }
-      })
-    } else {
-      emnapiCtx.feature.setImmediate(() => {
-        emnapiTSFN.dispatch(func)
-      })
+        })
+      } else {
+        // On the main thread we can skip the cross-thread hop and schedule the
+        // coalesced drain chain directly.
+        emnapiTSFN.enqueue(func)
+      }
     }
   }
 }
diff --git a/packages/emnapi/src/typings/struct-wasm32.d.ts b/packages/emnapi/src/typings/struct-wasm32.d.ts
@@ -27,6 +27,8 @@ declare const enum NapiTSFNOffset32 {
   queue_size = 60,
   queue = 64,
   async = 72,
+  async_pending = 132,
+  async_u_fd = 96,
   thread_count = 136,
   state = 140,
   dispatch_state = 144,
diff --git a/packages/emnapi/src/typings/struct-wasm64.d.ts b/packages/emnapi/src/typings/struct-wasm64.d.ts
@@ -27,6 +27,8 @@ declare const enum NapiTSFNOffset64 {
   queue_size = 88,
   queue = 96,
   async = 112,
+  async_pending = 232,
+  async_u_fd = 160,
   thread_count = 240,
   state = 248,
   dispatch_state = 252,
diff --git a/packages/test/CMakeLists.txt b/packages/test/CMakeLists.txt
@@ -75,6 +75,12 @@ else()
   set(IS_MEMORY64 OFF)
 endif()
 
+if(DEFINED ENV{EMNAPI_WASI_THREADS_LINK_BASIC_MT})
+  set(_EMNAPI_WASI_THREADS_LINK_BASIC_MT $ENV{EMNAPI_WASI_THREADS_LINK_BASIC_MT})
+else()
+  set(_EMNAPI_WASI_THREADS_LINK_BASIC_MT OFF)
+endif()
+
 if(IS_EMSCRIPTEN)
   if(IS_MEMORY64)
     add_compile_options("-sMEMORY64=1")
@@ -169,7 +175,7 @@ function(add_test NAME SOURCE_LIST PTHREAD)
           target_link_libraries(${NAME} PRIVATE "emnapi-basic")
         endif()
       else()
-        if(IS_WASI_THREADS)
+        if(IS_WASI_THREADS AND _EMNAPI_WASI_THREADS_LINK_BASIC_MT)
           target_link_libraries(${NAME} PRIVATE "emnapi-basic-mt")
           target_link_options(${NAME} PRIVATE "-Wl,--import-memory,--shared-memory,--export=emnapi_async_worker_create,--export=emnapi_async_worker_init")
         else()
@@ -227,7 +233,7 @@ function(add_naa_test NAME SOURCE_LIST DEFINES ENABLE_EXCEPTION)
         target_link_libraries(${NAME} PRIVATE "emnapi-basic")
       endif()
     else()
-      if(IS_WASI_THREADS)
+      if(IS_WASI_THREADS AND _EMNAPI_WASI_THREADS_LINK_BASIC_MT)
         target_link_libraries(${NAME} PRIVATE "emnapi-basic-mt")
         target_link_options(${NAME} PRIVATE "-Wl,--import-memory,--shared-memory,--export=emnapi_async_worker_create,--export=emnapi_async_worker_init")
       else()
diff --git a/packages/test/script/test.js b/packages/test/script/test.js
@@ -1,5 +1,13 @@
 if (process.env.EMNAPI_TEST_WASI_THREADS) {
   process.env.EMNAPI_TEST_WASI = 1
+
+  const { spawnSync } = require('child_process')
+  const path = require('path')
+  const wasmPath = path.join(__dirname, '../.build/wasm32-wasi-threads/Debug/async.wasm')
+  const module = new WebAssembly.Module(require('fs').readFileSync(wasmPath))
+  console.log('libemnapi-basic-mt: ' + (WebAssembly.Module.imports(module).filter(d => {
+    return d.name === 'napi_create_async_work'
+  }).length > 0))
 }
 
 const { spawnSync } = require('child_process')
