X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=ef5ebeadd6ad08e3ce6a37eb04aa425280a618af;hb=59ce75a1e75c675c4cd9cd1908522c4ce144fd85;hp=a0fc76c371584d3e69855fe732d720db482cc0bf;hpb=8ea697554e82d9f2854acee9fddb8d2fe2d8f9c4;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index a0fc76c3..ef5ebead 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -495,10 +495,7 @@ export abstract class AbstractPool< private checkMessageWorkerId (message: MessageValue): void { if (message.workerId == null) { throw new Error('Worker message received without worker id') - } else if ( - message.workerId != null && - this.getWorkerNodeKeyByWorkerId(message.workerId) === -1 - ) { + } else if (this.getWorkerNodeKeyByWorkerId(message.workerId) === -1) { throw new Error( `Worker message received from unknown worker '${message.workerId}'` ) @@ -622,7 +619,7 @@ export abstract class AbstractPool< for (const [workerNodeKey] of this.workerNodes.entries()) { this.workerNodes[workerNodeKey].addEventListener( 'emptyqueue', - this.handleEmptyQueueEvent + this.handleEmptyQueueEvent as EventListener ) } } @@ -631,7 +628,7 @@ export abstract class AbstractPool< for (const [workerNodeKey] of this.workerNodes.entries()) { this.workerNodes[workerNodeKey].removeEventListener( 'emptyqueue', - this.handleEmptyQueueEvent + this.handleEmptyQueueEvent as EventListener ) } } @@ -640,7 +637,7 @@ export abstract class AbstractPool< for (const [workerNodeKey] of this.workerNodes.entries()) { this.workerNodes[workerNodeKey].addEventListener( 'backpressure', - this.handleBackPressureEvent + this.handleBackPressureEvent as EventListener ) } } @@ -649,7 +646,7 @@ export abstract class AbstractPool< for (const [workerNodeKey] of this.workerNodes.entries()) { this.workerNodes[workerNodeKey].removeEventListener( 'backpressure', - this.handleBackPressureEvent + this.handleBackPressureEvent as EventListener ) } } @@ -1276,6 +1273,8 @@ export abstract class AbstractPool< workerUsage.tasks.executing === 0 && this.tasksQueueSize(localWorkerNodeKey) === 0))) ) { + // Flag the worker node as not ready immediately + this.flagWorkerNodeAsNotReady(localWorkerNodeKey) this.destroyWorkerNode(localWorkerNodeKey).catch(error => { this.emitter?.emit(PoolEvents.error, error) }) @@ -1366,13 +1365,13 @@ export abstract class AbstractPool< if (this.opts.tasksQueueOptions?.taskStealing === true) { this.workerNodes[workerNodeKey].addEventListener( 'emptyqueue', - this.handleEmptyQueueEvent + this.handleEmptyQueueEvent as EventListener ) } if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) { this.workerNodes[workerNodeKey].addEventListener( 'backpressure', - this.handleBackPressureEvent + this.handleBackPressureEvent as EventListener ) } } @@ -1601,7 +1600,7 @@ export abstract class AbstractPool< * @returns The worker information. */ protected getWorkerInfo (workerNodeKey: number): WorkerInfo { - return this.workerNodes[workerNodeKey].info + return this.workerNodes[workerNodeKey]?.info } /** @@ -1641,6 +1640,10 @@ export abstract class AbstractPool< } } + protected flagWorkerNodeAsNotReady (workerNodeKey: number): void { + this.getWorkerInfo(workerNodeKey).ready = false + } + /** @inheritDoc */ public hasWorkerNodeBackPressure (workerNodeKey: number): boolean { return (