From c81424b89ca5184eea982c355791f658b4064ce7 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 10 Sep 2023 17:11:54 +0200 Subject: [PATCH] refactor: cleanup pool options defaults handling MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- src/worker/WorkerConstants.ts | 12 +++++------- src/worker/WorkerSet.ts | 21 ++++++++++++++------- 2 files changed, 19 insertions(+), 14 deletions(-) diff --git a/src/worker/WorkerConstants.ts b/src/worker/WorkerConstants.ts index 8c66df02..c7f8065a 100644 --- a/src/worker/WorkerConstants.ts +++ b/src/worker/WorkerConstants.ts @@ -1,4 +1,4 @@ -import { type ThreadPoolOptions, availableParallelism } from 'poolifier'; +import { availableParallelism } from 'poolifier'; import type { WorkerOptions } from './WorkerTypes'; @@ -20,10 +20,8 @@ export const DEFAULT_WORKER_OPTIONS: WorkerOptions = Object.freeze({ poolMinSize: DEFAULT_POOL_MIN_SIZE, poolMaxSize: DEFAULT_POOL_MAX_SIZE, elementsPerWorker: DEFAULT_ELEMENTS_PER_WORKER, - poolOptions: {}, + poolOptions: { + enableEvents: true, + restartWorkerOnError: true, + }, }); - -export const DEFAULT_POOL_OPTIONS: ThreadPoolOptions = { - enableEvents: true, - restartWorkerOnError: true, -}; diff --git a/src/worker/WorkerSet.ts b/src/worker/WorkerSet.ts index 39e8eca3..7ddc96c7 100644 --- a/src/worker/WorkerSet.ts +++ b/src/worker/WorkerSet.ts @@ -4,7 +4,7 @@ import { EventEmitter } from 'node:events'; import { SHARE_ENV, Worker } from 'node:worker_threads'; import { WorkerAbstract } from './WorkerAbstract'; -import { DEFAULT_POOL_OPTIONS, EMPTY_FUNCTION, workerSetVersion } from './WorkerConstants'; +import { EMPTY_FUNCTION, workerSetVersion } from './WorkerConstants'; import { type SetInfo, type WorkerData, @@ -19,6 +19,7 @@ import { sleep } from './WorkerUtils'; export class WorkerSet extends WorkerAbstract { public readonly emitter!: EventEmitter; private readonly workerSet: Set; + private started: boolean; private workerStartup: boolean; /** @@ -41,14 +42,11 @@ export class WorkerSet extends WorkerAbstract { if (this.workerOptions.elementsPerWorker <= 0) { throw new RangeError('Elements per worker must be greater than zero'); } - this.workerOptions.poolOptions = { - ...DEFAULT_POOL_OPTIONS, - ...this.workerOptions.poolOptions, - }; this.workerSet = new Set(); if (this.workerOptions.poolOptions?.enableEvents) { this.emitter = new EventEmitter(); } + this.started = false; this.workerStartup = false; } @@ -79,6 +77,7 @@ export class WorkerSet extends WorkerAbstract { this.addWorkerSetElement(); // Add worker set element sequentially to optimize memory at startup this.workerOptions.workerStartDelay! > 0 && (await sleep(this.workerOptions.workerStartDelay!)); + this.started = true; } /** @inheritDoc */ @@ -92,13 +91,17 @@ export class WorkerSet extends WorkerAbstract { }); await worker.terminate(); await waitWorkerExit; + this.started = false; } } /** @inheritDoc */ public async addElement(elementData: WorkerData): Promise { + if (!this.started) { + throw new Error('Cannot add a WorkerSet element: not started'); + } if (!this.workerSet) { - throw new Error("Cannot add a WorkerSet element: workers' set does not exist"); + throw new Error("Cannot add a WorkerSet element: 'workerSet' property does not exist"); } const workerSetElement = await this.getWorkerSetElement(); workerSetElement.worker.postMessage({ @@ -132,7 +135,11 @@ export class WorkerSet extends WorkerAbstract { worker.on('error', this.workerOptions.poolOptions?.errorHandler ?? EMPTY_FUNCTION); worker.on('error', (error) => { this.emitter?.emit(WorkerSetEvents.error, error); - if (this.workerOptions.poolOptions?.restartWorkerOnError && !this.workerStartup) { + if ( + this.workerOptions.poolOptions?.restartWorkerOnError && + this.started && + !this.workerStartup + ) { this.addWorkerSetElement(); } }); -- 2.34.1