Response = unknown
> implements IPoolInternal<Worker, Data, Response> {
/** {@inheritDoc} */
- public readonly workers: Map<number, WorkerType<Worker>> = new Map<
- number,
- WorkerType<Worker>
- >()
+ public readonly workers: Array<WorkerType<Worker>> = []
/** {@inheritDoc} */
public readonly emitter?: PoolEmitter
- /**
- * Id of the next worker.
- */
- protected nextWorkerId: number = 0
-
/**
* The promise map.
*
this.registerWorkerMessageListener(workerCreated, message => {
if (
isKillBehavior(KillBehaviors.HARD, message.kill) ||
- this.getWorkerRunningTasks(workerCreated) === 0
+ this.getWorkerTasksUsage(workerCreated)?.running === 0
) {
// Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
void this.destroyWorker(workerCreated)
* @param worker - The worker.
* @returns The worker key.
*/
- private getWorkerKey (worker: Worker): number | undefined {
- return [...this.workers].find(([, value]) => value.worker === worker)?.[0]
- }
-
- /** {@inheritDoc} */
- public getWorkerRunningTasks (worker: Worker): number | undefined {
- return this.getWorkerTasksUsage(worker)?.running
- }
-
- /** {@inheritDoc} */
- public getWorkerRunTasks (worker: Worker): number | undefined {
- return this.getWorkerTasksUsage(worker)?.run
- }
-
- /** {@inheritDoc} */
- public getWorkerAverageTasksRunTime (worker: Worker): number | undefined {
- return this.getWorkerTasksUsage(worker)?.avgRunTime
+ private getWorkerKey (worker: Worker): number {
+ return this.workers.findIndex(workerItem => workerItem.worker === worker)
}
/** {@inheritDoc} */
workerChoiceStrategy: WorkerChoiceStrategy
): void {
this.opts.workerChoiceStrategy = workerChoiceStrategy
- for (const [key, value] of this.workers) {
- this.setWorker(key, value.worker, {
+ for (const workerItem of this.workers) {
+ this.setWorker(workerItem.worker, {
run: 0,
running: 0,
runTime: 0,
/** {@inheritDoc} */
public findFreeWorker (): Worker | false {
- for (const value of this.workers.values()) {
- if (value.tasksUsage.running === 0) {
+ for (const workerItem of this.workers) {
+ if (workerItem.tasksUsage.running === 0) {
// A worker is free, return the matching worker
- return value.worker
+ return workerItem.worker
}
}
return false
/** {@inheritDoc} */
public async destroy (): Promise<void> {
await Promise.all(
- [...this.workers].map(async ([, value]) => {
- await this.destroyWorker(value.worker)
+ this.workers.map(async workerItem => {
+ await this.destroyWorker(workerItem.worker)
})
)
}
* @param worker - The worker.
*/
protected beforePromiseWorkerResponseHook (worker: Worker): void {
- this.increaseWorkerRunningTasks(worker)
+ ++(this.getWorkerTasksUsage(worker) as TasksUsage).running
}
/**
message: MessageValue<Response>,
promise: PromiseWorkerResponseWrapper<Worker, Response>
): void {
- this.decreaseWorkerRunningTasks(promise.worker)
- this.stepWorkerRunTasks(promise.worker, 1)
- this.updateWorkerTasksRunTime(promise.worker, message.taskRunTime)
+ const workerTasksUsage = this.getWorkerTasksUsage(
+ promise.worker
+ ) as TasksUsage
+ --workerTasksUsage.running
+ ++workerTasksUsage.run
+ if (
+ this.workerChoiceStrategyContext.getWorkerChoiceStrategy()
+ .requiredStatistics.runTime
+ ) {
+ workerTasksUsage.runTime += message.taskRunTime ?? 0
+ if (workerTasksUsage.run !== 0) {
+ workerTasksUsage.avgRunTime =
+ workerTasksUsage.runTime / workerTasksUsage.run
+ }
+ }
}
/**
* @param worker - The worker that will be removed.
*/
protected removeWorker (worker: Worker): void {
- this.workers.delete(this.getWorkerKey(worker) as number)
- --this.nextWorkerId
+ this.workers.splice(this.getWorkerKey(worker), 1)
}
/**
this.removeWorker(worker)
})
- this.setWorker(this.nextWorkerId, worker, {
+ this.setWorker(worker, {
run: 0,
running: 0,
runTime: 0,
avgRunTime: 0
})
- ++this.nextWorkerId
this.afterWorkerSetup(worker)
}
}
- /**
- * Increases the number of tasks that the given worker has applied.
- *
- * @param worker - Worker which running tasks is increased.
- */
- private increaseWorkerRunningTasks (worker: Worker): void {
- this.stepWorkerRunningTasks(worker, 1)
- }
-
- /**
- * Decreases the number of tasks that the given worker has applied.
- *
- * @param worker - Worker which running tasks is decreased.
- */
- private decreaseWorkerRunningTasks (worker: Worker): void {
- this.stepWorkerRunningTasks(worker, -1)
- }
-
- /**
- * Gets tasks usage of the given worker.
- *
- * @param worker - Worker which tasks usage is returned.
- */
- private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined {
- if (this.checkWorker(worker)) {
- const workerKey = this.getWorkerKey(worker) as number
- const workerEntry = this.workers.get(workerKey) as WorkerType<Worker>
- return workerEntry.tasksUsage
- }
- }
-
- /**
- * Steps the number of tasks that the given worker has applied.
- *
- * @param worker - Worker which running tasks are stepped.
- * @param step - Number of running tasks step.
- */
- private stepWorkerRunningTasks (worker: Worker, step: number): void {
- // prettier-ignore
- (this.getWorkerTasksUsage(worker) as TasksUsage).running += step
- }
-
- /**
- * Steps the number of tasks that the given worker has run.
- *
- * @param worker - Worker which has run tasks.
- * @param step - Number of run tasks step.
- */
- private stepWorkerRunTasks (worker: Worker, step: number): void {
- // prettier-ignore
- (this.getWorkerTasksUsage(worker) as TasksUsage).run += step
- }
-
- /**
- * Updates tasks runtime for the given worker.
- *
- * @param worker - Worker which run the task.
- * @param taskRunTime - Worker task runtime.
- */
- private updateWorkerTasksRunTime (
- worker: Worker,
- taskRunTime: number | undefined
- ): void {
- if (
- this.workerChoiceStrategyContext.getWorkerChoiceStrategy()
- .requiredStatistics.runTime
- ) {
- const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage
- workerTasksUsage.runTime += taskRunTime ?? 0
- if (workerTasksUsage.run !== 0) {
- workerTasksUsage.avgRunTime =
- workerTasksUsage.runTime / workerTasksUsage.run
- }
+ /** {@inheritDoc} */
+ public getWorkerTasksUsage (worker: Worker): TasksUsage | undefined {
+ const workerKey = this.getWorkerKey(worker)
+ if (workerKey !== -1) {
+ return this.workers[workerKey].tasksUsage
}
+ throw new Error('Worker could not be found in the pool')
}
/**
* Sets the given worker.
*
- * @param workerKey - The worker key.
* @param worker - The worker.
* @param tasksUsage - The worker tasks usage.
*/
- private setWorker (
- workerKey: number,
- worker: Worker,
- tasksUsage: TasksUsage
- ): void {
- this.workers.set(workerKey, {
+ private setWorker (worker: Worker, tasksUsage: TasksUsage): void {
+ this.workers.push({
worker,
tasksUsage
})
}
-
- /**
- * Checks if the given worker is registered in the pool.
- *
- * @param worker - Worker to check.
- * @returns `true` if the worker is registered in the pool.
- */
- private checkWorker (worker: Worker): boolean {
- if (this.getWorkerKey(worker) == null) {
- throw new Error('Worker could not be found in the pool')
- }
- return true
- }
}