wip: progressing and functional test worker

[ci skip]

Author: tegefaulkes

package-lock.json

@@ -37,9 +37,9 @@
     "bench": "tsc -p ./tsconfig.build.json && shx rm -rf ./benches/results && tsx ./benches/index.ts"
   },
   "dependencies": {
-    "@matrixai/async-init": "^2.0.0",
-    "@matrixai/errors": "^2.0.1",
-    "@matrixai/logger": "^4.0.1",
+    "@matrixai/async-init": "^2.1.2",
+    "@matrixai/errors": "^2.1.3",
+    "@matrixai/logger": "^4.0.3",
     "ts-node": "^10.9.1",
     "rxjs": "^7.8.2"
   },

package.json

@@ -50,8 +50,9 @@ class WorkerPool {
 
   protected addWorker() {
     const worker = this.workerFactory();
-    const messageHandler = (result: unknown) => {
-      worker[taskInfoSymbol](result, undefined);
+    const messageHandler = (result: { result?: unknown; error?: Error }) => {
+      if (result.error != null) worker[taskInfoSymbol](undefined, result.error);
+      else worker[taskInfoSymbol](result.result);
       worker[taskInfoSymbol] = undefined;
       this.freeWorkers.push(worker);
       this.$workerFreed.next();
@@ -99,7 +100,6 @@ class WorkerPool {
 
   public async terminate(force: boolean) {
     this.terminating = true;
-    this.handleDestroySubscription.unsubscribe();
     if (!force) {
       // Prevent new tasks and wait for exising queue to drain
       await new Promise<void>((resolve) => {
@@ -122,6 +122,8 @@ class WorkerPool {
       });
     }
 
+    // Prevent terminations from creating new workers
+    this.handleDestroySubscription.unsubscribe();
     const workerTerminatePs: Array<Promise<number>> = [];
     for (const worker of this.workers) {
       workerTerminatePs.push(worker.terminate());

src/WorkerPool.ts

@@ -10,24 +10,32 @@ 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(task);
+    console.log('got task!', task);
     const method = workerManifest[task.type];
     if (method == null) {
-      const e = Error('TMP IMP missing handler');
-      console.error(e);
-      throw e;
+      parentPort!.postMessage({ error: Error('TMP IMP missing handler') });
+      return;
     }
-    const result = await method(task);
-    parentPort?.postMessage(result);
+    let result: unknown;
+    try {
+      result = await method(task.data);
+    } catch (error) {
+      parentPort!.postMessage({ error });
+      return;
+    }
+    parentPort!.postMessage({ result });
   };
   const handleMessageError = (e) => {
-    console.error(`Worker ${threadId} got error`, e);
+    console.log(`Worker ${threadId} got error`, e);
   };
+  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/expose.ts

@@ -8,6 +8,7 @@ const worker = {
     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> => {

src/worker.ts

@@ -1,49 +1,53 @@
 import type { WorkerFactory } from '#types.js';
 import { Worker } from 'node:worker_threads';
-import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
+// Import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
+import path from 'node:path';
 import WorkerPool from '#WorkerPool.js';
 
+async function sleep(ms: number) {
+  return new Promise((resolve) => {
+    setTimeout(resolve, ms);
+  });
+}
+
 describe('WorkerPool', () => {
-  const logger = new Logger('WorkerPool Test', LogLevel.WARN, [
-    new StreamHandler(),
-  ]);
+  // Const logger = new Logger('WorkerPool Test', LogLevel.WARN, [
+  //   new StreamHandler(),
+  // ]);
 
-  const delayedEchoScript = `
-const nodeWorkers = require("node:worker_threads");
-console.log("Hello from worker!: ", nodeWorkers.isMainThread, nodeWorkers.threadId);
-nodeWorkers.parentPort.on('message', async (v) => {
-  // console.log('message', v);
-  await new Promise(resolve => {
-    setTimeout(resolve, 10);
-  })
-  nodeWorkers.parentPort.postMessage(v);
-});
-`;
+  let pool: WorkerPool;
 
   const workerFactory: WorkerFactory = () => {
-    return new Worker(delayedEchoScript, { eval: true });
+    return new Worker(path.join(__dirname, '../dist/worker.js'));
   };
 
+  afterEach(async () => {
+    await pool.terminate(true);
+    console.log('DONE!');
+  });
+
   test('asd', async () => {
-    const pool = new WorkerPool(3, workerFactory);
-    // Pool.$poolStatus.subscribe((v) => {
-    //   console.log('!queue state!', v);
-    // });
+    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({ method: 'one', data: 1 }, (result, error) => {
-        // Console.log(result, error);
+      pool.runTask({ type: 'fac', data: 5 }, (result, error) => {
+        console.log({ result, error });
         doneCount++;
         if (doneCount === numTasks) resolveP(result);
       });
     }
-    const result = await taskP;
-    // Console.log(result);
+    console.log('waiting');
+    await taskP;
+    console.log('waiting done');
     await pool.terminate(false);
+    console.log('term done');
   });
 });