fix: fixes to pool and worker initialization

Signed-off-by: Jérôme Benoit <jerome.benoit@piment-noir.org>

Author: jerome-benoit

CHANGELOG.md

@@ -10,9 +10,8 @@ and this project adheres to
 
 ### Fixed
 
-- Ensure worker ready response can be received only once.
-
-<!-- - Ensure pool ready event can be emitted only once. -->
+- Ensure pool cannot be initialized from a worker.
+- Ensure worker initialization is done once.
 
 ## [0.0.6] - 2023-10-20
 

deno.json

@@ -1,6 +1,7 @@
 {
   "$schema": "https://raw.githubusercontent.com/denoland/deno/master/cli/schemas/config-file.v1.json",
   "compilerOptions": {
+    "lib": ["deno.worker"],
     "strict": true
   },
   "tasks": {

examples/javascript/multiFunctionExample.js

@@ -6,10 +6,6 @@ import {
 const pool = new FixedThreadPool(
   availableParallelism(),
   new URL('./multiFunctionWorker.js', import.meta.url),
-  {
-    errorHandler: (e) => console.error(e),
-    onlineHandler: () => console.info('worker is online'),
-  },
 )
 
 pool

src/pools/abstract-pool.ts

@@ -133,6 +133,11 @@ export abstract class AbstractPool<
     protected readonly fileURL: URL,
     protected readonly opts: PoolOptions<Data>,
   ) {
+    if (!this.isMain()) {
+      throw new Error(
+        'Cannot start a pool from a worker with the same type as the pool',
+      )
+    }
     checkFileURL(this.fileURL)
     this.checkNumberOfWorkers(this.numberOfWorkers)
     this.checkPoolOptions(this.opts)
@@ -998,6 +1003,11 @@ export abstract class AbstractPool<
     /* Intentionally empty */
   }
 
+  /**
+   * Should return whether the worker is the main worker or not.
+   */
+  protected abstract isMain(): boolean
+
   /**
    * Hook executed before the worker task execution.
    * Can be overridden.
@@ -1536,11 +1546,6 @@ export abstract class AbstractPool<
     const workerInfo = this.getWorkerInfo(
       this.getWorkerNodeKeyByWorkerId(message.workerId),
     )
-    if (!this.started && workerInfo.ready) {
-      throw new Error(
-        `Ready response already received by worker ${message.workerId}`,
-      )
-    }
     workerInfo.ready = message.ready as boolean
     workerInfo.taskFunctionNames = message.taskFunctionNames
     if (this.ready) {

src/pools/thread/fixed.ts

@@ -1,4 +1,5 @@
 import type { MessageValue } from '../../utility-types.ts'
+import { isWebWorker } from '../../utils.ts'
 import { AbstractPool } from '../abstract-pool.ts'
 import { type PoolOptions, type PoolType, PoolTypes } from '../pool.ts'
 import { type WorkerType, WorkerTypes } from '../worker.ts'
@@ -41,6 +42,10 @@ export class FixedThreadPool<
   ) {
     super(numberOfThreads, fileURL, opts)
   }
+  /** @inheritDoc */
+  protected isMain(): boolean {
+    return !isWebWorker()
+  }
 
   /** @inheritDoc */
   protected async destroyWorkerNode(workerNodeKey: number): Promise<void> {

src/utils.ts

@@ -258,3 +258,10 @@ export const once = <T, A extends any[], R>(
     return result
   }
 }
+
+export const isWebWorker = () => {
+  return (
+    typeof WorkerGlobalScope !== 'undefined' &&
+    self instanceof WorkerGlobalScope
+  )
+}

src/worker/abstract-worker.ts

@@ -58,7 +58,7 @@ export abstract class AbstractWorker<
   /**
    * Worker id.
    */
-  protected abstract id: string
+  protected abstract id?: string
   /**
    * Task function(s) processed by the worker when the pool's `execution` function is invoked.
    */
@@ -77,23 +77,29 @@ export abstract class AbstractWorker<
   protected activeInterval?: number
   /**
    * Constructs a new poolifier worker.
-   *
+   * @param isMain - Whether this is the main worker or not.
    * @param mainWorker - Reference to main worker.
    * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked. The first function is the default function.
    * @param opts - Options for the worker.
    */
   public constructor(
+    private readonly isMain: boolean,
     private readonly mainWorker: MainWorker,
     taskFunctions: TaskFunction<Data, Response> | TaskFunctions<Data, Response>,
     protected opts: WorkerOptions = DEFAULT_WORKER_OPTIONS,
   ) {
+    if (this.isMain == null) {
+      throw new Error('isMain parameter is mandatory')
+    }
     this.checkTaskFunctions(taskFunctions)
     this.checkWorkerOptions(this.opts)
-    this.getMainWorker().addEventListener(
-      'message',
-      this.handleReadyMessageEvent.bind(this),
-      { once: true },
-    )
+    if (!this.isMain) {
+      this.getMainWorker().addEventListener(
+        'message',
+        this.handleReadyMessageEvent.bind(this),
+        { once: true },
+      )
+    }
   }
 
   private checkWorkerOptions(opts: WorkerOptions): void {

src/worker/thread-worker.ts

@@ -2,6 +2,7 @@ import type { MessageValue, MsgEvent } from '../utility-types.ts'
 import { AbstractWorker } from './abstract-worker.ts'
 import type { WorkerOptions } from './worker-options.ts'
 import type { TaskFunction, TaskFunctions } from './task-functions.ts'
+import { isWebWorker } from '../utils.ts'
 
 /**
  * A thread worker used by a poolifier `ThreadPool`.
@@ -26,7 +27,7 @@ export class ThreadWorker<
    */
   private port!: MessagePort
   /** @inheritdoc */
-  public id!: string
+  public id?: string
   /**
    * Constructs a new poolifier thread worker.
    *
@@ -38,6 +39,7 @@ export class ThreadWorker<
     opts: WorkerOptions = {},
   ) {
     super(
+      !isWebWorker(),
       self,
       taskFunctions,
       opts,
@@ -48,7 +50,9 @@ export class ThreadWorker<
   protected handleReadyMessageEvent(
     message: MsgEvent<MessageValue<Data>>,
   ): void {
-    if (
+    if (this.id != null) {
+      return
+    } else if (
       message.data?.workerId != null &&
       message.data?.ready === false &&
       message.data?.port != null

tests/pools/abstract-pool.test.mjs

@@ -24,6 +24,11 @@ Deno.test({
   name: 'Abstract pool test suite',
   fn: async (t) => {
     const numberOfWorkers = 2
+    class StubPoolWithIsMain extends FixedThreadPool {
+      isMain() {
+        return false
+      }
+    }
 
     await t.step('Verify that pool can be created and destroyed', async () => {
       const pool = new FixedThreadPool(
@@ -37,6 +42,23 @@ Deno.test({
       await pool.destroy()
     })
 
+    await t.step(
+      'Verify that pool cannot be created from a non main thread/process',
+      () => {
+        expect(
+          () =>
+            new StubPoolWithIsMain(
+              numberOfWorkers,
+              './tests/worker-files/thread/testWorker.mjs',
+            ),
+        ).toThrow(
+          new Error(
+            'Cannot start a pool from a worker with the same type as the pool',
+          ),
+        )
+      },
+    )
+
     await t.step('Verify that pool statuses properties are set', async () => {
       const pool = new FixedThreadPool(
         numberOfWorkers,