import {
DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
EMPTY_FUNCTION,
+ isKillBehavior,
isPlainObject,
- median
+ median,
+ round
} from '../utils'
-import { KillBehaviors, isKillBehavior } from '../worker/worker-options'
+import { KillBehaviors } from '../worker/worker-options'
import { CircularArray } from '../circular-array'
import { Queue } from '../queue'
import {
type PoolType,
PoolTypes,
type TasksQueueOptions,
- type WorkerType
+ type WorkerType,
+ WorkerTypes
} from './pool'
import type {
IWorker,
Response
>
+ /**
+ * The start timestamp of the pool.
+ */
+ private readonly startTimestamp
+
/**
* Constructs a new poolifier pool.
*
this.setupHook()
- for (let i = 1; i <= this.numberOfWorkers; i++) {
+ while (this.workerNodes.length < this.numberOfWorkers) {
this.createAndSetupWorker()
}
+
+ this.startTimestamp = performance.now()
}
private checkFilePath (filePath: string): void {
}
}
- private get starting (): boolean {
- return this.workerNodes.some(workerNode => !workerNode.info.started)
- }
-
- private get started (): boolean {
- return this.workerNodes.some(workerNode => workerNode.info.started)
- }
-
/** @inheritDoc */
public get info (): PoolInfo {
return {
worker: this.worker,
minSize: this.minSize,
maxSize: this.maxSize,
+ utilization: round(this.utilization),
workerNodes: this.workerNodes.length,
idleWorkerNodes: this.workerNodes.reduce(
(accumulator, workerNode) =>
}
}
+ /**
+ * Gets the pool run time.
+ *
+ * @returns The pool run time in milliseconds.
+ */
+ private get runTime (): number {
+ return performance.now() - this.startTimestamp
+ }
+
+ /**
+ * Gets the approximate pool utilization.
+ *
+ * @returns The pool utilization.
+ */
+ private get utilization (): number {
+ const poolRunTimeCapacity = this.runTime * this.maxSize
+ const totalTasksRunTime = this.workerNodes.reduce(
+ (accumulator, workerNode) =>
+ accumulator + workerNode.usage.runTime.aggregate,
+ 0
+ )
+ const totalTasksWaitTime = this.workerNodes.reduce(
+ (accumulator, workerNode) =>
+ accumulator + workerNode.usage.waitTime.aggregate,
+ 0
+ )
+ return (totalTasksRunTime + totalTasksWaitTime) / poolRunTimeCapacity
+ }
+
/**
* Pool type.
*
if (this.emitter != null) {
this.emitter.emit(PoolEvents.error, error)
}
- if (this.opts.restartWorkerOnError === true && !this.starting) {
+ if (this.opts.restartWorkerOnError === true) {
this.createAndSetupWorker()
}
})
return message => {
if (message.workerId != null && message.started != null) {
// Worker started message received
- this.workerNodes[
- this.getWorkerNodeKey(this.getWorkerById(message.workerId) as Worker)
- ].info.started = message.started
+ const worker = this.getWorkerById(message.workerId)
+ if (worker != null) {
+ this.workerNodes[this.getWorkerNodeKey(worker)].info.started =
+ message.started
+ } else {
+ throw new Error(
+ `Worker started message received from unknown worker '${message.workerId}'`
+ )
+ }
} else if (message.id != null) {
// Task execution response received
const promiseResponse = this.promiseResponseMap.get(message.id)
private pushWorkerNode (worker: Worker): number {
this.workerNodes.push({
worker,
- info: { id: worker.threadId ?? worker.id, started: false },
+ info: { id: this.getWorkerId(worker), started: true },
usage: this.getWorkerUsage(),
tasksQueue: new Queue<Task<Data>>()
})
return this.workerNodes.length
}
+ /**
+ * Gets the worker id.
+ *
+ * @param worker - The worker.
+ * @returns The worker id.
+ */
+ private getWorkerId (worker: Worker): number | undefined {
+ if (this.worker === WorkerTypes.thread) {
+ return worker.threadId
+ } else if (this.worker === WorkerTypes.cluster) {
+ return worker.id
+ }
+ }
+
// /**
// * Sets the given worker in the pool worker nodes.
// *