} from './pool'
import type {
IWorker,
+ MessageHandler,
Task,
- TaskStatistics,
WorkerNode,
WorkerUsage
} from './worker'
import {
+ Measurements,
WorkerChoiceStrategies,
type WorkerChoiceStrategy,
type WorkerChoiceStrategyOptions
* Base class that implements some shared logic for all poolifier pools.
*
* @typeParam Worker - Type of worker which manages this pool.
- * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
- * @typeParam Response - Type of execution response. This can only be serializable data.
+ * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
+ * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
*/
export abstract class AbstractPool<
Worker extends IWorker,
'Invalid worker choice strategy options: must have a weight for each worker node'
)
}
+ if (
+ workerChoiceStrategyOptions.measurement != null &&
+ !Object.values(Measurements).includes(
+ workerChoiceStrategyOptions.measurement
+ )
+ ) {
+ throw new Error(
+ `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'`
+ )
+ }
}
private checkValidTasksQueueOptions (
if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
throw new TypeError('Invalid tasks queue options: must be a plain object')
}
- if ((tasksQueueOptions?.concurrency as number) <= 0) {
+ if (
+ tasksQueueOptions?.concurrency != null &&
+ !Number.isSafeInteger(tasksQueueOptions.concurrency)
+ ) {
+ throw new TypeError(
+ 'Invalid worker tasks concurrency: must be an integer'
+ )
+ }
+ if (
+ tasksQueueOptions?.concurrency != null &&
+ tasksQueueOptions.concurrency <= 0
+ ) {
throw new Error(
- `Invalid worker tasks concurrency '${
- tasksQueueOptions.concurrency as number
- }'`
+ `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'`
)
}
}
0
),
queuedTasks: this.workerNodes.reduce(
- (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size,
+ (accumulator, workerNode) =>
+ accumulator + workerNode.workerUsage.tasks.queued,
0
),
maxQueuedTasks: this.workerNodes.reduce(
(accumulator, workerNode) =>
- accumulator + workerNode.tasksQueue.maxSize,
+ accumulator + workerNode.workerUsage.tasks.maxQueued,
0
),
failedTasks: this.workerNodes.reduce(
if (workerChoiceStrategyOptions != null) {
this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
}
- for (const workerNode of this.workerNodes) {
+ for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) {
this.setWorkerNodeTasksUsage(
workerNode,
- this.getWorkerUsage(workerNode.worker)
+ this.getWorkerUsage(workerNodeKey)
)
this.setWorkerStatistics(workerNode.worker)
}
} else {
this.executeTask(workerNodeKey, submittedTask)
}
- this.workerChoiceStrategyContext.update(workerNodeKey)
this.checkAndEmitEvents()
// eslint-disable-next-line @typescript-eslint/return-await
return res
protected abstract destroyWorker (worker: Worker): void | Promise<void>
/**
- * Setup hook to execute code before worker node are created in the abstract constructor.
- * Can be overridden
+ * Setup hook to execute code before worker nodes are created in the abstract constructor.
+ * Can be overridden.
*
* @virtual
*/
* @param worker - The worker which should register a listener.
* @param listener - The message listener callback.
*/
- protected abstract registerWorkerMessageListener<
- Message extends Data | Response
- >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
+ private registerWorkerMessageListener<Message extends Data | Response>(
+ worker: Worker,
+ listener: (message: MessageValue<Message>) => void
+ ): void {
+ worker.on('message', listener as MessageHandler<Worker>)
+ }
/**
* Creates a new worker.
/**
* Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
- *
- * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
+ * Can be overridden.
*
* @param worker - The newly created worker.
*/
- protected abstract afterWorkerSetup (worker: Worker): void
+ protected afterWorkerSetup (worker: Worker): void {
+ // Listen to worker messages.
+ this.registerWorkerMessageListener(worker, this.workerListener())
+ }
/**
* Creates a new worker and sets it up completely in the pool worker nodes.
protected createAndSetupDynamicWorker (): Worker {
const worker = this.createAndSetupWorker()
this.registerWorkerMessageListener(worker, message => {
- const currentWorkerNodeKey = this.getWorkerNodeKey(worker)
+ const workerNodeKey = this.getWorkerNodeKey(worker)
if (
isKillBehavior(KillBehaviors.HARD, message.kill) ||
(message.kill != null &&
- this.workerNodes[currentWorkerNodeKey].workerUsage.tasks.executing ===
- 0)
+ ((this.opts.enableTasksQueue === false &&
+ this.workerNodes[workerNodeKey].workerUsage.tasks.executing ===
+ 0) ||
+ (this.opts.enableTasksQueue === true &&
+ this.workerNodes[workerNodeKey].workerUsage.tasks.executing ===
+ 0 &&
+ this.tasksQueueSize(workerNodeKey) === 0)))
) {
// Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
- this.flushTasksQueue(currentWorkerNodeKey)
- // FIXME: wait for tasks to be finished
void (this.destroyWorker(worker) as Promise<void>)
}
})
const promiseResponse = this.promiseResponseMap.get(message.id)
if (promiseResponse != null) {
if (message.taskError != null) {
- promiseResponse.reject(message.taskError.message)
if (this.emitter != null) {
this.emitter.emit(PoolEvents.taskError, message.taskError)
}
+ promiseResponse.reject(message.taskError.message)
} else {
promiseResponse.resolve(message.data as Response)
}
this.dequeueTask(workerNodeKey) as Task<Data>
)
}
+ this.workerChoiceStrategyContext.update(workerNodeKey)
}
}
}
* @returns The worker nodes length.
*/
private pushWorkerNode (worker: Worker): number {
- return this.workerNodes.push({
+ this.workerNodes.push({
worker,
- workerUsage: this.getWorkerUsage(worker),
+ workerUsage: this.getWorkerUsage(),
tasksQueue: new Queue<Task<Data>>()
})
+ const workerNodeKey = this.getWorkerNodeKey(worker)
+ this.setWorkerNodeTasksUsage(
+ this.workerNodes[workerNodeKey],
+ this.getWorkerUsage(workerNodeKey)
+ )
+ return this.workerNodes.length
}
// /**
return this.workerNodes[workerNodeKey].tasksQueue.size
}
+ private tasksMaxQueueSize (workerNodeKey: number): number {
+ return this.workerNodes[workerNodeKey].tasksQueue.maxSize
+ }
+
private flushTasksQueue (workerNodeKey: number): void {
if (this.tasksQueueSize(workerNodeKey) > 0) {
for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
)
}
}
+ this.workerNodes[workerNodeKey].tasksQueue.clear()
}
private flushTasksQueues (): void {
})
}
- private getWorkerUsage (worker: Worker): WorkerUsage {
+ private getWorkerUsage (workerNodeKey?: number): WorkerUsage {
+ const getTasksQueueSize = (workerNodeKey?: number): number => {
+ return workerNodeKey != null ? this.tasksQueueSize(workerNodeKey) : 0
+ }
+ const getTasksMaxQueueSize = (workerNodeKey?: number): number => {
+ return workerNodeKey != null ? this.tasksMaxQueueSize(workerNodeKey) : 0
+ }
return {
- tasks: this.getTaskStatistics(worker),
+ tasks: {
+ executed: 0,
+ executing: 0,
+ get queued (): number {
+ return getTasksQueueSize(workerNodeKey)
+ },
+ get maxQueued (): number {
+ return getTasksMaxQueueSize(workerNodeKey)
+ },
+ failed: 0
+ },
runTime: {
aggregate: 0,
average: 0,
}
}
}
-
- private getTaskStatistics (worker: Worker): TaskStatistics {
- const queueSize =
- this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size
- return {
- executed: 0,
- executing: 0,
- get queued (): number {
- return queueSize ?? 0
- },
- failed: 0
- }
- }
}
repositories / poolifier.git / blobdiff