X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker-node.ts;h=d29a6e323f35208e22b6f87ca4610fb0af1b3e34;hb=6a677734f19a229fe6a863acd0e6aa0b0d762d6f;hp=bf2600d57782873bea28f50f413a7de945a35b35;hpb=d35e571704515a8b729d3455e4784054f07c368f;p=poolifier.git diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts index bf2600d5..d29a6e32 100644 --- a/src/pools/worker-node.ts +++ b/src/pools/worker-node.ts @@ -1,28 +1,31 @@ -import { MessageChannel } from 'node:worker_threads' import { EventEmitter } from 'node:events' -import { CircularArray } from '../circular-array.js' +import { MessageChannel } from 'node:worker_threads' + +import { CircularBuffer } from '../circular-buffer.js' +import { PriorityQueue } from '../priority-queue.js' import type { Task } from '../utility-types.js' -import { DEFAULT_TASK_NAME, getWorkerId, getWorkerType } from '../utils.js' -import { Deque } from '../deque.js' +import { DEFAULT_TASK_NAME } from '../utils.js' +import { + checkWorkerNodeArguments, + createWorker, + getWorkerId, + getWorkerType, +} from './utils.js' import { - type ErrorHandler, - type ExitHandler, + type EventHandler, type IWorker, type IWorkerNode, - type MessageHandler, - type OnlineHandler, + MeasurementHistorySize, type StrategyData, type WorkerInfo, type WorkerNodeOptions, type WorkerType, WorkerTypes, - type WorkerUsage + type WorkerUsage, } from './worker.js' -import { checkWorkerNodeArguments, createWorker } from './utils.js' /** * Worker node. - * * @typeParam Worker - Type of worker. * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. */ @@ -41,13 +44,12 @@ export class WorkerNode public messageChannel?: MessageChannel /** @inheritdoc */ public tasksQueueBackPressureSize: number - private readonly tasksQueue: Deque> - private onBackPressureStarted: boolean + private readonly tasksQueue: PriorityQueue> + private setBackPressureFlag: boolean private readonly taskFunctionsUsage: Map /** * Constructs a new worker node. - * * @param type - The worker type. * @param filePath - Path to the worker file. * @param opts - The worker node options. @@ -57,19 +59,28 @@ export class WorkerNode checkWorkerNodeArguments(type, filePath, opts) this.worker = createWorker(type, filePath, { env: opts.env, - workerOptions: opts.workerOptions + workerOptions: opts.workerOptions, }) this.info = this.initWorkerInfo(this.worker) this.usage = this.initWorkerUsage() if (this.info.type === WorkerTypes.thread) { this.messageChannel = new MessageChannel() } - this.tasksQueueBackPressureSize = opts.tasksQueueBackPressureSize - this.tasksQueue = new Deque>() - this.onBackPressureStarted = false + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + this.tasksQueueBackPressureSize = opts.tasksQueueBackPressureSize! + this.tasksQueue = new PriorityQueue>( + opts.tasksQueueBucketSize, + opts.tasksQueuePriority + ) + this.setBackPressureFlag = false this.taskFunctionsUsage = new Map() } + /** @inheritdoc */ + public setTasksQueuePriority (enablePriority: boolean): void { + this.tasksQueue.enablePriority = enablePriority + } + /** @inheritdoc */ public tasksQueueSize (): number { return this.tasksQueue.size @@ -77,34 +88,39 @@ export class WorkerNode /** @inheritdoc */ public enqueueTask (task: Task): number { - const tasksQueueSize = this.tasksQueue.push(task) - if (this.hasBackPressure() && !this.onBackPressureStarted) { - this.onBackPressureStarted = true - this.emit('backPressure', { workerId: this.info.id as number }) - this.onBackPressureStarted = false + const tasksQueueSize = this.tasksQueue.enqueue(task, task.priority) + if ( + !this.setBackPressureFlag && + this.hasBackPressure() && + !this.info.backPressure + ) { + this.setBackPressureFlag = true + this.info.backPressure = true + this.emit('backPressure', { workerId: this.info.id }) + this.setBackPressureFlag = false } return tasksQueueSize } /** @inheritdoc */ - public unshiftTask (task: Task): number { - const tasksQueueSize = this.tasksQueue.unshift(task) - if (this.hasBackPressure() && !this.onBackPressureStarted) { - this.onBackPressureStarted = true - this.emit('backPressure', { workerId: this.info.id as number }) - this.onBackPressureStarted = false + public dequeueTask (bucket?: number): Task | undefined { + const task = this.tasksQueue.dequeue(bucket) + if ( + !this.setBackPressureFlag && + !this.hasBackPressure() && + this.info.backPressure + ) { + this.setBackPressureFlag = true + this.info.backPressure = false + this.setBackPressureFlag = false } - return tasksQueueSize + return task } /** @inheritdoc */ - public dequeueTask (): Task | undefined { - return this.tasksQueue.shift() - } - - /** @inheritdoc */ - public popTask (): Task | undefined { - return this.tasksQueue.pop() + public dequeueLastPrioritizedTask (): Task | undefined { + // Start from the last empty or partially filled bucket + return this.dequeueTask(this.tasksQueue.buckets + 1) } /** @inheritdoc */ @@ -117,12 +133,6 @@ export class WorkerNode return this.tasksQueue.size >= this.tasksQueueBackPressureSize } - /** @inheritdoc */ - public resetUsage (): void { - this.usage = this.initWorkerUsage() - this.taskFunctionsUsage.clear() - } - /** @inheritdoc */ public async terminate (): Promise { const waitWorkerExit = new Promise(resolve => { @@ -134,6 +144,7 @@ export class WorkerNode this.removeAllListeners() switch (this.info.type) { case WorkerTypes.thread: + this.worker.unref?.() await this.worker.terminate?.() break case WorkerTypes.cluster: @@ -149,11 +160,7 @@ export class WorkerNode /** @inheritdoc */ public registerWorkerEventHandler ( event: string, - handler: - | OnlineHandler - | MessageHandler - | ErrorHandler - | ExitHandler + handler: EventHandler ): void { this.worker.on(event, handler) } @@ -161,32 +168,28 @@ export class WorkerNode /** @inheritdoc */ public registerOnceWorkerEventHandler ( event: string, - handler: - | OnlineHandler - | MessageHandler - | ErrorHandler - | ExitHandler + handler: EventHandler ): void { this.worker.once(event, handler) } /** @inheritdoc */ public getTaskFunctionWorkerUsage (name: string): WorkerUsage | undefined { - if (!Array.isArray(this.info.taskFunctionNames)) { + if (!Array.isArray(this.info.taskFunctionsProperties)) { throw new Error( - `Cannot get task function worker usage for task function name '${name}' when task function names list is not yet defined` + `Cannot get task function worker usage for task function name '${name}' when task function properties list is not yet defined` ) } if ( - Array.isArray(this.info.taskFunctionNames) && - this.info.taskFunctionNames.length < 3 + Array.isArray(this.info.taskFunctionsProperties) && + this.info.taskFunctionsProperties.length < 3 ) { throw new Error( - `Cannot get task function worker usage for task function name '${name}' when task function names list has less than 3 elements` + `Cannot get task function worker usage for task function name '${name}' when task function properties list has less than 3 elements` ) } if (name === DEFAULT_TASK_NAME) { - name = this.info.taskFunctionNames[1] + name = this.info.taskFunctionsProperties[1].name } if (!this.taskFunctionsUsage.has(name)) { this.taskFunctionsUsage.set(name, this.initTaskFunctionWorkerUsage(name)) @@ -212,10 +215,12 @@ export class WorkerNode private initWorkerInfo (worker: Worker): WorkerInfo { return { id: getWorkerId(worker), - type: getWorkerType(worker) as WorkerType, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + type: getWorkerType(worker)!, dynamic: false, ready: false, - stealing: false + stealing: false, + backPressure: false, } } @@ -238,22 +243,22 @@ export class WorkerNode }, sequentiallyStolen: 0, stolen: 0, - failed: 0 + failed: 0, }, runTime: { - history: new CircularArray() + history: new CircularBuffer(MeasurementHistorySize), }, waitTime: { - history: new CircularArray() + history: new CircularBuffer(MeasurementHistorySize), }, elu: { idle: { - history: new CircularArray() + history: new CircularBuffer(MeasurementHistorySize), }, active: { - history: new CircularArray() - } - } + history: new CircularBuffer(MeasurementHistorySize), + }, + }, } } @@ -263,7 +268,8 @@ export class WorkerNode for (const task of this.tasksQueue) { if ( (task.name === DEFAULT_TASK_NAME && - name === (this.info.taskFunctionNames as string[])[1]) || + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + name === this.info.taskFunctionsProperties![1].name) || (task.name !== DEFAULT_TASK_NAME && name === task.name) ) { ++taskFunctionQueueSize @@ -280,22 +286,22 @@ export class WorkerNode }, sequentiallyStolen: 0, stolen: 0, - failed: 0 + failed: 0, }, runTime: { - history: new CircularArray() + history: new CircularBuffer(MeasurementHistorySize), }, waitTime: { - history: new CircularArray() + history: new CircularBuffer(MeasurementHistorySize), }, elu: { idle: { - history: new CircularArray() + history: new CircularBuffer(MeasurementHistorySize), }, active: { - history: new CircularArray() - } - } + history: new CircularBuffer(MeasurementHistorySize), + }, + }, } } }