X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2FWorkerSet.ts;h=19ed8393f171a64d37e4e3d3e871cacf81f17539;hb=2bb7a73e94db7533c1ba47db677d709e6f07dfdb;hp=dd7cf2d7685ac3012143e856873408307d85a8fa;hpb=5edd8ba0f8978cfb3ca9d80f299d9748c6c5970e;p=e-mobility-charging-stations-simulator.git diff --git a/src/worker/WorkerSet.ts b/src/worker/WorkerSet.ts index dd7cf2d7..19ed8393 100644 --- a/src/worker/WorkerSet.ts +++ b/src/worker/WorkerSet.ts @@ -3,6 +3,8 @@ import { EventEmitter } from 'node:events'; import { SHARE_ENV, Worker } from 'node:worker_threads'; +import type { ThreadPoolOptions } from 'poolifier'; + import { WorkerAbstract } from './WorkerAbstract'; import { WorkerConstants } from './WorkerConstants'; import { @@ -15,8 +17,13 @@ import { } from './WorkerTypes'; import { sleep } from './WorkerUtils'; +const DEFAULT_POOL_OPTIONS: ThreadPoolOptions = { + enableEvents: true, + restartWorkerOnError: true, +}; + export class WorkerSet extends WorkerAbstract { - public readonly emitter: EventEmitter; + public readonly emitter!: EventEmitter; private readonly workerSet: Set; /** @@ -25,17 +32,14 @@ export class WorkerSet extends WorkerAbstract { * @param workerScript - * @param workerOptions - */ - constructor(workerScript: string, workerOptions?: WorkerOptions) { + constructor(workerScript: string, workerOptions: WorkerOptions) { super(workerScript, workerOptions); this.workerOptions.poolOptions = { - ...{ - enableEvents: true, - restartWorkerOnError: true, - }, + ...DEFAULT_POOL_OPTIONS, ...this.workerOptions.poolOptions, }; this.workerSet = new Set(); - if (this.workerOptions?.poolOptions?.enableEvents) { + if (this.workerOptions.poolOptions?.enableEvents) { this.emitter = new EventEmitter(); } } @@ -50,7 +54,7 @@ export class WorkerSet extends WorkerAbstract { (accumulator, workerSetElement) => accumulator + workerSetElement.numberOfWorkerElements, 0, ), - elementsPerWorker: this.maxElementsPerWorker, + elementsPerWorker: this.maxElementsPerWorker!, }; } @@ -66,7 +70,7 @@ export class WorkerSet extends WorkerAbstract { public async start(): Promise { this.addWorkerSetElement(); // Add worker set element sequentially to optimize memory at startup - this.workerOptions.workerStartDelay > 0 && (await sleep(this.workerOptions.workerStartDelay)); + this.workerOptions.workerStartDelay! > 0 && (await sleep(this.workerOptions.workerStartDelay!)); } /** @inheritDoc */ @@ -90,13 +94,13 @@ export class WorkerSet extends WorkerAbstract { } const workerSetElement = await this.getWorkerSetElement(); workerSetElement.worker.postMessage({ - id: WorkerMessageEvents.startWorkerElement, + event: WorkerMessageEvents.startWorkerElement, data: elementData, }); ++workerSetElement.numberOfWorkerElements; // Add element sequentially to optimize memory at startup - if (this.workerOptions.elementStartDelay > 0) { - await sleep(this.workerOptions.elementStartDelay); + if (this.workerOptions.elementStartDelay! > 0) { + await sleep(this.workerOptions.elementStartDelay!); } } @@ -106,34 +110,32 @@ export class WorkerSet extends WorkerAbstract { private addWorkerSetElement(): WorkerSetElement { const worker = new Worker(this.workerScript, { env: SHARE_ENV, - ...this.workerOptions.poolOptions.workerOptions, + ...this.workerOptions.poolOptions?.workerOptions, }); worker.on( 'message', - this.workerOptions?.poolOptions?.messageHandler ?? WorkerConstants.EMPTY_FUNCTION, + this.workerOptions.poolOptions?.messageHandler ?? WorkerConstants.EMPTY_FUNCTION, ); worker.on( 'error', - this.workerOptions?.poolOptions?.errorHandler ?? WorkerConstants.EMPTY_FUNCTION, + this.workerOptions.poolOptions?.errorHandler ?? WorkerConstants.EMPTY_FUNCTION, ); worker.on('error', (error) => { - if (this.emitter !== undefined) { - this.emitter.emit(WorkerSetEvents.error, error); - } - if (this.workerOptions?.poolOptions?.restartWorkerOnError) { + this.emitter?.emit(WorkerSetEvents.error, error); + if (this.workerOptions.poolOptions?.restartWorkerOnError) { this.addWorkerSetElement(); } }); worker.on( 'online', - this.workerOptions?.poolOptions?.onlineHandler ?? WorkerConstants.EMPTY_FUNCTION, + this.workerOptions.poolOptions?.onlineHandler ?? WorkerConstants.EMPTY_FUNCTION, ); worker.on( 'exit', - this.workerOptions?.poolOptions?.exitHandler ?? WorkerConstants.EMPTY_FUNCTION, + this.workerOptions.poolOptions?.exitHandler ?? WorkerConstants.EMPTY_FUNCTION, ); worker.once('exit', () => - this.removeWorkerSetElement(this.getWorkerSetElementByWorker(worker)), + this.removeWorkerSetElement(this.getWorkerSetElementByWorker(worker)!), ); const workerSetElement: WorkerSetElement = { worker, numberOfWorkerElements: 0 }; this.workerSet.add(workerSetElement); @@ -145,9 +147,9 @@ export class WorkerSet extends WorkerAbstract { } private async getWorkerSetElement(): Promise { - let chosenWorkerSetElement: WorkerSetElement; + let chosenWorkerSetElement: WorkerSetElement | undefined; for (const workerSetElement of this.workerSet) { - if (workerSetElement.numberOfWorkerElements < this.workerOptions.elementsPerWorker) { + if (workerSetElement.numberOfWorkerElements < this.workerOptions.elementsPerWorker!) { chosenWorkerSetElement = workerSetElement; break; } @@ -155,13 +157,14 @@ export class WorkerSet extends WorkerAbstract { if (!chosenWorkerSetElement) { chosenWorkerSetElement = this.addWorkerSetElement(); // Add worker set element sequentially to optimize memory at startup - this.workerOptions.workerStartDelay > 0 && (await sleep(this.workerOptions.workerStartDelay)); + this.workerOptions.workerStartDelay! > 0 && + (await sleep(this.workerOptions.workerStartDelay!)); } return chosenWorkerSetElement; } private getWorkerSetElementByWorker(worker: Worker): WorkerSetElement | undefined { - let workerSetElt: WorkerSetElement; + let workerSetElt: WorkerSetElement | undefined; for (const workerSetElement of this.workerSet) { if (workerSetElement.worker.threadId === worker.threadId) { workerSetElt = workerSetElement;