+import crypto from 'node:crypto'
import type {
MessageValue,
PromiseWorkerResponseWrapper
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.
*
* When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
*/
protected promiseMap: Map<
- number,
+ string,
PromiseWorkerResponseWrapper<Worker, Response>
- > = new Map<number, PromiseWorkerResponseWrapper<Worker, Response>>()
-
- /**
- * Id of the next message.
- */
- protected nextMessageId: number = 0
+ > = new Map<string, PromiseWorkerResponseWrapper<Worker, Response>>()
/**
* Worker choice strategy instance implementing the worker choice algorithm.
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)
}
/**
- * Gets worker key.
+ * Gets the given worker key.
*
* @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.workers.get(this.getWorkerKey(worker) as number)?.tasksUsage
- ?.running
- }
-
- /** {@inheritDoc} */
- public getWorkerAverageTasksRunTime (worker: Worker): number | undefined {
- return this.workers.get(this.getWorkerKey(worker) as number)?.tasksUsage
- ?.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 execute (data: Data): Promise<Response> {
- // Configure worker to handle message with the specified task
const worker = this.chooseWorker()
- const res = this.internalExecute(worker, this.nextMessageId)
+ const messageId = crypto.randomUUID()
+ const res = this.internalExecute(worker, messageId)
this.checkAndEmitBusy()
this.sendToWorker(worker, {
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions
data: data ?? ({} as Data),
- id: this.nextMessageId
+ id: messageId
})
- ++this.nextMessageId
// eslint-disable-next-line @typescript-eslint/return-await
return res
}
/** {@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)
private async internalExecute (
worker: Worker,
- messageId: number
+ messageId: string
): Promise<Response> {
this.beforePromiseWorkerResponseHook(worker)
return await new Promise<Response>((resolve, reject) => {
}
}
- /**
- * 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)
- }
-
- /**
- * Get 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
- }
}
repositories / poolifier.git / blobdiff