X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=67b4b5ff9e4575ead2d824537fb106d6d0b5ba11;hb=a038b5175071af7c51dae9e25c7423c7572f3949;hp=b16c886b26c933c77aeee5ff32a3c9a61272aab3;hpb=b0a4db63f53f4df5d60abc592f32043468626574;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index b16c886b..67b4b5ff 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -66,17 +66,17 @@ export abstract class AbstractWorker< * * @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, + private readonly mainWorker: MainWorker, taskFunctions: | WorkerFunction | TaskFunctions, - protected readonly mainWorker: MainWorker, protected readonly opts: WorkerOptions = { /** * The kill behavior option on this worker or its default value. @@ -93,7 +93,7 @@ export abstract class AbstractWorker< 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)) } } @@ -119,7 +119,15 @@ export abstract class AbstractWorker< } 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)) { @@ -133,11 +141,12 @@ export abstract class AbstractWorker< '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') @@ -190,13 +199,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 @@ -232,7 +242,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 Array.from(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. @@ -257,10 +276,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 WorkerFunction ) return true } catch { @@ -268,40 +284,39 @@ 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 - The received message. */ - protected messageListener (message: MessageValue): void { + protected messageListener (message: MessageValue): void { if (message.workerId === this.id) { - if (message.ready != null) { - // Startup message received - this.sendReadyResponse() - } else if (message.statistics != null) { + 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() + !this.isMain && message.checkActive + ? this.startCheckActive() + : this.stopCheckActive() } else if (message.id != null && message.data != null) { // Task message received this.run(message) } else if (message.kill === true) { // Kill message received - this.stopCheckActive() + !this.isMain && this.stopCheckActive() this.emitDestroy() } } } - /** - * Sends to the main worker the ready response. - */ - protected sendReadyResponse (): void { - !this.isMain && this.sendToMainWorker({ ready: true, workerId: this.id }) - } - /** * Starts the worker check active interval. */ @@ -310,14 +325,17 @@ export abstract class AbstractWorker< this.activeInterval = setInterval( this.checkActive.bind(this), (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2 - ) + ).unref() } /** * Stops the worker check active interval. */ private stopCheckActive (): void { - this.activeInterval != null && clearInterval(this.activeInterval) + if (this.activeInterval != null) { + clearInterval(this.activeInterval) + delete this.activeInterval + } } /** @@ -370,6 +388,9 @@ 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) @@ -410,9 +431,7 @@ export abstract class AbstractWorker< id: task.id }) } finally { - if (!this.isMain && this.activeInterval != null) { - this.lastTaskTimestamp = performance.now() - } + this.updateLastTaskTimestamp() } } @@ -451,9 +470,7 @@ export abstract class AbstractWorker< }) }) .finally(() => { - if (!this.isMain && this.activeInterval != null) { - this.lastTaskTimestamp = performance.now() - } + this.updateLastTaskTimestamp() }) .catch(EMPTY_FUNCTION) } @@ -503,4 +520,10 @@ export abstract class AbstractWorker< throw new Error('Performance statistics computation requirements not set') } } + + private updateLastTaskTimestamp (): void { + if (!this.isMain && this.activeInterval != null) { + this.lastTaskTimestamp = performance.now() + } + } }