feat(load-emscripten-module-node): support pthread modules

Author: thewtex

packages/core/typescript/itk-wasm/src/pipeline/internal/load-emscripten-module-node.ts

@@ -2,6 +2,7 @@ import fs from 'fs'
 import EmscriptenModule from '../itk-wasm-emscripten-module.js'
 import { pathToFileURL } from 'url'
 import { ZSTDDecoder } from '@thewtex/zstddec'
+import pthreadSupportAvailable from '../pthread-support-available.js'
 
 const zstdDecoder = new ZSTDDecoder()
 await zstdDecoder.init()
@@ -19,9 +20,49 @@ async function loadEmscriptenModuleNode (
   if (modulePath.endsWith('.wasm.zst')) {
     modulePrefix = modulePath.substring(0, modulePath.length - 9)
   }
-  const compressedWasmBinaryPath = `${modulePrefix}.wasm.zst`
-  const compressedWasmBinary = fs.readFileSync(compressedWasmBinaryPath)
-  const wasmBinary = zstdDecoder.decode(new Uint8Array(compressedWasmBinary))
+
+  // Check for pthread support and use the appropriate WASM file
+  const hasPthreadSupport = pthreadSupportAvailable()
+  let wasmFileName = `${modulePrefix}.wasm.zst`
+  let wasmBinary: Uint8Array
+
+  if (hasPthreadSupport) {
+    const threadsWasmPath = `${modulePrefix}.threads.wasm.zst`
+    if (fs.existsSync(threadsWasmPath)) {
+      wasmFileName = threadsWasmPath
+      const compressedWasmBinary = fs.readFileSync(wasmFileName)
+      wasmBinary = zstdDecoder.decode(new Uint8Array(compressedWasmBinary))
+    } else {
+      // Fall back to checking for compressed non-threaded version
+      if (fs.existsSync(wasmFileName)) {
+        const compressedWasmBinary = fs.readFileSync(wasmFileName)
+        wasmBinary = zstdDecoder.decode(new Uint8Array(compressedWasmBinary))
+      } else {
+        // Fall back to uncompressed WASM file
+        const uncompressedWasmPath = `${modulePrefix}.wasm`
+        if (fs.existsSync(uncompressedWasmPath)) {
+          wasmBinary = fs.readFileSync(uncompressedWasmPath)
+        } else {
+          throw new Error(`No WASM file found for module: ${modulePrefix}`)
+        }
+      }
+    }
+  } else {
+    // Check for compressed non-threaded version first
+    if (fs.existsSync(wasmFileName)) {
+      const compressedWasmBinary = fs.readFileSync(wasmFileName)
+      wasmBinary = zstdDecoder.decode(new Uint8Array(compressedWasmBinary))
+    } else {
+      // Fall back to uncompressed WASM file
+      const uncompressedWasmPath = `${modulePrefix}.wasm`
+      if (fs.existsSync(uncompressedWasmPath)) {
+        wasmBinary = fs.readFileSync(uncompressedWasmPath)
+      } else {
+        throw new Error(`No WASM file found for module: ${modulePrefix}`)
+      }
+    }
+  }
+
   const fullModulePath = pathToFileURL(`${modulePrefix}.js`).href
   const result = await import(
     /* webpackIgnore: true */ /* @vite-ignore */ fullModulePath

packages/core/typescript/itk-wasm/test/node/pipeline/run-pipeline-node-test.js

@@ -477,7 +477,7 @@ test('runPipelineNode writes and reads an itk.TransformList via memory io', asyn
   verifyTransform(outputs[0].data)
 })
 
-test.only('runPipelineNode writes and reads a CompositeTransform itk.TransformList via memory io', async (t) => {
+test('runPipelineNode writes and reads a CompositeTransform itk.TransformList via memory io', async (t) => {
   const verifyTransform = (transformList) => {
     t.is(
       transformList.length,
@@ -643,6 +643,39 @@ test.only('runPipelineNode writes and reads a CompositeTransform itk.TransformLi
   verifyTransform(outputs[0].data)
 })
 
+test('runPipelineNode runs pthreads-enabled pipeline', async (t) => {
+  const pipelinePath = path.resolve(
+    'test',
+    'pipelines',
+    'emscripten-threads-build',
+    'pthreads-pipeline',
+    'pthreads-test'
+  )
+  const args = ['--memory-io', '--number-of-threads', '4', '0']
+  const desiredOutputs = [{ type: InterfaceTypes.JsonCompatible }]
+  const inputs = []
+
+  const { outputs } = await runPipelineNode(
+    pipelinePath,
+    args,
+    desiredOutputs,
+    inputs
+  )
+
+  const result = outputs[0].data
+  t.is(typeof result, 'object', 'output should be an object')
+  t.is(
+    typeof result.createdThreads,
+    'number',
+    'createdThreads should be a number'
+  )
+  t.true(result.createdThreads >= 0, 'should have created at least 0 threads')
+  t.true(
+    result.createdThreads <= 4,
+    'should not have created more than 4 threads'
+  )
+})
+
 test('runPipelineNode creates a composite transform with expected parameters', async (t) => {
   const verifyCompositeTransform = (transformList) => {
     t.is(

packages/core/typescript/itk-wasm/test/pipelines/CMakeLists.txt

@@ -10,7 +10,5 @@ add_subdirectory("mesh-read-write-pipeline")
 add_subdirectory("transform-read-write-pipeline")
 add_subdirectory("read-image")
 add_subdirectory("stdout-stderr-pipeline")
-if (NOT EMSCRIPTEN)
-  add_subdirectory("pthreads-pipeline")
-  add_subdirectory("cxx-threads-pipeline")
-endif()
+add_subdirectory("pthreads-pipeline")
+add_subdirectory("cxx-threads-pipeline")