repositories
/
poolifier.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
docs: refine benchmark README.md
[poolifier.git]
/
src
/
pools
/
abstract-pool.ts
diff --git
a/src/pools/abstract-pool.ts
b/src/pools/abstract-pool.ts
index 9dcc6eddd28fcc9b0e27e6eebbe55e9dc31b6458..fb1bf8458ad7a970f27ae7a4dcbb300faf5bf848 100644
(file)
--- a/
src/pools/abstract-pool.ts
+++ b/
src/pools/abstract-pool.ts
@@
-92,14
+92,14
@@
export abstract class AbstractPool<
*/
protected readonly max?: number
*/
protected readonly max?: number
- /**
- * Whether the pool is starting or not.
- */
- private readonly starting: boolean
/**
* Whether the pool is started or not.
*/
private started: boolean
/**
* Whether the pool is started or not.
*/
private started: boolean
+ /**
+ * Whether the pool is starting or not.
+ */
+ private starting: boolean
/**
* The start timestamp of the pool.
*/
/**
* The start timestamp of the pool.
*/
@@
-145,10
+145,11
@@
export abstract class AbstractPool<
this.setupHook()
this.setupHook()
- this.starting = true
- this.startPool()
+ this.started = false
this.starting = false
this.starting = false
- this.started = true
+ if (this.opts.startWorkers === true) {
+ this.start()
+ }
this.startTimestamp = performance.now()
}
this.startTimestamp = performance.now()
}
@@
-212,6
+213,7
@@
export abstract class AbstractPool<
private checkPoolOptions (opts: PoolOptions<Worker>): void {
if (isPlainObject(opts)) {
private checkPoolOptions (opts: PoolOptions<Worker>): void {
if (isPlainObject(opts)) {
+ this.opts.startWorkers = opts.startWorkers ?? true
this.opts.workerChoiceStrategy =
opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
this.opts.workerChoiceStrategy =
opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
@@
-314,11
+316,6
@@
export abstract class AbstractPool<
`Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero`
)
}
`Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero`
)
}
- if (tasksQueueOptions?.queueMaxSize != null) {
- throw new Error(
- 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
- )
- }
if (
tasksQueueOptions?.size != null &&
!Number.isSafeInteger(tasksQueueOptions?.size)
if (
tasksQueueOptions?.size != null &&
!Number.isSafeInteger(tasksQueueOptions?.size)
@@
-334,24
+331,13
@@
export abstract class AbstractPool<
}
}
}
}
- private startPool (): void {
- while (
- this.workerNodes.reduce(
- (accumulator, workerNode) =>
- !workerNode.info.dynamic ? accumulator + 1 : accumulator,
- 0
- ) < this.numberOfWorkers
- ) {
- this.createAndSetupWorkerNode()
- }
- }
-
/** @inheritDoc */
public get info (): PoolInfo {
return {
version,
type: this.type,
worker: this.worker,
/** @inheritDoc */
public get info (): PoolInfo {
return {
version,
type: this.type,
worker: this.worker,
+ started: this.started,
ready: this.ready,
strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
minSize: this.minSize,
ready: this.ready,
strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
minSize: this.minSize,
@@
-418,14
+404,14
@@
export abstract class AbstractPool<
minimum: round(
min(
...this.workerNodes.map(
minimum: round(
min(
...this.workerNodes.map(
-
(workerNode)
=> workerNode.usage.runTime?.minimum ?? Infinity
+
workerNode
=> workerNode.usage.runTime?.minimum ?? Infinity
)
)
),
maximum: round(
max(
...this.workerNodes.map(
)
)
),
maximum: round(
max(
...this.workerNodes.map(
-
(workerNode)
=> workerNode.usage.runTime?.maximum ?? -Infinity
+
workerNode
=> workerNode.usage.runTime?.maximum ?? -Infinity
)
)
),
)
)
),
@@
-461,14
+447,14
@@
export abstract class AbstractPool<
minimum: round(
min(
...this.workerNodes.map(
minimum: round(
min(
...this.workerNodes.map(
-
(workerNode)
=> workerNode.usage.waitTime?.minimum ?? Infinity
+
workerNode
=> workerNode.usage.waitTime?.minimum ?? Infinity
)
)
),
maximum: round(
max(
...this.workerNodes.map(
)
)
),
maximum: round(
max(
...this.workerNodes.map(
-
(workerNode)
=> workerNode.usage.waitTime?.maximum ?? -Infinity
+
workerNode
=> workerNode.usage.waitTime?.maximum ?? -Infinity
)
)
),
)
)
),
@@
-590,7
+576,7
@@
export abstract class AbstractPool<
*/
private getWorkerNodeKeyByWorker (worker: Worker): number {
return this.workerNodes.findIndex(
*/
private getWorkerNodeKeyByWorker (worker: Worker): number {
return this.workerNodes.findIndex(
-
(workerNode)
=> workerNode.worker === worker
+
workerNode
=> workerNode.worker === worker
)
}
)
}
@@
-602,7
+588,7
@@
export abstract class AbstractPool<
*/
private getWorkerNodeKeyByWorkerId (workerId: number): number {
return this.workerNodes.findIndex(
*/
private getWorkerNodeKeyByWorkerId (workerId: number): number {
return this.workerNodes.findIndex(
-
(workerNode)
=> workerNode.info.id === workerId
+
workerNode
=> workerNode.info.id === workerId
)
}
)
}
@@
-675,7
+661,9
@@
export abstract class AbstractPool<
return {
...{
size: Math.pow(this.maxSize, 2),
return {
...{
size: Math.pow(this.maxSize, 2),
- concurrency: 1
+ concurrency: 1,
+ taskStealing: true,
+ tasksStealingOnBackPressure: true
},
...tasksQueueOptions
}
},
...tasksQueueOptions
}
@@
-706,7
+694,7
@@
export abstract class AbstractPool<
if (this.opts.enableTasksQueue === true) {
return (
this.workerNodes.findIndex(
if (this.opts.enableTasksQueue === true) {
return (
this.workerNodes.findIndex(
-
(workerNode)
=>
+
workerNode
=>
workerNode.info.ready &&
workerNode.usage.tasks.executing <
(this.opts.tasksQueueOptions?.concurrency as number)
workerNode.info.ready &&
workerNode.usage.tasks.executing <
(this.opts.tasksQueueOptions?.concurrency as number)
@@
-715,7
+703,7
@@
export abstract class AbstractPool<
} else {
return (
this.workerNodes.findIndex(
} else {
return (
this.workerNodes.findIndex(
-
(workerNode)
=>
+
workerNode
=>
workerNode.info.ready && workerNode.usage.tasks.executing === 0
) === -1
)
workerNode.info.ready && workerNode.usage.tasks.executing === 0
) === -1
)
@@
-751,7
+739,7
@@
export abstract class AbstractPool<
): Promise<Response> {
return await new Promise<Response>((resolve, reject) => {
if (!this.started) {
): Promise<Response> {
return await new Promise<Response>((resolve, reject) => {
if (!this.started) {
- reject(new Error('Cannot execute a task on
destroy
ed pool'))
+ reject(new Error('Cannot execute a task on
not start
ed pool'))
return
}
if (name != null && typeof name !== 'string') {
return
}
if (name != null && typeof name !== 'string') {
@@
-772,14
+760,13
@@
export abstract class AbstractPool<
}
const timestamp = performance.now()
const workerNodeKey = this.chooseWorkerNode()
}
const timestamp = performance.now()
const workerNodeKey = this.chooseWorkerNode()
- const workerInfo = this.getWorkerInfo(workerNodeKey)
const task: Task<Data> = {
name: name ?? DEFAULT_TASK_NAME,
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions
data: data ?? ({} as Data),
transferList,
timestamp,
const task: Task<Data> = {
name: name ?? DEFAULT_TASK_NAME,
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions
data: data ?? ({} as Data),
transferList,
timestamp,
- workerId:
workerInfo
.id as number,
+ workerId:
this.getWorkerInfo(workerNodeKey)
.id as number,
taskId: randomUUID()
}
this.promiseResponseMap.set(task.taskId as string, {
taskId: randomUUID()
}
this.promiseResponseMap.set(task.taskId as string, {
@@
-799,6
+786,22
@@
export abstract class AbstractPool<
})
}
})
}
+ /** @inheritdoc */
+ public start (): void {
+ this.starting = true
+ while (
+ this.workerNodes.reduce(
+ (accumulator, workerNode) =>
+ !workerNode.info.dynamic ? accumulator + 1 : accumulator,
+ 0
+ ) < this.numberOfWorkers
+ ) {
+ this.createAndSetupWorkerNode()
+ }
+ this.starting = false
+ this.started = true
+ }
+
/** @inheritDoc */
public async destroy (): Promise<void> {
await Promise.all(
/** @inheritDoc */
public async destroy (): Promise<void> {
await Promise.all(
@@
-815,7
+818,7
@@
export abstract class AbstractPool<
workerId: number
): Promise<void> {
await new Promise<void>((resolve, reject) => {
workerId: number
): Promise<void> {
await new Promise<void>((resolve, reject) => {
- this.registerWorkerMessageListener(workerNodeKey,
(message)
=> {
+ this.registerWorkerMessageListener(workerNodeKey,
message
=> {
if (message.kill === 'success') {
resolve()
} else if (message.kill === 'failure') {
if (message.kill === 'success') {
resolve()
} else if (message.kill === 'failure') {
@@
-1064,7
+1067,7
@@
export abstract class AbstractPool<
worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
- worker.on('error',
(error)
=> {
+ worker.on('error',
error
=> {
const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
const workerInfo = this.getWorkerInfo(workerNodeKey)
workerInfo.ready = false
const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
const workerInfo = this.getWorkerInfo(workerNodeKey)
workerInfo.ready = false
@@
-1072,8
+1075,8
@@
export abstract class AbstractPool<
this.emitter?.emit(PoolEvents.error, error)
if (
this.opts.restartWorkerOnError === true &&
this.emitter?.emit(PoolEvents.error, error)
if (
this.opts.restartWorkerOnError === true &&
-
!this.starting
&&
- this.started
+
this.started
&&
+ !this.starting
) {
if (workerInfo.dynamic) {
this.createAndSetupDynamicWorkerNode()
) {
if (workerInfo.dynamic) {
this.createAndSetupDynamicWorkerNode()
@@
-1104,7
+1107,7
@@
export abstract class AbstractPool<
*/
protected createAndSetupDynamicWorkerNode (): number {
const workerNodeKey = this.createAndSetupWorkerNode()
*/
protected createAndSetupDynamicWorkerNode (): number {
const workerNodeKey = this.createAndSetupWorkerNode()
- this.registerWorkerMessageListener(workerNodeKey,
(message)
=> {
+ this.registerWorkerMessageListener(workerNodeKey,
message
=> {
const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
message.workerId
)
const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
message.workerId
)
@@
-1119,7
+1122,7
@@
export abstract class AbstractPool<
workerUsage.tasks.executing === 0 &&
this.tasksQueueSize(localWorkerNodeKey) === 0)))
) {
workerUsage.tasks.executing === 0 &&
this.tasksQueueSize(localWorkerNodeKey) === 0)))
) {
- this.destroyWorkerNode(localWorkerNodeKey).catch(
(error)
=> {
+ this.destroyWorkerNode(localWorkerNodeKey).catch(
error
=> {
this.emitter?.emit(PoolEvents.error, error)
})
}
this.emitter?.emit(PoolEvents.error, error)
})
}
@@
-1167,10
+1170,14
@@
export abstract class AbstractPool<
// Send the statistics message to worker.
this.sendStatisticsMessageToWorker(workerNodeKey)
if (this.opts.enableTasksQueue === true) {
// Send the statistics message to worker.
this.sendStatisticsMessageToWorker(workerNodeKey)
if (this.opts.enableTasksQueue === true) {
- this.workerNodes[workerNodeKey].onEmptyQueue =
- this.taskStealingOnEmptyQueue.bind(this)
- this.workerNodes[workerNodeKey].onBackPressure =
- this.tasksStealingOnBackPressure.bind(this)
+ if (this.opts.tasksQueueOptions?.taskStealing === true) {
+ this.workerNodes[workerNodeKey].onEmptyQueue =
+ this.taskStealingOnEmptyQueue.bind(this)
+ }
+ if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) {
+ this.workerNodes[workerNodeKey].onBackPressure =
+ this.tasksStealingOnBackPressure.bind(this)
+ }
}
}
}
}
@@
-1253,7
+1260,7
@@
export abstract class AbstractPool<
workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
)
const sourceWorkerNode = workerNodes.find(
workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
)
const sourceWorkerNode = workerNodes.find(
-
(workerNode)
=>
+
workerNode
=>
workerNode.info.ready &&
workerNode.info.id !== workerId &&
workerNode.usage.tasks.queued > 0
workerNode.info.ready &&
workerNode.info.id !== workerId &&
workerNode.usage.tasks.queued > 0
@@
-1319,7
+1326,7
@@
export abstract class AbstractPool<
* @returns The listener function to execute when a message is received from a worker.
*/
protected workerListener (): (message: MessageValue<Response>) => void {
* @returns The listener function to execute when a message is received from a worker.
*/
protected workerListener (): (message: MessageValue<Response>) => void {
- return
(message)
=> {
+ return
message
=> {
this.checkMessageWorkerId(message)
if (message.ready != null && message.taskFunctions != null) {
// Worker ready response received from worker
this.checkMessageWorkerId(message)
if (message.ready != null && message.taskFunctions != null) {
// Worker ready response received from worker
@@
-1457,7
+1464,7
@@
export abstract class AbstractPool<
return (
this.opts.enableTasksQueue === true &&
this.workerNodes.findIndex(
return (
this.opts.enableTasksQueue === true &&
this.workerNodes.findIndex(
-
(workerNode)
=> !workerNode.hasBackPressure()
+
workerNode
=> !workerNode.hasBackPressure()
) === -1
)
}
) === -1
)
}