feat(analyze): concurrent demo analyses

Configurable from the Analyze settings tab.

ref #1189

Author: akiver

src/common/analyses.ts

@@ -0,0 +1 @@
+export const MAX_CONCURRENT_ANALYSES = 8;

src/node/settings/default-settings.ts

@@ -40,6 +40,7 @@ export const defaultSettings: Settings = {
     redirectDemoToMatch: false,
   },
   analyze: {
+    maxConcurrentAnalyses: 4,
     analyzePositions: false,
   },
   playback: {

src/node/settings/settings.ts

@@ -39,6 +39,7 @@ type UISettings = {
 };
 
 type AnalyzeSettings = {
+  maxConcurrentAnalyses?: number;
   analyzePositions: boolean;
 };
 

src/server/analyses-listener.ts

@@ -11,10 +11,11 @@ import { analyzeDemo } from 'csdm/node/demo/analyze-demo';
 import { getSettings } from 'csdm/node/settings/get-settings';
 import { getErrorCodeFromError } from './get-error-code-from-error';
 import type { ErrorCode } from 'csdm/common/error-code';
+import { MAX_CONCURRENT_ANALYSES } from 'csdm/common/analyses';
 
 class AnalysesListener {
   private analyses: Analysis[] = [];
-  private currentAnalysis: Analysis | undefined;
+  private currentAnalyses: Analysis[] = [];
   private outputFolderPath: string; // Folder path where CSV files will be write on the host
 
   public constructor() {
@@ -60,47 +61,66 @@ class AnalysesListener {
   }
 
   public getAnalyses = () => {
-    if (this.currentAnalysis !== undefined) {
-      return [...this.analyses, this.currentAnalysis];
-    }
-    return this.analyses;
+    return [...this.analyses, ...this.currentAnalyses];
   };
 
   public hasAnalysesInProgress = () => {
-    return this.hasPendingAnalyses() || this.currentAnalysis !== undefined;
+    return this.hasPendingAnalyses() || this.currentAnalyses.length > 0;
   };
 
   public clear() {
     this.analyses = [];
-    this.currentAnalysis = undefined;
+    this.currentAnalyses = [];
   }
 
   private hasPendingAnalyses = () => {
     return this.analyses.length > 0;
   };
 
   private async loopUntilAnalysesDone() {
-    if (this.currentAnalysis) {
-      return;
+    const promises: Promise<void>[] = [];
+
+    const settings = await getSettings();
+    const maxConcurrentAnalyses = Math.min(
+      MAX_CONCURRENT_ANALYSES,
+      settings.analyze.maxConcurrentAnalyses ?? MAX_CONCURRENT_ANALYSES / 2,
+    );
+    while (this.analyses.length > 0 && this.currentAnalyses.length < maxConcurrentAnalyses) {
+      const analysis = this.analyses.shift();
+      if (analysis) {
+        this.currentAnalyses.push(analysis);
+        const analysisPromise = this.processAnalysis(analysis, settings.analyze.analyzePositions)
+          .catch((error) => {
+            logger.error('Unhandled error during analysis');
+            logger.error(error);
+          })
+          .finally(() => {
+            this.currentAnalyses = this.currentAnalyses.filter(
+              ({ demoChecksum }) => demoChecksum !== analysis.demoChecksum,
+            );
+          });
+
+        promises.push(analysisPromise);
+      }
     }
 
-    this.currentAnalysis = this.analyses.shift();
-    while (this.currentAnalysis) {
-      await this.processAnalysis(this.currentAnalysis);
-      this.currentAnalysis = this.analyses.shift();
+    if (promises.length > 0) {
+      await Promise.race(promises);
+      if (this.analyses.length > 0) {
+        await this.loopUntilAnalysesDone();
+      }
     }
   }
 
-  private readonly processAnalysis = async (analysis: Analysis) => {
+  private readonly processAnalysis = async (analysis: Analysis, analyzePositions: boolean) => {
     const { demoChecksum: checksum, demoPath, source } = analysis;
     try {
-      this.updateCurrentAnalysisStatus(AnalysisStatus.Analyzing);
-      const settings = await getSettings();
+      this.updateAnalysisStatus(analysis, AnalysisStatus.Analyzing);
       await analyzeDemo({
         demoPath,
-        outputFolderPath: this.outputFolderPath,
+        outputFolderPath: this.getAnalysisOutputFolderPath(analysis),
         source,
-        analyzePositions: settings.analyze.analyzePositions,
+        analyzePositions,
         onStdout: (data) => {
           logger.log(data);
           analysis.output += data;
@@ -118,72 +138,72 @@ class AnalysesListener {
           });
         },
       });
-      this.updateCurrentAnalysisStatus(AnalysisStatus.AnalyzeSuccess);
+      this.updateAnalysisStatus(analysis, AnalysisStatus.AnalyzeSuccess);
 
-      await this.insertMatch(checksum, demoPath);
+      await this.insertMatch(analysis, checksum, demoPath);
     } catch (error) {
       logger.error('Error while analyzing demo');
       if (error) {
         logger.error(error);
       }
       const isCorruptedDemo = error instanceof CorruptedDemoError;
-      if (!isCorruptedDemo && this.currentAnalysis && error instanceof Error) {
-        this.currentAnalysis.output += error.message;
+      if (!isCorruptedDemo && error instanceof Error) {
+        analysis.output += error.message;
       }
-      this.updateCurrentAnalysisStatus(AnalysisStatus.AnalyzeError);
+      this.updateAnalysisStatus(analysis, AnalysisStatus.AnalyzeError);
       // If the demo is corrupted, we still want to try to insert it in the database.
       if (isCorruptedDemo) {
-        await this.insertMatch(checksum, demoPath);
+        await this.insertMatch(analysis, checksum, demoPath);
       }
     }
   };
 
-  private async insertMatch(checksum: string, demoPath: string) {
+  private async insertMatch(analysis: Analysis, checksum: string, demoPath: string) {
     try {
-      this.updateCurrentAnalysisStatus(AnalysisStatus.Inserting);
+      this.updateAnalysisStatus(analysis, AnalysisStatus.Inserting);
       const match = await processMatchInsertion({
         checksum,
         demoPath,
-        outputFolderPath: this.outputFolderPath,
+        outputFolderPath: this.getAnalysisOutputFolderPath(analysis),
       });
-      this.updateCurrentAnalysisStatus(AnalysisStatus.InsertSuccess);
+      this.updateAnalysisStatus(analysis, AnalysisStatus.InsertSuccess);
       server.sendMessageToRendererProcess({
         name: RendererServerMessageName.MatchInserted,
         payload: match,
       });
     } catch (error) {
       logger.error('Error while inserting match');
       logger.error(error);
-      if (this.currentAnalysis) {
-        let errorOutput: string;
-        if (error instanceof Error) {
-          errorOutput = error.message;
-          if (error.stack) {
-            errorOutput += `\n${error.stack}`;
-          }
-          if (error.cause) {
-            errorOutput += `\n${error.cause}`;
-          }
-        } else {
-          errorOutput = String(error);
+      let errorOutput: string;
+      if (error instanceof Error) {
+        errorOutput = error.message;
+        if (error.stack) {
+          errorOutput += `\n${error.stack}`;
+        }
+        if (error.cause) {
+          errorOutput += `\n${error.cause}`;
         }
-        this.currentAnalysis.output += errorOutput;
+      } else {
+        errorOutput = String(error);
       }
+      analysis.output += errorOutput;
 
-      this.updateCurrentAnalysisStatus(AnalysisStatus.InsertError, getErrorCodeFromError(error));
+      this.updateAnalysisStatus(analysis, AnalysisStatus.InsertError, getErrorCodeFromError(error));
     }
   }
 
-  private updateCurrentAnalysisStatus = (status: AnalysisStatus, errorCode?: ErrorCode) => {
-    if (this.currentAnalysis !== undefined) {
-      this.currentAnalysis.status = status;
-      this.currentAnalysis.errorCode = errorCode;
-      server.sendMessageToRendererProcess({
-        name: RendererServerMessageName.AnalysisUpdated,
-        payload: this.currentAnalysis,
-      });
-    }
+  private updateAnalysisStatus = (analysis: Analysis, status: AnalysisStatus, errorCode?: ErrorCode) => {
+    analysis.status = status;
+    analysis.errorCode = errorCode;
+    server.sendMessageToRendererProcess({
+      name: RendererServerMessageName.AnalysisUpdated,
+      payload: analysis,
+    });
   };
+
+  private getAnalysisOutputFolderPath(analysis: Analysis) {
+    return path.join(this.outputFolderPath, analysis.demoChecksum);
+  }
 }
 
 export const analysesListener = new AnalysesListener();

src/ui/match/video/ffmpeg/ffmpeg-audio-bitrate-select.tsx

@@ -10,7 +10,7 @@ export function FfmpegAudioBitrateSelect() {
   const options: SelectOption<number>[] = [128, 256, 320].map((bitrate) => {
     return {
       value: bitrate,
-      label: bitrate.toString(),
+      label: bitrate,
     };
   });
 
@@ -25,7 +25,7 @@ export function FfmpegAudioBitrateSelect() {
         onChange={async (audioBitrate) => {
           await updateSettings({
             ffmpegSettings: {
-              audioBitrate,
+              audioBitrate: Number(audioBitrate),
             },
           });
         }}

src/ui/settings/analyze/analyze-settings.tsx

@@ -1,10 +1,12 @@
 import React from 'react';
 import { SettingsView } from 'csdm/ui/settings/settings-view';
 import { ToggleAnalyzePositions } from './toggle-analyze-positions';
+import { MaxConcurrentAnalysesSelect } from './max-concurrent-analyses-select';
 
 export function AnalyzeSettings() {
   return (
     <SettingsView>
+      <MaxConcurrentAnalysesSelect />
       <ToggleAnalyzePositions />
     </SettingsView>
   );

src/ui/settings/analyze/max-concurrent-analyses-select.tsx

@@ -0,0 +1,38 @@
+import React from 'react';
+import { Trans } from '@lingui/react/macro';
+import type { SelectOption } from 'csdm/ui/components/inputs/select';
+import { Select } from 'csdm/ui/components/inputs/select';
+import { SettingsEntry } from 'csdm/ui/settings/settings-entry';
+import { useUpdateSettings } from '../use-update-settings';
+import { useAnalyzeSettings } from './use-analyze-settings';
+import { MAX_CONCURRENT_ANALYSES } from 'csdm/common/analyses';
+
+export function MaxConcurrentAnalysesSelect() {
+  const { maxConcurrentAnalyses } = useAnalyzeSettings();
+  const updateSettings = useUpdateSettings();
+
+  const options: SelectOption<number>[] = Array.from({ length: MAX_CONCURRENT_ANALYSES }, (n, i) => ({
+    value: i + 1,
+    label: i + 1,
+  }));
+
+  return (
+    <SettingsEntry
+      interactiveComponent={
+        <Select
+          options={options}
+          value={maxConcurrentAnalyses}
+          onChange={async (maxConcurrentAnalyses) => {
+            await updateSettings({
+              analyze: {
+                maxConcurrentAnalyses: Number(maxConcurrentAnalyses),
+              },
+            });
+          }}
+        />
+      }
+      description={<Trans>Maximum number of concurrent analyses.</Trans>}
+      title={<Trans context="Settings title">Maximum number of analyses that can run at the same time.</Trans>}
+    />
+  );
+}

src/ui/translations/de/messages.po

@@ -6811,6 +6811,15 @@ msgstr ""
 msgid "Ignore bans happened before first match"
 msgstr ""
 
+#: src/ui/settings/analyze/max-concurrent-analyses-select.tsx
+msgid "Maximum number of concurrent analyses."
+msgstr ""
+
+#: src/ui/settings/analyze/max-concurrent-analyses-select.tsx
+msgctxt "Settings title"
+msgid "Maximum number of analyses that can run at the same time."
+msgstr ""
+
 #: src/ui/settings/analyze/toggle-analyze-positions.tsx
 msgid "Analyze player/grenade positions during analysis. Positions are required only for the 2D viewer. When enabled it increases time to insert matches into database and disk space usage."
 msgstr ""

src/ui/translations/en/messages.po

@@ -6810,6 +6810,15 @@ msgstr "Bans that happened before the player was first seen in a match will be i
 msgid "Ignore bans happened before first match"
 msgstr "Ignore bans happened before first match"
 
+#: src/ui/settings/analyze/max-concurrent-analyses-select.tsx
+msgid "Maximum number of concurrent analyses."
+msgstr "Maximum number of concurrent analyses."
+
+#: src/ui/settings/analyze/max-concurrent-analyses-select.tsx
+msgctxt "Settings title"
+msgid "Maximum number of analyses that can run at the same time."
+msgstr "Maximum number of analyses that can run at the same time."
+
 #: src/ui/settings/analyze/toggle-analyze-positions.tsx
 msgid "Analyze player/grenade positions during analysis. Positions are required only for the 2D viewer. When enabled it increases time to insert matches into database and disk space usage."
 msgstr "Analyze player/grenade positions during analysis. Positions are required only for the 2D viewer. When enabled it increases time to insert matches into database and disk space usage."

src/ui/translations/es/messages.po

@@ -6811,6 +6811,15 @@ msgstr ""
 msgid "Ignore bans happened before first match"
 msgstr ""
 
+#: src/ui/settings/analyze/max-concurrent-analyses-select.tsx
+msgid "Maximum number of concurrent analyses."
+msgstr ""
+
+#: src/ui/settings/analyze/max-concurrent-analyses-select.tsx
+msgctxt "Settings title"
+msgid "Maximum number of analyses that can run at the same time."
+msgstr ""
+
 #: src/ui/settings/analyze/toggle-analyze-positions.tsx
 msgid "Analyze player/grenade positions during analysis. Positions are required only for the 2D viewer. When enabled it increases time to insert matches into database and disk space usage."
 msgstr ""