X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker.ts;h=29050455088c6efd49d8147aed07c41f2a5a004e;hb=2a87030e2fa5e42167e2aa13bd8c1cbb3d27ffc1;hp=664b6a7695ef42a088732812a7eac270e9f87a00;hpb=53b1b2fc0c935c958a4b6a36223dbdc44c24d926;p=poolifier.git diff --git a/src/pools/worker.ts b/src/pools/worker.ts index 664b6a76..29050455 100644 --- a/src/pools/worker.ts +++ b/src/pools/worker.ts @@ -1,5 +1,11 @@ +import type { MessageChannel } from 'node:worker_threads' 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. + */ +export type OnlineHandler = (this: Worker) => void /** * Callback invoked if the worker has received a message. @@ -17,11 +23,6 @@ export type ErrorHandler = ( error: Error ) => void -/** - * Callback invoked when the worker has started successfully. - */ -export type OnlineHandler = (this: Worker) => void - /** * Callback invoked when the worker exits successfully. */ @@ -31,52 +32,35 @@ export type ExitHandler = ( ) => void /** - * Message object that is passed as a task between main worker and worker. + * Measurement statistics. * - * @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 +export interface MeasurementStatistics { /** - * Timestamp. + * Measurement aggregate. */ - readonly timestamp?: number + aggregate?: number /** - * Message UUID. + * Measurement minimum. */ - readonly id?: string -} - -/** - * Measurement statistics. - * - * @internal - */ -export interface MeasurementStatistics { + minimum?: number /** - * Measurement aggregate. + * Measurement maximum. */ - aggregate: number + maximum?: number /** * Measurement average. */ - average: number + average?: number /** * Measurement median. */ - median: number + median?: number /** * Measurement history. */ - history: CircularArray + readonly history: CircularArray } /** @@ -85,9 +69,9 @@ export interface MeasurementStatistics { * @internal */ export interface EventLoopUtilizationMeasurementStatistics { - idle: MeasurementStatistics - active: MeasurementStatistics - utilization: number + readonly idle: MeasurementStatistics + readonly active: MeasurementStatistics + utilization?: number } /** @@ -111,13 +95,30 @@ export interface TaskStatistics { /** * Maximum number of queued tasks. */ - readonly maxQueued: number + readonly maxQueued?: 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. * @@ -127,11 +128,23 @@ export interface WorkerInfo { /** * Worker id. */ - id: number | undefined + readonly id: number | undefined + /** + * Worker type. + */ + readonly type: WorkerType + /** + * Dynamic flag. + */ + dynamic: boolean /** - * Started flag. + * Ready flag. */ - started: boolean + ready: boolean + /** + * Task function names. + */ + taskFunctions?: string[] } /** @@ -143,19 +156,28 @@ export interface WorkerUsage { /** * Tasks statistics. */ - tasks: TaskStatistics + readonly tasks: TaskStatistics /** * Tasks runtime statistics. */ - runTime: MeasurementStatistics + readonly runTime: MeasurementStatistics /** * Tasks wait time statistics. */ - waitTime: MeasurementStatistics + readonly waitTime: MeasurementStatistics /** * Tasks event loop utilization statistics. */ - elu: EventLoopUtilizationMeasurementStatistics + readonly elu: EventLoopUtilizationMeasurementStatistics +} + +/** + * Worker choice strategy data. + * + * @internal + */ +export interface StrategyData { + virtualTaskEndTimestamp?: number } /** @@ -173,19 +195,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 callback. + * + * @param workerId - The worker id. + * @internal + */ +export type WorkerNodeEventCallback = (workerId: number) => void + /** * Worker node interface. * @@ -193,21 +223,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 { /** - * Worker node worker. + * Worker. */ readonly worker: Worker /** - * Worker node worker info. + * Worker info. + */ + readonly info: WorkerInfo + /** + * Worker usage statistics. + */ + readonly usage: WorkerUsage + /** + * Worker choice strategy data. + * This is used to store data that is specific to the worker choice strategy. + */ + strategyData?: StrategyData + /** + * Message channel (worker_threads only). */ - info: WorkerInfo + readonly messageChannel?: MessageChannel /** - * Worker node worker usage statistics. + * Tasks queue back pressure size. + * This is the number of tasks that can be enqueued before the worker node has back pressure. */ - usage: WorkerUsage + tasksQueueBackPressureSize: number /** - * Worker node tasks queue. + * Callback invoked when worker node tasks queue is back pressured. + */ + onBackPressure?: WorkerNodeEventCallback + /** + * Callback invoked when worker node tasks queue is empty. + */ + onEmptyQueue?: WorkerNodeEventCallback + /** + * 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. + */ + readonly clearTasksQueue: () => void + /** + * 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 tasksQueue: Queue> + readonly getTaskFunctionWorkerUsage: (name: string) => WorkerUsage | undefined }