From: Jérôme Benoit Date: Fri, 30 Jun 2023 22:55:35 +0000 (+0200) Subject: fix: ensure worker started message cannot be received for non existing X-Git-Tag: v2.6.7~16^2~10 X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=83fa0a36c7d67362593915f3b19867f8fbbcbcce;p=poolifier.git fix: ensure worker started message cannot be received for non existing worker Signed-off-by: Jérôme Benoit --- diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 6fde2491..ba05d929 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -19,7 +19,8 @@ import { type PoolType, PoolTypes, type TasksQueueOptions, - type WorkerType + type WorkerType, + WorkerTypes } from './pool' import type { IWorker, @@ -784,9 +785,13 @@ export abstract class AbstractPool< return message => { if (message.workerId != null && message.started != null) { // Worker started message received - this.workerNodes[ - this.getWorkerNodeKey(this.getWorkerById(message.workerId) as Worker) - ].info.started = message.started + const worker = this.getWorkerById(message.workerId) + if (worker != null) { + this.workerNodes[this.getWorkerNodeKey(worker)].info.started = + message.started + } else { + throw new Error('Worker started message received from unknown worker') + } } else if (message.id != null) { // Task execution response received const promiseResponse = this.promiseResponseMap.get(message.id) @@ -850,7 +855,7 @@ export abstract class AbstractPool< private pushWorkerNode (worker: Worker): number { this.workerNodes.push({ worker, - info: { id: worker.threadId ?? worker.id, started: false }, + info: { id: this.getWorkerId(worker), started: false }, usage: this.getWorkerUsage(), tasksQueue: new Queue>() }) @@ -862,6 +867,20 @@ export abstract class AbstractPool< return this.workerNodes.length } + /** + * Gets the worker id. + * + * @param worker - The worker. + * @returns The worker id. + */ + private getWorkerId (worker: Worker): number | undefined { + if (this.worker === WorkerTypes.thread) { + return worker.threadId + } else if (this.worker === WorkerTypes.cluster) { + return worker.id + } + } + // /** // * Sets the given worker in the pool worker nodes. // * diff --git a/src/pools/worker.ts b/src/pools/worker.ts index c92478d4..e0636d99 100644 --- a/src/pools/worker.ts +++ b/src/pools/worker.ts @@ -125,7 +125,7 @@ export interface TaskStatistics { */ export interface WorkerInfo { /** - * Worker Id. + * Worker id. */ id: number | undefined /** @@ -163,7 +163,7 @@ export interface WorkerUsage { */ export interface IWorker { /** - * Worker Id. + * Worker id. */ id?: number threadId?: number diff --git a/src/utility-types.ts b/src/utility-types.ts index 1f724274..0443b4ce 100644 --- a/src/utility-types.ts +++ b/src/utility-types.ts @@ -54,7 +54,7 @@ export interface WorkerStatistics { export interface MessageValue extends Task { /** - * Worker Id. + * Worker id. */ readonly workerId?: number /** diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 242c4698..0646a9bf 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -37,7 +37,7 @@ export abstract class AbstractWorker< Response = unknown > extends AsyncResource { /** - * Worker Id. + * Worker id. */ protected abstract id: number /**