X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=6bf84b875622ebaece3d8b4b8dae3c637593ed68;hb=ee9f0ad39b47fe80bb08d159024dad0fae81f6c5;hp=7aa626f1610a3cd4f4e30ca23232b640a1eca410;hpb=5c4d16da7677746e563fdcfe7f82cbb842d1c9e6;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 7aa626f1..6bf84b87 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -20,11 +20,11 @@ import { type WorkerOptions } from './worker-options' import type { + TaskAsyncFunction, + TaskFunction, TaskFunctions, - WorkerAsyncFunction, - WorkerFunction, - WorkerSyncFunction -} from './worker-functions' + TaskSyncFunction +} from './task-functions' const DEFAULT_MAX_INACTIVE_TIME = 60000 const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT @@ -48,7 +48,7 @@ export abstract class AbstractWorker< /** * Task function(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. */ @@ -58,42 +58,44 @@ 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. * * @param type - The type of async event. * @param isMain - Whether this is the main worker or not. - * @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 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, - taskFunctions: - | WorkerFunction - | TaskFunctions, - protected readonly mainWorker: MainWorker, + private readonly mainWorker: MainWorker, + taskFunctions: TaskFunction | TaskFunctions, protected readonly opts: WorkerOptions = { /** * The kill behavior option on this worker or its default value. */ 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 + maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME, + /** + * The function to call when the worker is killed. + */ + killHandler: EMPTY_FUNCTION } ) { super(type) this.checkWorkerOptions(this.opts) this.checkTaskFunctions(taskFunctions) if (!this.isMain) { - this.mainWorker?.on('message', this.messageListener.bind(this)) + this.getMainWorker()?.on('message', this.handleReadyMessage.bind(this)) } } @@ -102,6 +104,7 @@ export abstract class AbstractWorker< this.opts.maxInactiveTime = opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME delete this.opts.async + this.opts.killHandler = opts.killHandler ?? EMPTY_FUNCTION } /** @@ -110,29 +113,41 @@ export abstract class AbstractWorker< * @param taskFunctions - The task function(s) parameter that should be checked. */ private checkTaskFunctions ( - taskFunctions: - | WorkerFunction - | TaskFunctions + taskFunctions: TaskFunction | TaskFunctions ): void { if (taskFunctions == null) { throw new Error('taskFunctions parameter is mandatory') } - this.taskFunctions = new Map>() + this.taskFunctions = new Map>() if (typeof taskFunctions === 'function') { - this.taskFunctions.set(DEFAULT_TASK_NAME, taskFunctions.bind(this)) + const boundFn = taskFunctions.bind(this) + this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) + this.taskFunctions.set( + typeof taskFunctions.name === 'string' && + taskFunctions.name.trim().length > 0 + ? taskFunctions.name + : 'fn1', + boundFn + ) } 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' ) } - this.taskFunctions.set(name, fn.bind(this)) + const boundFn = fn.bind(this) if (firstEntry) { - this.taskFunctions.set(DEFAULT_TASK_NAME, fn.bind(this)) + this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) firstEntry = false } + this.taskFunctions.set(name, boundFn) } if (firstEntry) { throw new Error('taskFunctions parameter object is empty') @@ -171,7 +186,7 @@ export abstract class AbstractWorker< */ public addTaskFunction ( name: string, - fn: WorkerFunction + fn: TaskFunction ): boolean { if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') @@ -185,13 +200,14 @@ export abstract class AbstractWorker< throw new TypeError('fn parameter is not a function') } try { + const boundFn = fn.bind(this) if ( this.taskFunctions.get(name) === this.taskFunctions.get(DEFAULT_TASK_NAME) ) { - this.taskFunctions.set(DEFAULT_TASK_NAME, fn.bind(this)) + this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) } - this.taskFunctions.set(name, fn.bind(this)) + this.taskFunctions.set(name, boundFn) return true } catch { return false @@ -227,7 +243,16 @@ export abstract class AbstractWorker< } /** - * Sets the default task function to use when no task function name is specified. + * Lists the names of the worker's task functions. + * + * @returns The names of the worker's task functions. + */ + public listTaskFunctions (): string[] { + return [...this.taskFunctions.keys()] + } + + /** + * 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. @@ -252,10 +277,7 @@ export abstract class AbstractWorker< try { this.taskFunctions.set( DEFAULT_TASK_NAME, - this.taskFunctions.get(name)?.bind(this) as WorkerFunction< - Data, - Response - > + this.taskFunctions.get(name) as TaskFunction ) return true } catch { @@ -263,63 +285,100 @@ 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 - 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() - } else if (message.statistics != null) { + protected messageListener (message: MessageValue): void { + if (this.isMain) { + throw new Error('Cannot handle message to worker in main worker') + } else if (message.workerId != null && message.workerId !== this.id) { + throw new Error( + `Message worker id ${message.workerId} does not match the worker id ${this.id}` + ) + } else if (message.workerId === this.id) { + 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.id != null && message.data != null) { + } else if (message.checkActive != null) { + // Check active message received + message.checkActive ? this.startCheckActive() : this.stopCheckActive() + } else if (message.taskId != null && message.data != null) { // Task message received this.run(message) } else if (message.kill === true) { // Kill message received - this.stopCheckAlive() - this.emitDestroy() + this.handleKillMessage(message) } } } /** - * Notifies the main worker that this worker is ready to process tasks. + * Handles a kill message sent by the main worker. + * + * @param message - The kill message. */ - protected workerReady (): void { - !this.isMain && this.sendToMainWorker({ ready: true, workerId: this.id }) + protected handleKillMessage (message: MessageValue): void { + this.stopCheckActive() + if (isAsyncFunction(this.opts.killHandler)) { + (this.opts.killHandler?.() as Promise) + .then(() => { + this.sendToMainWorker({ kill: 'success', workerId: this.id }) + return null + }) + .catch(() => { + this.sendToMainWorker({ kill: 'failure', workerId: this.id }) + }) + .finally(() => { + this.emitDestroy() + }) + .catch(EMPTY_FUNCTION) + } else { + try { + // eslint-disable-next-line @typescript-eslint/no-invalid-void-type + this.opts.killHandler?.() as void + this.sendToMainWorker({ kill: 'success', workerId: this.id }) + } catch { + this.sendToMainWorker({ kill: 'failure', workerId: this.id }) + } finally { + this.emitDestroy() + } + } } /** - * 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 { + if (this.activeInterval != null) { + clearInterval(this.activeInterval) + delete 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) @@ -341,7 +400,7 @@ export abstract class AbstractWorker< } /** - * Sends a message to the main worker. + * Sends a message to main worker. * * @param message - The response message. */ @@ -375,13 +434,13 @@ export abstract class AbstractWorker< } /** - * Runs the given function synchronously. + * Runs the given task function synchronously. * * @param fn - Task function that will be executed. * @param task - Input data for the task function. */ protected runSync ( - fn: WorkerSyncFunction, + fn: TaskSyncFunction, task: Task ): void { try { @@ -392,7 +451,7 @@ export abstract class AbstractWorker< data: res, taskPerformance, workerId: this.id, - id: task.id + taskId: task.taskId }) } catch (e) { const errorMessage = this.handleError(e as Error | string) @@ -403,38 +462,36 @@ export abstract class AbstractWorker< data: task.data }, workerId: this.id, - id: task.id + taskId: task.taskId }) } finally { - if (!this.isMain && this.aliveInterval != null) { - this.lastTaskTimestamp = performance.now() - } + this.updateLastTaskTimestamp() } } /** - * Runs the given function asynchronously. + * Runs the given task function asynchronously. * * @param fn - Task function that will be executed. * @param task - Input data for the task function. */ protected runAsync ( - fn: WorkerAsyncFunction, + fn: TaskAsyncFunction, task: Task ): void { let taskPerformance = this.beginTaskPerformance(task.name) fn(task.data) - .then(res => { + .then((res) => { taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, taskPerformance, workerId: this.id, - id: task.id + taskId: task.taskId }) return null }) - .catch(e => { + .catch((e) => { const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { @@ -443,13 +500,11 @@ export abstract class AbstractWorker< data: task.data }, workerId: this.id, - id: task.id + taskId: task.taskId }) }) .finally(() => { - if (!this.isMain && this.aliveInterval != null) { - this.lastTaskTimestamp = performance.now() - } + this.updateLastTaskTimestamp() }) .catch(EMPTY_FUNCTION) } @@ -461,7 +516,7 @@ export abstract class AbstractWorker< * @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 { + private getTaskFunction (name?: string): TaskFunction { name = name ?? DEFAULT_TASK_NAME const fn = this.taskFunctions.get(name) if (fn == null) { @@ -499,4 +554,10 @@ export abstract class AbstractWorker< throw new Error('Performance statistics computation requirements not set') } } + + private updateLastTaskTimestamp (): void { + if (this.activeInterval != null) { + this.lastTaskTimestamp = performance.now() + } + } }