wip: completing tests

Author: tegefaulkes

src/WorkerManager.ts

@@ -1,7 +1,7 @@
 import type { WorkerFactory } from './types.js';
 import Logger from '@matrixai/logger';
 import { CreateDestroy, ready } from '@matrixai/async-init/CreateDestroy.js';
-import { WorkerTaskInformation } from './types.js';
+import { type WorkerTaskInput } from './types.js';
 import * as errors from './errors.js';
 import WorkerPool from './WorkerPool.js';
 
@@ -63,12 +63,12 @@ class WorkerManager {
   }
 
   @ready(new errors.ErrorWorkerManagerDestroyed())
-  public async call(task: WorkerTaskInformation): Promise<unknown> {
+  public async call(task: WorkerTaskInput): Promise<unknown> {
     return await this.queue(task);
   }
 
   @ready(new errors.ErrorWorkerManagerDestroyed())
-  public queue(task: WorkerTaskInformation): Promise<unknown> {
+  public queue(task: WorkerTaskInput): Promise<unknown> {
     return new Promise((resolve, reject) => {
       this.pool.runTask(task, (result, error) => {
         if (error != null) return reject(error);

src/WorkerPool.ts

@@ -19,7 +19,7 @@ class WorkerTask extends AsyncResource {
     this.callback = callback;
   }
 
-  public done(result, error: Error) {
+  public done(result: unknown, error: Error) {
     this.runInAsyncScope(this.callback, null, result, error);
     this.emitDestroy();
   }
@@ -96,7 +96,7 @@ class WorkerPool {
       this.freeWorkers.push(worker);
       this.$workerFreed.next();
     };
-    const errorHandler = (e) => {
+    const errorHandler = (e: Error) => {
       workerError = e;
       if (worker[taskInfoSymbol]) worker[taskInfoSymbol].done(undefined, e);
       else this.$workerError.next(e);

src/expose.ts

@@ -35,15 +35,9 @@ function expose(workerManifest: WorkerManifest) {
     }
     parentPort!.postMessage({ data: result.data }, result.transferList);
   };
-  const handleMessageError = (e) => {
-    console.error('Message error!', e);
-    // Do nothing for now
-  };
   parentPort!.on('message', handleMessage);
-  parentPort!.on('messageerror', handleMessageError);
   parentPort!.once('close', () => {
     parentPort!.off('message', handleMessage);
-    parentPort!.off('messageerror', handleMessageError);
   });
 }
 

src/worker.ts

@@ -28,6 +28,11 @@ const worker = {
     });
     return { data: undefined };
   },
+  transferBuffer: async (data: ArrayBuffer) => {
+    const buffer = Buffer.from(data, 0, data.byteLength);
+    buffer.fill(0xf);
+    return { data, transferList: [data] };
+  },
 } satisfies WorkerManifest;
 
 expose(worker);

tests/WorkerManager.test.ts

@@ -105,68 +105,32 @@ describe('WorkerManager', () => {
     await workerManager.settled();
     await workerManager.destroy();
   });
-  // Test('zero-copy buffer transfer', async () => {
-  //   workerManager = await WorkerManager.createWorkerManager(
-  //     {
-  //       workerFactory,
-  //       cores: 1,
-  //       logger,
-  //     },
-  //   );
-  //   const buffer = await workerManager.call(async (w) => {
-  //     // Start with a Node Buffer that is "pooled"
-  //     const inputBuffer = Buffer.from('hello 1');
-  //     // Slice copy out the ArrayBuffer
-  //     const input = inputBuffer.buffer.slice(
-  //       inputBuffer.byteOffset,
-  //       inputBuffer.byteOffset + inputBuffer.byteLength,
-  //     );
-  //     // When the underlying ArrayBuffer is detached
-  //     // this Buffer's byteLength will also become 0
-  //     const inputBuffer_ = Buffer.from(input);
-  //     expect(inputBuffer_.byteLength).toBe(input.byteLength);
-  //     // Zero-copy transfer moves "ownership"
-  //     // input is detached from main thread
-  //     // output is detached from worker thread
-  //     const output = await w.transferBuffer(Transfer(input));
-  //     // Detached ArrayBuffers have byte lengths of 0
-  //     expect(input.byteLength).toBe(0);
-  //     expect(inputBuffer_.byteLength).toBe(0);
-  //     // Zero-copy wrap to use Node Buffer API
-  //     const outputBuffer = Buffer.from(output);
-  //     return outputBuffer;
-  //   });
-  //   expect(buffer).toEqual(Buffer.from('hello 2'));
-  //   await workerManager.destroy();
-  // });
-  //   test('scratch', async () => {
-  //     console.log('start');
-  //     console.log(
-  //       'Hello from main',
-  //       nodeWorkers.isMainThread,
-  //       nodeWorkers.threadId,
-  //     );
-  //
-  //     const script = `
-  // const nodeWorkers = require("node:worker_threads");
-  // console.log("Hello from worker!: ", nodeWorkers.isMainThread, nodeWorkers.threadId);
-  // nodeWorkers.parentPort.on('message', v => {
-  //   console.log('message', v);
-  //   nodeWorkers.parentPort.postMessage(v);
-  // });
-  //     `;
-  //     await new Promise((resolve, reject) => {
-  //       const worker = new nodeWorkers.Worker(script, { eval: true });
-  //       worker.on('message', (v) => {
-  //         logger.warn(`message: ${v}`);
-  //         worker.terminate();
-  //       });
-  //       worker.on('error', reject);
-  //       worker.on('exit', resolve);
-  //       worker.postMessage('some message!');
-  //     });
-  //     console.log('stop');
-  //   });
-});
+  test('zero-copy buffer transfer', async () => {
+    workerManager = await WorkerManager.createWorkerManager({
+      workerFactory,
+      cores: 1,
+      logger,
+    });
+    // Creating a new buffer
+    const inputBuffer = Buffer.from('hello 1');
+    // Extracting the underlying ArrayBuffer
+    const input = inputBuffer.buffer.slice(
+      inputBuffer.byteOffset,
+      inputBuffer.byteOffset + inputBuffer.byteLength,
+    );
+    // Making call with transfer
+    const output = await workerManager.call({
+      type: 'transferBuffer',
+      data: input,
+      transferList: [input],
+    });
+    // The input ArrayBuffer is detached so the length is now 0
+    expect(input.byteLength).toBe(0);
+    // The output should be filled with 0xF
+    expect(Buffer.from(output as ArrayBuffer)).toEqual(
+      Buffer.alloc(inputBuffer.byteLength, 0xf),
+    );
 
-test('asd', async () => {});
+    await workerManager.destroy();
+  });
+});

tests/WorkerPool.test.ts

@@ -7,9 +7,6 @@ import * as errors from '#errors.js';
 
 const dirname = url.fileURLToPath(new URL('.', import.meta.url));
 
-// TODO: zero copy of data
-// TODO: apply the async resource.
-
 describe('WorkerPool', () => {
   let pool: WorkerPool;
   const workerFactory: WorkerFactory = () => {