Deep Dive: How Node.js 26 Worker Threads Work with WebAssembly 2.0 and Rust 1.95

// Cargo.toml for Rust 1.95 Wasm 2.0 module
// [package]
// name = "wasm-hasher"
// version = "0.1.0"
// edition = "2021"
// 
// [lib]
// crate-type = ["cdylib"]
// 
// [dependencies]
// wasm-bindgen = "0.2.95"
// sha2 = "0.10.8"
// wasm-threads = "0.1.0" // Wasm 2.0 thread support
// 
// [target.wasm32-wasip2]
// runner = []

// lib.rs: Wasm 2.0 module compiled from Rust 1.95
// Exports a function to batch-compute SHA-256 hashes from a SharedArrayBuffer
// Uses Wasm 2.0 shared memory to avoid serialization overhead

use wasm_bindgen::prelude::*;
use sha2::{Sha256, Digest};
use wasm_threads::shared_memory::SharedSlice;
use std::sync::atomic::{AtomicU32, Ordering};

// Wasm 2.0 shared memory region: 1MB pre-allocated, accessible from Node.js Worker Threads
// Declared with #[wasm_bindgen] to expose to JS, uses shared(1mb) attribute for Wasm 2.0
#[wasm_bindgen]
pub struct Hasher {
    // Atomic counter to track processed items across threads
    processed: AtomicU32,
    // Shared buffer handle passed from Node.js via SharedArrayBuffer
    input_buffer: SharedSlice,
    // Output buffer for hash results (32 bytes per hash)
    output_buffer: SharedSlice,
}

#[wasm_bindgen]
impl Hasher {
    // Constructor: initializes hasher with shared input/output buffers
    // Parameters:
    // - input_ptr: Pointer to SharedArrayBuffer base (from Node.js)
    // - input_len: Length of input buffer in bytes
    // - output_ptr: Pointer to output SharedArrayBuffer base
    // - output_len: Length of output buffer in bytes
    // Returns: Result for error handling
    #[wasm_bindgen(constructor)]
    pub fn new(
        input_ptr: *mut u8,
        input_len: usize,
        output_ptr: *mut u8,
        output_len: usize,
    ) -> Result {
        // Validate input pointers are non-null
        if input_ptr.is_null() || output_ptr.is_null() {
            return Err(JsError::new("Input/output pointers cannot be null"));
        }
        // Validate output buffer can hold at least 32 bytes (1 SHA-256 hash)
        if output_len < 32 {
            return Err(JsError::new("Output buffer must be at least 32 bytes"));
        }

        // Wrap raw pointers in Wasm 2.0 shared slices
        // Safety: Pointers are validated by Node.js caller, memory is shared via SAB
        let input_buffer = unsafe {
            SharedSlice::from_raw_parts(input_ptr, input_len)
                .map_err(|e| JsError::new(&format!("Invalid input buffer: {}", e)))?
        };
        let output_buffer = unsafe {
            SharedSlice::from_raw_parts(output_ptr, output_len)
                .map_err(|e| JsError::new(&format!("Invalid output buffer: {}", e)))?
        };

        Ok(Hasher {
            processed: AtomicU32::new(0),
            input_buffer,
            output_buffer,
        })
    }

    // Process next batch of data: reads 64-byte chunks from input, writes 32-byte hashes to output
    // Returns: number of hashes processed, or error if buffers are full
    #[wasm_bindgen]
    pub fn process_batch(&self) -> Result {
        let mut hash_count = 0;
        let input = &self.input_buffer;
        let output = &self.output_buffer;
        let total_input = input.len();
        let mut offset = self.processed.load(Ordering::Relaxed) as usize * 64;

        // Loop until input is exhausted or output is full
        while offset + 64 <= total_input && (hash_count * 32) + 32 <= output.len() {
            // Read 64-byte chunk (Sha256 block size)
            let chunk = &input[offset..offset + 64];
            // Compute SHA-256 hash
            let mut hasher = Sha256::new();
            hasher.update(chunk);
            let hash = hasher.finalize();
            // Write 32-byte hash to output buffer
            let out_offset = hash_count as usize * 32;
            output[out_offset..out_offset + 32].copy_from_slice(&hash);
            // Update counters
            hash_count += 1;
            offset += 64;
            self.processed.fetch_add(1, Ordering::SeqCst);
        }

        Ok(hash_count)
    }

    // Get total number of processed hashes
    #[wasm_bindgen]
    pub fn get_processed_count(&self) -> u32 {
        self.processed.load(Ordering::Relaxed)
    }
}

// Error handling: custom panic hook to return errors to Node.js instead of aborting
#[cfg(target_arch = "wasm32")]
#[no_mangle]
pub extern "C" fn __wasm_panic_hook() {
    // Log panic to Node.js console via Wasm 2.0 import (omitted for brevity, see full repo)
}

