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
+ WorkerStatistics,
} from '../utility-types.js'
import {
+ buildTaskFunctionProperties,
DEFAULT_TASK_NAME,
EMPTY_FUNCTION,
isAsyncFunction,
- isPlainObject
+ isPlainObject,
} from '../utils.js'
-import { KillBehaviors, type WorkerOptions } from './worker-options.js'
import type {
TaskAsyncFunction,
TaskFunction,
+ TaskFunctionObject,
TaskFunctionOperationResult,
TaskFunctions,
- TaskSyncFunction
+ TaskSyncFunction,
} from './task-functions.js'
import {
checkTaskFunctionName,
- checkValidTaskFunctionEntry,
- checkValidWorkerOptions
+ checkValidTaskFunctionObjectEntry,
+ checkValidWorkerOptions,
} from './utils.js'
+import { KillBehaviors, type WorkerOptions } from './worker-options.js'
const DEFAULT_MAX_INACTIVE_TIME = 60000
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.
*/
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<string, TaskFunction<Data, Response>>
+ protected taskFunctions!: Map<string, TaskFunctionObject<Data, Response>>
/**
* Timestamp of the last task processed by this worker.
*/
/**
* Performance statistics computation requirements.
*/
- protected statistics!: WorkerStatistics
+ protected statistics?: WorkerStatistics
/**
* Handler id of the `activeInterval` worker activity check.
*/
/**
* 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<Data, Response> | TaskFunctions<Data, Response>,
protected opts: WorkerOptions = DEFAULT_WORKER_OPTIONS
) {
/**
* 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<Data, Response> | TaskFunctions<Data, Response>
+ taskFunctions:
+ | TaskFunction<Data, Response>
+ | TaskFunctions<Data, Response>
+ | undefined
): void {
if (taskFunctions == null) {
throw new Error('taskFunctions parameter is mandatory')
}
- this.taskFunctions = new Map<string, TaskFunction<Data, Response>>()
+ this.taskFunctions = new Map<string, TaskFunctionObject<Data, Response>>()
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<Data, Response>(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<Data, Response>(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')
/**
* 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.
*/
/**
* 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<Data, Response>
+ fn: TaskFunction<Data, Response> | TaskFunctionObject<Data, Response>
): TaskFunctionOperationResult {
try {
checkTaskFunctionName(name)
'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<Data, Response>
}
- const boundFn = fn.bind(this)
+ checkValidTaskFunctionObjectEntry<Data, Response>(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 }
/**
* 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.
*/
)
}
const deleteStatus = this.taskFunctions.delete(name)
- this.sendTaskFunctionNamesToMainWorker()
+ this.sendTaskFunctionsPropertiesToMainWorker()
return { status: deleteStatus }
} catch (error) {
return { status: false, error: error as Error }
}
/**
- * 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.
*/
'Cannot set the default task function to a non-existing task function'
)
}
- this.taskFunctions.set(
- DEFAULT_TASK_NAME,
- this.taskFunctions.get(name) as TaskFunction<Data, Response>
- )
- 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 }
/**
* Handles the ready message sent by the main worker.
- *
* @param message - The ready message.
*/
protected abstract handleReadyMessage (message: MessageValue<Data>): void
/**
* Worker message listener.
- *
* @param message - The received message.
*/
protected messageListener (message: MessageValue<Data>): 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)
}
protected handleTaskFunctionOperationMessage (
message: MessageValue<Data>
): 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<Data, Response>,
+ ...(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<Data>): void {
+ protected handleKillMessage (message: MessageValue<Data>): void {
this.stopCheckActive()
if (isAsyncFunction(this.opts.killHandler)) {
- (this.opts.killHandler?.() as Promise<void>)
+ ;(this.opts.killHandler as () => Promise<void>)()
.then(() => {
this.sendToMainWorker({ kill: 'success' })
return undefined
})
} 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' })
/**
* 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.
*/
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()}`
)
}
}
/**
* 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.
*/
/**
* Sends a message to main worker.
- *
* @param message - The response message.
*/
protected abstract sendToMainWorker (
): 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.
*/
/**
* Runs the given task.
- *
* @param task - The task to execute.
*/
protected readonly run = (task: Task<Data>): void => {
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<Data, Response>, task)
} else {
/**
* Runs the given task function synchronously.
- *
* @param fn - Task function that will be executed.
* @param task - Input data for the task function.
*/
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()
/**
* Runs the given task function asynchronously.
- *
* @param fn - Task function that will be executed.
* @param task - Input data for the task function.
*/
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(() => {
}
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),
+ }),
}
}
repositories / poolifier.git / blobdiff