X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker-node.ts;h=f9b4016c29b078143d6f56d9acfcaf24b06e688d;hb=e8b896e4a31e99644eeed068e9eddd5dabc0b12c;hp=079881b94cf165a16f4fe2e6be9c4223579a8346;hpb=79b197bb222403411621c41775d9d640f4150633;p=poolifier.git diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts index 079881b9..f9b4016c 100644 --- a/src/pools/worker-node.ts +++ b/src/pools/worker-node.ts @@ -5,6 +5,8 @@ import { DEFAULT_TASK_NAME, EMPTY_FUNCTION, exponentialDelay, + getWorkerId, + getWorkerType, sleep } from '../utils' import { Deque } from '../deque' @@ -12,6 +14,7 @@ import { type IWorker, type IWorkerNode, type WorkerInfo, + type WorkerNodeEventCallback, type WorkerType, WorkerTypes, type WorkerUsage @@ -30,39 +33,29 @@ implements IWorkerNode { /** @inheritdoc */ public readonly info: WorkerInfo /** @inheritdoc */ - public messageChannel?: MessageChannel - /** @inheritdoc */ public usage: WorkerUsage /** @inheritdoc */ + public messageChannel?: MessageChannel + /** @inheritdoc */ public tasksQueueBackPressureSize: number /** @inheritdoc */ - public onBackPressure?: (workerId: number) => void + public onBackPressure?: WorkerNodeEventCallback /** @inheritdoc */ - public onEmptyQueue?: (workerId: number) => void - private readonly taskFunctionsUsage: Map + public onEmptyQueue?: WorkerNodeEventCallback private readonly tasksQueue: Deque> private onEmptyQueueCount: number + private readonly taskFunctionsUsage: Map /** * Constructs a new worker node. * * @param worker - The worker. - * @param workerType - The worker type. * @param tasksQueueBackPressureSize - The tasks queue back pressure size. */ - constructor ( - worker: Worker, - workerType: WorkerType, - tasksQueueBackPressureSize: number - ) { + constructor (worker: Worker, tasksQueueBackPressureSize: number) { if (worker == null) { throw new TypeError('Cannot construct a worker node without a worker') } - if (workerType == null) { - throw new TypeError( - 'Cannot construct a worker node without a worker type' - ) - } if (tasksQueueBackPressureSize == null) { throw new TypeError( 'Cannot construct a worker node without a tasks queue back pressure size' @@ -74,15 +67,15 @@ implements IWorkerNode { ) } this.worker = worker - this.info = this.initWorkerInfo(worker, workerType) - if (workerType === WorkerTypes.thread) { + this.info = this.initWorkerInfo(worker) + this.usage = this.initWorkerUsage() + if (this.info.type === WorkerTypes.thread) { this.messageChannel = new MessageChannel() } - this.usage = this.initWorkerUsage() - this.taskFunctionsUsage = new Map() - this.tasksQueue = new Deque>() this.tasksQueueBackPressureSize = tasksQueueBackPressureSize + this.tasksQueue = new Deque>() this.onEmptyQueueCount = 0 + this.taskFunctionsUsage = new Map() } /** @inheritdoc */ @@ -185,16 +178,16 @@ implements IWorkerNode { this.onEmptyQueueCount = 0 return } - (this.onEmptyQueue as (workerId: number) => void)(this.info.id as number) + (this.onEmptyQueue as WorkerNodeEventCallback)(this.info.id as number) ++this.onEmptyQueueCount await sleep(exponentialDelay(this.onEmptyQueueCount)) await this.startOnEmptyQueue() } - private initWorkerInfo (worker: Worker, workerType: WorkerType): WorkerInfo { + private initWorkerInfo (worker: Worker): WorkerInfo { return { - id: this.getWorkerId(worker, workerType), - type: workerType, + id: getWorkerId(worker), + type: getWorkerType(worker) as WorkerType, dynamic: false, ready: false } @@ -277,22 +270,4 @@ implements IWorkerNode { } } } - - /** - * Gets the worker id. - * - * @param worker - The worker. - * @param workerType - The worker type. - * @returns The worker id. - */ - private getWorkerId ( - worker: Worker, - workerType: WorkerType - ): number | undefined { - if (workerType === WorkerTypes.thread) { - return worker.threadId - } else if (workerType === WorkerTypes.cluster) { - return worker.id - } - } }