fix(query-orchestrator): Queue - improve performance for high concurrency (cube-js#9705)

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 he effect of competition between nodes, it's important to reduce the number of tries to process by active jobs.

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,
     ];
   }
 

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, {
-            stageQueryKey: 1,
-            requestId: 'request-id',
-            spanId: 'span-id'
-          });
+          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;
         })();
 

packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts

@@ -244,11 +244,9 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
       expect(result).toEqual(['10', '21', '32', '43']);
     });
 
-    const nonCubeStoreTest = options.cacheAndQueueDriver !== 'cubestore' ? test : xtest;
+    const onlyLocalTest = options.cacheAndQueueDriver !== 'cubestore' ? test : xtest;
 
-    // this works with cube store, but there is an issue with timings
-    // TODO(ovr): fix me
-    nonCubeStoreTest('orphaned', async () => {
+    test('orphaned', async () => {
       // recover if previous test broken something
       for (let i = 1; i <= 4; i++) {
         await queue.executeInQueue('delay', `11${i}`, { delay: 50, result: `${i}` }, 0);
@@ -259,7 +257,7 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
 
       let result = queue.executeInQueue('delay', '111', { delay: 800, result: '1' }, 0);
       delayFn(null, 50).then(() => queue.executeInQueue('delay', '112', { delay: 800, result: '2' }, 0)).catch(e => e);
-      delayFn(null, 75).then(() => queue.executeInQueue('delay', '113', { delay: 500, result: '3' }, 0)).catch(e => e);
+      delayFn(null, 75).then(() => queue.executeInQueue('delay', '113', { delay: 800, result: '3' }, 0)).catch(e => e);
       // orphaned timeout should be applied
       delayFn(null, 100).then(() => queue.executeInQueue('delay', '114', { delay: 900, result: '4' }, 0)).catch(e => e);
 
@@ -269,7 +267,7 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
       result = await queue.executeInQueue('delay', '113', { delay: 900, result: '3' }, 0);
       expect(result).toBe('32');
 
-      await delayFn(null, 200);
+      await delayFn(null, 500);
       expect(cancelledQuery).toBe('114');
       await queue.executeInQueue('delay', '114', { delay: 50, result: '4' }, 0);
     });
@@ -368,7 +366,7 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
       expect(result).toBe('select * from bar');
     });
 
-    nonCubeStoreTest('queue driver lock obtain race condition', async () => {
+    onlyLocalTest('queue driver lock obtain race condition', async () => {
       const redisClient: any = await queue.queueDriver.createConnection();
       const redisClient2: any = await queue.queueDriver.createConnection();
       const priority = 10;
@@ -423,7 +421,7 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
       await queue.queueDriver.release(redisClient2);
     });
 
-    nonCubeStoreTest('activated but lock is not acquired', async () => {
+    onlyLocalTest('activated but lock is not acquired', async () => {
       const redisClient = await queue.queueDriver.createConnection();
       const redisClient2 = await queue.queueDriver.createConnection();
       const priority = 10;