// Node.js 26 Worker Thread script (worker.mjs)
// Loads Wasm 2.0 module compiled from Rust 1.95, processes data from main thread
import { parentPort, workerData } from 'worker_threads';
import { readFileSync } from 'fs';
import { instantiateStreaming } from 'node:wasm'; // Node 26 native Wasm 2.0 API

// Error handling: wrap all worker logic in try/catch to report to main thread
try {
  // 1. Load pre-compiled Wasm module from workerData (avoids re-instantiation)
  const wasmBuffer = workerData.wasmBuffer;
  if (!wasmBuffer || !(wasmBuffer instanceof ArrayBuffer)) {
    throw new Error('Invalid Wasm buffer provided in workerData');
  }

  // 2. Instantiate Wasm 2.0 module with Node.js 26's native instantiateStreaming
  // Uses shared memory flag to enable Wasm 2.0 shared ArrayBuffer access
  const wasmModule = await instantiateStreaming(wasmBuffer, {
    env: {
      // Import Node.js console for Wasm logging (Wasm 2.0 import/export)
      log: (ptr, len) => {
        const str = Buffer.from(wasmModule.exports.memory.buffer, ptr, len).toString();
        console.log(`[Wasm Worker] ${str}`);
      }
    }
  }, { sharedMemory: true }); // Enable Wasm 2.0 shared memory

  // 3. Initialize SharedArrayBuffers passed from main thread
  const inputSab = workerData.inputSab;
  const outputSab = workerData.outputSab;
  if (!(inputSab instanceof SharedArrayBuffer) || !(outputSab instanceof SharedArrayBuffer)) {
    throw new Error('Input/output SharedArrayBuffer not provided');
  }

  // 4. Instantiate Rust Hasher struct from Wasm exports
  const { Hasher } = wasmModule.exports;
  const hasher = new Hasher(
    inputSab, // Pass SharedArrayBuffer directly (Wasm 2.0 shared memory)
    inputSab.byteLength,
    outputSab,
    outputSab.byteLength
  );

  // 5. Listen for messages from main thread to start processing
  parentPort.on('message', async (msg) => {
    if (msg.type === 'PROCESS') {
      try {
        const startTime = performance.now();
        const processed = hasher.process_batch();
        const duration = performance.now() - startTime;
        // Report results to main thread
        parentPort.postMessage({
          type: 'RESULT',
          processed,
          durationMs: duration,
          totalProcessed: hasher.get_processed_count()
        });
      } catch (err) {
        parentPort.postMessage({
          type: 'ERROR',
          error: err.message,
          stack: err.stack
        });
      }
    } else if (msg.type === 'SHUTDOWN') {
      // Cleanup Wasm instance
      wasmModule.exports.memory.destroy?.(); // Node 26 Wasm memory cleanup
      parentPort.postMessage({ type: 'SHUTDOWN_ACK' });
      process.exit(0);
    }
  });

  // Signal to main thread that worker is ready
  parentPort.postMessage({ type: 'READY' });

} catch (err) {
  // Report initialization errors to main thread
  parentPort.postMessage({
    type: 'INIT_ERROR',
    error: err.message,
    stack: err.stack
  });
  process.exit(1);
}

// Node.js 26 Main Thread Script (main.mjs)
// Spawns Worker Threads, manages Wasm 2.0 modules, runs benchmarks
import { Worker } from 'worker_threads';
import { readFileSync } from 'fs';
import { performance } from 'node:perf_hooks';

// Configuration
const WORKER_COUNT = 4; // Match CPU core count (Node 26 recommends core count for WT)
const WASM_PATH = './target/wasm32-wasip2/release/wasm_hasher.wasm';
const INPUT_SIZE = 1024 * 1024 * 10; // 10MB input data
const BATCH_SIZE = 64; // Sha256 block size, matches Rust Wasm module

// Benchmark results storage
const benchmarkResults = {
  mainThread: null,
  workerThread: null,
  speedup: null
};

// 1. Load pre-compiled Wasm module once, share with all workers
console.log('Loading Wasm module from', WASM_PATH);
const wasmBuffer = readFileSync(WASM_PATH);
const wasmSharedBuffer = wasmBuffer.buffer.slice(0); // Copy to ArrayBuffer for sharing

// 2. Generate test input data (random bytes)
console.log('Generating', INPUT_SIZE / 1024 / 1024, 'MB input data');
const inputData = new Uint8Array(INPUT_SIZE);
crypto.getRandomValues(inputData);

// 3. Create SharedArrayBuffers for input/output
const inputSab = new SharedArrayBuffer(INPUT_SIZE);
const outputSab = new SharedArrayBuffer((INPUT_SIZE / BATCH_SIZE) * 32); // 32 bytes per hash
// Copy input data to shared buffer
new Uint8Array(inputSab).set(inputData);

