## [Unreleased]
+## [2.6.33] - 2023-08-24
+
+### Fixed
+
+- Fix queued tasks rescheduling.
+
### Changed
- Rename tasks queue options `queueMaxSize` to `size`.
+### Added
+
+- Task stealing scheduling algorithm if tasks queueing is enabled.
+
## [2.6.32] - 2023-08-23
### Fixed
- Tasks distribution strategies :white_check_mark:
- Lockless tasks queueing :white_check_mark:
- Queued tasks rescheduling:
- - Tasks redistribution on worker error :white_check_mark:
+ - Task stealing :white_check_mark:
- Tasks stealing under back pressure :white_check_mark:
+ - Tasks redistribution on worker error :white_check_mark:
- General guidelines on pool choice :white_check_mark:
- Error handling out of the box :white_check_mark:
- Widely tested :white_check_mark:
"dependencies": {
"fastify": "^4.21.0",
"fastify-plugin": "^4.5.1",
- "poolifier": "^2.6.28"
+ "poolifier": "^2.6.32"
},
"devDependencies": {
- "@types/node": "^20.5.1",
+ "@types/node": "^20.5.4",
"typescript": "^5.1.6"
}
}
specifier: ^4.5.1
version: 4.5.1
poolifier:
- specifier: ^2.6.28
- version: 2.6.28
+ specifier: ^2.6.32
+ version: 2.6.32
devDependencies:
'@types/node':
- specifier: ^20.5.1
- version: 20.5.1
+ specifier: ^20.5.4
+ version: 20.5.4
typescript:
specifier: ^5.1.6
version: 5.1.6
fast-json-stringify: 5.8.0
dev: false
- /@types/node@20.5.1:
- resolution: {integrity: sha512-4tT2UrL5LBqDwoed9wZ6N3umC4Yhz3W3FloMmiiG4JwmUJWpie0c7lcnUNd4gtMKuDEO4wRVS8B6Xa0uMRsMKg==}
+ /@types/node@20.5.4:
+ resolution: {integrity: sha512-Y9vbIAoM31djQZrPYjpTLo0XlaSwOIsrlfE3LpulZeRblttsLQRFRlBAppW0LOxyT3ALj2M5vU1ucQQayQH3jA==}
dev: true
/abort-controller@3.0.0:
thread-stream: 2.4.0
dev: false
- /poolifier@2.6.28:
- resolution: {integrity: sha512-dm3N/BcJ6n1m9aDidXYx53mI7nQc1QvTGdIL0rA29hOjLsfWq6+7pz977vnxvfGxKG5+pCG+VIT7PcJENQSzng==}
+ /poolifier@2.6.32:
+ resolution: {integrity: sha512-2bXB7C5Uazckw8Li7ZAoSHfBaMHA2h0I5VlsprubPcFrr/KRVi1EUg/j6wPFlZs4tcHouD4ZJOcJjySXpV5EgQ==}
engines: {node: '>=16.14.0', pnpm: '>=8.6.0'}
requiresBuild: true
dev: false
{
"$schema": "https://json.schemastore.org/package",
"name": "poolifier",
- "version": "2.6.32",
+ "version": "2.6.33",
"description": "Fast and small Node.js Worker_Threads and Cluster Worker Pool",
"license": "MIT",
"main": "./lib/index.js",
sonar.organization=pioardi
sonar.javascript.lcov.reportPaths=coverage/lcov.info
sonar.projectName=poolifier
-sonar.projectVersion=2.6.32
+sonar.projectVersion=2.6.33
sonar.host.url=https://sonarcloud.io
sonar.sources=src
sonar.tests=tests
// 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)
}
}
private redistributeQueuedTasks (workerNodeKey: number): void {
- const workerNodes = this.workerNodes.filter(
- (_, workerNodeId) => workerNodeId !== workerNodeKey
- )
while (this.tasksQueueSize(workerNodeKey) > 0) {
- let targetWorkerNodeKey: number = workerNodeKey
+ let destinationWorkerNodeKey: number = workerNodeKey
let minQueuedTasks = Infinity
let executeTask = false
- for (const [workerNodeId, workerNode] of workerNodes.entries()) {
+ for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
if (
+ workerNode.info.ready &&
+ workerNodeId !== workerNodeKey &&
workerNode.usage.tasks.executing <
- (this.opts.tasksQueueOptions?.concurrency as number)
+ (this.opts.tasksQueueOptions?.concurrency as number)
) {
executeTask = true
}
- if (workerNode.info.ready && workerNode.usage.tasks.queued === 0) {
- targetWorkerNodeKey = workerNodeId
+ if (
+ workerNode.info.ready &&
+ workerNodeId !== workerNodeKey &&
+ workerNode.usage.tasks.queued === 0
+ ) {
+ destinationWorkerNodeKey = workerNodeId
break
}
if (
workerNode.info.ready &&
+ workerNodeId !== workerNodeKey &&
workerNode.usage.tasks.queued < minQueuedTasks
) {
minQueuedTasks = workerNode.usage.tasks.queued
- targetWorkerNodeKey = workerNodeId
+ destinationWorkerNodeKey = workerNodeId
}
}
+ const task = {
+ ...(this.dequeueTask(workerNodeKey) as Task<Data>),
+ workerId: (this.getWorkerInfo(destinationWorkerNodeKey) as WorkerInfo)
+ .id as number
+ }
if (executeTask) {
- this.executeTask(
- targetWorkerNodeKey,
- this.dequeueTask(workerNodeKey) as Task<Data>
- )
+ this.executeTask(destinationWorkerNodeKey, task)
} else {
- this.enqueueTask(
- targetWorkerNodeKey,
- this.dequeueTask(workerNodeKey) as Task<Data>
- )
+ this.enqueueTask(destinationWorkerNodeKey, task)
+ }
+ }
+ }
+
+ private taskStealingOnEmptyQueue (workerId: number): void {
+ const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
+ const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey]
+ const workerNodes = this.workerNodes
+ .slice()
+ .sort(
+ (workerNodeA, workerNodeB) =>
+ workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
+ )
+ for (const sourceWorkerNode of workerNodes) {
+ if (
+ sourceWorkerNode.info.ready &&
+ sourceWorkerNode.info.id !== workerId &&
+ sourceWorkerNode.usage.tasks.queued > 0
+ ) {
+ const task = {
+ ...(sourceWorkerNode.popTask() as Task<Data>),
+ workerId: destinationWorkerNode.info.id as number
+ }
+ if (
+ destinationWorkerNode.usage.tasks.executing <
+ (this.opts.tasksQueueOptions?.concurrency as number)
+ ) {
+ this.executeTask(destinationWorkerNodeKey, task)
+ } else {
+ this.enqueueTask(destinationWorkerNodeKey, task)
+ }
+ break
}
}
}
const sourceWorkerNode =
this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
const workerNodes = this.workerNodes
- .filter((workerNode) => workerNode.info.id !== workerId)
+ .slice()
.sort(
(workerNodeA, workerNodeB) =>
workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued
for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
if (
workerNode.info.ready &&
+ workerNode.info.id !== workerId &&
sourceWorkerNode.usage.tasks.queued > 0 &&
!workerNode.hasBackPressure()
) {
+ const task = {
+ ...(sourceWorkerNode.popTask() as Task<Data>),
+ workerId: workerNode.info.id as number
+ }
if (
workerNode.usage.tasks.executing <
(this.opts.tasksQueueOptions?.concurrency as number)
) {
- this.executeTask(
- workerNodeKey,
- sourceWorkerNode.popTask() as Task<Data>
- )
+ this.executeTask(workerNodeKey, task)
} else {
- this.enqueueTask(
- workerNodeKey,
- sourceWorkerNode.popTask() as Task<Data>
- )
+ this.enqueueTask(workerNodeKey, task)
}
}
}
-export const version = '2.6.32'
+export const version = '2.6.33'
public tasksQueueBackPressureSize: number
/** @inheritdoc */
public onBackPressure?: (workerId: number) => void
+ /** @inheritdoc */
+ public onEmptyQueue?: (workerId: number) => void
private readonly taskFunctionsUsage: Map<string, WorkerUsage>
private readonly tasksQueue: Deque<Task<Data>>
return this.tasksQueue.size
}
- /**
- * Tasks queue maximum size.
- *
- * @returns The tasks queue maximum size.
- */
- private tasksQueueMaxSize (): number {
- return this.tasksQueue.maxSize
- }
-
/** @inheritdoc */
public enqueueTask (task: Task<Data>): number {
const tasksQueueSize = this.tasksQueue.push(task)
/** @inheritdoc */
public dequeueTask (): Task<Data> | undefined {
- return this.tasksQueue.shift()
+ const task = this.tasksQueue.shift()
+ if (this.onEmptyQueue != null && this.tasksQueue.size === 0) {
+ once(this.onEmptyQueue, this)(this.info.id as number)
+ }
+ return task
}
/** @inheritdoc */
public popTask (): Task<Data> | undefined {
- return this.tasksQueue.pop()
+ const task = this.tasksQueue.pop()
+ if (this.onEmptyQueue != null && this.tasksQueue.size === 0) {
+ once(this.onEmptyQueue, this)(this.info.id as number)
+ }
+ return task
}
/** @inheritdoc */
private initWorkerUsage (): WorkerUsage {
const getTasksQueueSize = (): number => {
- return this.tasksQueueSize()
+ return this.tasksQueue.size
}
const getTasksQueueMaxSize = (): number => {
- return this.tasksQueueMaxSize()
+ return this.tasksQueue.maxSize
}
return {
tasks: {
* @param workerId - The worker id.
*/
onBackPressure?: (workerId: number) => void
+ /**
+ * Callback invoked when worker node tasks queue is empty.
+ *
+ * @param workerId - The worker id.
+ */
+ onEmptyQueue?: (workerId: number) => void
/**
* Tasks queue size.
*
expect(queuePool.info.backPressure).toBe(false)
await Promise.all(promises)
for (const workerNode of queuePool.workerNodes) {
- expect(workerNode.usage.tasks.executing).toBe(0)
+ expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0)
+ expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual(
+ numberOfWorkers * maxMultiplier
+ )
expect(workerNode.usage.tasks.executed).toBe(maxMultiplier)
expect(workerNode.usage.tasks.queued).toBe(0)
expect(workerNode.usage.tasks.maxQueued).toBe(
expect(queuePool.info.backPressure).toBe(false)
await Promise.all(promises)
for (const workerNode of queuePool.workerNodes) {
- expect(workerNode.usage.tasks.executing).toBe(0)
+ expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0)
+ expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual(
+ numberOfThreads * maxMultiplier
+ )
expect(workerNode.usage.tasks.executed).toBe(maxMultiplier)
expect(workerNode.usage.tasks.queued).toBe(0)
expect(workerNode.usage.tasks.maxQueued).toBe(
repositories / poolifier.git / commitdiff
