feat: offload drain events to microtask

src/index.ts

@@ -465,7 +465,7 @@ class ThreadPool {
       this.histogram?.recordWaitTime(now - task.created)
       task.started = now;
       candidate[kWorkerData].postTask(task);
-      this._maybeDrain();
+      queueMicrotask(() => this._maybeDrain());
       // If candidate, let's try to distribute more tasks
       return true;
     }
@@ -530,7 +530,7 @@ class ThreadPool {
           resolve(result);
         }
 
-        this._maybeDrain();
+        queueMicrotask(() => this._maybeDrain());
       },
       signal,
       this.publicInterface.asyncResource.asyncId());
@@ -575,7 +575,7 @@ class ThreadPool {
         this.taskQueue.push(taskInfo);
       }
 
-      this._maybeDrain();
+      queueMicrotask(() => this._maybeDrain());
       return ret;
     }
 
@@ -595,7 +595,7 @@ class ThreadPool {
       }
     };
 
-    this._maybeDrain();
+    queueMicrotask(() => this._maybeDrain());
     return ret;
   }
 
@@ -616,10 +616,10 @@ class ThreadPool {
 
     if (maxCapacity === currentUsage) {
       this._needsDrain = true;
-      this.publicInterface.emit('needsDrain');
+      queueMicrotask(() => this.publicInterface.emit('needsDrain'));
     } else if (maxCapacity > currentUsage && this._needsDrain) {
       this._needsDrain = false;
-      this.publicInterface.emit('drain');
+      queueMicrotask(() => this.publicInterface.emit('drain'));
     }
   }
 

test/post-task.test.ts

@@ -1,5 +1,5 @@
 import assert from 'node:assert';
-import { test } from 'node:test';
+import { test, TestContext } from 'node:test';
 import { MessageChannel } from 'node:worker_threads';
 import { resolve } from 'node:path';
 import Piscina from '..';
@@ -77,46 +77,44 @@ test('postTask() validates abortSignal', () => {
     /signal argument must be an object/);
 });
 
-test('Piscina emits drain', async () => {
+test('Piscina emits drain', async (t: TestContext) => {
   const pool = new Piscina({
     filename: resolve(__dirname, 'fixtures/eval.js'),
     maxThreads: 1
   });
 
-  let drained = false;
-  let needsDrain = true;
+  t.plan(2);
+
   pool.on('drain', () => {
-    drained = true;
-    needsDrain = pool.needsDrain;
+    t.assert.ok(true);
+    t.assert.ok(!pool.needsDrain);
   });
 
   await Promise.all([pool.run('123'), pool.run('123'), pool.run('123')]);
-
-  assert.ok(drained);
-  assert.ok(!needsDrain);
 });
 
-test('Piscina exposes/emits needsDrain to true when capacity is exceeded', (t, done) => {
+test('Piscina exposes/emits needsDrain to true when capacity is exceeded', async (t: TestContext) => {
   const pool = new Piscina({
     filename: resolve(__dirname, 'fixtures/eval.js'),
     maxQueue: 3,
     maxThreads: 1
   });
 
+  t.plan(2);
+
   pool.once('drain', () => {
-    assert.ok(true);
-    done();
+    t.assert.ok(true);
   });
   pool.once('needsDrain', () => {
-    assert.ok(true);
+    t.assert.ok(pool.needsDrain);
   });
 
-  pool.run('123');
-  pool.run('123');
-  pool.run('123');
-  pool.run('123');
-
-  assert.ok(pool.needsDrain);
+  await Promise.all([
+    pool.run('123'),
+    pool.run('123'),
+    pool.run('123'),
+    pool.run('123')
+  ]);
 });
 
 test('Piscina can use async loaded workers', async () => {