+import type { UUID } from 'node:crypto'
import type { Transferable } from 'node:worker_threads'
import { AsyncResource } from 'node:async_hooks'
*
* 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<Response>
- > = new Map<
- `${string}-${string}-${string}-${string}-${string}`,
- PromiseResponseWrapper<Response>
- >()
+ protected promiseResponseMap: Map<UUID, PromiseResponseWrapper<Response>> =
+ new Map<UUID, PromiseResponseWrapper<Response>>()
/**
* Whether the pool is started or not.
needWorkerChoiceStrategiesUpdate = true
}
if (
- this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
+ this.shallUpdateTaskFunctionWorkerUsage(
+ this.getWorkerInfo(workerNodeKey)
+ ) &&
message.taskPerformance?.name != null &&
this.workerNodes[workerNodeKey].getTaskFunctionWorkerUsage(
message.taskPerformance.name
)
}
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
}
workerNode.info.ready = false
let workerNodeTasks:
- | Map<
- `${string}-${string}-${string}-${string}-${string}`,
- PromiseResponseWrapper<Response>
- >
+ | Map<UUID, PromiseResponseWrapper<Response>>
| undefined
if (settlingTasks) {
const workerNodeKey = this.workerNodes.indexOf(workerNode)
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!)
}
}
- 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
}
private readonly handleWorkerError = (
- taskId: `${string}-${string}-${string}-${string}-${string}`,
+ taskId: UUID,
workerError: WorkerError
): Error => {
const { aborted, error, message, name, stack } = workerError
)
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) => {
}
private rejectTask (
- taskId: `${string}-${string}-${string}-${string}-${string}`,
+ taskId: UUID,
promiseResponse: PromiseResponseWrapper<Response>,
error: unknown
): void {
* @param error - The worker failure.
*/
private rejectWorkerNodeTasks (
- workerNodeTasks: Map<
- `${string}-${string}-${string}-${string}-${string}`,
- PromiseResponseWrapper<Response>
- >,
+ workerNodeTasks: Map<UUID, PromiseResponseWrapper<Response>>,
error?: Error
): void {
for (const [taskId, promiseResponse] of workerNodeTasks) {
}
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
/**
* 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) &&
}
if (
taskName != null &&
- this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
+ this.shallUpdateTaskFunctionWorkerUsage(
+ this.getWorkerInfo(workerNodeKey)
+ ) &&
workerNode.getTaskFunctionWorkerUsage(taskName) != null
) {
const taskFunctionWorkerUsage =
++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
+import type { UUID } from 'node:crypto'
import type { EventEmitter } from 'node:events'
import type { MessageChannel, WorkerOptions } from 'node:worker_threads'
* @internal
*/
export interface WorkerNodeEventDetail {
- taskId?: `${string}-${string}-${string}-${string}-${string}`
+ taskId?: UUID
workerId?: number
workerNodeKey?: number
}
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'
/**
* Task UUID.
*/
- readonly taskId?: `${string}-${string}-${string}-${string}-${string}`
+ readonly taskId?: UUID
/**
* Timestamp.
*/
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'
/**
* 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<UUID, () => void>
/**
* Task function object(s) processed by the worker when the pool's `execute` method is invoked.
throw new Error('isMain parameter is mandatory')
}
this.checkTaskFunctions(taskFunctions)
- this.taskAbortFunctions = new Map<
- `${string}-${string}-${string}-${string}-${string}`,
- () => void
- >()
+ this.taskAbortFunctions = new Map<UUID, () => void>()
this.checkWorkerOptions(this.opts)
if (!this.isMain) {
this.getMainWorker().once('message', this.handleReadyMessage.bind(this))
*/
private getAbortableTaskFunction (
name: string,
- taskId: `${string}-${string}-${string}-${string}-${string}`
+ taskId: UUID
): TaskAsyncFunction<Data, Response> {
return async (data?: Data): Promise<Response> =>
await new Promise<Response>(
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
--- /dev/null
+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'
+ )
+ }
+ }
+ })
+})
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
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 }
}
}
-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
await bounded(exit)
expect(pool.promiseResponseMap.size).toBe(0)
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Worker exit pool cleanup remained pending'
+ )
}
})
}
expect(result).toBeInstanceOf(Error)
expect(pool.promiseResponseMap.size).toBe(0)
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Silent exit pool cleanup remained pending'
+ )
}
})
}
})
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')
},
})
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')
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'
+ )
}
})
expect(completions).toStrictEqual(executions)
expect(pool.promiseResponseMap.size).toBe(0)
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Affinity pool cleanup remained pending'
+ )
}
})
}
expect(outcomes[1]).toBeInstanceOf(Error)
await bounded(exit)
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Default affinity pool cleanup remained pending'
+ )
}
})
expect(outcomes[2]).toStrictEqual(eligibleData)
await bounded(exit)
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Eligible migration pool cleanup remained pending'
+ )
}
})
},
})
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')
expect(await bounded(stolenResult)).toBeInstanceOf(Error)
await bounded(destinationExit)
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Stolen task pool cleanup remained pending'
+ )
}
})
},
})
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')
await bounded(exit)
expect(pool.promiseResponseMap.size).toBe(0)
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Redistribution abort pool cleanup remained pending'
+ )
}
})
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'
+ )
}
})
},
})
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')
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 () => {
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')
expect(await bounded(survivorResult)).toBeInstanceOf(Error)
expect(removedTaskSettled).toBe(false)
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Intentional termination pool cleanup remained pending'
+ )
}
})
expect(error).toBe(rawWorkerError)
}
} finally {
- await pool.destroy()
+ await bounded(
+ pool.destroy(),
+ 'Native failure pool cleanup remained pending'
+ )
}
})
}
} 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'
+ )
+ }
}
})
}
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([
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
)
)
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
)
)
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
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()
+ }
}
})
} 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')
}
}
})
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,
}
module.exports = {
+ bounded,
executeTaskFunction,
factorial,
fibonacci,
jsonIntegerSerialization,
sleep,
sleepTaskFunction,
+ waitForMessage,
waitPoolEvents,
waitWorkerEvents,
}
--- /dev/null
+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,
+ }
+)
--- /dev/null
+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,
+})
repositories / poolifier.git / commitdiff
