dynamic rater prefetch

Author: howardchung

svc/cycler.ts

@@ -34,7 +34,7 @@ async function cycle() {
   while (true) {
     // Start with a base number for gcdata/rater reqs and add additional retrievers based on parser capacity
     // Each retriever handles about 1 req/sec so divide by the avg number of seconds per parse
-    const count = Math.ceil((await getCapacity()) / 12) + 4;
+    const count = Math.ceil((await getCapacity()) / 12) + 5;
     shuffle(zones);
     const zone = zones[0];
     const config = {

svc/parser.ts

@@ -8,7 +8,7 @@
 import config from '../config.ts';
 import { runReliableQueue } from './store/queue.ts';
 import c from 'ansi-colors';
-import { buildReplayUrl, redisCount } from './util/utility.ts';
+import { buildReplayUrl, getParserCapacity, redisCount } from './util/utility.ts';
 import redis from './store/redis.ts';
 import { apiFetcher } from './fetcher/getApiData.ts';
 import { parsedFetcher } from './fetcher/getParsedData.ts';
@@ -17,8 +17,6 @@ import { getPGroup } from './util/pgroup.ts';
 import moment from 'moment';
 import { queueReconcile } from './util/insert.ts';
 
-const { PARSER_PARALLELISM } = config;
-
 async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
   const start = Date.now();
   let apiTime = 0;
@@ -164,15 +162,9 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
     }
   }
 }
-async function getCapacity() {
-  if (config.USE_SERVICE_REGISTRY) {
-    return redis.zcard('registry:parser');
-  }
-  return Number(PARSER_PARALLELISM);
-}
 runReliableQueue(
   'parse',
-  Number(PARSER_PARALLELISM),
+  Number(config.PARSER_PARALLELISM),
   parseProcessor,
-  getCapacity,
+  getParserCapacity,
 );

svc/rater.ts

@@ -1,6 +1,6 @@
 import { gcFetcher } from './fetcher/getGcData.ts';
 import db from './store/db.ts';
-import { average, isRadiant, redisCount } from './util/utility.ts';
+import { average, getRetrieverCapacity, isRadiant, redisCount } from './util/utility.ts';
 
 const DEFAULT_RATING = 4000;
 const kFactor = 32;
@@ -98,7 +98,8 @@ async function prefetchGcData() {
     const { rows } = await db.raw<{
       rows: { match_seq_num: number; match_id: number; pgroup: PGroup }[];
     }>(
-      'SELECT match_seq_num, match_id, pgroup from rating_queue WHERE gcdata IS NULL ORDER BY match_seq_num LIMIT 5',
+      'SELECT match_seq_num, match_id, pgroup from rating_queue WHERE gcdata IS NULL ORDER BY match_seq_num LIMIT ?',
+      [await getRetrieverCapacity()]
     );
     if (rows.length) {
       // Attempt to fetch for each

svc/util/utility.ts

@@ -725,7 +725,17 @@ function makeUrlArray(input: string) {
   return output;
 }
 
-export function getRetrieverCount() {
+export async function getParserCapacity() {
+  if (config.USE_SERVICE_REGISTRY) {
+    return redis.zcard('registry:parser');
+  }
+  return Number(config.PARSER_PARALLELISM);
+}
+
+export async function getRetrieverCapacity() {
+  if (config.USE_SERVICE_REGISTRY) {
+    return redis.zcard('registry:retriever');
+  }
   return RETRIEVER_ARRAY.length;
 }