X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker-node.ts;h=f9b4016c29b078143d6f56d9acfcaf24b06e688d;hb=e8b896e4a31e99644eeed068e9eddd5dabc0b12c;hp=f3ec3be3d83f4541da3713cbd46406d19be41ac0;hpb=6e9c39d3eedabc9f12808490983d36e551a75a28;p=poolifier.git diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts index f3ec3be3..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,14 +14,12 @@ import { type IWorker, type IWorkerNode, type WorkerInfo, + type WorkerNodeEventCallback, type WorkerType, WorkerTypes, type WorkerUsage } from './worker' -type EmptyQueueCallback = (workerId: number) => void -type BackPressureCallback = EmptyQueueCallback - /** * Worker node. * @@ -39,9 +39,9 @@ implements IWorkerNode { /** @inheritdoc */ public tasksQueueBackPressureSize: number /** @inheritdoc */ - public onBackPressure?: BackPressureCallback + public onBackPressure?: WorkerNodeEventCallback /** @inheritdoc */ - public onEmptyQueue?: EmptyQueueCallback + public onEmptyQueue?: WorkerNodeEventCallback private readonly tasksQueue: Deque> private onEmptyQueueCount: number private readonly taskFunctionsUsage: Map @@ -50,22 +50,12 @@ implements IWorkerNode { * 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' @@ -77,9 +67,9 @@ implements IWorkerNode { ) } this.worker = worker - this.info = this.initWorkerInfo(worker, workerType) + this.info = this.initWorkerInfo(worker) this.usage = this.initWorkerUsage() - if (workerType === WorkerTypes.thread) { + if (this.info.type === WorkerTypes.thread) { this.messageChannel = new MessageChannel() } this.tasksQueueBackPressureSize = tasksQueueBackPressureSize @@ -188,16 +178,16 @@ implements IWorkerNode { this.onEmptyQueueCount = 0 return } - (this.onEmptyQueue as EmptyQueueCallback)(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 } @@ -280,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 - } - } }