type WorkerChoiceStrategy,
type WorkerChoiceStrategyOptions
} from './selection-strategies/selection-strategies-types.js'
-import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context.js'
+import { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js'
import {
checkFilePath,
+ checkValidPriority,
checkValidTasksQueueOptions,
checkValidWorkerChoiceStrategy,
getDefaultTasksQueueOptions,
/**
* The task execution response promise map:
* - `key`: The message id of each submitted task.
- * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
+ * - `value`: An object that contains task's worker node key, execution response promise resolve and reject callbacks, async resource.
*
* When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
*/
new Map<string, PromiseResponseWrapper<Response>>()
/**
- * Worker choice strategy context referencing a worker choice algorithm implementation.
+ * Worker choice strategies context referencing worker choice algorithms implementation.
*/
- protected workerChoiceStrategyContext?: WorkerChoiceStrategyContext<
+ protected workerChoiceStrategiesContext?: WorkerChoiceStrategiesContext<
Worker,
Data,
Response
/**
* The start timestamp of the pool.
*/
- private readonly startTimestamp
+ private startTimestamp?: number
/**
* Constructs a new poolifier pool.
if (this.opts.enableEvents === true) {
this.initializeEventEmitter()
}
- this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
+ this.workerChoiceStrategiesContext = new WorkerChoiceStrategiesContext<
Worker,
Data,
Response
>(
this,
- this.opts.workerChoiceStrategy,
+ // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+ [this.opts.workerChoiceStrategy!],
this.opts.workerChoiceStrategyOptions
)
if (this.opts.startWorkers === true) {
this.start()
}
-
- this.startTimestamp = performance.now()
}
private checkPoolType (): void {
started: this.started,
ready: this.ready,
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
- strategy: this.opts.workerChoiceStrategy!,
- strategyRetries: this.workerChoiceStrategyContext?.retriesCount ?? 0,
+ defaultStrategy: this.opts.workerChoiceStrategy!,
+ strategyRetries: this.workerChoiceStrategiesContext?.retriesCount ?? 0,
minSize: this.minimumNumberOfWorkers,
maxSize: this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
- ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+ ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
.runTime.aggregate === true &&
- this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
.waitTime.aggregate && {
utilization: round(this.utilization)
}),
accumulator + workerNode.usage.tasks.failed,
0
),
- ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+ ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
.runTime.aggregate === true && {
runTime: {
minimum: round(
)
)
),
- ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
.runTime.average && {
average: round(
average(
)
)
}),
- ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
.runTime.median && {
median: round(
median(
})
}
}),
- ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+ ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
.waitTime.aggregate === true && {
waitTime: {
minimum: round(
)
)
),
- ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
.waitTime.average && {
average: round(
average(
)
)
}),
- ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
.waitTime.median && {
median: round(
median(
* @returns The pool utilization.
*/
private get utilization (): number {
+ if (this.startTimestamp == null) {
+ return 0
+ }
const poolTimeCapacity =
(performance.now() - this.startTimestamp) *
(this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
workerChoiceStrategy: WorkerChoiceStrategy,
workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
): void {
+ let requireSync = false
checkValidWorkerChoiceStrategy(workerChoiceStrategy)
- this.opts.workerChoiceStrategy = workerChoiceStrategy
- this.workerChoiceStrategyContext?.setWorkerChoiceStrategy(
- this.opts.workerChoiceStrategy
- )
if (workerChoiceStrategyOptions != null) {
- this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
+ requireSync = !this.setWorkerChoiceStrategyOptions(
+ workerChoiceStrategyOptions
+ )
}
- for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) {
- workerNode.resetUsage()
- this.sendStatisticsMessageToWorker(workerNodeKey)
+ if (workerChoiceStrategy !== this.opts.workerChoiceStrategy) {
+ this.opts.workerChoiceStrategy = workerChoiceStrategy
+ this.workerChoiceStrategiesContext?.setDefaultWorkerChoiceStrategy(
+ this.opts.workerChoiceStrategy,
+ this.opts.workerChoiceStrategyOptions
+ )
+ requireSync = true
+ }
+ if (requireSync) {
+ this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
+ this.getWorkerWorkerChoiceStrategies(),
+ this.opts.workerChoiceStrategyOptions
+ )
+ for (const workerNodeKey of this.workerNodes.keys()) {
+ this.sendStatisticsMessageToWorker(workerNodeKey)
+ }
}
}
/** @inheritDoc */
public setWorkerChoiceStrategyOptions (
workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined
- ): void {
+ ): boolean {
this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
if (workerChoiceStrategyOptions != null) {
this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions
+ this.workerChoiceStrategiesContext?.setOptions(
+ this.opts.workerChoiceStrategyOptions
+ )
+ this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
+ this.getWorkerWorkerChoiceStrategies(),
+ this.opts.workerChoiceStrategyOptions
+ )
+ for (const workerNodeKey of this.workerNodes.keys()) {
+ this.sendStatisticsMessageToWorker(workerNodeKey)
+ }
+ return true
}
- this.workerChoiceStrategyContext?.setOptions(
- this.opts.workerChoiceStrategyOptions
- )
+ return false
}
/** @inheritDoc */
}
private setTaskStealing (): void {
- for (const [workerNodeKey] of this.workerNodes.entries()) {
+ for (const workerNodeKey of this.workerNodes.keys()) {
this.workerNodes[workerNodeKey].on('idle', this.handleWorkerNodeIdleEvent)
}
}
private unsetTaskStealing (): void {
- for (const [workerNodeKey] of this.workerNodes.entries()) {
+ for (const workerNodeKey of this.workerNodes.keys()) {
this.workerNodes[workerNodeKey].off(
'idle',
this.handleWorkerNodeIdleEvent
}
private setTasksStealingOnBackPressure (): void {
- for (const [workerNodeKey] of this.workerNodes.entries()) {
+ for (const workerNodeKey of this.workerNodes.keys()) {
this.workerNodes[workerNodeKey].on(
'backPressure',
this.handleWorkerNodeBackPressureEvent
}
private unsetTasksStealingOnBackPressure (): void {
- for (const [workerNodeKey] of this.workerNodes.entries()) {
+ for (const workerNodeKey of this.workerNodes.keys()) {
this.workerNodes[workerNodeKey].off(
'backPressure',
this.handleWorkerNodeBackPressureEvent
}
}
}
- for (const [workerNodeKey] of this.workerNodes.entries()) {
+ for (const workerNodeKey of this.workerNodes.keys()) {
this.registerWorkerMessageListener(
workerNodeKey,
taskFunctionOperationsListener
/** @inheritDoc */
public hasTaskFunction (name: string): boolean {
- for (const workerNode of this.workerNodes) {
- if (
- Array.isArray(workerNode.info.taskFunctionsProperties) &&
- workerNode.info.taskFunctionsProperties.some(
- taskFunctionProperties => taskFunctionProperties.name === name
- )
- ) {
- return true
- }
- }
- return false
+ return this.listTaskFunctionsProperties().some(
+ taskFunctionProperties => taskFunctionProperties.name === name
+ )
}
/** @inheritDoc */
if (typeof fn.taskFunction !== 'function') {
throw new TypeError('taskFunction property must be a function')
}
+ checkValidPriority(fn.priority)
+ checkValidWorkerChoiceStrategy(fn.strategy)
const opResult = await this.sendTaskFunctionOperationToWorkers({
taskFunctionOperation: 'add',
taskFunctionProperties: buildTaskFunctionProperties(name, fn),
taskFunction: fn.taskFunction.toString()
})
this.taskFunctions.set(name, fn)
+ this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
+ this.getWorkerWorkerChoiceStrategies()
+ )
return opResult
}
})
this.deleteTaskFunctionWorkerUsages(name)
this.taskFunctions.delete(name)
+ this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
+ this.getWorkerWorkerChoiceStrategies()
+ )
return opResult
}
return []
}
+ /**
+ * Gets task function strategy, if any.
+ *
+ * @param name - The task function name.
+ * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
+ */
+ private readonly getTaskFunctionWorkerWorkerChoiceStrategy = (
+ name?: string
+ ): WorkerChoiceStrategy | undefined => {
+ if (name != null) {
+ return this.listTaskFunctionsProperties().find(
+ (taskFunctionProperties: TaskFunctionProperties) =>
+ taskFunctionProperties.name === name
+ )?.strategy
+ }
+ }
+
+ /**
+ * Gets worker node task function priority, if any.
+ *
+ * @param workerNodeKey - The worker node key.
+ * @param name - The task function name.
+ * @returns The task function worker choice priority if the task function worker choice priority is defined, `undefined` otherwise.
+ */
+ private readonly getWorkerNodeTaskFunctionPriority = (
+ workerNodeKey: number,
+ name?: string
+ ): number | undefined => {
+ if (name != null) {
+ return this.getWorkerInfo(workerNodeKey)?.taskFunctionsProperties?.find(
+ (taskFunctionProperties: TaskFunctionProperties) =>
+ taskFunctionProperties.name === name
+ )?.priority
+ }
+ }
+
+ /**
+ * Gets the worker choice strategies registered in this pool.
+ *
+ * @returns The worker choice strategies.
+ */
+ private readonly getWorkerWorkerChoiceStrategies =
+ (): Set<WorkerChoiceStrategy> => {
+ return new Set([
+ // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+ this.opts.workerChoiceStrategy!,
+ ...(this.listTaskFunctionsProperties()
+ .map(
+ (taskFunctionProperties: TaskFunctionProperties) =>
+ taskFunctionProperties.strategy
+ )
+ .filter(
+ (strategy: WorkerChoiceStrategy | undefined) => strategy != null
+ ) as WorkerChoiceStrategy[])
+ ])
+ }
+
/** @inheritDoc */
public async setDefaultTaskFunction (name: string): Promise<boolean> {
return await this.sendTaskFunctionOperationToWorkers({
return
}
const timestamp = performance.now()
- const workerNodeKey = this.chooseWorkerNode()
+ const taskFunctionStrategy =
+ this.getTaskFunctionWorkerWorkerChoiceStrategy(name)
+ const workerNodeKey = this.chooseWorkerNode(taskFunctionStrategy)
const task: Task<Data> = {
name: name ?? DEFAULT_TASK_NAME,
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions
data: data ?? ({} as Data),
+ priority: this.getWorkerNodeTaskFunctionPriority(workerNodeKey, name),
+ strategy: taskFunctionStrategy,
transferList,
timestamp,
taskId: randomUUID()
}
this.starting = true
this.startMinimumNumberOfWorkers()
+ this.startTimestamp = performance.now()
this.starting = false
this.started = true
}
this.emitter?.emit(PoolEvents.destroy, this.info)
this.emitter?.emitDestroy()
this.readyEventEmitted = false
+ delete this.startTimestamp
this.destroying = false
this.started = false
}
const workerUsage = this.workerNodes[workerNodeKey].usage
++workerUsage.tasks.executing
updateWaitTimeWorkerUsage(
- this.workerChoiceStrategyContext,
+ this.workerChoiceStrategiesContext,
workerUsage,
task
)
].getTaskFunctionWorkerUsage(task.name!)!
++taskFunctionWorkerUsage.tasks.executing
updateWaitTimeWorkerUsage(
- this.workerChoiceStrategyContext,
+ this.workerChoiceStrategiesContext,
taskFunctionWorkerUsage,
task
)
const workerUsage = this.workerNodes[workerNodeKey].usage
updateTaskStatisticsWorkerUsage(workerUsage, message)
updateRunTimeWorkerUsage(
- this.workerChoiceStrategyContext,
+ this.workerChoiceStrategiesContext,
workerUsage,
message
)
updateEluWorkerUsage(
- this.workerChoiceStrategyContext,
+ this.workerChoiceStrategiesContext,
workerUsage,
message
)
].getTaskFunctionWorkerUsage(message.taskPerformance!.name)!
updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
updateRunTimeWorkerUsage(
- this.workerChoiceStrategyContext,
+ this.workerChoiceStrategiesContext,
taskFunctionWorkerUsage,
message
)
updateEluWorkerUsage(
- this.workerChoiceStrategyContext,
+ this.workerChoiceStrategiesContext,
taskFunctionWorkerUsage,
message
)
needWorkerChoiceStrategyUpdate = true
}
if (needWorkerChoiceStrategyUpdate) {
- this.workerChoiceStrategyContext?.update(workerNodeKey)
+ this.workerChoiceStrategiesContext?.update(workerNodeKey)
}
}
}
/**
- * Chooses a worker node for the next task.
- *
- * The default worker choice strategy uses a round robin algorithm to distribute the tasks.
+ * Chooses a worker node for the next task given the worker choice strategy.
*
+ * @param workerChoiceStrategy - The worker choice strategy.
* @returns The chosen worker node key
*/
- private chooseWorkerNode (): number {
+ private chooseWorkerNode (
+ workerChoiceStrategy?: WorkerChoiceStrategy
+ ): number {
if (this.shallCreateDynamicWorker()) {
const workerNodeKey = this.createAndSetupDynamicWorkerNode()
if (
- this.workerChoiceStrategyContext?.getStrategyPolicy()
- .dynamicWorkerUsage === true
+ this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage ===
+ true
) {
return workerNodeKey
}
}
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
- return this.workerChoiceStrategyContext!.execute()
+ return this.workerChoiceStrategiesContext!.execute(workerChoiceStrategy)
}
/**
const workerNode = this.workerNodes[workerNodeKey]
workerNode.info.dynamic = true
if (
- this.workerChoiceStrategyContext?.getStrategyPolicy()
- .dynamicWorkerReady === true ||
- this.workerChoiceStrategyContext?.getStrategyPolicy()
- .dynamicWorkerUsage === true
+ this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerReady ===
+ true ||
+ this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage ===
+ true
) {
workerNode.info.ready = true
}
this.sendToWorker(workerNodeKey, {
statistics: {
runTime:
- this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()
+ this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
.runTime.aggregate ?? false,
elu:
- this.workerChoiceStrategyContext?.getTaskStatisticsRequirements().elu
- .aggregate ?? false
+ this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
+ .elu.aggregate ?? false
}
})
}
)
if (sourceWorkerNode != null) {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
- const task = sourceWorkerNode.popTask()!
+ const task = sourceWorkerNode.dequeueTask(1)!
this.handleTask(workerNodeKey, task)
this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey)
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
}
workerInfo.stealing = true
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
- const task = sourceWorkerNode.popTask()!
+ const task = sourceWorkerNode.dequeueTask(1)!
this.handleTask(workerNodeKey, task)
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!)
this.opts.tasksQueueOptions?.size ??
getDefaultTasksQueueOptions(
this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
- ).size
+ ).size,
+ tasksQueueBucketSize:
+ (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers) * 2
}
)
// Flag the worker node as ready at pool startup.
const workerNodeKey = this.workerNodes.indexOf(workerNode)
if (workerNodeKey !== -1) {
this.workerNodes.splice(workerNodeKey, 1)
- this.workerChoiceStrategyContext?.remove(workerNodeKey)
+ this.workerChoiceStrategiesContext?.remove(workerNodeKey)
}
this.checkAndEmitEmptyEvent()
}
return tasksQueueSize
}
- private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
- return this.workerNodes[workerNodeKey].dequeueTask()
+ private dequeueTask (
+ workerNodeKey: number,
+ bucket?: number
+ ): Task<Data> | undefined {
+ return this.workerNodes[workerNodeKey].dequeueTask(bucket)
}
private tasksQueueSize (workerNodeKey: number): number {
}
private flushTasksQueues (): void {
- for (const [workerNodeKey] of this.workerNodes.entries()) {
+ for (const workerNodeKey of this.workerNodes.keys()) {
this.flushTasksQueue(workerNodeKey)
}
}
repositories / poolifier.git / blobdiff