From: igor-nanox Date: Fri, 16 Oct 2026 22:02:36 +0000 (+0300) Subject: fix(pool): keep idle stealing retries on their worker (#3346) X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=e0612414a9a74a798f12acb20edea170603e4696;p=poolifier.git fix(pool): keep idle stealing retries on their worker (#3346) * fix: account task responses to the replying worker node Resolve the worker node from the response worker id and re-resolve node keys after promise continuations and stolen-task delays, since keys shift when workers are removed. * refactor(pool): reuse UUID and shared regression helpers * docs: remove regression helper contribution guidance * test(pool): unify regression deadlines and message waits * fix(pool): preserve worker info overrides in task statistics * docs(api): remove idle retry implementation detail --------- Co-authored-by: Jérôme Benoit --- diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 317cb621a..57303027b 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1,3 +1,4 @@ +import type { UUID } from 'node:crypto' import type { Transferable } from 'node:worker_threads' import { AsyncResource } from 'node:async_hooks' @@ -384,13 +385,8 @@ export abstract class AbstractPool< * * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id. */ - protected promiseResponseMap: Map< - `${string}-${string}-${string}-${string}-${string}`, - PromiseResponseWrapper - > = new Map< - `${string}-${string}-${string}-${string}-${string}`, - PromiseResponseWrapper - >() + protected promiseResponseMap: Map> = + new Map>() /** * Whether the pool is started or not. @@ -957,7 +953,9 @@ export abstract class AbstractPool< needWorkerChoiceStrategiesUpdate = true } if ( - this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + this.shallUpdateTaskFunctionWorkerUsage( + this.getWorkerInfo(workerNodeKey) + ) && message.taskPerformance?.name != null && this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage( message.taskPerformance.name @@ -1038,7 +1036,9 @@ export abstract class AbstractPool< ) } if ( - this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + this.shallUpdateTaskFunctionWorkerUsage( + this.getWorkerInfo(workerNodeKey) + ) && // eslint-disable-next-line @typescript-eslint/no-non-null-assertion this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(task.name!) != null @@ -1184,10 +1184,7 @@ export abstract class AbstractPool< } workerNode.info.ready = false let workerNodeTasks: - | Map< - `${string}-${string}-${string}-${string}-${string}`, - PromiseResponseWrapper - > + | Map> | undefined if (settlingTasks) { const workerNodeKey = this.workerNodes.indexOf(workerNode) @@ -1780,10 +1777,7 @@ export abstract class AbstractPool< if (workerNode.usage.tasks.executing > 0) { --workerNode.usage.tasks.executing } - if ( - Array.isArray(workerNode.info.taskFunctionsProperties) && - workerNode.info.taskFunctionsProperties.length > 2 - ) { + if (this.shallUpdateTaskFunctionWorkerUsage(workerNode.info)) { const taskFunctionWorkerUsage = // eslint-disable-next-line @typescript-eslint/no-non-null-assertion workerNode.getTaskFunctionWorkerUsage(task.name!) @@ -1812,10 +1806,7 @@ export abstract class AbstractPool< } } - private readonly getAbortError = ( - taskName: string, - taskId: `${string}-${string}-${string}-${string}-${string}` - ): Error => { + private readonly getAbortError = (taskName: string, taskId: UUID): Error => { const abortError = this.promiseResponseMap.get(taskId)?.abortSignal ?.reason as Error | string return abortError instanceof Error @@ -2065,7 +2056,7 @@ export abstract class AbstractPool< } private readonly handleWorkerError = ( - taskId: `${string}-${string}-${string}-${string}-${string}`, + taskId: UUID, workerError: WorkerError ): Error => { const { aborted, error, message, name, stack } = workerError @@ -2172,7 +2163,15 @@ export abstract class AbstractPool< ) sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen)) .then(() => { - this.handleWorkerNodeIdleEvent(eventDetail, stolenTask) + // Worker nodes can be removed during the delay, shifting the keys. + const currentWorkerNodeKey = this.workerNodes.indexOf(workerNode) + if (currentWorkerNodeKey === -1) { + return undefined + } + this.handleWorkerNodeIdleEvent( + { ...eventDetail, workerNodeKey: currentWorkerNodeKey }, + stolenTask + ) return undefined }) .catch((error: unknown) => { @@ -2444,7 +2443,7 @@ export abstract class AbstractPool< } private rejectTask ( - taskId: `${string}-${string}-${string}-${string}-${string}`, + taskId: UUID, promiseResponse: PromiseResponseWrapper, error: unknown ): void { @@ -2468,10 +2467,7 @@ export abstract class AbstractPool< * @param error - The worker failure. */ private rejectWorkerNodeTasks ( - workerNodeTasks: Map< - `${string}-${string}-${string}-${string}-${string}`, - PromiseResponseWrapper - >, + workerNodeTasks: Map>, error?: Error ): void { for (const [taskId, promiseResponse] of workerNodeTasks) { @@ -2517,7 +2513,9 @@ export abstract class AbstractPool< } if ( taskName != null && - this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + this.shallUpdateTaskFunctionWorkerUsage( + this.getWorkerInfo(workerNodeKey) + ) && workerNode.getTaskFunctionWorkerUsage(taskName) != null ) { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion @@ -2741,11 +2739,12 @@ export abstract class AbstractPool< /** * Whether the worker node shall update its task function worker usage or not. - * @param workerNodeKey - The worker node key. + * @param workerInfo - The worker information. * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise. */ - private shallUpdateTaskFunctionWorkerUsage (workerNodeKey: number): boolean { - const workerInfo = this.getWorkerInfo(workerNodeKey) + private shallUpdateTaskFunctionWorkerUsage ( + workerInfo: undefined | WorkerInfo + ): boolean { return ( workerInfo != null && Array.isArray(workerInfo.taskFunctionsProperties) && @@ -2860,7 +2859,9 @@ export abstract class AbstractPool< } if ( taskName != null && - this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + this.shallUpdateTaskFunctionWorkerUsage( + this.getWorkerInfo(workerNodeKey) + ) && workerNode.getTaskFunctionWorkerUsage(taskName) != null ) { const taskFunctionWorkerUsage = @@ -2889,7 +2890,9 @@ export abstract class AbstractPool< ++workerNode.usage.tasks.stolen } if ( - this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + this.shallUpdateTaskFunctionWorkerUsage( + this.getWorkerInfo(workerNodeKey) + ) && workerNode.getTaskFunctionWorkerUsage(taskName) != null ) { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion diff --git a/src/pools/worker.ts b/src/pools/worker.ts index e4a26efe2..941ad9922 100644 --- a/src/pools/worker.ts +++ b/src/pools/worker.ts @@ -1,3 +1,4 @@ +import type { UUID } from 'node:crypto' import type { EventEmitter } from 'node:events' import type { MessageChannel, WorkerOptions } from 'node:worker_threads' @@ -374,7 +375,7 @@ export interface WorkerInfo { * @internal */ export interface WorkerNodeEventDetail { - taskId?: `${string}-${string}-${string}-${string}-${string}` + taskId?: UUID workerId?: number workerNodeKey?: number } diff --git a/src/utility-types.ts b/src/utility-types.ts index ed5ff6ef6..52bec3bb2 100644 --- a/src/utility-types.ts +++ b/src/utility-types.ts @@ -1,4 +1,5 @@ import type { AsyncResource } from 'node:async_hooks' +import type { UUID } from 'node:crypto' import type { EventLoopUtilization } from 'node:perf_hooks' import type { MessagePort, Transferable } from 'node:worker_threads' @@ -133,7 +134,7 @@ export interface Task { /** * Task UUID. */ - readonly taskId?: `${string}-${string}-${string}-${string}-${string}` + readonly taskId?: UUID /** * Timestamp. */ diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index d70a073d3..311161077 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -1,4 +1,5 @@ import type { Worker } from 'node:cluster' +import type { UUID } from 'node:crypto' import type { MessagePort } from 'node:worker_threads' import { performance } from 'node:perf_hooks' @@ -83,10 +84,7 @@ export abstract class AbstractWorker< /** * Task abort functions processed by the worker when task operation 'abort' is received. */ - protected taskAbortFunctions: Map< - `${string}-${string}-${string}-${string}-${string}`, - () => void - > + protected taskAbortFunctions: Map void> /** * Task function object(s) processed by the worker when the pool's `execute` method is invoked. @@ -110,10 +108,7 @@ export abstract class AbstractWorker< throw new Error('isMain parameter is mandatory') } this.checkTaskFunctions(taskFunctions) - this.taskAbortFunctions = new Map< - `${string}-${string}-${string}-${string}-${string}`, - () => void - >() + this.taskAbortFunctions = new Map void>() this.checkWorkerOptions(this.opts) if (!this.isMain) { this.getMainWorker().once('message', this.handleReadyMessage.bind(this)) @@ -675,7 +670,7 @@ export abstract class AbstractWorker< */ private getAbortableTaskFunction ( name: string, - taskId: `${string}-${string}-${string}-${string}-${string}` + taskId: UUID ): TaskAsyncFunction { return async (data?: Data): Promise => await new Promise( diff --git a/tests/pools/cluster/fixed.test.mjs b/tests/pools/cluster/fixed.test.mjs index aac84a21f..9c40ab1bb 100644 --- a/tests/pools/cluster/fixed.test.mjs +++ b/tests/pools/cluster/fixed.test.mjs @@ -133,10 +133,7 @@ describe('Fixed cluster pool test suite', () => { expect(queuePool.info.stolenTasks).toBe(0) await Promise.all(promises) for (const workerNode of queuePool.workerNodes) { - expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0) - expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual( - numberOfWorkers * maxMultiplier - ) + expect(workerNode.usage.tasks.executing).toBe(0) expect(workerNode.usage.tasks.queued).toBe(0) expect(workerNode.usage.tasks.maxQueued).toBe( maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency diff --git a/tests/pools/task-accounting.test.mjs b/tests/pools/task-accounting.test.mjs new file mode 100644 index 000000000..1b9b88c43 --- /dev/null +++ b/tests/pools/task-accounting.test.mjs @@ -0,0 +1,393 @@ +import { EventEmitter, once } from 'node:events' +import { setTimeout as delay } from 'node:timers/promises' +import { describe, expect, it, vi } from 'vitest' + +import { + DynamicThreadPool, + FixedThreadPool, + PoolEvents, + WorkerChoiceStrategies, +} from '../../lib/index.mjs' +import { bounded, waitForMessage } from '../test-utils.cjs' + +const workerFile = './tests/worker-files/thread/taskAccountingWorker.mjs' + +const createGate = () => new SharedArrayBuffer(4) + +const release = gate => { + Atomics.store(new Int32Array(gate), 0, 1) + Atomics.notify(new Int32Array(gate), 0) +} + +const createOptions = started => ({ + enableTasksQueue: true, + messageHandler: ({ label }) => started.add(label), + tasksQueueOptions: { + concurrency: 1, + tasksStealingOnBackPressure: false, + taskStealing: false, + }, +}) + +const waitForStart = async (started, label) => { + await vi.waitFor(() => expect(started.has(label)).toBe(true)) +} + +const finishTasksAndDestroy = async (pool, tasks) => { + try { + // Let released tasks settle before intentional termination can discard responses. + // Bound cleanup without replacing an assertion failure with a cleanup timeout. + await Promise.race([ + Promise.allSettled(tasks), + delay(5000, undefined, { ref: false }), + ]) + } finally { + await bounded(pool.destroy(), 'Accounting pool cleanup remained pending') + } +} + +describe('Task response worker accounting', () => { + it.each([false, true])( + 'accounts for a shifted dynamic worker after idle retirement (failure: %s)', + async fail => { + const started = new Set() + const pool = new DynamicThreadPool( + 1, + 3, + workerFile, + createOptions(started) + ) + const gates = [createGate(), createGate(), createGate()] + const tasks = [] + try { + await bounded( + once(pool.emitter, PoolEvents.ready), + 'Accounting pool readiness remained pending' + ) + for (let i = 0; i < gates.length; i++) { + tasks.push( + pool + .execute({ + fail: i === 2 && fail, + gate: gates[i], + label: `task-${i}`, + }) + .then( + value => ({ value }), + error => ({ error }) + ) + ) + await waitForStart(started, `task-${i}`) + } + const shiftedWorker = pool.workerNodes[2] + const retired = once(pool.workerNodes[1].worker, 'exit', { + signal: AbortSignal.timeout(5000), + }) + release(gates[1]) + await bounded(tasks[1], 'Retiring worker task remained pending') + await retired + expect(pool.workerNodes.indexOf(shiftedWorker)).toBe(1) + + release(gates[2]) + const result = await bounded( + tasks[2], + 'Shifted worker task remained pending' + ) + if (fail) { + expect(result.error.message).toBe('task-2') + } else { + expect(result.value).toBe('task-2') + } + expect(shiftedWorker.usage.tasks.executing).toBe(0) + expect(shiftedWorker.usage.tasks.executed).toBe(fail ? 0 : 1) + expect(shiftedWorker.usage.tasks.failed).toBe(fail ? 1 : 0) + + // The minimum worker remains unaffected by surplus-worker retirement. + release(gates[0]) + await bounded(tasks[0], 'Minimum worker task remained pending') + expect(pool.workerNodes[0].usage.tasks.executing).toBe(0) + expect(pool.workerNodes[0].usage.tasks.executed).toBe(1) + await vi.waitFor(() => expect(pool.workerNodes).toHaveLength(1), { + timeout: 5000, + }) + } finally { + gates.forEach(release) + await finishTasksAndDestroy(pool, tasks) + } + } + ) + + it('drains the actual worker queue after its array index changes', async () => { + const started = new Set() + const pool = new DynamicThreadPool(1, 3, workerFile, createOptions(started)) + const gates = [createGate(), createGate(), createGate()] + const tasks = [] + try { + await bounded( + once(pool.emitter, PoolEvents.ready), + 'Queue regression pool readiness remained pending' + ) + for (let i = 0; i < gates.length; i++) { + tasks.push(pool.execute({ gate: gates[i], label: `task-${i}` })) + await waitForStart(started, `task-${i}`) + } + const shiftedWorker = pool.workerNodes[2] + vi.spyOn(pool, 'chooseWorkerNode').mockReturnValueOnce(2) + const queued = pool.execute({ label: 'queued' }) + tasks.push(queued) + expect(shiftedWorker.usage.tasks.queued).toBe(1) + const retired = once(pool.workerNodes[1].worker, 'exit', { + signal: AbortSignal.timeout(5000), + }) + release(gates[1]) + await bounded(tasks[1], 'Retiring worker task remained pending') + await retired + release(gates[2]) + await bounded(tasks[2], 'Shifted worker task remained pending') + await waitForStart(started, 'queued') + expect(await bounded(queued, 'Queued task remained pending')).toBe( + 'queued' + ) + expect(shiftedWorker.usage.tasks.executing).toBe(0) + expect(shiftedWorker.usage.tasks.executed).toBe(2) + expect(shiftedWorker.usage.tasks.queued).toBe(0) + } finally { + gates.forEach(release) + await finishTasksAndDestroy(pool, tasks) + } + }) + + it('accounts for tasks stolen from a different worker queue', async () => { + const started = new Set() + const pool = new FixedThreadPool(2, workerFile, createOptions(started)) + const gates = [createGate(), createGate()] + const tasks = [] + try { + await bounded( + once(pool.emitter, PoolEvents.ready), + 'Stealing regression pool readiness remained pending' + ) + for (let i = 0; i < gates.length; i++) { + vi.spyOn(pool, 'chooseWorkerNode').mockReturnValueOnce(i) + tasks.push(pool.execute({ gate: gates[i], label: `task-${i}` })) + await waitForStart(started, `task-${i}`) + } + const [source, destination] = pool.workerNodes + vi.spyOn(pool, 'chooseWorkerNode').mockReturnValueOnce(0) + const stolen = pool.execute({ label: 'stolen' }) + tasks.push(stolen) + // Exercise the same migration helper used by idle/back-pressure stealing. + pool.stealTask(source, 1) + expect(source.usage.tasks.queued).toBe(0) + expect(destination.usage.tasks.queued).toBe(1) + release(gates[1]) + await bounded(tasks[1], 'Destination worker task remained pending') + await waitForStart(started, 'stolen') + expect(await bounded(stolen, 'Stolen task remained pending')).toBe( + 'stolen' + ) + expect(source.usage.tasks.executing).toBe(1) + expect(source.usage.tasks.executed).toBe(0) + expect(destination.usage.tasks.executing).toBe(0) + expect(destination.usage.tasks.executed).toBe(2) + } finally { + gates.forEach(release) + await finishTasksAndDestroy(pool, tasks) + } + }) + + it('preserves worker info overrides in task function statistics', async () => { + class ProjectedInfoPool extends FixedThreadPool { + projectInfo = false + + getWorkerInfo (workerNodeKey) { + const workerInfo = super.getWorkerInfo(workerNodeKey) + return this.projectInfo && workerInfo != null + ? { + ...workerInfo, + taskFunctionsProperties: + workerInfo.taskFunctionsProperties?.slice(0, 2), + } + : workerInfo + } + } + + const pool = new ProjectedInfoPool(1, workerFile) + try { + await bounded( + once(pool.emitter, PoolEvents.ready), + 'Override regression pool readiness remained pending' + ) + const workerNode = pool.workerNodes[0] + const metadata = waitForMessage( + workerNode.messageChannel.port1, + message => + message.taskFunctionsProperties?.length === 3 && + message.taskFunctionsProperties.some(({ name }) => name === 'extra') + ) + const [, added] = await bounded( + Promise.all([metadata, pool.addTaskFunction('extra', data => data)]), + 'Extra task function metadata remained pending' + ) + expect(added).toBe(true) + const taskFunctionUsage = workerNode.getTaskFunctionWorkerUsage('extra') + const controlData = { phase: 'control' } + expect( + await bounded( + pool.execute(controlData, 'extra'), + 'Control task remained pending' + ) + ).toStrictEqual(controlData) + expect(workerNode.usage.tasks.executed).toBe(1) + expect(taskFunctionUsage.tasks.executed).toBe(1) + expect(taskFunctionUsage.tasks.executing).toBe(0) + + pool.projectInfo = true + const projectedData = { phase: 'projected' } + expect( + await bounded( + pool.execute(projectedData, 'extra'), + 'Projected task remained pending' + ) + ).toStrictEqual(projectedData) + expect(workerNode.usage.tasks.executed).toBe(2) + expect(workerNode.usage.tasks.executing).toBe(0) + expect(taskFunctionUsage.tasks.executed).toBe(1) + expect(taskFunctionUsage.tasks.executing).toBe(0) + } finally { + pool.projectInfo = false + await bounded( + pool.destroy(), + 'Override regression pool cleanup remained pending' + ) + } + }) +}) + +describe('Idle stealing retry accounting', () => { + it('keeps the idle stealing retry on its worker after native retirement shifts its index', { + retry: 0, + }, async () => { + const messages = new EventEmitter() + const received = [] + const gates = { + anchor: createGate(), + block: createGate(), + kill: createGate(), + loader1: createGate(), + loader2: createGate(), + survivor: createGate(), + } + const waitForMessage = async predicate => { + const signal = AbortSignal.timeout(5000) + while (!received.some(predicate)) { + await once(messages, 'message', { signal }) + } + return received.find(predicate) + } + const pool = new DynamicThreadPool( + 1, + 4, + './tests/worker-files/thread/taskStealingRetryWorker.mjs', + { + enableTasksQueue: true, + messageHandler: message => { + received.push(message) + messages.emit('message') + }, + tasksQueueOptions: { + concurrency: 1, + tasksStealingOnBackPressure: false, + taskStealing: true, + }, + workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED, + workerOptions: { workerData: { killGate: gates.kill } }, + } + ) + const tasks = [] + const execute = (gate, label, name) => { + const task = pool.execute({ gate, label }, name) + // Attach rejection handling immediately, including when setup fails. + tasks.push( + task.then( + value => ({ value }), + error => ({ error }) + ) + ) + } + try { + await bounded( + once(pool.emitter, PoolEvents.ready), + 'Idle retry pool readiness remained pending' + ) + execute(gates.block, 'block') + await waitForMessage(message => message.label === 'block') + execute(gates.loader1, 'loader-1') + execute(gates.loader2, 'loader-2') + execute(gates.anchor, 'anchor', 'anchor') + execute(gates.survivor, 'survivor', 'survivor') + const [, retiree, retrying, wrongTarget] = pool.workerNodes + await waitForMessage(message => message.label === 'anchor') + await waitForMessage(message => message.label === 'survivor') + + // The actual SOFT retirement is already waiting for its public killHandler. + // Boot duration cannot remove R before the retry is armed. + await waitForMessage( + message => + message.label === 'kill-handler' && + message.workerId === retiree.info.id + ) + expect(pool.workerNodes.indexOf(retrying)).toBe(2) + const retired = once(retiree.worker, 'exit', { + signal: AbortSignal.timeout(5000), + }) + release(gates.anchor) + await waitForMessage( + message => + ['loader-1', 'loader-2'].includes(message.label) && + message.workerId === retrying.info.id + ) + expect(retrying.usage.tasks.sequentiallyStolen).toBe(1) + expect(retrying.info.continuousStealing).toBe(true) + expect(pool.workerNodes).toHaveLength(4) + release(gates.kill) + await retired + expect(pool.workerNodes.indexOf(retrying)).toBe(1) + expect(pool.workerNodes.indexOf(wrongTarget)).toBe(2) + // Prove the splice happened inside the pending retry window, not after it. + expect(retrying.usage.tasks.sequentiallyStolen).toBe(1) + expect(retrying.info.continuousStealing).toBe(true) + expect(wrongTarget.usage.tasks.stolen).toBe(0) + + await vi.waitFor( + () => { + expect( + retrying.usage.tasks.sequentiallyStolen === 0 || + wrongTarget.usage.tasks.stolen > 0 + ).toBe(true) + }, + { interval: 5, timeout: 2000 } + ) + // A stale-key retry steals a real queued loader into the busy wrong worker. + expect(wrongTarget.usage.tasks.stolen).toBe(0) + expect(wrongTarget.usage.tasks.queued).toBe(0) + expect(retrying.usage.tasks.sequentiallyStolen).toBe(0) + expect(pool.info.stolenTasks).toBe(1) + } finally { + Object.values(gates).forEach(release) + try { + await Promise.race([ + Promise.all(tasks), + delay(5000, undefined, { ref: false }).then(() => { + throw new Error('Gated tasks did not finish') + }), + ]) + } finally { + await bounded( + pool.destroy(), + 'Idle retry pool cleanup remained pending' + ) + } + } + }) +}) diff --git a/tests/pools/thread/fixed.test.mjs b/tests/pools/thread/fixed.test.mjs index ef4c9e689..99bad2d83 100644 --- a/tests/pools/thread/fixed.test.mjs +++ b/tests/pools/thread/fixed.test.mjs @@ -132,10 +132,7 @@ describe('Fixed thread pool test suite', () => { expect(queuePool.info.stolenTasks).toBe(0) await Promise.all(promises) for (const workerNode of queuePool.workerNodes) { - expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0) - expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual( - numberOfThreads * maxMultiplier - ) + expect(workerNode.usage.tasks.executing).toBe(0) expect(workerNode.usage.tasks.queued).toBe(0) expect(workerNode.usage.tasks.maxQueued).toBe( maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency diff --git a/tests/pools/worker-exit.test.mjs b/tests/pools/worker-exit.test.mjs index 275fd027f..b521d1610 100644 --- a/tests/pools/worker-exit.test.mjs +++ b/tests/pools/worker-exit.test.mjs @@ -14,23 +14,7 @@ import { FixedThreadPool, WorkerChoiceStrategies, } from '../../lib/index.mjs' - -const bounded = async promise => { - let timeout - try { - return await Promise.race([ - promise, - new Promise((resolve, reject) => { - timeout = setTimeout( - () => reject(new Error('Task remained pending')), - 5000 - ) - }), - ]) - } finally { - clearTimeout(timeout) - } -} +import { bounded, waitForMessage } from '../test-utils.cjs' const send = (workerNode, message) => { const workerMessage = { ...message, workerId: workerNode.info.id } @@ -41,17 +25,8 @@ const send = (workerNode, message) => { } } -const started = (workerNode, id) => { - return new Promise(resolve => { - const listener = message => { - if (message.started === id) { - workerNode.worker.off('message', listener) - resolve() - } - } - workerNode.worker.on('message', listener) - }) -} +const started = (workerNode, id) => + waitForMessage(workerNode.worker, message => message.started === id) const cleanupRegressionPool = async (pool, capturedNodes) => { pool.opts.restartWorkerOnError = false @@ -111,7 +86,10 @@ for (const [workerType, PoolClass, workerFile] of [ await bounded(exit) expect(pool.promiseResponseMap.size).toBe(0) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Worker exit pool cleanup remained pending' + ) } }) } @@ -126,7 +104,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(result).toBeInstanceOf(Error) expect(pool.promiseResponseMap.size).toBe(0) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Silent exit pool cleanup remained pending' + ) } }) } @@ -190,7 +171,10 @@ for (const [workerType, PoolClass, workerFile] of [ }) let destruction try { - await pool.execute({ warmup: true }) + await bounded( + pool.execute({ warmup: true }), + 'Exit redistribution warmup remained pending' + ) const [source, destination] = pool.workerNodes const sourceStarted = started(source, 'source') const destinationStarted = started(destination, 'destination') @@ -225,7 +209,10 @@ for (const [workerType, PoolClass, workerFile] of [ }, }) try { - await pool.execute({ warmup: true }) + await bounded( + pool.execute({ warmup: true }), + 'Worker index shift warmup remained pending' + ) const [failedNode, healthyNode] = pool.workerNodes const failedStarted = started(failedNode, 'failed') const healthyStarted = started(healthyNode, 'healthy') @@ -248,7 +235,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(await bounded(queued)).toStrictEqual(queuedData) expect(healthyNode.usage.tasks.executing).toBe(0) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Worker index shift pool cleanup remained pending' + ) } }) @@ -339,7 +329,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(completions).toStrictEqual(executions) expect(pool.promiseResponseMap.size).toBe(0) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Affinity pool cleanup remained pending' + ) } }) } @@ -380,7 +373,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(outcomes[1]).toBeInstanceOf(Error) await bounded(exit) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Default affinity pool cleanup remained pending' + ) } }) @@ -424,7 +420,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(outcomes[2]).toStrictEqual(eligibleData) await bounded(exit) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Eligible migration pool cleanup remained pending' + ) } }) @@ -437,7 +436,10 @@ for (const [workerType, PoolClass, workerFile] of [ }, }) try { - await pool.execute({ warmup: true }) + await bounded( + pool.execute({ warmup: true }), + 'Stolen task warmup remained pending' + ) const [source, destination] = pool.workerNodes const sourceStarted = started(source, 'source') const destinationStarted = started(destination, 'destination') @@ -468,7 +470,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(await bounded(stolenResult)).toBeInstanceOf(Error) await bounded(destinationExit) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Stolen task pool cleanup remained pending' + ) } }) @@ -483,7 +488,10 @@ for (const [workerType, PoolClass, workerFile] of [ }, }) try { - await pool.execute({ warmup: true }) + await bounded( + pool.execute({ warmup: true }), + 'Redistribution abort warmup remained pending' + ) const [source, destination] = pool.workerNodes const sourceStarted = started(source, 'source') const destinationStarted = started(destination, 'destination') @@ -540,7 +548,10 @@ for (const [workerType, PoolClass, workerFile] of [ await bounded(exit) expect(pool.promiseResponseMap.size).toBe(0) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Redistribution abort pool cleanup remained pending' + ) } }) @@ -599,7 +610,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(source.tasksQueue.size).toBe(0) expect(pool.promiseResponseMap.size).toBe(0) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Queued abort pool cleanup remained pending' + ) } }) @@ -612,7 +626,10 @@ for (const [workerType, PoolClass, workerFile] of [ }, }) try { - await pool.execute({ warmup: true }) + await bounded( + pool.execute({ warmup: true }), + 'Abort routing warmup remained pending' + ) const [source, destination] = pool.workerNodes const sourceStarted = started(source, 'source') const destinationStarted = started(destination, 'destination') @@ -637,7 +654,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(await bounded(destinationResult)).toStrictEqual(destinationData) expect(destination.tasksQueue.size).toBe(0) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Abort routing pool cleanup remained pending' + ) } }) it('can destroy the pool immediately after catching a worker failure', async () => { @@ -670,7 +690,10 @@ for (const [workerType, PoolClass, workerFile] of [ it('does not attribute intentionally terminated tasks to a later crashed worker', async () => { const pool = await createPool(2) try { - await pool.execute({ warmup: true }) + await bounded( + pool.execute({ warmup: true }), + 'Intentional termination warmup remained pending' + ) const [removed, survivor] = pool.workerNodes const removedStarted = started(removed, 'removed') const survivorStarted = started(survivor, 'survivor') @@ -690,7 +713,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(await bounded(survivorResult)).toBeInstanceOf(Error) expect(removedTaskSettled).toBe(false) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Intentional termination pool cleanup remained pending' + ) } }) @@ -1057,7 +1083,10 @@ for (const [workerType, PoolClass, workerFile] of [ expect(error).toBe(rawWorkerError) } } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Native failure pool cleanup remained pending' + ) } }) } @@ -1178,7 +1207,12 @@ for (const [workerType, PoolClass, workerFile] of [ } finally { workerNode.messageChannel?.port1.close() workerNode.messageChannel?.port2.close() - if (pool.info.started) await pool.destroy() + if (pool.info.started) { + await bounded( + pool.destroy(), + 'Buffered response pool cleanup remained pending' + ) + } } }) } @@ -1472,37 +1506,25 @@ for (const [workerType, PoolClass, workerFile] of [ const data = { recovered: true } expect(await bounded(pool.execute(data))).toStrictEqual(data) } finally { - await pool.destroy() + await bounded( + pool.destroy(), + 'Dynamic recovery pool cleanup remained pending' + ) } }) }) } describe('thread worker exit bootstrap affinity metadata', { retry: 0 }, () => { - const bootstrapMessage = (workerNode, predicate) => - new Promise(resolve => { - const listener = message => { - if (predicate(message)) { - workerNode.worker.off('message', listener) - resolve(message) - } - } - workerNode.worker.on('message', listener) - }) - const metadataReady = workerNode => { if (workerNode.info.taskFunctionsProperties?.length > 0) { return Promise.resolve() } - return new Promise(resolve => { - const listener = message => { - if (message.ready === true && message.taskFunctionsProperties != null) { - workerNode.messageChannel.port1.off('message', listener) - resolve() - } - } - workerNode.messageChannel.port1.on('message', listener) - }) + return waitForMessage( + workerNode.messageChannel.port1, + message => + message.ready === true && message.taskFunctionsProperties != null + ) } it.each([ @@ -1570,8 +1592,8 @@ describe('thread worker exit bootstrap affinity metadata', { retry: 0 }, () => { capturedNodes.push(pool.workerNodes[0]) const heldStarted = [started(capturedNodes[0], 'held-0')] const firstClaim = await bounded( - bootstrapMessage( - capturedNodes[0], + waitForMessage( + capturedNodes[0].worker, message => message.bootstrapRole === 0 ) ) @@ -1588,8 +1610,8 @@ describe('thread worker exit bootstrap affinity metadata', { retry: 0 }, () => { capturedNodes.push(pool.workerNodes[1]) heldStarted.push(started(capturedNodes[1], 'held-1')) const secondClaim = await bounded( - bootstrapMessage( - capturedNodes[1], + waitForMessage( + capturedNodes[1].worker, message => message.bootstrapRole === 1 ) ) diff --git a/tests/pools/worker-node.test.mjs b/tests/pools/worker-node.test.mjs index 2c2a491ac..1e9443da9 100644 --- a/tests/pools/worker-node.test.mjs +++ b/tests/pools/worker-node.test.mjs @@ -9,6 +9,7 @@ import { WorkerNode } from '../../lib/pools/worker-node.mjs' import { MeasurementHistorySize } from '../../lib/pools/worker.mjs' import { PriorityQueue } from '../../lib/queues/priority-queue.mjs' import { DEFAULT_TASK_NAME } from '../../lib/utils.mjs' +import { bounded } from '../test-utils.cjs' describe('Worker node test suite', () => { let clusterWorkerNode, threadWorkerNode @@ -51,27 +52,31 @@ describe('Worker node test suite', () => { tasksQueuePriority: true, } ) - let timeout try { - await once(workerNode.worker, 'online') - await workerNode.worker.terminate() + await bounded( + once(workerNode.worker, 'online'), + 'Worker node online remained pending' + ) + await bounded( + workerNode.worker.terminate(), + 'Native worker termination remained pending' + ) await expect( - Promise.race([ + bounded( workerNode.terminate(), - new Promise((resolve, reject) => { - timeout = setTimeout( - () => - reject(new Error('Worker node termination remained pending')), - 5000 - ) - }), - ]) + 'Worker node termination remained pending' + ) ).resolves.toBeUndefined() } finally { - clearTimeout(timeout) - await workerNode.worker.terminate() - workerNode.messageChannel?.port1.close() - workerNode.messageChannel?.port2.close() + try { + await bounded( + workerNode.worker.terminate(), + 'Native worker cleanup remained pending' + ) + } finally { + workerNode.messageChannel?.port1.close() + workerNode.messageChannel?.port2.close() + } } }) @@ -94,37 +99,30 @@ describe('Worker node test suite', () => { } finally { cluster.setupPrimary(clusterSettings) } - let timeout - const boundedWait = async (promise, stage) => { - try { - await Promise.race([ - promise, - new Promise((resolve, reject) => { - timeout = setTimeout( - () => reject(new Error(`Worker node ${stage} remained pending`)), - 5000 - ) - }), - ]) - } finally { - clearTimeout(timeout) - } - } try { - await boundedWait(once(workerNode.worker, 'online'), 'online') + await bounded( + once(workerNode.worker, 'online'), + 'Worker node online remained pending' + ) const disconnect = once(workerNode.worker, 'disconnect') workerNode.worker.send({ disconnect: true }) - await boundedWait(disconnect, 'disconnect') + await bounded(disconnect, 'Worker node disconnect remained pending') expect(workerNode.worker.isConnected()).toBe(false) expect(workerNode.worker.isDead()).toBe(false) - await boundedWait(workerNode.terminate(), 'disconnected termination') + await bounded( + workerNode.terminate(), + 'Worker node disconnected termination remained pending' + ) expect(workerNode.worker.isDead()).toBe(true) - await boundedWait(workerNode.terminate(), 'post-exit termination') + await bounded( + workerNode.terminate(), + 'Worker node post-exit termination remained pending' + ) } finally { if (!workerNode.worker.isDead()) { const exit = once(workerNode.worker, 'exit') workerNode.worker.process.kill() - await boundedWait(exit, 'native cleanup exit') + await bounded(exit, 'Worker node native cleanup exit remained pending') } } }) diff --git a/tests/test-utils.cjs b/tests/test-utils.cjs index 4b73d633b..0bae6b76a 100644 --- a/tests/test-utils.cjs +++ b/tests/test-utils.cjs @@ -1,5 +1,34 @@ const { TaskFunctions } = require('./test-types.cjs') +const bounded = async ( + promise, + message = 'Task remained pending', + timeoutMs = 5000 +) => { + let timeout + try { + return await Promise.race([ + promise, + new Promise((resolve, reject) => { + timeout = setTimeout(() => reject(new Error(message)), timeoutMs) + }), + ]) + } finally { + clearTimeout(timeout) + } +} + +const waitForMessage = (emitter, predicate) => + new Promise(resolve => { + const listener = message => { + if (predicate(message)) { + emitter.off('message', listener) + resolve(message) + } + } + emitter.on('message', listener) + }) + const waitWorkerEvents = async ( pool, workerEvent, @@ -150,12 +179,14 @@ const executeTaskFunction = data => { } module.exports = { + bounded, executeTaskFunction, factorial, fibonacci, jsonIntegerSerialization, sleep, sleepTaskFunction, + waitForMessage, waitPoolEvents, waitWorkerEvents, } diff --git a/tests/worker-files/thread/taskAccountingWorker.mjs b/tests/worker-files/thread/taskAccountingWorker.mjs new file mode 100644 index 000000000..2deaf9c1a --- /dev/null +++ b/tests/worker-files/thread/taskAccountingWorker.mjs @@ -0,0 +1,24 @@ +import { parentPort } from 'node:worker_threads' + +import { ThreadWorker } from '../../../lib/index.mjs' + +export default new ThreadWorker( + async ({ fail, gate, label }) => { + parentPort.postMessage({ label }) + if (gate != null) { + await Atomics.waitAsync(new Int32Array(gate), 0, 0).value + } + if (fail === true) { + throw new Error(label) + } + return label + }, + { + killHandler: () => { + // Preserve native SOFT retirement without racing a kill ACK against node removal. + // eslint-disable-next-line n/no-process-exit + process.exit(0) + }, + maxInactiveTime: 100, + } +) diff --git a/tests/worker-files/thread/taskStealingRetryWorker.mjs b/tests/worker-files/thread/taskStealingRetryWorker.mjs new file mode 100644 index 000000000..55e01860b --- /dev/null +++ b/tests/worker-files/thread/taskStealingRetryWorker.mjs @@ -0,0 +1,31 @@ +import { parentPort, threadId, workerData } from 'node:worker_threads' + +import { ThreadWorker } from '../../../lib/index.mjs' + +const waitForGate = async gate => { + await Atomics.waitAsync(new Int32Array(gate), 0, 0).value +} + +const run = async ({ gate, label }) => { + parentPort.postMessage({ label, workerId: threadId }) + await waitForGate(gate) + return { label, workerId: threadId } +} + +// Entry order defines the default function; keep loader first independently of key sorting. +const taskFunctions = Object.fromEntries([ + ['loader', { taskFunction: run, workerNodeKeys: [0] }], + ['anchor', { taskFunction: run, workerNodeKeys: [0, 2] }], + ['survivor', { taskFunction: run, workerNodeKeys: [0, 3] }], +]) + +export default new ThreadWorker(taskFunctions, { + killHandler: () => { + parentPort.postMessage({ label: 'kill-handler', workerId: threadId }) + Atomics.wait(new Int32Array(workerData.killGate), 0, 0) + // Exit from the public SOFT-kill hook without racing a kill ACK against node removal. + // eslint-disable-next-line n/no-process-exit + process.exit(0) + }, + maxInactiveTime: 1000, +})