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}'`
)
}
}
} 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) ||
(message.kill != null &&
- this.workerNodes[currentWorkerNodeKey].workerUsage.tasks.executing ===
- 0)
+ ((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)
- this.flushTasksQueue(currentWorkerNodeKey)
- // FIXME: wait for tasks to be finished
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)
}
}
}