feat: add stricterFIFO option for queue fairness (#1100)

Co-authored-by: Cursor <cursoragent@cursor.com>
Signed-off-by: huntiez <huntieezz@outlook.com>

Author: huntiezz

README.md

@@ -463,6 +463,10 @@ This class extends [`EventEmitter`][] from Node.js.
     complete all in-flight tasks when `close()` is called. The default is `30000`
   - `recordTiming`: (`boolean`) By default, run and wait time will be recorded
     for the pool. To disable, set to `false`.
+  - `stricterFIFO`: (`boolean`) When `true`, tasks that cannot be dispatched
+    immediately are returned to the front of the queue instead of the back.
+    This avoids head-of-line blocking under sustained load (especially with a
+    single worker). Defaults to `false`.
 
 Use caution when setting resource limits. Setting limits that are too low may
 result in the `Piscina` worker threads being unusable.

src/index.ts

@@ -70,6 +70,7 @@ interface Options {
   trackUnmanagedFds? : boolean,
   closeTimeout?: number,
   recordTiming?: boolean,
+  stricterFIFO?: boolean,
   loadBalancer?: PiscinaLoadBalancer,
   workerHistogram?: boolean,
 }
@@ -87,6 +88,7 @@ interface FilledOptions extends Options {
   niceIncrement : number,
   closeTimeout : number,
   recordTiming : boolean,
+  stricterFIFO : boolean,
   workerHistogram: boolean,
 }
 
@@ -122,6 +124,7 @@ const kDefaultOptions : FilledOptions = {
   trackUnmanagedFds: true,
   closeTimeout: 30000,
   recordTiming: true,
+  stricterFIFO: false,
   workerHistogram: false
 };
 
@@ -407,8 +410,8 @@ class ThreadPool {
       if (distributed) {
         // If task was distributed, we should continue to distribute more tasks
         continue;
-      } 
-      
+      }
+
       if (this.workers.size < this.options.maxThreads) {
         // We spawn if possible
         // TODO: scheduler will intercept this.
@@ -458,13 +461,27 @@ class ThreadPool {
       return true;
     }
 
+    this._returnUndistributedTask(task, !this.options.stricterFIFO);
+    return false;
+  }
+
+  _returnUndistributedTask (task: TaskInfo, toTail = false) : void {
     if (task.abortSignal != null) {
-      this.skipQueue.push(task);
+      if (toTail) {
+        this.skipQueue.push(task);
+      } else {
+        this.skipQueue.unshift(task);
+      }
+      return;
+    }
+
+    if (toTail) {
+      this.taskQueue.push(task);
+    } else if (this.taskQueue.unshift != null) {
+      this.taskQueue.unshift(task);
     } else {
       this.taskQueue.push(task);
     }
-
-    return false;
   }
 
   runTask (
@@ -787,6 +804,9 @@ export default class Piscina<Exports extends Record<string, (payload: any) => an
     if (opts.workerHistogram != null && (typeof opts.workerHistogram !== 'boolean')) {
       throw Errors.ValidationError('options.workerHistogram must be a boolean');
     }
+    if (opts.stricterFIFO !== undefined && (typeof opts.stricterFIFO !== 'boolean')) {
+      throw new TypeError('options.stricterFIFO must be a boolean');
+    }
 
     this.#pool = new ThreadPool(this, opts);
   }

src/task_queue/array_queue.ts

@@ -17,6 +17,10 @@ export class ArrayTaskQueue implements TaskQueue {
     this.tasks.push(task);
   }
 
+  unshift (task: Task): void {
+    this.tasks.unshift(task);
+  }
+
   remove (task: Task): void {
     const index = this.tasks.indexOf(task);
     assert.notStrictEqual(index, -1);

src/task_queue/common.ts

@@ -5,6 +5,7 @@ export interface TaskQueue {
     shift(): Task | null;
     remove(task: Task): void;
     push(task: Task): void;
+    unshift?(task: Task): void;
 }
 
 // Public Interface

src/task_queue/fixed_queue.ts

@@ -77,6 +77,11 @@ class FixedCircularBuffer {
     this.top = (this.top + 1) & kMask;
   }
 
+  unshift (data:Task) {
+    this.bottom = (this.bottom - 1) & kMask;
+    this.list[this.bottom] = data;
+  }
+
   shift () {
     const nextItem = this.list[this.bottom];
     if (nextItem === undefined) { return null; }
@@ -129,6 +134,16 @@ export class FixedQueue implements TaskQueue {
     this.#size++;
   }
 
+  unshift (data:Task) {
+    if (this.tail.isFull()) {
+      const newTail = new FixedCircularBuffer();
+      newTail.next = this.tail;
+      this.tail = newTail;
+    }
+    this.tail.unshift(data);
+    this.#size++;
+  }
+
   shift (): Task | null {
     const tail = this.tail;
     const next = tail.shift();

test/fixtures/task-logger.js

@@ -0,0 +1,8 @@
+const tasks = [];
+
+module.exports = {
+  run: task => {
+    tasks.push(task);
+  },
+  getLog: () => tasks
+};

test/queues.test.ts

@@ -0,0 +1,58 @@
+import * as assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { kQueueOptions } from '../dist/symbols';
+import { FixedQueue, ArrayTaskQueue, PiscinaTask as Task } from '..';
+
+// @ts-expect-error - it misses several properties, but it's enough for the test
+class NumberedTask implements Task {
+  constructor(readonly index: number) {}
+  get [kQueueOptions] () {
+    return null;
+  }
+}
+
+for (const QueueClass of [FixedQueue, ArrayTaskQueue]) {
+  test(`${QueueClass.name} - unshift`, () => {
+    const bufferSize = 2048;
+    let queue = new QueueClass();
+
+    queue.unshift(new NumberedTask(1));
+    assert.equal(queue.size, 1);
+
+    for (let i = 2; i <= bufferSize + 10; ++i) {
+      queue.unshift(new NumberedTask(i));
+    }
+    assert.equal(queue.size, bufferSize + 10);
+
+    for (let i = 0; i < 5; ++i) {
+      assert.equal((queue.shift() as NumberedTask).index, bufferSize + 10 - i);
+    }
+    assert.equal(queue.size, bufferSize + 5);
+
+    for (let i = 0; i < bufferSize + 5; ++i) {
+      assert.equal((queue.shift() as NumberedTask).index, bufferSize + 5 - i);
+    }
+    assert.equal(queue.size, 0);
+
+    queue = new QueueClass();
+    queue.push(new NumberedTask(1));
+    queue.push(new NumberedTask(2));
+    queue.push(new NumberedTask(3));
+    queue.push(new NumberedTask(4));
+    queue.push(new NumberedTask(5));
+    assert.equal((queue.shift() as NumberedTask).index, 1);
+    assert.equal((queue.shift() as NumberedTask).index, 2);
+    assert.equal((queue.shift() as NumberedTask).index, 3);
+    assert.equal(queue.size, 2);
+    queue.unshift(new NumberedTask(6));
+    queue.unshift(new NumberedTask(7));
+    queue.unshift(new NumberedTask(8));
+    queue.unshift(new NumberedTask(9));
+    assert.equal((queue.shift() as NumberedTask).index, 9);
+    assert.equal((queue.shift() as NumberedTask).index, 8);
+    assert.equal((queue.shift() as NumberedTask).index, 7);
+    assert.equal((queue.shift() as NumberedTask).index, 6);
+    assert.equal((queue.shift() as NumberedTask).index, 4);
+    assert.equal((queue.shift() as NumberedTask).index, 5);
+  });
+}

test/stricter-fifo.test.ts

@@ -0,0 +1,29 @@
+import * as assert from 'node:assert/strict';
+import { resolve } from 'node:path';
+import { test } from 'node:test';
+import Piscina from '..';
+
+test('stricterFIFO keeps queued tasks in submission order', async () => {
+  const pool = new Piscina({
+    filename: resolve(__dirname, 'fixtures/task-logger.js'),
+    minThreads: 1,
+    maxThreads: 1,
+    stricterFIFO: true
+  });
+
+  const tasks = new Array(20).fill(0).map((_, i) => ({ i }));
+  const pending = tasks.map((task) => pool.run(task, { name: 'run' }));
+  await Promise.all(pending);
+
+  const log = await pool.run({}, { name: 'getLog' });
+  assert.deepEqual(log, tasks);
+
+  await pool.destroy();
+});
+
+test('stricterFIFO must be a boolean', () => {
+  assert.throws(
+    () => new Piscina({ stricterFIFO: 1 } as any),
+    /options.stricterFIFO must be a boolean/
+  );
+});