fix(query-orchestrator): QueryQueue - improve performance for high concurrency setups

Author: ovr

packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts

@@ -136,12 +136,31 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
   }
 
   public async getActiveAndToProcess(): Promise<GetActiveAndToProcessResponse> {
+    const active: QueryKeysTuple[] = [];
+    const toProcess: QueryKeysTuple[] = [];
+
+    const rows = await this.driver.query<CubeStoreListResponse>('QUEUE LIST ?', [
+      this.options.redisQueuePrefix
+    ]);
+    if (rows.length) {
+      for (const row of rows) {
+        if (row.status === 'active') {
+          active.push([
+            row.id as QueryKeyHash,
+            row.queue_id ? parseInt(row.queue_id, 10) : null,
+          ]);
+        } else {
+          toProcess.push([
+            row.id as QueryKeyHash,
+            row.queue_id ? parseInt(row.queue_id, 10) : null,
+          ]);
+        }
+      }
+    }
+
     return [
-      // We don't return active queries, because it's useless
-      // There is only one place where it's used, and it's QueryQueue.reconcileQueueImpl
-      // Cube Store provides strict guarantees that queue item cannot be active & pending in the same time
-      [],
-      await this.getToProcessQueries()
+      active,
+      toProcess,
     ];
   }
 
@@ -233,6 +252,9 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
     const payload = JSON.parse(row.payload);
 
     if (row.extra) {
+      console.log(row.extra, typeof row.extra);
+      console.log(payload, typeof payload);
+      console.log(row.payload, typeof row.payload);
       return Object.assign(payload, JSON.parse(row.extra));
     }
 

packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts

@@ -543,7 +543,16 @@ export class QueryQueue {
         }
       }));
 
-      const [_active, toProcess] = await queueConnection.getActiveAndToProcess();
+      const [active, toProcess] = await queueConnection.getActiveAndToProcess();
+
+      /**
+       * Important notice: Concurrency configuration works per a specific queue, not per node.
+       *
+       * In production clusters where it contains N nodes, it shares the same concurrency. It leads to a point
+       * where every node tries to pick up jobs as much as concurrency is defined for the whole cluster. To minimize
+       * the effect of competition between nodes, it's important to reduce the number of tries to process by active jobs.
+       */
+      const toProcessLimit = active.length >= this.concurrency ? 1 : this.concurrency - active.length;
 
       const tasks = toProcess
         .filter(([queryKey, _queueId]) => {
@@ -559,7 +568,7 @@ export class QueryQueue {
             return false;
           }
         })
-        .slice(0, this.concurrency)
+        .slice(0, toProcessLimit)
         .map(([queryKey, queueId]) => this.sendProcessMessageFn(queryKey, queueId));
 
       await Promise.all(tasks);

packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts

@@ -138,6 +138,10 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions
           } else {
             counters.events[event] = 1;
           }
+
+          if (event.includes('error')) {
+            console.log(event, _params);
+          }
         },
         queueDriverFactory,
         ...options
@@ -159,7 +163,8 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions
       const progressIntervalId = setInterval(() => {
         console.log('running', {
           ...counters,
-          processingPromisses: processingPromisses.length
+          processingPromisses: processingPromisses.length,
+          benchSettings,
         });
       }, 1000);
 
@@ -177,18 +182,25 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions
 
         const queueId = crypto.randomBytes(12).toString('hex');
         const running = (async () => {
-          await queue.executeInQueue('query', queueId, {
-            // eslint-disable-next-line no-bitwise
-            payload: 'a'.repeat(benchSettings.queuePayloadSize)
-          }, 1, {
+          try {
+            await queue.executeInQueue('query', queueId, {
+              // eslint-disable-next-line no-bitwise
+              payload: {
+                large_str: 'a'.repeat(benchSettings.queuePayloadSize)
+              },
+              orphanedTimeout: 120
+            }, 1, {
             stageQueryKey: 1,
             requestId: 'request-id',
             spanId: 'span-id'
-          });
+            });
+          } catch (e) {
+            console.error(e);
+          }
 
           counters.queueResolved++;
 
-          // loosing memory for result
+          // losing memory for a result
           return null;
         })();