X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=98b32fe4cb9525efba1ab151b8d7b7f3efb2310b;hb=2115798751b31f025acae1ff8c4f5cbcec8ad3af;hp=64627efb8823198c6cb09c81fbdded3bef046f01;hpb=15a8bf08edd98c1f9ef4a4d3f89d2cffbe30852e;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 64627efb..98b32fe4 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -84,7 +84,11 @@ export abstract class AbstractWorker< * 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) @@ -99,6 +103,7 @@ export abstract class AbstractWorker< this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR this.opts.maxInactiveTime = opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME + this.opts.killHandler = opts.killHandler ?? EMPTY_FUNCTION delete this.opts.async } @@ -119,7 +124,7 @@ export abstract class AbstractWorker< this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) this.taskFunctions.set( typeof taskFunctions.name === 'string' && - taskFunctions.name.trim().length > 0 + taskFunctions.name.trim().length > 0 ? taskFunctions.name : 'fn1', boundFn @@ -132,6 +137,11 @@ export abstract class AbstractWorker< 'A taskFunctions parameter object key is not a string' ) } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError( + 'A taskFunctions parameter object key an empty string' + ) + } if (typeof fn !== 'function') { throw new TypeError( 'A taskFunctions parameter object value is not a function' @@ -159,12 +169,15 @@ 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 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-typeerror} If the `name` parameter is not a string or an empty string. */ public hasTaskFunction (name: string): boolean { if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name parameter is an empty string') + } return this.taskFunctions.has(name) } @@ -175,7 +188,7 @@ 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 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-typeerror} If the `name` parameter is not a string or an empty 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. */ @@ -186,6 +199,9 @@ export abstract class AbstractWorker< if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name parameter is an empty string') + } if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot add a task function with the default reserved name' @@ -203,6 +219,7 @@ export abstract class AbstractWorker< this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) } this.taskFunctions.set(name, boundFn) + this.sendTaskFunctionsListToMainWorker() return true } catch { return false @@ -214,7 +231,7 @@ 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 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-typeerror} If the `name` parameter is not a string or an empty 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. */ @@ -222,6 +239,9 @@ export abstract class AbstractWorker< if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name parameter is an empty string') + } if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot remove the task function with the default reserved name' @@ -234,7 +254,9 @@ export abstract class AbstractWorker< 'Cannot remove the task function used as the default task function' ) } - return this.taskFunctions.delete(name) + const deleteStatus = this.taskFunctions.delete(name) + this.sendTaskFunctionsListToMainWorker() + return deleteStatus } /** @@ -251,7 +273,7 @@ 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 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-typeerror} If the `name` parameter is not a string or an empty 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. */ @@ -259,6 +281,9 @@ export abstract class AbstractWorker< if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name parameter is an empty string') + } if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot set the default task function reserved name as the default task function' @@ -293,24 +318,19 @@ export abstract class AbstractWorker< * @param message - The received message. */ protected messageListener (message: MessageValue): void { - if (message.workerId != null && message.workerId !== this.id) { - throw new Error('Message worker id does not match worker id') - } else if (message.workerId === this.id) { - if (message.statistics != null) { - // Statistics message received - this.statistics = message.statistics - } else if (message.checkActive != null) { - // Check active message received - !this.isMain && 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.handleKillMessage(message) - } + this.checkMessageWorkerId(message) + if (message.statistics != null) { + // Statistics message received + this.statistics = message.statistics + } 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.handleKillMessage(message) } } @@ -320,8 +340,47 @@ export abstract class AbstractWorker< * @param message - The kill message. */ protected handleKillMessage (message: MessageValue): void { - !this.isMain && this.stopCheckActive() - this.emitDestroy() + 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() + } + } + } + + /** + * 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. + */ + private checkMessageWorkerId (message: MessageValue): void { + if (message.workerId == null) { + throw new Error('Message worker id is not set') + } 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}` + ) + } } /** @@ -378,6 +437,16 @@ export abstract class AbstractWorker< message: MessageValue ): void + /** + * Sends the list of task function names to the main worker. + */ + protected sendTaskFunctionsListToMainWorker (): void { + this.sendToMainWorker({ + taskFunctions: this.listTaskFunctions(), + workerId: this.id + }) + } + /** * Handles an error and convert it to a string so it can be sent back to the main worker. * @@ -395,9 +464,6 @@ export abstract class AbstractWorker< * @throws {@link https://nodejs.org/api/errors.html#class-error} If the task function is not found. */ protected run (task: Task): void { - if (this.isMain) { - throw new Error('Cannot run a task in the main worker') - } const fn = this.getTaskFunction(task.name) if (isAsyncFunction(fn)) { this.runInAsyncScope(this.runAsync.bind(this), this, fn, task) @@ -416,26 +482,27 @@ export abstract class AbstractWorker< fn: TaskSyncFunction, task: Task ): void { + const { name, taskId, data } = task try { - let taskPerformance = this.beginTaskPerformance(task.name) - const res = fn(task.data) + let taskPerformance = this.beginTaskPerformance(name) + const res = fn(data) taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, taskPerformance, workerId: this.id, - taskId: task.taskId + taskId }) } catch (e) { const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { - name: task.name ?? DEFAULT_TASK_NAME, + name: name ?? DEFAULT_TASK_NAME, message: errorMessage, - data: task.data + data }, workerId: this.id, - taskId: task.taskId + taskId }) } finally { this.updateLastTaskTimestamp() @@ -452,28 +519,29 @@ export abstract class AbstractWorker< fn: TaskAsyncFunction, task: Task ): void { - let taskPerformance = this.beginTaskPerformance(task.name) - fn(task.data) - .then(res => { + const { name, taskId, data } = task + let taskPerformance = this.beginTaskPerformance(name) + fn(data) + .then((res) => { taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, taskPerformance, workerId: this.id, - taskId: task.taskId + taskId }) return null }) - .catch(e => { + .catch((e) => { const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { - name: task.name ?? DEFAULT_TASK_NAME, + name: name ?? DEFAULT_TASK_NAME, message: errorMessage, - data: task.data + data }, workerId: this.id, - taskId: task.taskId + taskId }) }) .finally(() => { @@ -529,7 +597,7 @@ export abstract class AbstractWorker< } private updateLastTaskTimestamp (): void { - if (!this.isMain && this.activeInterval != null) { + if (this.activeInterval != null) { this.lastTaskTimestamp = performance.now() } }