-import { isKillBehavior, KillBehaviors } from '../worker/worker-options'
import type { IWorker } from './abstract-pool'
import type { IPoolInternal } from './pool-internal'
/** @inheritdoc */
public choose (): Worker {
+ const isPoolDynamic = this.pool.dynamic
let minNumberOfTasks = Infinity
// A worker is always found because it picks the one with fewer tasks
let lessRecentlyUsedWorker!: Worker
for (const [worker, numberOfTasks] of this.pool.tasks) {
- if (numberOfTasks === 0) {
+ if (!isPoolDynamic && numberOfTasks === 0) {
return worker
} else if (numberOfTasks < minNumberOfTasks) {
minNumberOfTasks = numberOfTasks
}
}
-/**
- * Get the worker choice strategy instance.
- *
- * @param pool The pool instance.
- * @param workerChoiceStrategy The worker choice strategy.
- * @returns The worker choice strategy instance.
- */
-function getWorkerChoiceStrategy<Worker extends IWorker, Data, Response> (
- pool: IPoolInternal<Worker, Data, Response>,
- workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
-): IWorkerChoiceStrategy<Worker> {
- switch (workerChoiceStrategy) {
- case WorkerChoiceStrategies.ROUND_ROBIN:
- return new RoundRobinWorkerChoiceStrategy(pool)
- case WorkerChoiceStrategies.LESS_RECENTLY_USED:
- return new LessRecentlyUsedWorkerChoiceStrategy(pool)
- default:
- throw new Error(
- `Worker choice strategy '${workerChoiceStrategy}' not found`
- )
- }
-}
-
/**
* Dynamically choose a worker.
*
* Constructs a worker choice strategy for dynamical pools.
*
* @param pool The pool instance.
+ * @param createDynamicallyWorkerCallback The worker creation callback for dynamic pool.
* @param workerChoiceStrategy The worker choice strategy when the pull is full.
*/
public constructor (
private readonly pool: IPoolInternal<Worker, Data, Response>,
+ private createDynamicallyWorkerCallback: () => Worker,
workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
) {
- this.workerChoiceStrategy = getWorkerChoiceStrategy(
+ this.workerChoiceStrategy = SelectionStrategiesUtils.getWorkerChoiceStrategy(
this.pool,
workerChoiceStrategy
)
}
- /**
- * Find a free worker based on number of tasks the worker has applied.
- *
- * If a worker was found that has `0` tasks, it is detected as free and will be returned.
- *
- * If no free worker was found, `null` will be returned.
- *
- * @returns A free worker if there was one, otherwise `null`.
- */
- private findFreeWorkerBasedOnTasks (): Worker | null {
- for (const [worker, numberOfTasks] of this.pool.tasks) {
- if (numberOfTasks === 0) {
- // A worker is free, use it
- return worker
- }
- }
- return null
- }
-
/** @inheritdoc */
public choose (): Worker {
- const freeWorker = this.findFreeWorkerBasedOnTasks()
+ const freeWorker = SelectionStrategiesUtils.findFreeWorkerBasedOnTasks(
+ this.pool.tasks
+ )
if (freeWorker) {
return freeWorker
}
if (this.pool.workers.length === this.pool.max) {
- this.pool.emitter.emit('FullPool')
+ this.pool.emitter.emit('busy')
return this.workerChoiceStrategy.choose()
}
// All workers are busy, create a new worker
- const workerCreated = this.pool.createAndSetupWorker()
- this.pool.registerWorkerMessageListener(workerCreated, message => {
- const tasksInProgress = this.pool.tasks.get(workerCreated)
- if (
- isKillBehavior(KillBehaviors.HARD, message.kill) ||
- tasksInProgress === 0
- ) {
- // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
- void this.pool.destroyWorker(workerCreated)
- }
- })
- return workerCreated
+ return this.createDynamicallyWorkerCallback()
}
}
* Worker choice strategy context constructor.
*
* @param pool The pool instance.
+ * @param createDynamicallyWorkerCallback The worker creation callback for dynamic pool.
* @param workerChoiceStrategy The worker choice strategy.
*/
public constructor (
private readonly pool: IPoolInternal<Worker, Data, Response>,
+ private createDynamicallyWorkerCallback: () => Worker,
workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
) {
this.setWorkerChoiceStrategy(workerChoiceStrategy)
private getPoolWorkerChoiceStrategy (
workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
): IWorkerChoiceStrategy<Worker> {
- if (this.pool.isDynamic()) {
+ if (this.pool.dynamic) {
return new DynamicPoolWorkerChoiceStrategy(
this.pool,
+ this.createDynamicallyWorkerCallback,
workerChoiceStrategy
)
}
- return getWorkerChoiceStrategy(this.pool, workerChoiceStrategy)
+ return SelectionStrategiesUtils.getWorkerChoiceStrategy(
+ this.pool,
+ workerChoiceStrategy
+ )
}
/**
return this.workerChoiceStrategy.choose()
}
}
+
+/**
+ * Worker selection strategies helpers.
+ */
+class SelectionStrategiesUtils {
+ /**
+ * Find a free worker based on number of tasks the worker has applied.
+ *
+ * If a worker was found that has `0` tasks, it is detected as free and will be returned.
+ *
+ * If no free worker was found, `null` will be returned.
+ *
+ * @param workerTasksMap The pool worker tasks map.
+ * @returns A free worker if there was one, otherwise `null`.
+ */
+ public static findFreeWorkerBasedOnTasks<Worker extends IWorker> (
+ workerTasksMap: Map<Worker, number>
+ ): Worker | null {
+ for (const [worker, numberOfTasks] of workerTasksMap) {
+ if (numberOfTasks === 0) {
+ // A worker is free, use it
+ return worker
+ }
+ }
+ return null
+ }
+
+ /**
+ * Get the worker choice strategy instance.
+ *
+ * @param pool The pool instance.
+ * @param workerChoiceStrategy The worker choice strategy.
+ * @returns The worker choice strategy instance.
+ */
+ public static getWorkerChoiceStrategy<
+ Worker extends IWorker,
+ Data,
+ Response
+ > (
+ pool: IPoolInternal<Worker, Data, Response>,
+ workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
+ ): IWorkerChoiceStrategy<Worker> {
+ switch (workerChoiceStrategy) {
+ case WorkerChoiceStrategies.ROUND_ROBIN:
+ return new RoundRobinWorkerChoiceStrategy(pool)
+ case WorkerChoiceStrategies.LESS_RECENTLY_USED:
+ return new LessRecentlyUsedWorkerChoiceStrategy(pool)
+ default:
+ throw new Error(
+ `Worker choice strategy '${workerChoiceStrategy}' not found`
+ )
+ }
+ }
+}
repositories / poolifier.git / blobdiff