X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker.ts;h=5439606d420f0abd165b84c49157808de230aee6;hb=fa548cda5120ac0708d82f37cd0ce1260d7f96c1;hp=7f9cef0a6632aa354f07eb04db5dea6d7ad38c4e;hpb=f751010550f07200428ca708afb2ec1d242cf7a7;p=poolifier.git diff --git a/src/pools/worker.ts b/src/pools/worker.ts index 7f9cef0a..5439606d 100644 --- a/src/pools/worker.ts +++ b/src/pools/worker.ts @@ -1,8 +1,19 @@ +import type { MessageChannel } from 'node:worker_threads' +import type { EventEmitter } from 'node:events' import type { CircularArray } from '../circular-array' -import type { Queue } from '../queue' +import type { Task } from '../utility-types' + +/** + * Callback invoked when the worker has started successfully. + * + * @typeParam Worker - Type of worker. + */ +export type OnlineHandler = (this: Worker) => void /** * Callback invoked if the worker has received a message. + * + * @typeParam Worker - Type of worker. */ export type MessageHandler = ( this: Worker, @@ -11,50 +22,24 @@ export type MessageHandler = ( /** * Callback invoked if the worker raised an error. + * + * @typeParam Worker - Type of worker. */ export type ErrorHandler = ( this: Worker, error: Error ) => void -/** - * Callback invoked when the worker has started successfully. - */ -export type OnlineHandler = (this: Worker) => void - /** * Callback invoked when the worker exits successfully. + * + * @typeParam Worker - Type of worker. */ export type ExitHandler = ( this: Worker, exitCode: number ) => void -/** - * Message object that is passed as a task between main worker and worker. - * - * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. - * @internal - */ -export interface Task { - /** - * Task name. - */ - readonly name?: string - /** - * Task input data that will be passed to the worker. - */ - readonly data?: Data - /** - * Timestamp. - */ - readonly timestamp?: number - /** - * Message UUID. - */ - readonly id?: string -} - /** * Measurement statistics. * @@ -64,23 +49,23 @@ export interface MeasurementStatistics { /** * Measurement aggregate. */ - aggregate: number + aggregate?: number /** * Measurement minimum. */ - minimum: number + minimum?: number /** * Measurement maximum. */ - maximum: number + maximum?: number /** * Measurement average. */ - average: number + average?: number /** * Measurement median. */ - median: number + median?: number /** * Measurement history. */ @@ -119,13 +104,34 @@ export interface TaskStatistics { /** * Maximum number of queued tasks. */ - readonly maxQueued: number + readonly maxQueued?: number + /** + * Number of sequentially stolen tasks. + */ + sequentiallyStolen: number + /** + * Number of stolen tasks. + */ + stolen: number /** * Number of failed tasks. */ failed: number } +/** + * Enumeration of worker types. + */ +export const WorkerTypes = Object.freeze({ + thread: 'thread', + cluster: 'cluster' +} as const) + +/** + * Worker type. + */ +export type WorkerType = keyof typeof WorkerTypes + /** * Worker information. * @@ -137,13 +143,21 @@ export interface WorkerInfo { */ readonly id: number | undefined /** - * Started flag. + * Worker type. + */ + readonly type: WorkerType + /** + * Dynamic flag. + */ + dynamic: boolean + /** + * Ready flag. */ - started: boolean + ready: boolean /** - * Shared buffer. + * Task function names. */ - readonly sharedBuffer?: Int32Array + taskFunctionNames?: string[] } /** @@ -170,6 +184,15 @@ export interface WorkerUsage { readonly elu: EventLoopUtilizationMeasurementStatistics } +/** + * Worker choice strategy data. + * + * @internal + */ +export interface StrategyData { + virtualTaskEndTimestamp?: number +} + /** * Worker interface. */ @@ -185,17 +208,27 @@ export interface IWorker { * @param event - The event. * @param handler - The event handler. */ - on: ((event: 'message', handler: MessageHandler) => void) & + readonly on: ((event: 'online', handler: OnlineHandler) => void) & + ((event: 'message', handler: MessageHandler) => void) & ((event: 'error', handler: ErrorHandler) => void) & - ((event: 'online', handler: OnlineHandler) => void) & ((event: 'exit', handler: ExitHandler) => void) /** * Registers a listener to the exit event that will only be performed once. * - * @param event - `'exit'`. + * @param event - The `'exit'` event. * @param handler - The exit handler. */ - once: (event: 'exit', handler: ExitHandler) => void + readonly once: (event: 'exit', handler: ExitHandler) => void +} + +/** + * Worker node event detail. + * + * @internal + */ +export interface WorkerNodeEventDetail { + workerId: number + workerNodeKey?: number } /** @@ -205,21 +238,96 @@ export interface IWorker { * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. * @internal */ -export interface WorkerNode { +export interface IWorkerNode + extends EventEmitter { /** - * Worker node worker. + * Worker. */ readonly worker: Worker /** - * Worker node worker info. + * Worker info. */ readonly info: WorkerInfo /** - * Worker node worker usage statistics. + * Worker usage statistics. + */ + readonly usage: WorkerUsage + /** + * Worker choice strategy data. + * This is used to store data that are specific to the worker choice strategy. + */ + strategyData?: StrategyData + /** + * Message channel (worker_threads only). + */ + readonly messageChannel?: MessageChannel + /** + * Tasks queue back pressure size. + * This is the number of tasks that can be enqueued before the worker node has back pressure. + */ + tasksQueueBackPressureSize: number + /** + * Tasks queue size. + * + * @returns The tasks queue size. + */ + readonly tasksQueueSize: () => number + /** + * Enqueue task. + * + * @param task - The task to queue. + * @returns The tasks queue size. + */ + readonly enqueueTask: (task: Task) => number + /** + * Prepends a task to the tasks queue. + * + * @param task - The task to prepend. + * @returns The tasks queue size. + */ + readonly unshiftTask: (task: Task) => number + /** + * Dequeue task. + * + * @returns The dequeued task. + */ + readonly dequeueTask: () => Task | undefined + /** + * Pops a task from the tasks queue. + * + * @returns The popped task. + */ + readonly popTask: () => Task | undefined + /** + * Clears tasks queue. */ - usage: WorkerUsage + readonly clearTasksQueue: () => void /** - * Worker node tasks queue. + * Whether the worker node has back pressure (i.e. its tasks queue is full). + * + * @returns `true` if the worker node has back pressure, `false` otherwise. + */ + readonly hasBackPressure: () => boolean + /** + * Resets usage statistics. + */ + readonly resetUsage: () => void + /** + * Closes communication channel. + */ + readonly closeChannel: () => void + /** + * Gets task function worker usage statistics. + * + * @param name - The task function name. + * @returns The task function worker usage statistics if the task function worker usage statistics are initialized, `undefined` otherwise. + */ + readonly getTaskFunctionWorkerUsage: (name: string) => WorkerUsage | undefined + /** + * Deletes task function worker usage statistics. + * + * @param name - The task function name. + * @returns `true` if the task function worker usage statistics were deleted, `false` otherwise. */ - readonly tasksQueue: Queue> + readonly deleteTaskFunctionWorkerUsage: (name: string) => boolean }