X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fworker%2Fabstract-worker.ts;h=b16c886b26c933c77aeee5ff32a3c9a61272aab3;hb=79e68dbfc6bfa1de92849134b10bb6e909fab422;hp=fcdd1bf283971955b8cdb9cc6482856b73516a48;hpb=968a2e8ca7bfc81d61b5ffc820a6a76bf44b202f;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index fcdd1bf2..b16c886b 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -4,6 +4,7 @@ import type { MessagePort } from 'node:worker_threads' import { performance } from 'node:perf_hooks' import type { MessageValue, + Task, TaskPerformance, WorkerStatistics } from '../utility-types' @@ -57,9 +58,9 @@ export abstract class AbstractWorker< */ protected statistics!: WorkerStatistics /** - * Handler id of the `aliveInterval` worker alive check. + * Handler id of the `activeInterval` worker activity check. */ - protected aliveInterval?: NodeJS.Timeout + protected activeInterval?: NodeJS.Timeout /** * Constructs a new poolifier worker. * @@ -82,7 +83,7 @@ export abstract class AbstractWorker< */ killBehavior: DEFAULT_KILL_BEHAVIOR, /** - * The maximum time to keep this worker alive while idle. + * The maximum time to keep this worker active while idle. * The pool automatically checks and terminates this worker when the time expires. */ maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME @@ -122,6 +123,11 @@ export abstract class AbstractWorker< } else if (isPlainObject(taskFunctions)) { let firstEntry = true for (const [name, fn] of Object.entries(taskFunctions)) { + if (typeof name !== 'string') { + throw new TypeError( + 'A taskFunctions parameter object key is not a string' + ) + } if (typeof fn !== 'function') { throw new TypeError( 'A taskFunctions parameter object value is not a function' @@ -148,7 +154,7 @@ export abstract class AbstractWorker< * * @param name - The name of the task function to check. * @returns Whether the worker has a task function with the given name or not. - * @throws {@link TypeError} If the `name` parameter is not a string. + * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string. */ public hasTaskFunction (name: string): boolean { if (typeof name !== 'string') { @@ -164,9 +170,9 @@ export abstract class AbstractWorker< * @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. - * @throws {@link TypeError} If the `name` parameter is not a string. - * @throws {@link Error} If the `name` parameter is the default task function reserved name. - * @throws {@link TypeError} If the `fn` parameter is not a function. + * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string. + * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is the default task function reserved name. + * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `fn` parameter is not a function. */ public addTaskFunction ( name: string, @@ -202,9 +208,9 @@ export abstract class AbstractWorker< * * @param name - The name of the task function to remove. * @returns Whether the task function existed and was removed or not. - * @throws {@link TypeError} If the `name` parameter is not a string. - * @throws {@link Error} If the `name` parameter is the default task function reserved name. - * @throws {@link Error} If the `name` parameter is the task function used as default task function. + * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string. + * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is the default task function reserved name. + * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is the task function used as default task function. */ public removeTaskFunction (name: string): boolean { if (typeof name !== 'string') { @@ -230,9 +236,9 @@ export abstract class AbstractWorker< * * @param name - The name of the task function to use as default task function. * @returns Whether the default task function was set or not. - * @throws {@link TypeError} If the `name` parameter is not a string. - * @throws {@link Error} If the `name` parameter is the default task function reserved name. - * @throws {@link Error} If the `name` parameter is a non-existing task function. + * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string. + * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is the default task function reserved name. + * @throws {@link https://nodejs.org/api/errors.html#class-error} If the `name` parameter is a non-existing task function. */ public setDefaultTaskFunction (name: string): boolean { if (typeof name !== 'string') { @@ -265,65 +271,59 @@ export abstract class AbstractWorker< /** * Worker message listener. * - * @param message - Message received. + * @param message - The received message. */ protected messageListener (message: MessageValue): void { if (message.workerId === this.id) { if (message.ready != null) { // Startup message received - this.workerReady() + this.sendReadyResponse() } else if (message.statistics != null) { // Statistics message received this.statistics = message.statistics - } else if (message.checkAlive != null) { - // Check alive message received - message.checkAlive ? this.startCheckAlive() : this.stopCheckAlive() + } else if (message.checkActive != null) { + // Check active message received + message.checkActive ? this.startCheckActive() : this.stopCheckActive() } else if (message.id != null && message.data != null) { // Task message received - const fn = this.getTaskFunction(message.name) - if (isAsyncFunction(fn)) { - this.runInAsyncScope(this.runAsync.bind(this), this, fn, message) - } else { - this.runInAsyncScope(this.runSync.bind(this), this, fn, message) - } + this.run(message) } else if (message.kill === true) { // Kill message received - this.stopCheckAlive() + this.stopCheckActive() this.emitDestroy() } } } /** - * Notifies the main worker that this worker is ready to process tasks. + * Sends to the main worker the ready response. */ - protected workerReady (): void { + protected sendReadyResponse (): void { !this.isMain && this.sendToMainWorker({ ready: true, workerId: this.id }) } /** - * Starts the worker alive check interval. + * Starts the worker check active interval. */ - private startCheckAlive (): void { + private startCheckActive (): void { this.lastTaskTimestamp = performance.now() - this.aliveInterval = setInterval( - this.checkAlive.bind(this), + this.activeInterval = setInterval( + this.checkActive.bind(this), (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2 ) - this.checkAlive.bind(this)() } /** - * Stops the worker alive check interval. + * Stops the worker check active interval. */ - private stopCheckAlive (): void { - this.aliveInterval != null && clearInterval(this.aliveInterval) + private stopCheckActive (): void { + this.activeInterval != null && clearInterval(this.activeInterval) } /** * Checks if the worker should be terminated, because its living too long. */ - private checkAlive (): void { + private checkActive (): void { if ( performance.now() - this.lastTaskTimestamp > (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) @@ -364,62 +364,77 @@ export abstract class AbstractWorker< } /** - * Runs the given function synchronously. + * 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 { + const fn = this.getTaskFunction(task.name) + if (isAsyncFunction(fn)) { + this.runInAsyncScope(this.runAsync.bind(this), this, fn, task) + } else { + this.runInAsyncScope(this.runSync.bind(this), this, fn, task) + } + } + + /** + * Runs the given task function synchronously. * - * @param fn - Function that will be executed. - * @param message - Input data for the given function. + * @param fn - Task function that will be executed. + * @param task - Input data for the task function. */ protected runSync ( fn: WorkerSyncFunction, - message: MessageValue + task: Task ): void { try { - let taskPerformance = this.beginTaskPerformance(message.name) - const res = fn(message.data) + let taskPerformance = this.beginTaskPerformance(task.name) + const res = fn(task.data) taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, taskPerformance, workerId: this.id, - id: message.id + id: task.id }) } catch (e) { const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { - name: message.name ?? DEFAULT_TASK_NAME, + name: task.name ?? DEFAULT_TASK_NAME, message: errorMessage, - data: message.data + data: task.data }, workerId: this.id, - id: message.id + id: task.id }) } finally { - if (!this.isMain && this.aliveInterval != null) { + if (!this.isMain && this.activeInterval != null) { this.lastTaskTimestamp = performance.now() } } } /** - * Runs the given function asynchronously. + * Runs the given task function asynchronously. * - * @param fn - Function that will be executed. - * @param message - Input data for the given function. + * @param fn - Task function that will be executed. + * @param task - Input data for the task function. */ protected runAsync ( fn: WorkerAsyncFunction, - message: MessageValue + task: Task ): void { - let taskPerformance = this.beginTaskPerformance(message.name) - fn(message.data) + let taskPerformance = this.beginTaskPerformance(task.name) + fn(task.data) .then(res => { taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, taskPerformance, workerId: this.id, - id: message.id + id: task.id }) return null }) @@ -427,16 +442,16 @@ export abstract class AbstractWorker< const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { - name: message.name ?? DEFAULT_TASK_NAME, + name: task.name ?? DEFAULT_TASK_NAME, message: errorMessage, - data: message.data + data: task.data }, workerId: this.id, - id: message.id + id: task.id }) }) .finally(() => { - if (!this.isMain && this.aliveInterval != null) { + if (!this.isMain && this.activeInterval != null) { this.lastTaskTimestamp = performance.now() } }) @@ -444,9 +459,11 @@ export abstract class AbstractWorker< } /** - * Gets the task function in the given scope. + * Gets the task function with the given name. * * @param name - Name of the task function that will be returned. + * @returns The task function. + * @throws {@link https://nodejs.org/api/errors.html#class-error} If the task function is not found. */ private getTaskFunction (name?: string): WorkerFunction { name = name ?? DEFAULT_TASK_NAME