diff --git a/README.md b/README.md index a34896d..a6ab85f 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,7 @@ We have a similar API to Piscina, so for more information, you can read Piscina' - `isolateWorkers`: Default to `false`. Always starts with a fresh worker when running tasks to isolate the environment. - `workerId`: Each worker now has an id ( <= `maxThreads`) that can be imported from `tinypool` in the worker itself (or `process.__tinypool_state__.workerId`) +- `terminateTimeout`: Defaults to `null`. If terminating a worker takes `terminateTimeout` amount of milliseconds to execute, an error is raised. ## Authors diff --git a/src/index.ts b/src/index.ts index 2dc5b62..a63382e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -122,6 +122,7 @@ interface Options { minThreads?: number maxThreads?: number idleTimeout?: number + terminateTimeout?: number maxQueue?: number | 'auto' concurrentTasksPerWorker?: number useAtomics?: boolean @@ -447,14 +448,38 @@ class WorkerInfo extends AsynchronouslyCreatedResource { ) } - async destroy(): Promise { - await this.worker.terminate() - this.port.close() - this.clearIdleTimeout() - for (const taskInfo of this.taskInfos.values()) { - taskInfo.done(Errors.ThreadTermination()) - } - this.taskInfos.clear() + async destroy(timeout?: number): Promise { + let resolve: () => void + let reject: (err: Error) => void + + const ret = new Promise((res, rej) => { + resolve = res + reject = rej + }) + + const timer = timeout + ? setTimeout( + () => reject(new Error('Failed to terminate worker')), + timeout + ) + : null + + this.worker.terminate().then(() => { + if (timer !== null) { + clearTimeout(timer) + } + + this.port.close() + this.clearIdleTimeout() + for (const taskInfo of this.taskInfos.values()) { + taskInfo.done(Errors.ThreadTermination()) + } + this.taskInfos.clear() + + resolve() + }) + + return ret } clearIdleTimeout(): void { @@ -771,12 +796,12 @@ class ThreadPool { } } - _removeWorker(workerInfo: WorkerInfo): void { + _removeWorker(workerInfo: WorkerInfo): Promise { workerInfo.freeWorkerId() - workerInfo.destroy() - this.workers.delete(workerInfo) + + return workerInfo.destroy(this.options.terminateTimeout) } _onWorkerAvailable(workerInfo: WorkerInfo): void { @@ -845,14 +870,16 @@ class ThreadPool { this.completed++ if (err !== null) { reject(err) - } else { - resolve(result) } // When `isolateWorkers` is enabled, remove the worker after task is finished if (this.options.isolateWorkers && taskInfo.workerInfo) { this._removeWorker(taskInfo.workerInfo) - this._ensureEnoughWorkersForTaskQueue() + .then(() => this._ensureEnoughWorkersForTaskQueue()) + .then(() => resolve(result)) + .catch(reject) + } else { + resolve(result) } }, signal, diff --git a/test/termination-timeout.test.ts b/test/termination-timeout.test.ts new file mode 100644 index 0000000..e6146ea --- /dev/null +++ b/test/termination-timeout.test.ts @@ -0,0 +1,32 @@ +import { dirname, resolve } from 'path' +import { Tinypool } from 'tinypool' +import { fileURLToPath } from 'url' + +const __dirname = dirname(fileURLToPath(import.meta.url)) +const cleanups: (() => Promise)[] = [] + +afterEach(async () => { + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())) +}) + +test('termination timeout throws when worker does not terminate in time', async () => { + const pool = new Tinypool({ + filename: resolve(__dirname, 'fixtures/sleep.js'), + terminateTimeout: 10, + minThreads: 1, + maxThreads: 2, + isolateWorkers: true, + }) + + expect(pool.threads.length).toBe(1) + + const worker = pool.threads[0] + expect(worker).toBeTruthy() + + cleanups.push(worker!.terminate.bind(worker)) + worker!.terminate = () => new Promise(() => {}) + + await expect(pool.run('default')).rejects.toThrowError( + 'Failed to terminate worker' + ) +})