'dequeue',
'dequeued',
'ecma',
+ 'elu',
'enqueue',
'enum',
'errored',
'mjs',
'num',
'os',
+ 'perf',
'piscina',
'pnpm',
'poolifier',
'node:crypto',
'node:events',
'node:os',
+ 'node:perf_hooks',
'node:worker_threads'
],
plugins: [
import crypto from 'node:crypto'
+import { performance } from 'node:perf_hooks'
import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
import {
DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
waitTimeHistory: new CircularArray(),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
}
this.updateRunTimeTasksUsage(workerTasksUsage, message)
this.updateWaitTimeTasksUsage(workerTasksUsage, message)
+ this.updateEluTasksUsage(workerTasksUsage, message)
}
private updateRunTimeTasksUsage (
}
}
+ private updateEluTasksUsage (
+ workerTasksUsage: TasksUsage,
+ message: MessageValue<Response>
+ ): void {
+ if (this.workerChoiceStrategyContext.getRequiredStatistics().elu) {
+ if (workerTasksUsage.elu != null && message.elu != null) {
+ // TODO: cumulative or delta?
+ workerTasksUsage.elu = {
+ idle: workerTasksUsage.elu.idle + message.elu.idle,
+ active: workerTasksUsage.elu.active + message.elu.active,
+ utilization:
+ workerTasksUsage.elu.utilization + message.elu.utilization
+ }
+ } else if (message.elu != null) {
+ workerTasksUsage.elu = message.elu
+ }
+ }
+ }
+
/**
* Chooses a worker node for the next task.
*
waitTimeHistory: new CircularArray(),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
},
tasksQueue: new Queue<Task<Data>>()
})
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
}
/**
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
}
/**
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
}
/** @inheritDoc */
* Require tasks median wait time.
*/
medWaitTime: boolean
+ /**
+ * Event loop utilization.
+ */
+ elu: boolean
}
/**
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
}
/**
+import type { EventLoopUtilization } from 'node:perf_hooks'
import type { CircularArray } from '../circular-array'
import type { Queue } from '../queue'
* Number of tasks errored.
*/
error: number
+ /**
+ * Event loop utilization.
+ */
+ elu: EventLoopUtilization | undefined
}
/**
import type { Worker as ClusterWorker } from 'node:cluster'
import type { MessagePort } from 'node:worker_threads'
+import type { EventLoopUtilization } from 'node:perf_hooks'
import type { KillBehavior } from './worker/worker-options'
import type { IWorker, Task } from './pools/worker'
* Wait time.
*/
readonly waitTime?: number
+ /**
+ * Event loop utilization.
+ */
+ readonly elu?: EventLoopUtilization
/**
* Reference to main worker.
*/
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,
TaskFunctions,
const DEFAULT_MAX_INACTIVE_TIME = 60000
const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
+interface TaskPerformance {
+ timestamp: number
+ waitTime: number
+ runTime?: number
+ elu: EventLoopUtilization
+}
+
/**
* Base class that implements some shared logic for all poolifier workers.
*
message: MessageValue<Data>
): void {
try {
- const startTimestamp = performance.now()
- const waitTime = startTimestamp - (message.submissionTimestamp ?? 0)
+ const taskPerformance = this.beforeTaskRunHook(message)
const res = fn(message.data)
- const runTime = performance.now() - startTimestamp
+ const { runTime, waitTime, elu } = this.afterTaskRunHook(taskPerformance)
this.sendToMainWorker({
data: res,
runTime,
waitTime,
+ elu,
id: message.id
})
} catch (e) {
fn: WorkerAsyncFunction<Data, Response>,
message: MessageValue<Data>
): void {
- const startTimestamp = performance.now()
- const waitTime = startTimestamp - (message.submissionTimestamp ?? 0)
+ const taskPerformance = this.beforeTaskRunHook(message)
fn(message.data)
.then(res => {
- const runTime = performance.now() - startTimestamp
+ const { runTime, waitTime, elu } =
+ this.afterTaskRunHook(taskPerformance)
this.sendToMainWorker({
data: res,
runTime,
waitTime,
+ elu,
id: message.id
})
return null
}
return fn
}
+
+ private beforeTaskRunHook (message: MessageValue<Data>): TaskPerformance {
+ // TODO: conditional accounting
+ const timestamp = performance.now()
+ return {
+ timestamp,
+ waitTime: timestamp - (message.submissionTimestamp ?? 0),
+ elu: performance.eventLoopUtilization()
+ }
+ }
+
+ private afterTaskRunHook (taskPerformance: TaskPerformance): TaskPerformance {
+ return {
+ ...taskPerformance,
+ ...{
+ runTime: performance.now() - taskPerformance.timestamp,
+ elu: performance.eventLoopUtilization(taskPerformance.elu)
+ }
+ }
+ }
}
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
pool.setWorkerChoiceStrategyOptions({ medRunTime: true })
expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
medRunTime: true,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
pool.setWorkerChoiceStrategyOptions({ medRunTime: false })
expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
await pool.destroy()
})
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
await pool.destroy()
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
await Promise.all(promises)
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
await pool.destroy()
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.ran).toBeGreaterThan(0)
expect(workerNode.tasksUsage.ran).toBeLessThanOrEqual(maxMultiplier)
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.runTimeHistory.length).toBe(0)
expect(workerNode.tasksUsage.waitTimeHistory.length).toBe(0)
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
await pool.destroy()
pool = new DynamicThreadPool(
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
// We need to clean up the resources after our test
await pool.destroy()
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
expect(
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
expect(
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
await pool.destroy()
pool = new DynamicThreadPool(
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
// We need to clean up the resources after our test
await pool.destroy()
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
// We need to clean up the resources after our test
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
// We need to clean up the resources after our test
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
await pool.destroy()
pool = new DynamicThreadPool(
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
// We need to clean up the resources after our test
await pool.destroy()
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.ran).toBeGreaterThan(0)
expect(workerNode.tasksUsage.ran).toBeLessThanOrEqual(max * maxMultiplier)
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.ran).toBeGreaterThan(0)
expect(workerNode.tasksUsage.ran).toBeLessThanOrEqual(max * maxMultiplier)
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
await pool.destroy()
pool = new DynamicThreadPool(
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
// We need to clean up the resources after our test
await pool.destroy()
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0)
expect(workerNode.tasksUsage.avgRunTime).toBeGreaterThan(0)
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0)
expect(workerNode.tasksUsage.avgRunTime).toBeGreaterThan(0)
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0)
expect(workerNode.tasksUsage.medRunTime).toBeGreaterThan(0)
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
await pool.destroy()
pool = new DynamicThreadPool(
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
// We need to clean up the resources after our test
await pool.destroy()
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.ran).toBeGreaterThanOrEqual(0)
expect(workerNode.tasksUsage.ran).toBeLessThanOrEqual(max * maxMultiplier)
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.ran).toBeGreaterThan(0)
expect(workerNode.tasksUsage.ran).toBeLessThanOrEqual(max * maxMultiplier)
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
expect(workerNode.tasksUsage.ran).toBeGreaterThan(0)
expect(workerNode.tasksUsage.ran).toBeLessThanOrEqual(max * maxMultiplier)
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
await pool.destroy()
pool = new DynamicThreadPool(
medRunTime: false,
waitTime: false,
avgWaitTime: false,
- medWaitTime: false
+ medWaitTime: false,
+ elu: false
})
// We need to clean up the resources after our test
await pool.destroy()
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
expect(
waitTimeHistory: expect.any(CircularArray),
avgWaitTime: 0,
medWaitTime: 0,
- error: 0
+ error: 0,
+ elu: undefined
})
}
expect(