X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=b7c542aa469f40bf4feb50aa99599b52bb7bc44e;hb=refs%2Ftags%2Fv2.5.4;hp=788c5dfa4aabcc2c5343567dabc8be14fba5e4a2;hpb=a86b6df187001e7e2e5c248ddb828286f985096c;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 788c5dfa..b7c542aa 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -1,20 +1,35 @@ import { AsyncResource } from 'node:async_hooks' import type { Worker } from 'node:cluster' import type { MessagePort } from 'node:worker_threads' +import { type EventLoopUtilization, performance } from 'node:perf_hooks' +import type { MessageValue, WorkerStatistics } from '../utility-types' +import { EMPTY_FUNCTION, isPlainObject } from '../utils' import { - type MessageValue, - type TaskFunctions, - type WorkerAsyncFunction, - type WorkerFunction, - type WorkerSyncFunction -} from '../utility-types' -import { EMPTY_FUNCTION } from '../utils' -import type { KillBehavior, WorkerOptions } from './worker-options' -import { KillBehaviors } from './worker-options' + type KillBehavior, + KillBehaviors, + type WorkerOptions +} from './worker-options' +import type { + TaskFunctions, + WorkerAsyncFunction, + WorkerFunction, + WorkerSyncFunction +} from './worker-functions' +const DEFAULT_FUNCTION_NAME = 'default' const DEFAULT_MAX_INACTIVE_TIME = 60000 const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT +/** + * Task performance. + */ +export interface TaskPerformance { + timestamp: number + waitTime?: number + runTime?: number + elu?: EventLoopUtilization +} + /** * Base class that implements some shared logic for all poolifier workers. * @@ -35,6 +50,10 @@ export abstract class AbstractWorker< * Timestamp of the last task processed by this worker. */ protected lastTaskTimestamp!: number + /** + * Performance statistics computation. + */ + protected statistics!: WorkerStatistics /** * Handler id of the `aliveInterval` worker alive check. */ @@ -44,7 +63,7 @@ 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. + * @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 opts - Options for the worker. */ @@ -78,58 +97,53 @@ export abstract class AbstractWorker< ) this.checkAlive.bind(this)() } - - this.mainWorker?.on( - 'message', - (message: MessageValue) => { - this.messageListener(message) - } - ) + this.mainWorker?.on('message', this.messageListener.bind(this)) } private checkWorkerOptions (opts: WorkerOptions): void { this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR this.opts.maxInactiveTime = opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME - this.opts.async = opts.async ?? false + delete this.opts.async } /** * Checks if the `taskFunctions` parameter is passed to the constructor. * - * @param taskFunctions - The task function(s) that should be defined. + * @param taskFunctions - The task function(s) parameter that should be checked. */ private checkTaskFunctions ( taskFunctions: | WorkerFunction | TaskFunctions ): void { - if (taskFunctions == null) { throw new Error('taskFunctions parameter is mandatory') } - if ( - typeof taskFunctions !== 'function' && - typeof taskFunctions !== 'object' - ) { - throw new Error('taskFunctions parameter is not a function or an object') - } - if ( - typeof taskFunctions === 'object' && - taskFunctions.constructor !== Object && - Object.prototype.toString.call(taskFunctions) !== '[object Object]' - ) { - throw new Error('taskFunctions parameter is not an object literal') + if (taskFunctions == null) { + throw new Error('taskFunctions parameter is mandatory') } this.taskFunctions = new Map>() - if (typeof taskFunctions !== 'function') { + if (typeof taskFunctions === 'function') { + this.taskFunctions.set(DEFAULT_FUNCTION_NAME, taskFunctions.bind(this)) + } else if (isPlainObject(taskFunctions)) { + let firstEntry = true for (const [name, fn] of Object.entries(taskFunctions)) { if (typeof fn !== 'function') { - throw new Error( + throw new TypeError( 'A taskFunctions parameter object value is not a function' ) } this.taskFunctions.set(name, fn.bind(this)) + if (firstEntry) { + this.taskFunctions.set(DEFAULT_FUNCTION_NAME, fn.bind(this)) + firstEntry = false + } + } + if (firstEntry) { + throw new Error('taskFunctions parameter object is empty') } } else { - this.taskFunctions.set('default', taskFunctions.bind(this)) + throw new TypeError( + 'taskFunctions parameter is not a function or a plain object' + ) } } @@ -140,17 +154,12 @@ export abstract class AbstractWorker< */ protected messageListener (message: MessageValue): void { if (message.id != null && message.data != null) { - let fn: WorkerFunction | undefined - if (message.name == null) { - fn = this.taskFunctions.get('default') - } else { - fn = this.taskFunctions.get(message.name) - } // Task message received + const fn = this.getTaskFunction(message.name) if (fn?.constructor.name === 'AsyncFunction') { this.runInAsyncScope(this.runAsync.bind(this), this, fn, message) } else { - this.runInAsyncScope(this.run.bind(this), this, fn, message) + this.runInAsyncScope(this.runSync.bind(this), this, fn, message) } } else if (message.parent != null) { // Main worker reference message received @@ -159,6 +168,9 @@ export abstract class AbstractWorker< // Kill message received this.aliveInterval != null && clearInterval(this.aliveInterval) this.emitDestroy() + } else if (message.statistics != null) { + // Statistics message received + this.statistics = message.statistics } } @@ -209,22 +221,29 @@ export abstract class AbstractWorker< * @param fn - Function that will be executed. * @param message - Input data for the given function. */ - protected run ( + protected runSync ( fn: WorkerSyncFunction, message: MessageValue ): void { try { - const startTimestamp = performance.now() + const taskPerformance = this.beginTaskPerformance(message) const res = fn(message.data) - const runTime = performance.now() - startTimestamp + const { runTime, waitTime, elu } = + this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, - id: message.id, - runTime + runTime, + waitTime, + elu, + id: message.id }) } catch (e) { const err = this.handleError(e as Error) - this.sendToMainWorker({ error: err, id: message.id }) + this.sendToMainWorker({ + error: err, + errorData: message.data, + id: message.id + }) } finally { !this.isMain && (this.lastTaskTimestamp = performance.now()) } @@ -240,24 +259,70 @@ export abstract class AbstractWorker< fn: WorkerAsyncFunction, message: MessageValue ): void { - const startTimestamp = performance.now() + const taskPerformance = this.beginTaskPerformance(message) fn(message.data) .then(res => { - const runTime = performance.now() - startTimestamp + const { runTime, waitTime, elu } = + this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, - id: message.id, - runTime + runTime, + waitTime, + elu, + id: message.id }) return null }) .catch(e => { const err = this.handleError(e as Error) - this.sendToMainWorker({ error: err, id: message.id }) + this.sendToMainWorker({ + error: err, + errorData: message.data, + id: message.id + }) }) .finally(() => { !this.isMain && (this.lastTaskTimestamp = performance.now()) }) .catch(EMPTY_FUNCTION) } + + /** + * Gets the task function in the given scope. + * + * @param name - Name of the function that will be returned. + */ + private getTaskFunction (name?: string): WorkerFunction { + name = name ?? DEFAULT_FUNCTION_NAME + const fn = this.taskFunctions.get(name) + if (fn == null) { + throw new Error(`Task function '${name}' not found`) + } + return fn + } + + private beginTaskPerformance (message: MessageValue): TaskPerformance { + const timestamp = performance.now() + return { + timestamp, + ...(this.statistics.waitTime && { + waitTime: timestamp - (message.timestamp ?? timestamp) + }), + ...(this.statistics.elu && { elu: performance.eventLoopUtilization() }) + } + } + + private endTaskPerformance ( + taskPerformance: TaskPerformance + ): TaskPerformance { + return { + ...taskPerformance, + ...(this.statistics.runTime && { + runTime: performance.now() - taskPerformance.timestamp + }), + ...(this.statistics.elu && { + elu: performance.eventLoopUtilization(taskPerformance.elu) + }) + } + } }