From 962a8159b104d06b945fc28143c4468ff1c913d9 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Fri, 2 Jun 2023 09:45:05 +0200 Subject: [PATCH] feat: add event emitter to worker pool/set code MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- package.json | 2 +- src/charging-station/ChargingStationWorker.ts | 14 ++--- src/worker/WorkerAbstract.ts | 2 + src/worker/WorkerDynamicPool.ts | 5 ++ src/worker/WorkerSet.ts | 57 +++++++++++-------- src/worker/WorkerStaticPool.ts | 5 ++ 6 files changed, 50 insertions(+), 35 deletions(-) diff --git a/package.json b/package.json index 049eee85..a12c5535 100644 --- a/package.json +++ b/package.json @@ -2,7 +2,7 @@ "name": "e-mobility-charging-stations-simulator", "version": "1.2.15", "engines": { - "node": ">=16.9.0", + "node": ">=16.14.0", "pnpm": ">=8.6.0" }, "volta": { diff --git a/src/charging-station/ChargingStationWorker.ts b/src/charging-station/ChargingStationWorker.ts index 3703de9c..de0b7a9f 100644 --- a/src/charging-station/ChargingStationWorker.ts +++ b/src/charging-station/ChargingStationWorker.ts @@ -25,18 +25,14 @@ class ChargingStationWorker extends AsyncResource { // Add message listener to create and start charging station from the main thread parentPort?.on('message', (message: WorkerMessage) => { if (message.id === WorkerMessageEvents.startWorkerElement) { - this.run(message.data); + this.runInAsyncScope( + startChargingStation.bind(this) as (data: ChargingStationWorkerData) => void, + this, + message.data + ); } }); } - - private run(data: ChargingStationWorkerData): void { - this.runInAsyncScope( - startChargingStation.bind(this) as (data: ChargingStationWorkerData) => void, - this, - data - ); - } } export let chargingStationWorker: ChargingStationWorker; diff --git a/src/worker/WorkerAbstract.ts b/src/worker/WorkerAbstract.ts index 7753e64d..4934cff9 100644 --- a/src/worker/WorkerAbstract.ts +++ b/src/worker/WorkerAbstract.ts @@ -1,3 +1,4 @@ +import type EventEmitterAsyncResource from 'node:events'; import fs from 'node:fs'; import { WorkerConstants } from './WorkerConstants'; @@ -8,6 +9,7 @@ export abstract class WorkerAbstract { protected readonly workerOptions: WorkerOptions; public abstract readonly size: number; public abstract readonly maxElementsPerWorker: number | undefined; + public abstract readonly emitter: EventEmitterAsyncResource | undefined; /** * `WorkerAbstract` constructor. diff --git a/src/worker/WorkerDynamicPool.ts b/src/worker/WorkerDynamicPool.ts index af8b7e2a..b7b39819 100644 --- a/src/worker/WorkerDynamicPool.ts +++ b/src/worker/WorkerDynamicPool.ts @@ -1,3 +1,4 @@ +import type EventEmitterAsyncResource from 'node:events'; import type { Worker } from 'node:worker_threads'; import { DynamicThreadPool, type ErrorHandler, type ExitHandler } from 'poolifier'; @@ -40,6 +41,10 @@ export class WorkerDynamicPool extends WorkerAbstract { return undefined; } + get emitter(): EventEmitterAsyncResource | undefined { + return this.pool?.emitter; + } + /** @inheritDoc */ public async start(): Promise { // This is intentional diff --git a/src/worker/WorkerSet.ts b/src/worker/WorkerSet.ts index da1e5dd5..ff172d63 100644 --- a/src/worker/WorkerSet.ts +++ b/src/worker/WorkerSet.ts @@ -1,5 +1,6 @@ // Partial Copyright Jerome Benoit. 2021-2023. All Rights Reserved. +import EventEmitterAsyncResource from 'node:events'; import { Worker } from 'node:worker_threads'; import { WorkerAbstract } from './WorkerAbstract'; @@ -14,6 +15,7 @@ import { import { defaultErrorHandler, defaultExitHandler, sleep } from './WorkerUtils'; export class WorkerSet extends WorkerAbstract { + public readonly emitter: EventEmitterAsyncResource; private readonly workerSet: Set; /** @@ -25,6 +27,7 @@ export class WorkerSet extends WorkerAbstract { constructor(workerScript: string, workerOptions?: WorkerOptions) { super(workerScript, workerOptions); this.workerSet = new Set(); + this.emitter = new EventEmitterAsyncResource(); } get size(): number { @@ -40,22 +43,13 @@ export class WorkerSet extends WorkerAbstract { if (!this.workerSet) { throw new Error("Cannot add a WorkerSet element: workers' set does not exist"); } - let lastWorkerSetElement = this.getLastWorkerSetElement(); - if ( - this.workerSet.size === 0 || - lastWorkerSetElement.numberOfWorkerElements >= this.workerOptions.elementsPerWorker - ) { - this.startWorker(); - // Start worker sequentially to optimize memory at startup - this.workerOptions.workerStartDelay > 0 && (await sleep(this.workerOptions.workerStartDelay)); - lastWorkerSetElement = this.getLastWorkerSetElement(); - } - lastWorkerSetElement.worker.postMessage({ + const workerSetElement = await this.getWorkerSetElement(); + workerSetElement.worker.postMessage({ id: WorkerMessageEvents.startWorkerElement, data: elementData, }); - ++lastWorkerSetElement.numberOfWorkerElements; - // Start element sequentially to optimize memory at startup + ++workerSetElement.numberOfWorkerElements; + // Add element sequentially to optimize memory at startup if (this.workerOptions.elementStartDelay > 0) { await sleep(this.workerOptions.elementStartDelay); } @@ -63,8 +57,8 @@ export class WorkerSet extends WorkerAbstract { /** @inheritDoc */ public async start(): Promise { - this.startWorker(); - // Start worker sequentially to optimize memory at startup + this.addWorkerSetElement(); + // Add worker set element sequentially to optimize memory at startup this.workerOptions.workerStartDelay > 0 && (await sleep(this.workerOptions.workerStartDelay)); } @@ -77,9 +71,9 @@ export class WorkerSet extends WorkerAbstract { } /** - * Start a new `Worker`. + * Add a new `WorkerSetElement`. */ - private startWorker(): void { + private addWorkerSetElement(): WorkerSetElement { const worker = new Worker(this.workerScript); worker.on( 'message', @@ -88,21 +82,34 @@ export class WorkerSet extends WorkerAbstract { ) as MessageHandler ); worker.on('error', defaultErrorHandler.bind(this) as (err: Error) => void); - worker.on('error', () => this.startWorker()); + worker.on('error', (error) => { + this.emitter.emit('error', error); + this.addWorkerSetElement(); + }); worker.on('exit', defaultExitHandler.bind(this) as (exitCode: number) => void); worker.on('exit', () => this.workerSet.delete(this.getWorkerSetElementByWorker(worker))); - this.workerSet.add({ worker, numberOfWorkerElements: 0 }); + const workerSetElement: WorkerSetElement = { worker, numberOfWorkerElements: 0 }; + this.workerSet.add(workerSetElement); + return workerSetElement; } - private getLastWorkerSetElement(): WorkerSetElement { - let workerSetElement: WorkerSetElement; - for (workerSetElement of this.workerSet) { - /* This is intentional */ + private async getWorkerSetElement(): Promise { + let chosenWorkerSetElement: WorkerSetElement; + for (const workerSetElement of this.workerSet) { + if (workerSetElement.numberOfWorkerElements < this.workerOptions.elementsPerWorker) { + chosenWorkerSetElement = workerSetElement; + break; + } } - return workerSetElement; + if (!chosenWorkerSetElement) { + chosenWorkerSetElement = this.addWorkerSetElement(); + // Add worker set element sequentially to optimize memory at startup + this.workerOptions.workerStartDelay > 0 && (await sleep(this.workerOptions.workerStartDelay)); + } + return chosenWorkerSetElement; } - private getWorkerSetElementByWorker(worker: Worker): WorkerSetElement { + private getWorkerSetElementByWorker(worker: Worker): WorkerSetElement | undefined { let workerSetElt: WorkerSetElement; for (const workerSetElement of this.workerSet) { if (workerSetElement.worker.threadId === worker.threadId) { diff --git a/src/worker/WorkerStaticPool.ts b/src/worker/WorkerStaticPool.ts index 6034c6e0..af611b63 100644 --- a/src/worker/WorkerStaticPool.ts +++ b/src/worker/WorkerStaticPool.ts @@ -1,3 +1,4 @@ +import type EventEmitterAsyncResource from 'node:events'; import type { Worker } from 'node:worker_threads'; import { type ErrorHandler, type ExitHandler, FixedThreadPool } from 'poolifier'; @@ -39,6 +40,10 @@ export class WorkerStaticPool extends WorkerAbstract { return undefined; } + get emitter(): EventEmitterAsyncResource | undefined { + return this.pool?.emitter; + } + /** @inheritDoc */ public async start(): Promise { // This is intentional -- 2.34.1