X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=ebbefc75ba4cc8c4e09bcc3073e372085466c181;hb=refs%2Fheads%2Fmaster;hp=dfcadf280f689c01cdb4fb9cb583d8138fe19135;hpb=b73887071ac359b50a7f117a76b3edf268841aac;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index dfcadf28..ebbefc75 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -1,31 +1,35 @@ 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, + TaskFunctionObject, TaskFunctionOperationResult, TaskFunctions, - TaskSyncFunction -} from './task-functions' + TaskSyncFunction, +} from './task-functions.js' import { checkTaskFunctionName, - checkValidTaskFunctionEntry, - checkValidWorkerOptions -} from './utils' + checkValidTaskFunctionObjectEntry, + checkValidWorkerOptions, +} from './utils.js' +import { KillBehaviors, type WorkerOptions } from './worker-options.js' const DEFAULT_MAX_INACTIVE_TIME = 60000 const DEFAULT_WORKER_OPTIONS: WorkerOptions = { @@ -41,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. @@ -61,9 +64,9 @@ export abstract class AbstractWorker< */ 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. */ @@ -71,22 +74,22 @@ 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 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 ( - 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 ) { @@ -108,36 +111,44 @@ export abstract class AbstractWorker< /** * 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)) { - 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') @@ -151,7 +162,6 @@ 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. */ @@ -167,14 +177,13 @@ 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 + fn: TaskFunction | TaskFunctionObject ): TaskFunctionOperationResult { try { checkTaskFunctionName(name) @@ -183,18 +192,19 @@ export abstract class AbstractWorker< '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 } @@ -203,7 +213,6 @@ 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. */ @@ -224,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 } @@ -232,34 +241,42 @@ 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. */ @@ -276,11 +293,9 @@ 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 - ) - this.sendTaskFunctionNamesToMainWorker() + // 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 } @@ -289,31 +304,37 @@ export abstract class AbstractWorker< /** * 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) } @@ -322,52 +343,63 @@ export abstract class AbstractWorker< protected handleTaskFunctionOperationMessage ( message: MessageValue ): void { - const { taskFunctionOperation, taskFunctionName, taskFunction } = message - let response!: TaskFunctionOperationResult + const { taskFunctionOperation, taskFunctionProperties, taskFunction } = + message + if (taskFunctionProperties == null) { + throw new Error( + 'Cannot handle task function operation message without task function properties' + ) + } + let response: TaskFunctionOperationResult switch (taskFunctionOperation) { case 'add': - response = this.addTaskFunction( - taskFunctionName as string, + response = this.addTaskFunction(taskFunctionProperties.name, { // eslint-disable-next-line @typescript-eslint/no-implied-eval, no-new-func - new Function(`return ${taskFunction as string}`)() as TaskFunction< - Data, - Response - > - ) + 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(taskFunctionName as string) + response = this.removeTaskFunction(taskFunctionProperties.name) break case 'default': - response = this.setDefaultTaskFunction(taskFunctionName as string) + 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, - taskFunctionName, - ...(!response.status && - response?.error != null && { + taskFunctionOperationStatus: status, + taskFunctionProperties, + ...(!status && + error != null && { workerError: { - name: taskFunctionName as string, - message: this.handleError(response.error as Error | string) - } - }) + 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 { + 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 undefined @@ -377,8 +409,7 @@ export abstract class AbstractWorker< }) } 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' }) @@ -388,7 +419,6 @@ export abstract class AbstractWorker< /** * 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. */ @@ -397,7 +427,7 @@ export abstract class AbstractWorker< throw new Error('Message worker id is not set') } 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()}` ) } } @@ -437,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. */ @@ -450,7 +479,6 @@ export abstract class AbstractWorker< /** * Sends a message to main worker. - * * @param message - The response message. */ protected abstract sendToMainWorker ( @@ -458,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. */ @@ -478,7 +505,6 @@ export abstract class AbstractWorker< /** * Runs the given task. - * * @param task - The task to execute. */ protected readonly run = (task: Task): void => { @@ -487,15 +513,17 @@ export abstract class AbstractWorker< 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) + const fn = this.taskFunctions.get(taskFunctionName)?.taskFunction if (isAsyncFunction(fn)) { this.runAsync(fn as TaskAsyncFunction, task) } else { @@ -505,7 +533,6 @@ export abstract class AbstractWorker< /** * Runs the given task function synchronously. - * * @param fn - Task function that will be executed. * @param task - Input data for the task function. */ @@ -521,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() @@ -539,7 +567,6 @@ 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. */ @@ -555,18 +582,19 @@ export abstract class AbstractWorker< this.sendToMainWorker({ data: res, taskPerformance, - taskId + taskId, }) 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(() => { @@ -576,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), + }), } }