feat: add terminateTimeout option (#50)

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
 

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<void> {
-    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<void> {
+    let resolve: () => void
+    let reject: (err: Error) => void
+
+    const ret = new Promise<void>((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<void> {
     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,

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<unknown>)[] = []
+
+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'
+  )
+})