X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=ebbefc75ba4cc8c4e09bcc3073e372085466c181;hb=refs%2Fheads%2Fmaster;hp=bc62ade55ef15efac79c8c9e8835a45cb697aa40;hpb=76b178d95efa0dc23addd65ab87e94736f1ed23a;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index bc62ade5..ebbefc75 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -1,27 +1,35 @@ -import { AsyncResource } from 'node:async_hooks' import type { Worker } from 'node:cluster' -import type { MessagePort } from 'node:worker_threads' import { performance } from 'node:perf_hooks' +import type { MessagePort } from 'node:worker_threads' + import type { MessageValue, Task, + TaskFunctionProperties, TaskPerformance, - WorkerStatistics -} from '../utility-types' + WorkerStatistics, +} from '../utility-types.js' import { + buildTaskFunctionProperties, DEFAULT_TASK_NAME, EMPTY_FUNCTION, isAsyncFunction, - isPlainObject -} from '../utils' -import { KillBehaviors, type WorkerOptions } from './worker-options' + isPlainObject, +} from '../utils.js' import type { TaskAsyncFunction, TaskFunction, - TaskFunctionOperationReturnType, + TaskFunctionObject, + TaskFunctionOperationResult, TaskFunctions, - TaskSyncFunction -} from './task-functions' + TaskSyncFunction, +} from './task-functions.js' +import { + checkTaskFunctionName, + checkValidTaskFunctionObjectEntry, + checkValidWorkerOptions, +} from './utils.js' +import { KillBehaviors, type WorkerOptions } from './worker-options.js' const DEFAULT_MAX_INACTIVE_TIME = 60000 const DEFAULT_WORKER_OPTIONS: WorkerOptions = { @@ -37,12 +45,11 @@ const DEFAULT_WORKER_OPTIONS: WorkerOptions = { /** * The function to call when the worker is killed. */ - killHandler: EMPTY_FUNCTION + killHandler: EMPTY_FUNCTION, } /** * Base class that implements some shared logic for all poolifier workers. - * * @typeParam MainWorker - Type of main worker. * @typeParam Data - Type of data this worker receives from pool's execution. This can only be structured-cloneable data. * @typeParam Response - Type of response the worker sends back to the main worker. This can only be structured-cloneable data. @@ -51,15 +58,15 @@ export abstract class AbstractWorker< MainWorker extends Worker | MessagePort, Data = unknown, Response = unknown -> extends AsyncResource { +> { /** * Worker id. */ protected abstract id: number /** - * Task function(s) processed by the worker when the pool's `execution` function is invoked. + * Task function object(s) processed by the worker when the pool's `execution` function is invoked. */ - protected taskFunctions!: Map> + protected taskFunctions!: Map> /** * Timestamp of the last task processed by this worker. */ @@ -67,123 +74,81 @@ export abstract class AbstractWorker< /** * Performance statistics computation requirements. */ - protected statistics!: WorkerStatistics + protected statistics?: WorkerStatistics /** * Handler id of the `activeInterval` worker activity check. */ protected activeInterval?: NodeJS.Timeout + /** * Constructs a new poolifier worker. - * - * @param type - The type of async event. * @param isMain - Whether this is the main worker or not. * @param mainWorker - Reference to main worker. * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked. The first function is the default function. * @param opts - Options for the worker. */ public constructor ( - type: string, - protected readonly isMain: boolean, - private readonly mainWorker: MainWorker, + protected readonly isMain: boolean | undefined, + private readonly mainWorker: MainWorker | undefined | null, taskFunctions: TaskFunction | TaskFunctions, protected opts: WorkerOptions = DEFAULT_WORKER_OPTIONS ) { - super(type) if (this.isMain == null) { throw new Error('isMain parameter is mandatory') } this.checkTaskFunctions(taskFunctions) this.checkWorkerOptions(this.opts) if (!this.isMain) { + // Should be once() but Node.js on windows has a bug that prevents it from working this.getMainWorker().on('message', this.handleReadyMessage.bind(this)) } } private checkWorkerOptions (opts: WorkerOptions): void { - if (opts != null && !isPlainObject(opts)) { - throw new TypeError('opts worker options parameter is not a plain object') - } - if ( - opts?.killBehavior != null && - !Object.values(KillBehaviors).includes(opts.killBehavior) - ) { - throw new TypeError( - `killBehavior option '${opts.killBehavior}' is not valid` - ) - } - if ( - opts?.maxInactiveTime != null && - !Number.isSafeInteger(opts.maxInactiveTime) - ) { - throw new TypeError('maxInactiveTime option is not an integer') - } - if (opts?.maxInactiveTime != null && opts.maxInactiveTime < 5) { - throw new TypeError( - 'maxInactiveTime option is not a positive integer greater or equal than 5' - ) - } - if (opts?.killHandler != null && typeof opts.killHandler !== 'function') { - throw new TypeError('killHandler option is not a function') - } - if (opts?.async != null) { - throw new Error('async option is deprecated') - } + checkValidWorkerOptions(opts) this.opts = { ...DEFAULT_WORKER_OPTIONS, ...opts } } - private checkValidTaskFunctionEntry ( - name: string, - fn: TaskFunction - ): void { - if (typeof name !== 'string') { - throw new TypeError( - 'A taskFunctions parameter object key is not a string' - ) - } - if (typeof name === 'string' && name.trim().length === 0) { - throw new TypeError( - 'A taskFunctions parameter object key is an empty string' - ) - } - if (typeof fn !== 'function') { - throw new TypeError( - 'A taskFunctions parameter object value is not a function' - ) - } - } - /** * Checks if the `taskFunctions` parameter is passed to the constructor and valid. - * * @param taskFunctions - The task function(s) parameter that should be checked. */ private checkTaskFunctions ( - taskFunctions: TaskFunction | TaskFunctions + taskFunctions: + | TaskFunction + | TaskFunctions + | undefined ): void { if (taskFunctions == null) { throw new Error('taskFunctions parameter is mandatory') } - this.taskFunctions = new Map>() + this.taskFunctions = new Map>() if (typeof taskFunctions === 'function') { - const boundFn = taskFunctions.bind(this) - this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) + const fnObj = { taskFunction: taskFunctions.bind(this) } + this.taskFunctions.set(DEFAULT_TASK_NAME, fnObj) this.taskFunctions.set( typeof taskFunctions.name === 'string' && - taskFunctions.name.trim().length > 0 + taskFunctions.name.trim().length > 0 ? taskFunctions.name : 'fn1', - boundFn + fnObj ) } else if (isPlainObject(taskFunctions)) { let firstEntry = true - for (const [name, fn] of Object.entries(taskFunctions)) { - this.checkValidTaskFunctionEntry(name, fn) - const boundFn = fn.bind(this) + for (let [name, fnObj] of Object.entries(taskFunctions)) { + if (typeof fnObj === 'function') { + fnObj = { taskFunction: fnObj } satisfies TaskFunctionObject< + Data, + Response + > + } + checkValidTaskFunctionObjectEntry(name, fnObj) + fnObj.taskFunction = fnObj.taskFunction.bind(this) if (firstEntry) { - this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) + this.taskFunctions.set(DEFAULT_TASK_NAME, fnObj) firstEntry = false } - this.taskFunctions.set(name, boundFn) + this.taskFunctions.set(name, fnObj) } if (firstEntry) { throw new Error('taskFunctions parameter object is empty') @@ -197,13 +162,12 @@ export abstract class AbstractWorker< /** * Checks if the worker has a task function with the given name. - * * @param name - The name of the task function to check. * @returns Whether the worker has a task function with the given name or not. */ - public hasTaskFunction (name: string): TaskFunctionOperationReturnType { + public hasTaskFunction (name: string): TaskFunctionOperationResult { try { - this.checkTaskFunctionName(name) + checkTaskFunctionName(name) } catch (error) { return { status: false, error: error as Error } } @@ -213,34 +177,34 @@ export abstract class AbstractWorker< /** * Adds a task function to the worker. * If a task function with the same name already exists, it is replaced. - * * @param name - The name of the task function to add. * @param fn - The task function to add. * @returns Whether the task function was added or not. */ public addTaskFunction ( name: string, - fn: TaskFunction - ): TaskFunctionOperationReturnType { + fn: TaskFunction | TaskFunctionObject + ): TaskFunctionOperationResult { try { - this.checkTaskFunctionName(name) + checkTaskFunctionName(name) if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot add a task function with the default reserved name' ) } - if (typeof fn !== 'function') { - throw new TypeError('fn parameter is not a function') + if (typeof fn === 'function') { + fn = { taskFunction: fn } satisfies TaskFunctionObject } - const boundFn = fn.bind(this) + checkValidTaskFunctionObjectEntry(name, fn) + fn.taskFunction = fn.taskFunction.bind(this) if ( this.taskFunctions.get(name) === this.taskFunctions.get(DEFAULT_TASK_NAME) ) { - this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) + this.taskFunctions.set(DEFAULT_TASK_NAME, fn) } - this.taskFunctions.set(name, boundFn) - this.sendTaskFunctionNamesToMainWorker() + this.taskFunctions.set(name, fn) + this.sendTaskFunctionsPropertiesToMainWorker() return { status: true } } catch (error) { return { status: false, error: error as Error } @@ -249,13 +213,12 @@ export abstract class AbstractWorker< /** * Removes a task function from the worker. - * * @param name - The name of the task function to remove. * @returns Whether the task function existed and was removed or not. */ - public removeTaskFunction (name: string): TaskFunctionOperationReturnType { + public removeTaskFunction (name: string): TaskFunctionOperationResult { try { - this.checkTaskFunctionName(name) + checkTaskFunctionName(name) if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot remove the task function with the default reserved name' @@ -270,7 +233,7 @@ export abstract class AbstractWorker< ) } const deleteStatus = this.taskFunctions.delete(name) - this.sendTaskFunctionNamesToMainWorker() + this.sendTaskFunctionsPropertiesToMainWorker() return { status: deleteStatus } } catch (error) { return { status: false, error: error as Error } @@ -278,40 +241,48 @@ export abstract class AbstractWorker< } /** - * Lists the names of the worker's task functions. - * - * @returns The names of the worker's task functions. + * Lists the properties of the worker's task functions. + * @returns The properties of the worker's task functions. */ - public listTaskFunctionNames (): string[] { - const names: string[] = [...this.taskFunctions.keys()] - let defaultTaskFunctionName: string = DEFAULT_TASK_NAME - for (const [name, fn] of this.taskFunctions) { + public listTaskFunctionsProperties (): TaskFunctionProperties[] { + let defaultTaskFunctionName = DEFAULT_TASK_NAME + for (const [name, fnObj] of this.taskFunctions) { if ( name !== DEFAULT_TASK_NAME && - fn === this.taskFunctions.get(DEFAULT_TASK_NAME) + fnObj === this.taskFunctions.get(DEFAULT_TASK_NAME) ) { defaultTaskFunctionName = name break } } + const taskFunctionsProperties: TaskFunctionProperties[] = [] + for (const [name, fnObj] of this.taskFunctions) { + if (name === DEFAULT_TASK_NAME || name === defaultTaskFunctionName) { + continue + } + taskFunctionsProperties.push(buildTaskFunctionProperties(name, fnObj)) + } return [ - names[names.indexOf(DEFAULT_TASK_NAME)], - defaultTaskFunctionName, - ...names.filter( - name => name !== DEFAULT_TASK_NAME && name !== defaultTaskFunctionName - ) + buildTaskFunctionProperties( + DEFAULT_TASK_NAME, + this.taskFunctions.get(DEFAULT_TASK_NAME) + ), + buildTaskFunctionProperties( + defaultTaskFunctionName, + this.taskFunctions.get(defaultTaskFunctionName) + ), + ...taskFunctionsProperties, ] } /** * Sets the default task function to use in the worker. - * * @param name - The name of the task function to use as default task function. * @returns Whether the default task function was set or not. */ - public setDefaultTaskFunction (name: string): TaskFunctionOperationReturnType { + public setDefaultTaskFunction (name: string): TaskFunctionOperationResult { try { - this.checkTaskFunctionName(name) + checkTaskFunctionName(name) if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot set the default task function reserved name as the default task function' @@ -322,52 +293,48 @@ export abstract class AbstractWorker< 'Cannot set the default task function to a non-existing task function' ) } - this.taskFunctions.set( - DEFAULT_TASK_NAME, - this.taskFunctions.get(name) as TaskFunction - ) + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + this.taskFunctions.set(DEFAULT_TASK_NAME, this.taskFunctions.get(name)!) + this.sendTaskFunctionsPropertiesToMainWorker() return { status: true } } catch (error) { return { status: false, error: error as Error } } } - private checkTaskFunctionName (name: string): void { - if (typeof name !== 'string') { - throw new TypeError('name parameter is not a string') - } - if (typeof name === 'string' && name.trim().length === 0) { - throw new TypeError('name parameter is an empty string') - } - } - /** * Handles the ready message sent by the main worker. - * * @param message - The ready message. */ protected abstract handleReadyMessage (message: MessageValue): void /** * Worker message listener. - * * @param message - The received message. */ protected messageListener (message: MessageValue): void { this.checkMessageWorkerId(message) - if (message.statistics != null) { + const { + statistics, + checkActive, + taskFunctionOperation, + taskId, + data, + kill, + } = message + if (statistics != null) { // Statistics message received - this.statistics = message.statistics - } else if (message.checkActive != null) { + this.statistics = statistics + } else if (checkActive != null) { // Check active message received - message.checkActive ? this.startCheckActive() : this.stopCheckActive() - } else if (message.taskFunctionOperation != null) { + checkActive ? this.startCheckActive() : this.stopCheckActive() + } else if (taskFunctionOperation != null) { // Task function operation message received this.handleTaskFunctionOperationMessage(message) - } else if (message.taskId != null && message.data != null) { + } else if (taskId != null && data != null) { // Task message received this.run(message) - } else if (message.kill === true) { + } else if (kill === true) { // Kill message received this.handleKillMessage(message) } @@ -376,77 +343,91 @@ export abstract class AbstractWorker< protected handleTaskFunctionOperationMessage ( message: MessageValue ): void { - const { taskFunctionOperation, taskFunction, taskFunctionName } = message - let response!: TaskFunctionOperationReturnType - if (taskFunctionOperation === 'add') { - response = this.addTaskFunction( - taskFunctionName as string, - // eslint-disable-next-line @typescript-eslint/no-implied-eval, no-new-func - new Function(`return ${taskFunction as string}`)() as TaskFunction< - Data, - Response - > + const { taskFunctionOperation, taskFunctionProperties, taskFunction } = + message + if (taskFunctionProperties == null) { + throw new Error( + 'Cannot handle task function operation message without task function properties' ) - } else if (taskFunctionOperation === 'remove') { - response = this.removeTaskFunction(taskFunctionName as string) - } else if (taskFunctionOperation === 'default') { - response = this.setDefaultTaskFunction(taskFunctionName as string) } + let response: TaskFunctionOperationResult + switch (taskFunctionOperation) { + case 'add': + response = this.addTaskFunction(taskFunctionProperties.name, { + // eslint-disable-next-line @typescript-eslint/no-implied-eval, no-new-func + taskFunction: new Function( + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + `return ${taskFunction!}` + )() as TaskFunction, + ...(taskFunctionProperties.priority != null && { + priority: taskFunctionProperties.priority, + }), + ...(taskFunctionProperties.strategy != null && { + strategy: taskFunctionProperties.strategy, + }), + }) + break + case 'remove': + response = this.removeTaskFunction(taskFunctionProperties.name) + break + case 'default': + response = this.setDefaultTaskFunction(taskFunctionProperties.name) + break + default: + response = { status: false, error: new Error('Unknown task operation') } + break + } + const { status, error } = response this.sendToMainWorker({ taskFunctionOperation, - taskFunctionOperationStatus: response.status, - workerError: { - name: taskFunctionName as string, - message: this.handleError(response.error as Error | string) - } + taskFunctionOperationStatus: status, + taskFunctionProperties, + ...(!status && + error != null && { + workerError: { + name: taskFunctionProperties.name, + message: this.handleError(error as Error | string), + }, + }), }) } /** * Handles a kill message sent by the main worker. - * * @param message - The kill message. */ protected handleKillMessage (message: MessageValue): void { this.stopCheckActive() if (isAsyncFunction(this.opts.killHandler)) { - (this.opts.killHandler?.() as Promise) + ;(this.opts.killHandler as () => Promise)() .then(() => { this.sendToMainWorker({ kill: 'success' }) - return null + return undefined }) .catch(() => { this.sendToMainWorker({ kill: 'failure' }) }) - .finally(() => { - this.emitDestroy() - }) - .catch(EMPTY_FUNCTION) } else { try { - // eslint-disable-next-line @typescript-eslint/no-invalid-void-type - this.opts.killHandler?.() as void + ;(this.opts.killHandler as (() => void) | undefined)?.() this.sendToMainWorker({ kill: 'success' }) } catch { this.sendToMainWorker({ kill: 'failure' }) - } finally { - this.emitDestroy() } } } /** * Check if the message worker id is set and matches the worker id. - * * @param message - The message to check. * @throws {@link https://nodejs.org/api/errors.html#class-error} If the message worker id is not set or does not match the worker id. */ private checkMessageWorkerId (message: MessageValue): void { if (message.workerId == null) { throw new Error('Message worker id is not set') - } else if (message.workerId != null && message.workerId !== this.id) { + } else if (message.workerId !== this.id) { throw new Error( - `Message worker id ${message.workerId} does not match the worker id ${this.id}` + `Message worker id ${message.workerId.toString()} does not match the worker id ${this.id.toString()}` ) } } @@ -486,7 +467,6 @@ export abstract class AbstractWorker< /** * Returns the main worker. - * * @returns Reference to the main worker. * @throws {@link https://nodejs.org/api/errors.html#class-error} If the main worker is not set. */ @@ -499,7 +479,6 @@ export abstract class AbstractWorker< /** * Sends a message to main worker. - * * @param message - The response message. */ protected abstract sendToMainWorker ( @@ -507,17 +486,16 @@ export abstract class AbstractWorker< ): void /** - * Sends task function names to the main worker. + * Sends task functions properties to the main worker. */ - protected sendTaskFunctionNamesToMainWorker (): void { + protected sendTaskFunctionsPropertiesToMainWorker (): void { this.sendToMainWorker({ - taskFunctionNames: this.listTaskFunctionNames() + taskFunctionsProperties: this.listTaskFunctionsProperties(), }) } /** * Handles an error and convert it to a string so it can be sent back to the main worker. - * * @param error - The error raised by the worker. * @returns The error message. */ @@ -527,41 +505,41 @@ export abstract class AbstractWorker< /** * Runs the given task. - * * @param task - The task to execute. - * @throws {@link https://nodejs.org/api/errors.html#class-error} If the task function is not found. */ - protected run (task: Task): void { + protected readonly run = (task: Task): void => { const { name, taskId, data } = task - const fn = this.taskFunctions.get(name ?? DEFAULT_TASK_NAME) - if (fn == null) { + const taskFunctionName = name ?? DEFAULT_TASK_NAME + if (!this.taskFunctions.has(taskFunctionName)) { this.sendToMainWorker({ workerError: { - name: name as string, - message: `Task function '${name as string}' not found`, - data + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + name: name!, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + message: `Task function '${name!}' not found`, + data, }, - taskId + taskId, }) return } + const fn = this.taskFunctions.get(taskFunctionName)?.taskFunction if (isAsyncFunction(fn)) { - this.runInAsyncScope(this.runAsync.bind(this), this, fn, task) + this.runAsync(fn as TaskAsyncFunction, task) } else { - this.runInAsyncScope(this.runSync.bind(this), this, fn, task) + this.runSync(fn as TaskSyncFunction, task) } } /** * Runs the given task function synchronously. - * * @param fn - Task function that will be executed. * @param task - Input data for the task function. */ - protected runSync ( + protected readonly runSync = ( fn: TaskSyncFunction, task: Task - ): void { + ): void => { const { name, taskId, data } = task try { let taskPerformance = this.beginTaskPerformance(name) @@ -570,16 +548,17 @@ export abstract class AbstractWorker< this.sendToMainWorker({ data: res, taskPerformance, - taskId + taskId, }) } catch (error) { this.sendToMainWorker({ workerError: { - name: name as string, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + name: name!, message: this.handleError(error as Error | string), - data + data, }, - taskId + taskId, }) } finally { this.updateLastTaskTimestamp() @@ -588,14 +567,13 @@ export abstract class AbstractWorker< /** * Runs the given task function asynchronously. - * * @param fn - Task function that will be executed. * @param task - Input data for the task function. */ - protected runAsync ( + protected readonly runAsync = ( fn: TaskAsyncFunction, task: Task - ): void { + ): void => { const { name, taskId, data } = task let taskPerformance = this.beginTaskPerformance(name) fn(data) @@ -604,18 +582,19 @@ export abstract class AbstractWorker< this.sendToMainWorker({ data: res, taskPerformance, - taskId + taskId, }) - return null + return undefined }) - .catch(error => { + .catch((error: unknown) => { this.sendToMainWorker({ workerError: { - name: name as string, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + name: name!, message: this.handleError(error as Error | string), - data + data, }, - taskId + taskId, }) }) .finally(() => { @@ -625,32 +604,32 @@ export abstract class AbstractWorker< } private beginTaskPerformance (name?: string): TaskPerformance { - this.checkStatistics() + if (this.statistics == null) { + throw new Error('Performance statistics computation requirements not set') + } return { name: name ?? DEFAULT_TASK_NAME, timestamp: performance.now(), - ...(this.statistics.elu && { elu: performance.eventLoopUtilization() }) + ...(this.statistics.elu && { + elu: performance.eventLoopUtilization(), + }), } } private endTaskPerformance ( taskPerformance: TaskPerformance ): TaskPerformance { - this.checkStatistics() + if (this.statistics == null) { + throw new Error('Performance statistics computation requirements not set') + } return { ...taskPerformance, ...(this.statistics.runTime && { - runTime: performance.now() - taskPerformance.timestamp + runTime: performance.now() - taskPerformance.timestamp, }), ...(this.statistics.elu && { - elu: performance.eventLoopUtilization(taskPerformance.elu) - }) - } - } - - private checkStatistics (): void { - if (this.statistics == null) { - throw new Error('Performance statistics computation requirements not set') + elu: performance.eventLoopUtilization(taskPerformance.elu), + }), } }