chore: Add fastpriorityqueue to concurrencyController so it scales be…

…tter (#26657)
PostHog · Dec 4, 2024 · 1181b95 · 1181b95
1 parent 407947f
commit 1181b95
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 5 deletions.
diff --git a/frontend/src/lib/utils/concurrencyController.ts b/frontend/src/lib/utils/concurrencyController.ts
@@ -1,5 +1,5 @@
+import FastPriorityQueue from 'fastpriorityqueue'
 import { promiseResolveReject } from 'lib/utils'
-
 class ConcurrencyControllerItem<T> {
     _debugTag?: string
     _runFn: () => Promise<void>
@@ -52,7 +52,9 @@ export class ConcurrencyController {
     _concurrencyLimit: number
 
     _current: ConcurrencyControllerItem<any>[] = []
-    private _queue: ConcurrencyControllerItem<any>[] = []
+    private _queue: FastPriorityQueue<ConcurrencyControllerItem<any>> = new FastPriorityQueue(
+        (a, b) => a._priority < b._priority
+    )
 
     constructor(concurrencyLimit: number) {
         this._concurrencyLimit = concurrencyLimit
@@ -79,16 +81,15 @@ export class ConcurrencyController {
     }): Promise<T> => {
         const item = new ConcurrencyControllerItem(this, fn, abortController, priority, debugTag)
 
-        this._queue.push(item)
+        this._queue.add(item)
 
         this._tryRunNext()
 
         return item._promise
     }
 
     _runNext(): void {
-        this._queue.sort((a, b) => a._priority - b._priority)
-        const next = this._queue.shift()
+        const next = this._queue.poll()
         if (next) {
             next._runFn()
                 .catch(() => {

diff --git a/package.json b/package.json
@@ -132,6 +132,7 @@
         "expr-eval": "^2.0.2",
         "express": "^4.17.1",
         "fast-deep-equal": "^3.1.3",
+        "fastpriorityqueue": "^0.7.5",
         "fflate": "^0.7.4",
         "fs-extra": "^10.0.0",
         "fuse.js": "^6.6.2",

diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml