type TasksQueueOptions,
type WorkerType
} from './pool'
-import type {
- IWorker,
- Task,
- TaskStatistics,
- WorkerNode,
- WorkerUsage
-} from './worker'
+import type { IWorker, Task, WorkerNode, WorkerUsage } from './worker'
import {
+ Measurements,
WorkerChoiceStrategies,
type WorkerChoiceStrategy,
type WorkerChoiceStrategyOptions
'Invalid worker choice strategy options: must have a weight for each worker node'
)
}
+ if (
+ workerChoiceStrategyOptions.measurement != null &&
+ !Object.values(Measurements).includes(
+ workerChoiceStrategyOptions.measurement
+ )
+ ) {
+ throw new Error(
+ `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
+ )
+ }
}
private checkValidTasksQueueOptions (
if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
throw new TypeError('Invalid tasks queue options: must be a plain object')
}
- if ((tasksQueueOptions?.concurrency as number) <= 0) {
+ if (
+ tasksQueueOptions?.concurrency != null &&
+ !Number.isSafeInteger(tasksQueueOptions.concurrency)
+ ) {
+ throw new TypeError(
+ 'Invalid worker tasks concurrency: must be an integer'
+ )
+ }
+ if (
+ tasksQueueOptions?.concurrency != null &&
+ tasksQueueOptions.concurrency <= 0
+ ) {
throw new Error(
- `Invalid worker tasks concurrency '${
- tasksQueueOptions.concurrency as number
- }'`
+ `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
)
}
}
if (workerChoiceStrategyOptions != null) {
this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
}
- for (const workerNode of this.workerNodes) {
+ for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) {
this.setWorkerNodeTasksUsage(
workerNode,
- this.getWorkerUsage(workerNode.worker)
+ this.getWorkerUsage(workerNodeKey)
)
this.setWorkerStatistics(workerNode.worker)
}
} else {
this.executeTask(workerNodeKey, submittedTask)
}
- this.workerChoiceStrategyContext.update(workerNodeKey)
this.checkAndEmitEvents()
// eslint-disable-next-line @typescript-eslint/return-await
return res
protected createAndSetupDynamicWorker (): Worker {
const worker = this.createAndSetupWorker()
this.registerWorkerMessageListener(worker, message => {
- const currentWorkerNodeKey = this.getWorkerNodeKey(worker)
+ const workerNodeKey = this.getWorkerNodeKey(worker)
if (
isKillBehavior(KillBehaviors.HARD, message.kill) ||
- (this.opts.enableTasksQueue === false &&
- message.kill != null &&
- this.workerNodes[currentWorkerNodeKey].workerUsage.tasks.executing ===
- 0) ||
- (this.opts.enableTasksQueue === true &&
- message.kill != null &&
- this.workerNodes[currentWorkerNodeKey].workerUsage.tasks.executing ===
- 0 &&
- this.tasksQueueSize(currentWorkerNodeKey) === 0)
+ (message.kill != null &&
+ ((this.opts.enableTasksQueue === false &&
+ this.workerNodes[workerNodeKey].workerUsage.tasks.executing ===
+ 0) ||
+ (this.opts.enableTasksQueue === true &&
+ this.workerNodes[workerNodeKey].workerUsage.tasks.executing ===
+ 0 &&
+ this.tasksQueueSize(workerNodeKey) === 0)))
) {
// Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
void (this.destroyWorker(worker) as Promise<void>)
const promiseResponse = this.promiseResponseMap.get(message.id)
if (promiseResponse != null) {
if (message.taskError != null) {
- promiseResponse.reject(message.taskError.message)
if (this.emitter != null) {
this.emitter.emit(PoolEvents.taskError, message.taskError)
}
+ promiseResponse.reject(message.taskError.message)
} else {
promiseResponse.resolve(message.data as Response)
}
this.dequeueTask(workerNodeKey) as Task<Data>
)
}
+ this.workerChoiceStrategyContext.update(workerNodeKey)
}
}
}
* @returns The worker nodes length.
*/
private pushWorkerNode (worker: Worker): number {
- return this.workerNodes.push({
+ this.workerNodes.push({
worker,
- workerUsage: this.getWorkerUsage(worker),
+ workerUsage: this.getWorkerUsage(),
tasksQueue: new Queue<Task<Data>>()
})
+ const workerNodeKey = this.getWorkerNodeKey(worker)
+ this.setWorkerNodeTasksUsage(
+ this.workerNodes[workerNodeKey],
+ this.getWorkerUsage(workerNodeKey)
+ )
+ return this.workerNodes.length
}
// /**
})
}
- private getWorkerUsage (worker: Worker): WorkerUsage {
+ private getWorkerUsage (workerNodeKey?: number): WorkerUsage {
+ const getQueueSize = (workerNodeKey: number): number => {
+ return this.tasksQueueSize(workerNodeKey)
+ }
return {
- tasks: this.getTaskStatistics(worker),
+ tasks: {
+ executed: 0,
+ executing: 0,
+ get queued (): number {
+ return workerNodeKey == null ? 0 : getQueueSize(workerNodeKey)
+ },
+ failed: 0
+ },
runTime: {
aggregate: 0,
average: 0,
}
}
}
-
- private getTaskStatistics (worker: Worker): TaskStatistics {
- const queueSize =
- this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size
- return {
- executed: 0,
- executing: 0,
- get queued (): number {
- return queueSize ?? 0
- },
- failed: 0
- }
- }
}