X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker-node.ts;h=387d05f964270545b5b92f0e841768dea20a23df;hb=0dd90fe1089e9852ecea1188dcb8856dbc5de83f;hp=4799e11c03d1677e86ad675c31ffe7b64982c807;hpb=26fb3c18b678a1daab6b18a351a238fb5a3ed5df;p=poolifier.git diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts index 4799e11c..387d05f9 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 @@ -36,9 +39,9 @@ implements IWorkerNode { /** @inheritdoc */ public tasksQueueBackPressureSize: number /** @inheritdoc */ - public onBackPressure?: (workerId: number) => void + public onBackPressure?: WorkerNodeEventCallback /** @inheritdoc */ - public onEmptyQueue?: (workerId: number) => void + public onEmptyQueue?: WorkerNodeEventCallback private readonly tasksQueue: Deque> private onEmptyQueueCount: number private readonly taskFunctionsUsage: Map @@ -47,36 +50,14 @@ 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 - ) { - 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' - ) - } - if (!Number.isSafeInteger(tasksQueueBackPressureSize)) { - throw new TypeError( - 'Cannot construct a worker node with a tasks queue back pressure size that is not an integer' - ) - } + constructor (worker: Worker, tasksQueueBackPressureSize: number) { + this.checkWorkerNodeArguments(worker, tasksQueueBackPressureSize) 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 @@ -185,16 +166,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 } @@ -278,21 +259,27 @@ implements IWorkerNode { } } - /** - * Gets the worker id. - * - * @param worker - The worker. - * @param workerType - The worker type. - * @returns The worker id. - */ - private getWorkerId ( + private checkWorkerNodeArguments ( worker: Worker, - workerType: WorkerType - ): number | undefined { - if (workerType === WorkerTypes.thread) { - return worker.threadId - } else if (workerType === WorkerTypes.cluster) { - return worker.id + tasksQueueBackPressureSize: number + ): void { + if (worker == null) { + throw new TypeError('Cannot construct a worker node without a worker') + } + if (tasksQueueBackPressureSize == null) { + throw new TypeError( + 'Cannot construct a worker node without a tasks queue back pressure size' + ) + } + if (!Number.isSafeInteger(tasksQueueBackPressureSize)) { + throw new TypeError( + 'Cannot construct a worker node with a tasks queue back pressure size that is not an integer' + ) + } + if (tasksQueueBackPressureSize <= 0) { + throw new RangeError( + 'Cannot construct a worker node with a tasks queue back pressure size that is not a positive integer' + ) } } }