X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker-node.ts;h=d87a79c099c5dab34126e284ca2f14f4be72b3a1;hb=f12182ad6dc553c7a5dfeee01bcde65c0177f671;hp=3de3cb809cda9a3a3f2787579371a467e3f9d200;hpb=463226a44b0370911db746052bcad0bcc7878acf;p=poolifier.git diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts index 3de3cb80..d87a79c0 100644 --- a/src/pools/worker-node.ts +++ b/src/pools/worker-node.ts @@ -1,19 +1,28 @@ +import { EventEmitter } from 'node:events' import { MessageChannel } from 'node:worker_threads' -import { CircularArray } from '../circular-array' -import type { Task } from '../utility-types' -import { DEFAULT_TASK_NAME, getWorkerId, getWorkerType } from '../utils' -import { Deque } from '../deque' + +import { CircularBuffer } from '../circular-buffer.js' +import { PriorityQueue } from '../priority-queue.js' +import type { Task } from '../utility-types.js' +import { DEFAULT_TASK_NAME } from '../utils.js' +import { + checkWorkerNodeArguments, + createWorker, + getWorkerId, + getWorkerType +} from './utils.js' import { + type EventHandler, type IWorker, type IWorkerNode, + MeasurementHistorySize, type StrategyData, type WorkerInfo, - type WorkerNodeEventDetail, + type WorkerNodeOptions, type WorkerType, WorkerTypes, type WorkerUsage -} from './worker' -import { checkWorkerNodeArguments } from './utils' +} from './worker.js' /** * Worker node. @@ -22,7 +31,7 @@ import { checkWorkerNodeArguments } from './utils' * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. */ export class WorkerNode - extends EventTarget + extends EventEmitter implements IWorkerNode { /** @inheritdoc */ public readonly worker: Worker @@ -36,28 +45,33 @@ 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 worker - The worker. - * @param tasksQueueBackPressureSize - The tasks queue back pressure size. + * @param type - The worker type. + * @param filePath - Path to the worker file. + * @param opts - The worker node options. */ - constructor (worker: Worker, tasksQueueBackPressureSize: number) { + constructor (type: WorkerType, filePath: string, opts: WorkerNodeOptions) { super() - checkWorkerNodeArguments(worker, tasksQueueBackPressureSize) - this.worker = worker - this.info = this.initWorkerInfo(worker) + checkWorkerNodeArguments(type, filePath, opts) + this.worker = createWorker(type, filePath, { + env: opts.env, + 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 = 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) + this.setBackPressureFlag = false this.taskFunctionsUsage = new Map() } @@ -68,42 +82,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.dispatchEvent( - new CustomEvent('backPressure', { - detail: { 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.dispatchEvent( - new CustomEvent('backPressure', { - detail: { 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 - } - - /** @inheritdoc */ - public dequeueTask (): Task | undefined { - return this.tasksQueue.shift() + return task } /** @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,39 +128,62 @@ export class WorkerNode } /** @inheritdoc */ - public resetUsage (): void { - this.usage = this.initWorkerUsage() - this.taskFunctionsUsage.clear() + public async terminate (): Promise { + const waitWorkerExit = new Promise(resolve => { + this.registerOnceWorkerEventHandler('exit', () => { + resolve() + }) + }) + this.closeMessageChannel() + this.removeAllListeners() + switch (this.info.type) { + case WorkerTypes.thread: + this.worker.unref?.() + await this.worker.terminate?.() + break + case WorkerTypes.cluster: + this.registerOnceWorkerEventHandler('disconnect', () => { + this.worker.kill?.() + }) + this.worker.disconnect?.() + break + } + await waitWorkerExit } /** @inheritdoc */ - public closeChannel (): void { - if (this.messageChannel != null) { - this.messageChannel.port1.unref() - this.messageChannel.port2.unref() - this.messageChannel.port1.close() - this.messageChannel.port2.close() - delete this.messageChannel - } + public registerWorkerEventHandler ( + event: string, + handler: EventHandler + ): void { + this.worker.on(event, handler) + } + + /** @inheritdoc */ + public registerOnceWorkerEventHandler ( + event: string, + 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)) @@ -162,12 +196,25 @@ export class WorkerNode return this.taskFunctionsUsage.delete(name) } + private closeMessageChannel (): void { + if (this.messageChannel != null) { + this.messageChannel.port1.unref() + this.messageChannel.port2.unref() + this.messageChannel.port1.close() + this.messageChannel.port2.close() + delete this.messageChannel + } + } + 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 + ready: false, + stealing: false, + backPressure: false } } @@ -193,17 +240,17 @@ export class WorkerNode 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) } } } @@ -215,7 +262,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 @@ -235,17 +283,17 @@ export class WorkerNode 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) } } }