dequeue multiple tasks when needed

Author: mmomtchev

.vscode/launch.json

@@ -1,14 +1,24 @@
 {
 	"version": "0.2.0",
 	"configurations": [
+		{
+			"type": "node",
+			"request": "launch",
+			"name": "Launch mocha",
+			"skipFiles": [
+				"<node_internals>/**"
+			],
+			"program": "${workspaceFolder}/node_modules/mocha/lib/cli/cli.js"
+		},
 		{
 			"type": "node",
 			"request": "launch",
 			"name": "Launch test",
+			"runtimeExecutable": "ts-node",
 			"skipFiles": [
 				"<node_internals>/**"
 			],
-			"program": "${workspaceFolder}/test.js"
+			"program": "${workspaceFolder}/test/stress.ts"
 		}
 	]
 }

CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### [2.1.1]
 - Raise an exception when trying to add a task with an already existing id (instead of failing when trying to run it)
+- Ability to dequeue multiple tasks when scheduling (fixes not reaching the maximum concurrency limit when the cool down is very low but more than zero)
 
 ## [2.1.0] 2023-01-25
 - Builtin ultra-light heap implementation from [`heap-js`](https://github.com/ignlg/heap-js) by @ignlg

src/index.ts

@@ -72,22 +72,23 @@ export class Queue<T = unknown> {
    */
   tryRun(): void {
     debug('tryRun');
-    this.nextTimer = null;
-    if (!this.queueWaiting.peek() || this.queueRunning.size >= this.maxConcurrent) return;
-
-    /* Wait if it is too soon */
-    if (Date.now() - this.lastRun < this.minCycle) {
-      debug('will throttle', Date.now() % 1000, (this.minCycle + this.lastRun) % 1000, Date.now() - this.lastRun);
-      if (this.nextTimer === null) {
-        this.nextTimer = new Promise((resolve) => setTimeout(() => {
-          this.tryRun();
-          resolve();
-        }, this.minCycle - Date.now() + this.lastRun));
+    while (this.queueWaiting.peek() && this.queueRunning.size < this.maxConcurrent) {
+      /* Wait if it is too soon */
+      if (Date.now() - this.lastRun < this.minCycle) {
+        debug(`will throttle, now=${Date.now() % 1000}, next=${(this.minCycle + this.lastRun) % 1000}, elapsed=${Date.now() - this.lastRun}`);
+        if (this.nextTimer === null) {
+          this.nextTimer = new Promise((resolve) => setTimeout(() => {
+            this.nextTimer = null;
+            this.tryRun();
+            resolve();
+          }, this.minCycle - Date.now() + this.lastRun));
+        }
+        return;
       }
-    } else {
+
       /* Choose the next task to run and unblock its promise */
       const next = this.queueWaiting.pop();
-      debug('wont throttle', this.lastRun % 1000, Date.now() % 1000, 'next is ', next?.hash);
+      debug(`wont throttle, last=${this.lastRun % 1000}, now=${Date.now() % 1000}, next is ${next?.hash}`);
       if (next !== undefined) {
         let finishSignal;
         const finishWait = new Promise<void>((resolve) => {
@@ -153,7 +154,7 @@ export class Queue<T = unknown> {
     await wait;
 
     this.lastRun = Date.now();
-    debug(hash, 'will run', this.lastRun % 1000, Date.now() % 1000);
+    debug(`will run ${hash} last=${this.lastRun % 1000}, now=${Date.now() % 1000}`);
   }
 
   /**
@@ -191,6 +192,8 @@ export class Queue<T = unknown> {
 
   /**
    * Returns a promise that resolves when the queue is empty
+   * (or there are no more than <maxWaiting> waiting tasks
+   * if the argument is provided)
    * 
    * @method flush
    * @return {Promise<void>}

test/stress.ts

@@ -0,0 +1,49 @@
+import { Queue } from '../src/index';
+
+const CONCURRENCY = 1000;
+const DELAY = 100;
+const CYCLE = 1;
+const TOTAL = 1e5;
+const ECHO = 1e3;
+
+const q = new Queue<number>(CONCURRENCY, CYCLE);
+let counter = 0;
+let last = -1;
+let concurrency = 0;
+
+async function main() {
+  while (counter < TOTAL) {
+    const me = counter++;
+    if (me % ECHO == 0) {
+      console.log('- scheduled', me);
+    }
+    if (q.stat().waiting > ECHO * 2) {
+      console.log('- throttling');
+      await q.flush();
+    }
+    q.wait(me, 0).then(() => {
+      concurrency++;
+      if (concurrency > CONCURRENCY) {
+        throw new Error('overflow ' + concurrency);
+      }
+      if (me % ECHO == 0) {
+        console.log('- run', me, concurrency);
+      }
+      setTimeout(() => {
+        if (me % ECHO == 0) {
+          console.log('- finished', me, concurrency);
+        }
+        if (last >= me) {
+          throw new Error('desync');
+        }
+        last = me;
+        q.end(me);
+        concurrency--;
+      }, DELAY);
+    });
+  }
+  await q.flush();
+  console.log('last', last);
+}
+
+main();