From: Jérôme Benoit Date: Sat, 31 Oct 2026 02:13:07 +0000 (+0100) Subject: feat(pool): start task stealing on unused workers (#3387) X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=3b807363403ccc3d22dd5e69e34c03497f6fe4df;p=poolifier.git feat(pool): start task stealing on unused workers (#3387) * feat(pool): start task stealing on unused workers Wake idle workers from queued work and readiness transitions. Keep retries cancellable without delaying completion-driven steals. Closes #2283 * [autofix.ci] apply automated fixes * docs: remove redundant task stealing detail --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 9064825ab..4e0c0845d 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -39,7 +39,6 @@ import { median, min, round, - sleep, } from '../utils.js' import { KillBehaviors } from '../worker/worker-options.js' import { @@ -493,6 +492,13 @@ export abstract class AbstractPool< > > + private taskStealingRetries?: Map< + IWorkerNode, + { taskName?: string; timeout: NodeJS.Timeout } + > + + private taskStealingTimeout?: NodeJS.Timeout + /** * Worker nodes whose termination was requested by the pool. */ @@ -683,6 +689,7 @@ export abstract class AbstractPool< throw new Error('Cannot destroy an already destroying pool') } this.destroying = true + this.unsetTaskStealing() try { await Promise.allSettled( this.workerNodes.map(async (_, workerNodeKey) => { @@ -1611,6 +1618,17 @@ export abstract class AbstractPool< ) } + private canStartTaskStealing (workerNodeKey: number): boolean { + const workerNode = this.workerNodes[workerNodeKey] + return ( + this.isWorkerNodeIdle(workerNodeKey) && + !workerNode.info.continuousStealing && + !this.workerTaskFunctionBootstraps.has(workerNode) && + !this.failedWorkerTaskFunctionBootstraps.has(workerNode) && + !this.terminatingWorkerNodes.has(workerNode) + ) + } + private captureTaskResponses ( workerNode: IWorkerNode, taskResponses: MessageValue[] | undefined @@ -1844,6 +1862,17 @@ export abstract class AbstractPool< ) } + private clearTaskStealingRetry ( + workerNode: IWorkerNode + ): string | undefined { + const retry = this.taskStealingRetries?.get(workerNode) + if (retry != null) { + clearTimeout(retry.timeout) + this.taskStealingRetries?.delete(workerNode) + } + return retry?.taskName + } + private createWorkerMessageListenerCleanup ( workerNode: IWorkerNode, listener: (message: MessageValue) => void @@ -1928,6 +1957,7 @@ export abstract class AbstractPool< this.updateTaskWorkerNodeKey(workerNodeKey, task) const tasksQueueSize = this.workerNodes[workerNodeKey].enqueueTask(task) this.checkAndEmitTaskQueuingEvents() + this.scheduleTaskStealing() return tasksQueueSize } @@ -2353,7 +2383,7 @@ export abstract class AbstractPool< private readonly handleWorkerNodeIdleEvent = ( eventDetail: WorkerNodeEventDetail, - previousStolenTask?: Task + previousTaskName?: string ): void => { const { workerNodeKey } = eventDetail if (workerNodeKey == null) { @@ -2366,49 +2396,62 @@ export abstract class AbstractPool< if (workerNode == null) { return } - if ( - !workerNode.info.continuousStealing && - (this.cannotStealTask() || this.isStealingRatioReached()) - ) { - return - } + previousTaskName = + this.clearTaskStealingRetry(workerNode) ?? previousTaskName const workerNodeTasksUsage = workerNode.usage.tasks if ( - workerNode.info.continuousStealing && - !this.isWorkerNodeIdle(workerNodeKey) + this.opts.enableTasksQueue !== true || + this.opts.tasksQueueOptions?.taskStealing !== true || + this.opts.tasksQueueOptions.tasksStealingRatio === 0 || + this.cannotStealTask() || + !this.isWorkerNodeIdle(workerNodeKey) || + this.workerTaskFunctionBootstraps.has(workerNode) || + this.failedWorkerTaskFunctionBootstraps.has(workerNode) || + this.terminatingWorkerNodes.has(workerNode) ) { workerNode.info.continuousStealing = false if (workerNodeTasksUsage.sequentiallyStolen > 0) { this.resetTaskSequentiallyStolenStatisticsWorkerUsage( workerNodeKey, - previousStolenTask?.name + previousTaskName ) } return } + if (!workerNode.info.continuousStealing && this.isStealingRatioReached()) { + return + } workerNode.info.continuousStealing = true const stolenTask = this.workerNodeStealTask(workerNodeKey) this.updateTaskSequentiallyStolenStatisticsWorkerUsage( workerNodeKey, stolenTask?.name, - previousStolenTask?.name + previousTaskName ) - sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen)) - .then(() => { + const stolenTaskName = stolenTask?.name + const timeout = setTimeout(() => { + if (this.taskStealingRetries?.get(workerNode)?.timeout !== timeout) { + return + } + this.taskStealingRetries.delete(workerNode) + try { // Worker nodes can be removed during the delay, shifting the keys. const currentWorkerNodeKey = this.workerNodes.indexOf(workerNode) - if (currentWorkerNodeKey === -1) { - return undefined + if (currentWorkerNodeKey !== -1) { + this.handleWorkerNodeIdleEvent( + { workerNodeKey: currentWorkerNodeKey }, + stolenTaskName + ) } - this.handleWorkerNodeIdleEvent( - { ...eventDetail, workerNodeKey: currentWorkerNodeKey }, - stolenTask - ) - return undefined - }) - .catch((error: unknown) => { + } catch (error) { this.emitter?.emit(PoolEvents.error, error) - }) + } + }, exponentialDelay()).unref() + this.taskStealingRetries ??= new Map() + this.taskStealingRetries.set(workerNode, { + taskName: stolenTaskName, + timeout, + }) } private handleWorkerReadyResponse (message: MessageValue): void { @@ -2439,6 +2482,7 @@ export abstract class AbstractPool< this.sendStatisticsMessageToWorker(workerNodeKey) this.setTasksQueuePriority(workerNodeKey) this.checkAndEmitReadyEvent() + this.scheduleTaskStealing() } private initEventEmitter (): void { @@ -2795,6 +2839,12 @@ export abstract class AbstractPool< this.failedWorkerTaskFunctionBootstraps.delete(workerNode) const workerNodeKey = this.workerNodes.indexOf(workerNode) if (workerNodeKey !== -1) { + const taskName = this.clearTaskStealingRetry(workerNode) + workerNode.info.continuousStealing = false + this.resetTaskSequentiallyStolenStatisticsWorkerUsage( + workerNodeKey, + taskName + ) this.workerNodes.splice(workerNodeKey, 1) for (const promiseResponse of this.promiseResponseMap.values()) { if (promiseResponse.workerNodeKey === workerNodeKey) { @@ -2872,6 +2922,7 @@ export abstract class AbstractPool< this.sendStatisticsMessageToWorker(workerNodeKey) workerNode.info.ready = bootstrap.ready this.checkAndEmitReadyEvent() + this.scheduleTaskStealing() } catch (error) { // Retired bootstraps no longer block waiting submissions. if (this.workerNodes.includes(workerNode)) { @@ -2928,6 +2979,36 @@ export abstract class AbstractPool< await Promise.allSettled(promises) } + private scheduleTaskStealing (): void { + if ( + this.taskStealingTimeout != null || + this.opts.enableTasksQueue !== true || + this.opts.tasksQueueOptions?.taskStealing !== true || + this.opts.tasksQueueOptions.tasksStealingRatio === 0 || + this.cannotStealTask() + ) { + return + } + this.taskStealingTimeout = setTimeout(() => { + delete this.taskStealingTimeout + try { + if (this.cannotStealTask()) return + for (const workerNodeKey of this.workerNodes.keys()) { + if (!this.canStartTaskStealing(workerNodeKey)) continue + this.handleWorkerNodeIdleEvent({ workerNodeKey }) + if (this.cannotStealTask()) return + if (this.canStartTaskStealing(workerNodeKey)) { + // Retry admission when another worker relinquishes its stealing slot. + this.scheduleTaskStealing() + return + } + } + } catch (error) { + this.emitter?.emit(PoolEvents.error, error) + } + }, exponentialDelay()).unref() + } + private async sendKillMessageToWorker ( workerNodeKey: number, timeout = 1000 @@ -3229,6 +3310,7 @@ export abstract class AbstractPool< for (const workerNodeKey of this.workerNodes.keys()) { this.workerNodes[workerNodeKey].on('idle', this.handleWorkerNodeIdleEvent) } + this.scheduleTaskStealing() } private shallExecuteTask (workerNodeKey: number): boolean { @@ -3368,12 +3450,19 @@ export abstract class AbstractPool< } private unsetTaskStealing (): void { + clearTimeout(this.taskStealingTimeout) + delete this.taskStealingTimeout for (const workerNodeKey of this.workerNodes.keys()) { - this.workerNodes[workerNodeKey].off( - 'idle', - this.handleWorkerNodeIdleEvent + const workerNode = this.workerNodes[workerNodeKey] + const taskName = this.clearTaskStealingRetry(workerNode) + workerNode.info.continuousStealing = false + this.resetTaskSequentiallyStolenStatisticsWorkerUsage( + workerNodeKey, + taskName ) + workerNode.off('idle', this.handleWorkerNodeIdleEvent) } + delete this.taskStealingRetries } private updateTaskSequentiallyStolenStatisticsWorkerUsage ( diff --git a/src/utils.ts b/src/utils.ts index 30f346907..816793411 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -35,18 +35,6 @@ export const availableParallelism = (): number => { return availableParallelism } -/** - * Sleeps for the given amount of milliseconds. - * @param ms - The amount of milliseconds to sleep. - * @returns A promise that resolves after the given amount of milliseconds. - * @internal - */ -export const sleep = async (ms: number): Promise => { - await new Promise(resolve => { - setTimeout(resolve, ms) - }) -} - /** * Computes the retry delay in milliseconds using an exponential back off algorithm. * @param retryNumber - The number of retries that have already been attempted diff --git a/tests/pools/idle-task-stealing.test.mjs b/tests/pools/idle-task-stealing.test.mjs new file mode 100644 index 000000000..a255ce227 --- /dev/null +++ b/tests/pools/idle-task-stealing.test.mjs @@ -0,0 +1,373 @@ +import { once } from 'node:events' +import { useFakeTimers } from 'sinon' +import { describe, expect, it } from 'vitest' + +import { FixedThreadPool, PoolEvents } from '../../lib/index.mjs' +import { bounded } from '../test-utils.cjs' + +const release = gate => { + Atomics.store(new Int32Array(gate), 0, 1) + Atomics.notify(new Int32Array(gate), 0) +} + +const gatedTask = async ({ gate, label }) => { + const state = new Int32Array(gate) + Atomics.store(state, 1, 1) + Atomics.notify(state, 1) + await Atomics.waitAsync(state, 0, 0).value + return label +} + +describe('Unused idle worker task stealing', () => { + it.each([ + 'startup', + 'options', + 'queues', + 'late backlog', + 'disabled pending retry', + 'zero stealing ratio', + ])('starts stealing through %s', async activation => { + const tasksQueueOptions = { + concurrency: 1, + tasksStealingOnBackPressure: false, + tasksStealingRatio: activation === 'zero stealing ratio' ? 0 : 1, + taskStealing: + activation === 'startup' || activation === 'zero stealing ratio', + } + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskAccountingWorker.mjs', + { + enableTasksQueue: activation !== 'queues', + ...(activation !== 'queues' && { tasksQueueOptions }), + } + ) + const blockerGate = new SharedArrayBuffer(8) + const queuedGate = new SharedArrayBuffer(8) + release(queuedGate) + const tasks = [] + let clock + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await pool.addTaskFunction('affinity', { + taskFunction: gatedTask, + workerNodeKeys: [0], + }) + if (activation === 'queues') { + pool.enableTasksQueue(true, { + ...tasksQueueOptions, + taskStealing: true, + }) + } + if ( + [ + 'disabled pending retry', + 'late backlog', + 'zero stealing ratio', + ].includes(activation) + ) { + clock = useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + if (activation === 'late backlog') { + pool.setTasksQueueOptions({ taskStealing: true }) + // Exhaust an empty startup window before any workload exists. + await clock.tickAsync(240) + } + } + const blocker = pool.execute( + { gate: blockerGate, label: 'blocked' }, + 'affinity' + ) + tasks.push(blocker.catch(() => undefined)) + await Atomics.waitAsync(new Int32Array(blockerGate), 1, 0, 5000).value + expect(Atomics.load(new Int32Array(blockerGate), 1)).toBe(1) + const destination = pool.workerNodes[1] + expect(destination.usage.tasks.executed).toBe(0) + expect(destination.usage.tasks.executing).toBe(0) + const finished = once(destination, 'taskFinished', { + signal: AbortSignal.timeout(5000), + }) + tasks.push(finished.catch(() => undefined)) + const queued = pool.execute( + { gate: queuedGate, label: 'queued' }, + 'affinity' + ) + tasks.push(queued.catch(() => undefined)) + if (activation === 'options') { + pool.setTasksQueueOptions({ taskStealing: true }) + } else if (activation === 'disabled pending retry') { + pool.setTasksQueueOptions({ taskStealing: true }) + // Let execute() enqueue before cancelling its delayed stealing attempt. + await clock.tickAsync(0) + pool.setTasksQueueOptions({ taskStealing: false }) + await clock.tickAsync(240) + expect(pool.info.stolenTasks).toBe(0) + expect(destination.usage.tasks.executed).toBe(0) + expect(pool.info.queuedTasks).toBe(1) + pool.setTasksQueueOptions({ taskStealing: true }) + } else if (activation === 'zero stealing ratio') { + await clock.tickAsync(240) + expect(pool.info.stolenTasks).toBe(0) + expect(pool.info.queuedTasks).toBe(1) + pool.setTasksQueueOptions({ tasksStealingRatio: 1 }) + } + if (clock != null) await clock.tickAsync(240) + await finished + await expect(queued).resolves.toBe('queued') + expect(Atomics.load(new Int32Array(blockerGate), 0)).toBe(0) + expect(pool.workerNodes[0].usage.tasks.executing).toBe(1) + expect(destination.usage.tasks.executed).toBe(1) + expect(pool.info.stolenTasks).toBe(1) + } finally { + clock?.restore() + release(blockerGate) + release(queuedGate) + try { + await bounded( + Promise.all(tasks), + 'Gated stealing tasks remained pending' + ) + } finally { + await bounded( + pool.destroy(), + 'Idle stealing pool cleanup remained pending' + ) + } + } + }) + + it('waits for runtime task-function replay before stealing onto a replacement', async () => { + const replayGates = Array.from( + { length: 3 }, + () => new SharedArrayBuffer(4) + ) + release(replayGates[0]) + release(replayGates[1]) + const replayBlocked = Promise.withResolvers() + const blocked = Promise.withResolvers() + const blockerGate = new SharedArrayBuffer(4) + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + enableTasksQueue: true, + messageHandler: message => { + if (message.blocked) blocked.resolve() + if (message.replayBlocked && message.replayGeneration === 2) { + replayBlocked.resolve(message.workerId) + } + }, + tasksQueueOptions: { + concurrency: 1, + tasksStealingOnBackPressure: false, + taskStealing: true, + }, + workerOptions: { + workerData: { + asyncReplay: true, + blockWorkerKeys: [0], + replayCounter: new SharedArrayBuffer(4), + replayGates, + }, + }, + } + ) + const tasks = [] + let clock + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await pool.addTaskFunction('runtime', { + taskFunction: data => data, + workerNodeKeys: [0], + }) + const blocker = pool.execute({ gate: blockerGate }, 'block') + tasks.push(blocker.catch(() => undefined)) + await bounded(blocked.promise) + await pool.workerNodes[1].worker.terminate() + const workerId = await bounded(replayBlocked.promise) + const replacement = pool.workerNodes.find( + node => node.info.id === workerId + ) + expect(replacement.info.ready).toBe(false) + clock = useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + const queued = pool.execute('queued', 'runtime') + tasks.push(queued.catch(() => undefined)) + await clock.tickAsync(240) + expect(replacement.usage.tasks.executed).toBe(0) + expect(pool.info.queuedTasks).toBe(1) + clock.restore() + clock = undefined + release(replayGates[2]) + await expect( + bounded(queued, 'Replay did not wake idle stealing') + ).resolves.toBe('queued') + expect(Atomics.load(new Int32Array(blockerGate), 0)).toBe(0) + expect(pool.workerNodes[0].usage.tasks.executing).toBe(1) + expect(replacement.info.ready).toBe(true) + expect(replacement.usage.tasks.executed).toBe(1) + expect(pool.info.stolenTasks).toBe(1) + } finally { + clock?.restore() + replayGates.forEach(release) + release(blockerGate) + try { + await bounded( + Promise.all(tasks), + 'Replay stealing tasks remained pending' + ) + } finally { + await bounded( + pool.destroy(), + 'Replay stealing pool cleanup remained pending' + ) + } + } + }) + + it('retries idle-worker admission after active stealers become busy', async () => { + const pool = new FixedThreadPool( + 4, + './tests/worker-files/thread/taskAccountingWorker.mjs', + { + enableTasksQueue: true, + tasksQueueOptions: { + concurrency: 1, + tasksStealingOnBackPressure: false, + tasksStealingRatio: 0.25, + taskStealing: true, + }, + } + ) + const gates = Array.from({ length: 3 }, () => new SharedArrayBuffer(8)) + const openGate = new SharedArrayBuffer(8) + release(openGate) + const tasks = [] + let clock + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await pool.addTaskFunction('affinity', { + taskFunction: gatedTask, + workerNodeKeys: [0], + }) + const blocker = pool.execute( + { gate: gates[0], label: 'source' }, + 'affinity' + ) + tasks.push(blocker.catch(() => undefined)) + await Atomics.waitAsync(new Int32Array(gates[0]), 1, 0, 5000).value + expect(Atomics.load(new Int32Array(gates[0]), 1)).toBe(1) + clock = useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + for (const gate of gates.slice(1)) { + tasks.push( + pool + .execute({ gate, label: 'held-stealer' }, 'affinity') + .catch(() => undefined) + ) + } + await clock.tickAsync(120) + const finished = Promise.any( + pool.workerNodes + .slice(1) + .map(node => + once(node, 'taskFinished', { signal: AbortSignal.timeout(5000) }) + ) + ) + tasks.push(finished.catch(() => undefined)) + const queued = pool.execute( + { gate: openGate, label: 'queued' }, + 'affinity' + ) + tasks.push(queued.catch(() => undefined)) + await clock.tickAsync(1000) + await finished + await expect(queued).resolves.toBe('queued') + expect(Atomics.load(new Int32Array(gates[0]), 0)).toBe(0) + expect(pool.workerNodes[0].usage.tasks.executing).toBe(1) + } finally { + clock?.restore() + gates.forEach(release) + release(openGate) + try { + await bounded( + Promise.all(tasks), + 'Ratio-limited stealing tasks remained pending' + ) + } finally { + await bounded( + pool.destroy(), + 'Ratio-limited stealing pool cleanup remained pending' + ) + } + } + }) + + it('drains a short affinity backlog while the source remains blocked', async () => { + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskAccountingWorker.mjs', + { + enableTasksQueue: true, + tasksQueueOptions: { + concurrency: 1, + tasksStealingOnBackPressure: false, + taskStealing: true, + }, + } + ) + const blockerGate = new SharedArrayBuffer(8) + const openGate = new SharedArrayBuffer(8) + release(openGate) + const tasks = [] + const overflowWarnings = [] + const onWarning = warning => { + if (warning.name === 'TimeoutOverflowWarning') { + overflowWarnings.push(warning) + } + } + process.on('warning', onWarning) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await pool.addTaskFunction('affinity', { + taskFunction: gatedTask, + workerNodeKeys: [0], + }) + tasks.push( + pool + .execute({ gate: blockerGate, label: 'blocked' }, 'affinity') + .catch(() => undefined) + ) + await Atomics.waitAsync(new Int32Array(blockerGate), 1, 0, 5000).value + expect(Atomics.load(new Int32Array(blockerGate), 1)).toBe(1) + const labels = Array.from({ length: 64 }, (_, index) => `queued-${index}`) + const queued = labels.map(label => + pool.execute({ gate: openGate, label }, 'affinity') + ) + tasks.push(...queued.map(task => task.catch(() => undefined))) + await expect( + bounded( + Promise.all(queued), + 'Affinity backlog stopped on an idle worker' + ) + ).resolves.toEqual(labels) + expect(Atomics.load(new Int32Array(blockerGate), 0)).toBe(0) + expect(pool.workerNodes[0].usage.tasks.executing).toBe(1) + expect(pool.workerNodes[1].usage.tasks.executed).toBe(labels.length) + expect(pool.info.stolenTasks).toBe(labels.length) + expect(overflowWarnings).toEqual([]) + } finally { + process.off('warning', onWarning) + release(blockerGate) + release(openGate) + try { + await bounded( + Promise.all(tasks), + 'Short stealing tasks remained pending' + ) + } finally { + await bounded( + pool.destroy(), + 'Short stealing pool cleanup remained pending' + ) + } + } + }) +}) diff --git a/tests/pools/task-accounting.test.mjs b/tests/pools/task-accounting.test.mjs index 1b9b88c43..527f8a4c9 100644 --- a/tests/pools/task-accounting.test.mjs +++ b/tests/pools/task-accounting.test.mjs @@ -1,5 +1,6 @@ import { EventEmitter, once } from 'node:events' import { setTimeout as delay } from 'node:timers/promises' +import { useFakeTimers } from 'sinon' import { describe, expect, it, vi } from 'vitest' import { @@ -298,13 +299,14 @@ describe('Idle stealing retry accounting', () => { tasksQueueOptions: { concurrency: 1, tasksStealingOnBackPressure: false, - taskStealing: true, + taskStealing: false, }, workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED, workerOptions: { workerData: { killGate: gates.kill } }, } ) const tasks = [] + let clock const execute = (gate, label, name) => { const task = pool.execute({ gate, label }, name) // Attach rejection handling immediately, including when setup fails. @@ -337,6 +339,8 @@ describe('Idle stealing retry accounting', () => { message.label === 'kill-handler' && message.workerId === retiree.info.id ) + clock = useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + pool.setTasksQueueOptions({ taskStealing: true }) expect(pool.workerNodes.indexOf(retrying)).toBe(2) const retired = once(retiree.worker, 'exit', { signal: AbortSignal.timeout(5000), @@ -358,6 +362,9 @@ describe('Idle stealing retry accounting', () => { expect(retrying.usage.tasks.sequentiallyStolen).toBe(1) expect(retrying.info.continuousStealing).toBe(true) expect(wrongTarget.usage.tasks.stolen).toBe(0) + await clock.tickAsync(240) + clock.restore() + clock = undefined await vi.waitFor( () => { @@ -374,6 +381,7 @@ describe('Idle stealing retry accounting', () => { expect(retrying.usage.tasks.sequentiallyStolen).toBe(0) expect(pool.info.stolenTasks).toBe(1) } finally { + clock?.restore() Object.values(gates).forEach(release) try { await Promise.race([ diff --git a/tests/utils.test.mjs b/tests/utils.test.mjs index 77e1d659c..a730f499e 100644 --- a/tests/utils.test.mjs +++ b/tests/utils.test.mjs @@ -1,6 +1,5 @@ import { randomInt } from 'node:crypto' import os from 'node:os' -import { useFakeTimers } from 'sinon' import { describe, expect, it } from 'vitest' import { KillBehaviors } from '../lib/index.mjs' @@ -19,7 +18,6 @@ import { // once, round, secureRandom, - sleep, } from '../lib/utils.mjs' describe('Utils test suite', () => { @@ -44,24 +42,6 @@ describe('Utils test suite', () => { expect(parallelism).toBe(expectedParallelism) }) - it('Verify sleep() behavior', async () => { - const clock = useFakeTimers() - try { - const delay = 1000 - let resolved = false - const sleepPromise = sleep(delay).then(() => { - resolved = true - return true - }) - expect(resolved).toBe(false) - await clock.tickAsync(delay) - await sleepPromise - expect(resolved).toBe(true) - } finally { - clock.restore() - } - }) - it('Verify exponentialDelay() behavior', () => { const delay = exponentialDelay(randomInt(1000)) expect(typeof delay === 'number').toBe(true)