X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker.ts;h=9c8b77fc623b68ed2a25aed883531dfa5e25cf70;hb=5eb72b9e26eaffb43c67147fbc6b4d2b1b959d62;hp=3c44f05eae148883c5c83f91304f894d2297e5fc;hpb=2809112e53951451aeb76e86a2dcc263ba80e43c;p=poolifier.git diff --git a/src/pools/worker.ts b/src/pools/worker.ts index 3c44f05e..9c8b77fc 100644 --- a/src/pools/worker.ts +++ b/src/pools/worker.ts @@ -1,14 +1,19 @@ -import type { MessageChannel } from 'node:worker_threads' +import type { MessageChannel, WorkerOptions } from 'node:worker_threads' +import type { EventEmitter } from 'node:events' import type { CircularArray } from '../circular-array' 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, @@ -17,6 +22,8 @@ export type MessageHandler = ( /** * Callback invoked if the worker raised an error. + * + * @typeParam Worker - Type of worker. */ export type ErrorHandler = ( this: Worker, @@ -25,6 +32,8 @@ export type ErrorHandler = ( /** * Callback invoked when the worker exits successfully. + * + * @typeParam Worker - Type of worker. */ export type ExitHandler = ( this: Worker, @@ -96,6 +105,14 @@ export interface TaskStatistics { * Maximum number of queued tasks. */ readonly maxQueued?: number + /** + * Number of sequentially stolen tasks. + */ + sequentiallyStolen: number + /** + * Number of stolen tasks. + */ + stolen: number /** * Number of failed tasks. */ @@ -128,7 +145,7 @@ export interface WorkerInfo { /** * Worker type. */ - type: WorkerType + readonly type: WorkerType /** * Dynamic flag. */ @@ -137,10 +154,15 @@ export interface WorkerInfo { * Ready flag. */ ready: boolean + /** + * Stealing flag. + * This flag is set to `true` when worker node is stealing tasks from another worker node. + */ + stealing: boolean /** * Task function names. */ - taskFunctions?: string[] + taskFunctionNames?: string[] } /** @@ -167,32 +189,79 @@ export interface WorkerUsage { readonly elu: EventLoopUtilizationMeasurementStatistics } +/** + * Worker choice strategy data. + * + * @internal + */ +export interface StrategyData { + virtualTaskEndTimestamp?: number +} + /** * Worker interface. */ export interface IWorker { /** - * Worker id. + * Cluster worker id. */ readonly id?: number + /** + * Worker thread worker id. + */ readonly threadId?: number /** - * Registers an event listener. + * Registers an event handler. * * @param event - The event. * @param handler - The event handler. */ - readonly on: ((event: 'online', handler: OnlineHandler) => void) & - ((event: 'message', handler: MessageHandler) => void) & - ((event: 'error', handler: ErrorHandler) => void) & - ((event: 'exit', handler: ExitHandler) => void) + readonly on: ( + event: string, + handler: + | OnlineHandler + | MessageHandler + | ErrorHandler + | ExitHandler + ) => void /** - * Registers a listener to the exit event that will only be performed once. + * Registers once an event handler. * - * @param event - `'exit'`. - * @param handler - The exit handler. + * @param event - The event. + * @param handler - The event handler. + */ + readonly once: ( + event: string, + handler: + | OnlineHandler + | MessageHandler + | ErrorHandler + | ExitHandler + ) => void + /** + * Stop all JavaScript execution in the worker thread as soon as possible. + * Returns a Promise for the exit code that is fulfilled when the `'exit' event` is emitted. */ - readonly once: (event: 'exit', handler: ExitHandler) => void + readonly terminate?: () => Promise + /** + * Cluster worker disconnect. + */ + readonly disconnect?: () => void + /** + * Cluster worker kill. + */ + readonly kill?: (signal?: string) => void +} + +/** + * Worker node options. + * + * @internal + */ +export interface WorkerNodeOptions { + workerOptions?: WorkerOptions + env?: Record + tasksQueueBackPressureSize: number } /** @@ -202,7 +271,8 @@ export interface IWorker { * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. * @internal */ -export interface IWorkerNode { +export interface IWorkerNode + extends EventEmitter { /** * Worker. */ @@ -212,13 +282,23 @@ export interface IWorkerNode { */ readonly info: WorkerInfo /** - * Message channel. + * 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 thread only). */ readonly messageChannel?: MessageChannel /** - * 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 /** * Tasks queue size. * @@ -229,38 +309,96 @@ export interface IWorkerNode { * Enqueue task. * * @param task - The task to queue. - * @returns The task queue size. + * @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. + * 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 . + * Resets usage statistics. */ readonly resetUsage: () => void /** - * Closes communication channel. + * Terminates the worker node. */ - readonly closeChannel: () => void + readonly terminate: () => Promise + /** + * Registers a worker event handler. + * + * @param event - The event. + * @param handler - The event handler. + */ + readonly registerWorkerEventHandler: ( + event: string, + handler: + | OnlineHandler + | MessageHandler + | ErrorHandler + | ExitHandler + ) => void + /** + * Registers once a worker event handler. + * + * @param event - The event. + * @param handler - The event handler. + */ + readonly registerOnceWorkerEventHandler: ( + event: string, + handler: + | OnlineHandler + | MessageHandler + | ErrorHandler + | ExitHandler + ) => 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 getTaskWorkerUsage: (name: string) => WorkerUsage | undefined + 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 deleteTaskFunctionWorkerUsage: (name: string) => boolean +} + +/** + * Worker node event detail. + * + * @internal + */ +export interface WorkerNodeEventDetail { + workerId: number + workerNodeKey?: number }