X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2FWorkerSet.ts;h=4a43e27b0601c44d3035c77411640cefd2fe798a;hb=bb29949603108e1bc169ffbad4bc76d2d126dc9d;hp=b8e0e31fb641b94080933a87f06779009e95e43a;hpb=a4385edc2344433c20fcb92d01cbe5a330b850b4;p=e-mobility-charging-stations-simulator.git diff --git a/src/worker/WorkerSet.ts b/src/worker/WorkerSet.ts index b8e0e31f..4a43e27b 100644 --- a/src/worker/WorkerSet.ts +++ b/src/worker/WorkerSet.ts @@ -1,4 +1,4 @@ -// Partial Copyright Jerome Benoit. 2021-2023. All Rights Reserved. +// Partial Copyright Jerome Benoit. 2021-2024. All Rights Reserved. import { EventEmitterAsyncResource } from 'node:events' import { SHARE_ENV, Worker } from 'node:worker_threads' @@ -52,6 +52,7 @@ export class WorkerSet extends WorkerAbstract { version: workerSetVersion, type: 'set', worker: 'thread', + started: this.started, size: this.size, elementsExecuting: [...this.workerSet].reduce( (accumulator, workerSetElement) => accumulator + workerSetElement.numberOfWorkerElements, @@ -86,18 +87,19 @@ export class WorkerSet extends WorkerAbstract { public async stop (): Promise { for (const workerSetElement of this.workerSet) { const worker = workerSetElement.worker - const waitWorkerExit = new Promise((resolve) => { + const waitWorkerExit = new Promise(resolve => { worker.once('exit', () => { resolve() }) }) + worker.unref() await worker.terminate() await waitWorkerExit - this.emitter?.emit(WorkerSetEvents.stopped, this.info) - this.emitter?.emitDestroy() - this.emitter?.removeAllListeners() - this.started = false } + this.emitter?.emit(WorkerSetEvents.stopped, this.info) + this.started = false + this.emitter?.emitDestroy() + this.emitter?.removeAllListeners() } /** @inheritDoc */ @@ -105,12 +107,9 @@ export class WorkerSet extends WorkerAbstract { if (!this.started) { throw new Error('Cannot add a WorkerSet element: not started') } - if (this.workerSet == null) { - throw new Error("Cannot add a WorkerSet element: 'workerSet' property does not exist") - } const workerSetElement = await this.getWorkerSetElement() workerSetElement.worker.postMessage({ - event: WorkerMessageEvents.startWorkerElement, + event: WorkerMessageEvents.addWorkerElement, data: elementData }) ++workerSetElement.numberOfWorkerElements @@ -133,14 +132,14 @@ export class WorkerSet extends WorkerAbstract { }) worker.on('message', this.workerOptions.poolOptions?.messageHandler ?? EMPTY_FUNCTION) worker.on('message', (message: WorkerMessage) => { - if (message.event === WorkerMessageEvents.startedWorkerElement) { - this.emitter?.emit(WorkerSetEvents.elementStarted, this.info) - } else if (message.event === WorkerMessageEvents.startWorkerElementError) { + if (message.event === WorkerMessageEvents.addedWorkerElement) { + this.emitter?.emit(WorkerSetEvents.elementAdded, this.info) + } else if (message.event === WorkerMessageEvents.workerElementError) { this.emitter?.emit(WorkerSetEvents.elementError, message.data) } }) worker.on('error', this.workerOptions.poolOptions?.errorHandler ?? EMPTY_FUNCTION) - worker.on('error', (error) => { + worker.on('error', error => { this.emitter?.emit(WorkerSetEvents.error, error) if ( this.workerOptions.poolOptions?.restartWorkerOnError === true && @@ -149,6 +148,8 @@ export class WorkerSet extends WorkerAbstract { ) { this.addWorkerSetElement() } + worker.unref() + worker.terminate().catch(error => this.emitter?.emit(WorkerSetEvents.error, error)) }) worker.on('online', this.workerOptions.poolOptions?.onlineHandler ?? EMPTY_FUNCTION) worker.on('exit', this.workerOptions.poolOptions?.exitHandler ?? EMPTY_FUNCTION)