wip: expanding WorkerPool tests

[ci skip]

Author: tegefaulkes

src/errors.ts

@@ -8,4 +8,17 @@ class ErrorWorkerManagerDestroyed<T> extends ErrorWorkerManager<T> {
   static description = 'WorkerManager is destroyed';
 }
 
-export { ErrorWorkerManager, ErrorWorkerManagerDestroyed };
+class ErrorWorker<T> extends AbstractError<T> {
+  static description = 'Worker error';
+}
+
+class ErrorWorkerHandlerMissing<T> extends ErrorWorkerManager<T> {
+  static description = 'Worker has not exposed a handler of this name';
+}
+
+export {
+  ErrorWorkerManager,
+  ErrorWorkerManagerDestroyed,
+  ErrorWorker,
+  ErrorWorkerHandlerMissing,
+};

src/expose.ts

@@ -1,5 +1,6 @@
 import type { WorkerTask, WorkerManifest } from '#types.js';
 import { isMainThread, parentPort, threadId } from 'node:worker_threads';
+import * as workerErrors from '#errors.js';
 
 /**
  * Call inside a worker to set it up and expose the supported worker functions.
@@ -10,10 +11,11 @@ function expose(workerManifest: WorkerManifest) {
   // We don't want to run this in the main thread since it's not acting as a worker
   if (isMainThread) return;
   const handleMessage = async (task: WorkerTask) => {
-    console.log('got task!', task);
     const method = workerManifest[task.type];
     if (method == null) {
-      parentPort!.postMessage({ error: Error('TMP IMP missing handler') });
+      parentPort!.postMessage({
+        error: new workerErrors.ErrorWorkerHandlerMissing(),
+      });
       return;
     }
     let result: unknown;
@@ -25,17 +27,15 @@ function expose(workerManifest: WorkerManifest) {
     }
     parentPort!.postMessage({ result });
   };
-  const handleMessageError = (e) => {
-    console.log(`Worker ${threadId} got error`, e);
+  const handleMessageError = (e_) => {
+    // Do nothing for now
   };
   parentPort!.on('message', handleMessage);
   parentPort!.on('messageerror', handleMessageError);
   parentPort!.once('close', () => {
-    console.log(`Worker ${threadId} closed!`);
     parentPort!.off('message', handleMessage);
     parentPort!.off('messageerror', handleMessageError);
   });
-  console.log('expose done');
 }
 
 export { expose };

src/types.ts

@@ -7,9 +7,7 @@ type WorkerTask<T extends string = string, D extends any = any> = {
   data: D;
 };
 
-type WorkerFunction<I extends any = any, O extends any = any> = (
-  data: I,
-) => Promise<O>;
+type WorkerFunction<I = any, O = any> = (data: I) => Promise<O>;
 type WorkerManifest = Record<string, WorkerFunction>;
 
 export type { WorkerFactory, WorkerTask, WorkerFunction, WorkerManifest };

src/worker.ts

@@ -4,23 +4,29 @@ import type { WorkerManifest } from '#types.js';
 import { expose } from './expose.js';
 
 const worker = {
-  test: async (data: void) => {
+  test: async () => {
     return 'hello world!';
   },
   add: async (data: { a: number; b: number }): Promise<number> => {
-    console.log(data);
     return data.a + data.b;
   },
   sub: async (data: { a: number; b: number }): Promise<number> => {
     return data.a - data.b;
   },
   fac: async (data: number): Promise<number> => {
     let acc = 1;
-    for (let i = 1; i < data; i++) {
+    for (let i = 1; i <= data; i++) {
       acc = acc * i;
     }
     return acc;
   },
+  sleep: async (data: number) => {
+    await new Promise<void>((resolve) => {
+      const timer = setTimeout(resolve, data);
+      // Prevent holding the process open
+      timer.unref();
+    });
+  },
 } satisfies WorkerManifest;
 
 expose(worker);

tests/WorkerPool.test.ts

@@ -3,6 +3,7 @@ import { Worker } from 'node:worker_threads';
 // Import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
 import path from 'node:path';
 import WorkerPool from '#WorkerPool.js';
+import * as errors from '#errors.js';
 
 async function sleep(ms: number) {
   return new Promise((resolve) => {
@@ -22,32 +23,64 @@ describe('WorkerPool', () => {
   };
 
   afterEach(async () => {
-    await pool.terminate(true);
-    console.log('DONE!');
+    await pool?.terminate(true);
   });
 
-  test('asd', async () => {
+  test('worker pool must have atleast 1 worker', async () => {
+    expect(() => new WorkerPool(0, workerFactory)).toThrow(
+      errors.ErrorWorkerHandlerMissing,
+    );
+    expect(() => new WorkerPool(-1, workerFactory)).toThrow(
+      errors.ErrorWorkerHandlerMissing,
+    );
+    pool = new WorkerPool(1, workerFactory);
+    await pool.terminate(true);
+  });
+  test('can run tasks', async () => {
     pool = new WorkerPool(3, workerFactory);
-    pool.$poolStatus.subscribe((v) => {
-      console.log('!queue state!', v);
-    });
-    let resolveP;
-    const taskP = new Promise((resolve) => {
-      resolveP = resolve;
-    });
-    let doneCount = 0;
     const numTasks = 20;
     for (let i = 0; i < numTasks; i++) {
-      pool.runTask({ type: 'fac', data: 5 }, (result, error) => {
-        console.log({ result, error });
-        doneCount++;
-        if (doneCount === numTasks) resolveP(result);
+      pool.runTask({ type: 'test', data: undefined }, () => {
+        // Do nothing
       });
     }
-    console.log('waiting');
-    await taskP;
-    console.log('waiting done');
+    await pool.settled();
     await pool.terminate(false);
-    console.log('term done');
   });
+  test('tasks return expected results', async () => {
+    pool = new WorkerPool(1, workerFactory);
+    const task1 = new Promise((resolve, reject) => {
+      pool.runTask({ type: 'test', data: undefined }, (result, error) => {
+        if (error != null) return reject(error);
+        return resolve(result);
+      });
+    });
+    await expect(task1).resolves.toBe('hello world!');
+
+    const task2 = new Promise((resolve, reject) => {
+      pool.runTask({ type: 'add', data: { a: 2, b: 2 } }, (result, error) => {
+        if (error != null) return reject(error);
+        return resolve(result);
+      });
+    });
+    await expect(task2).resolves.toBe(4);
+
+    const task3 = new Promise((resolve, reject) => {
+      pool.runTask({ type: 'fac', data: 5 }, (result, error) => {
+        if (error != null) return reject(error);
+        return resolve(result);
+      });
+    });
+    await expect(task3).resolves.toBe(120);
+
+    const task4 = new Promise((resolve, reject) => {
+      pool.runTask({ type: 'sleep', data: 10 }, (result, error) => {
+        if (error != null) return reject(error);
+        return resolve(result);
+      });
+    });
+    await expect(task4).resolves.toBe(undefined);
+  });
+  // TODO: zero copy of data
+  // TODO: apply the async resource.
 });