X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=b16c886b26c933c77aeee5ff32a3c9a61272aab3;hb=79e68dbfc6bfa1de92849134b10bb6e909fab422;hp=5e7144877357806dacae29c1bf28993894616ca1;hpb=13a332e67a8a208b09c673d8a4a132df657ff4e6;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 5e714487..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' @@ -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