// 4. Spawn Worker Threads
const workers = [];
const workerPromises = [];

for (let i = 0; i < WORKER_COUNT; i++) {
  const worker = new Worker('./worker.mjs', {
    workerData: {
      wasmBuffer: wasmSharedBuffer,
      inputSab,
      outputSab,
      workerId: i
    }
  });

  // Track worker ready state
  const readyPromise = new Promise((resolve, reject) => {
    worker.on('message', (msg) => {
      if (msg.type === 'READY') {
        console.log(`Worker ${i} ready`);
        resolve(worker);
      } else if (msg.type === 'INIT_ERROR') {
        reject(new Error(`Worker ${i} init failed: ${msg.error}`));
      }
    });
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0) reject(new Error(`Worker ${i} exited with code ${code}`));
    });
  });

  workers.push(worker);
  workerPromises.push(readyPromise);
}

// Wait for all workers to initialize
console.log('Waiting for workers to initialize...');
await Promise.all(workerPromises);
console.log('All workers ready. Starting benchmark...');

// 5. Run main thread only benchmark (no workers, Wasm in main thread)
console.log('\n--- Main Thread Benchmark (No Workers) ---');
try {
  const mainStart = performance.now();
  // Instantiate Wasm in main thread (legacy approach)
  const { instantiateStreaming } = await import('node:wasm');
  const mainWasm = await instantiateStreaming(readFileSync(WASM_PATH), {}, { sharedMemory: true });
  const mainHasher = new mainWasm.exports.Hasher(
    inputSab,
    inputSab.byteLength,
    outputSab,
    outputSab.byteLength
  );
  const mainProcessed = mainHasher.process_batch();
  const mainDuration = performance.now() - mainStart;
  benchmarkResults.mainThread = {
    processed: mainProcessed,
    durationMs: mainDuration,
    throughput: (INPUT_SIZE / 1024 / 1024) / (mainDuration / 1000)
  };
  console.log(`Main thread processed ${mainProcessed} hashes in ${mainDuration.toFixed(2)}ms`);
  console.log(`Main thread throughput: ${benchmarkResults.mainThread.throughput.toFixed(2)} MB/s`);
} catch (err) {
  console.error('Main thread benchmark failed:', err.message);
}

// 6. Run Worker Thread benchmark
console.log('\n--- Worker Thread Benchmark (4 Workers) ---');
const workerStart = performance.now();
const workerPromises = workers.map((worker, i) => {
  return new Promise((resolve) => {
    worker.on('message', (msg) => {
      if (msg.type === 'RESULT') {
        console.log(`Worker ${i} processed ${msg.processed} hashes in ${msg.durationMs.toFixed(2)}ms`);
        resolve(msg);
      }
    });
    // Send process command to worker
    worker.postMessage({ type: 'PROCESS' });
  });
});

const workerResults = await Promise.all(workerPromises);
const workerDuration = performance.now() - workerStart;
const totalProcessed = workerResults.reduce((sum, r) => sum + r.processed, 0);
benchmarkResults.workerThread = {
  processed: totalProcessed,
  durationMs: workerDuration,
  throughput: (INPUT_SIZE / 1024 / 1024) / (workerDuration / 1000)
};
console.log(`Total worker processed ${totalProcessed} hashes in ${workerDuration.toFixed(2)}ms`);
console.log(`Worker throughput: ${benchmarkResults.workerThread.throughput.toFixed(2)} MB/s`);

// 7. Calculate speedup
if (benchmarkResults.mainThread && benchmarkResults.workerThread) {
  benchmarkResults.speedup = benchmarkResults.mainThread.durationMs / benchmarkResults.workerThread.durationMs;
  console.log(`\nSpeedup: ${benchmarkResults.speedup.toFixed(2)}x`);
}

// 8. Cleanup workers
workers.forEach(worker => worker.postMessage({ type: 'SHUTDOWN' }));
process.exit(0);

// Main thread: pre-compile Wasm once
const wasmBuffer = readFileSync('./wasm-hasher.wasm');
// Pass to workers via workerData
const worker = new Worker('./worker.mjs', {
  workerData: { wasmBuffer: wasmBuffer.buffer.slice(0) }
});

// Main thread: create SharedArrayBuffer
const sab = new SharedArrayBuffer(1024 * 1024 * 10);
// Worker thread: use Atomics to synchronize
const view = new Int32Array(sab);
Atomics.add(view, 0, 1); // Increment counter
Atomics.wait(view, 0, 0); // Wait until counter changes

// Worker thread: pin to core matching worker ID
import { setAffinity } from 'node-affinity';
const { workerId } = workerData;
setAffinity(workerId); // Pin to CPU core ${workerId}