}
if (requireSync) {
this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
- this.getWorkerWorkerChoiceStrategies(),
+ this.getWorkerChoiceStrategies(),
this.opts.workerChoiceStrategyOptions
)
for (const workerNodeKey of this.workerNodes.keys()) {
this.opts.workerChoiceStrategyOptions
)
this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
- this.getWorkerWorkerChoiceStrategies(),
+ this.getWorkerChoiceStrategies(),
this.opts.workerChoiceStrategyOptions
)
for (const workerNodeKey of this.workerNodes.keys()) {
})
this.taskFunctions.set(name, fn)
this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
- this.getWorkerWorkerChoiceStrategies()
+ this.getWorkerChoiceStrategies()
)
for (const workerNodeKey of this.workerNodes.keys()) {
this.sendStatisticsMessageToWorker(workerNodeKey)
}
this.taskFunctions.delete(name)
this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
- this.getWorkerWorkerChoiceStrategies()
+ this.getWorkerChoiceStrategies()
)
for (const workerNodeKey of this.workerNodes.keys()) {
this.sendStatisticsMessageToWorker(workerNodeKey)
}
/**
- * Gets task function strategy, if any.
+ * Gets task function worker choice 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 = (
+ private readonly getTaskFunctionWorkerChoiceStrategy = (
name?: string
): WorkerChoiceStrategy | undefined => {
- if (name != null) {
- return this.listTaskFunctionsProperties().find(
- (taskFunctionProperties: TaskFunctionProperties) =>
- taskFunctionProperties.name === name
- )?.strategy
+ name = name ?? DEFAULT_TASK_NAME
+ const taskFunctionsProperties = this.listTaskFunctionsProperties()
+ if (name === DEFAULT_TASK_NAME) {
+ name = taskFunctionsProperties[1]?.name
}
+ return taskFunctionsProperties.find(
+ (taskFunctionProperties: TaskFunctionProperties) =>
+ taskFunctionProperties.name === name
+ )?.strategy
+ }
+
+ /**
+ * Gets worker node task function worker choice strategy, if any.
+ *
+ * @param workerNodeKey - The worker node key.
+ * @param name - The task function name.
+ * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise.
+ */
+ private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy = (
+ workerNodeKey: number,
+ name?: string
+ ): WorkerChoiceStrategy | undefined => {
+ const workerInfo = this.getWorkerInfo(workerNodeKey)
+ if (workerInfo == null) {
+ return
+ }
+ name = name ?? DEFAULT_TASK_NAME
+ if (name === DEFAULT_TASK_NAME) {
+ name = workerInfo.taskFunctionsProperties?.[1]?.name
+ }
+ return workerInfo.taskFunctionsProperties?.find(
+ (taskFunctionProperties: TaskFunctionProperties) =>
+ taskFunctionProperties.name === name
+ )?.strategy
}
/**
workerNodeKey: number,
name?: string
): number | undefined => {
- if (name != null) {
- return this.getWorkerInfo(workerNodeKey)?.taskFunctionsProperties?.find(
- (taskFunctionProperties: TaskFunctionProperties) =>
- taskFunctionProperties.name === name
- )?.priority
+ const workerInfo = this.getWorkerInfo(workerNodeKey)
+ if (workerInfo == null) {
+ return
+ }
+ name = name ?? DEFAULT_TASK_NAME
+ if (name === DEFAULT_TASK_NAME) {
+ name = workerInfo.taskFunctionsProperties?.[1]?.name
}
+ return workerInfo.taskFunctionsProperties?.find(
+ (taskFunctionProperties: TaskFunctionProperties) =>
+ taskFunctionProperties.name === name
+ )?.priority
}
/**
*
* @returns The worker choice strategies.
*/
- private readonly getWorkerWorkerChoiceStrategies =
+ private readonly getWorkerChoiceStrategies =
(): Set<WorkerChoiceStrategy> => {
return new Set([
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
return
}
const timestamp = performance.now()
- const taskFunctionStrategy =
- this.getTaskFunctionWorkerWorkerChoiceStrategy(name)
- const workerNodeKey = this.chooseWorkerNode(taskFunctionStrategy)
+ const workerNodeKey = this.chooseWorkerNode(name)
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,
+ strategy: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy(
+ workerNodeKey,
+ name
+ ),
transferList,
timestamp,
taskId: randomUUID()
}
/**
- * Chooses a worker node for the next task given the worker choice strategy.
+ * Chooses a worker node for the next task.
*
- * @param workerChoiceStrategy - The worker choice strategy.
+ * @param name - The task function name.
* @returns The chosen worker node key
*/
- private chooseWorkerNode (
- workerChoiceStrategy?: WorkerChoiceStrategy
- ): number {
+ private chooseWorkerNode (name?: string): number {
if (this.shallCreateDynamicWorker()) {
const workerNodeKey = this.createAndSetupDynamicWorkerNode()
if (
}
}
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
- return this.workerChoiceStrategiesContext!.execute(workerChoiceStrategy)
+ return this.workerChoiceStrategiesContext!.execute(
+ this.getTaskFunctionWorkerChoiceStrategy(name)
+ )
}
/**