WIP add gpu and cpu job queues

Author: naglepuff

client/platform/desktop/backend/ipcService.ts

@@ -7,6 +7,7 @@ import {
   DesktopJobUpdate, RunPipeline, RunTraining, Settings, ExportDatasetArgs,
   DesktopMediaImportResponse,
   ExportTrainedPipeline,
+  MaxConcurrency,
 } from 'platform/desktop/constants';
 
 import linux from './native/linux';
@@ -114,12 +115,15 @@ export default function register() {
     const updater = (update: DesktopJobUpdate) => {
       event.sender.send('job-update', update);
     };
-    const ret = await common.finalizeMediaImport(settings.get(), args, updater);
-    return ret;
+    return common.finalizeMediaImport(settings.get(), args, updater);
   });
 
   ipcMain.handle('validate-settings', async (_, s: Settings) => {
-    const ret = await currentPlatform.validateViamePath(s);
+    let ret = await currentPlatform.validateViamePath(s);
+    if (Math.floor(s.concurrency) > MaxConcurrency) {
+      const prefix = typeof ret === 'string' ? `${ret}, ` : '';
+      ret = `${prefix}Concurrency must not exceed ${MaxConcurrency}`;
+    }
     return ret;
   });
   ipcMain.handle('run-pipeline', async (event, args: RunPipeline) => {
@@ -140,4 +144,10 @@ export default function register() {
     };
     return currentPlatform.train(settings.get(), args, updater);
   });
+  ipcMain.handle('convert', async (event, args: ConversionArgs) => {
+    const updater = (update: DesktopJobUpdate) => {
+      event.sender.send('job-update', update);
+    };
+    return currentPlatform.convert(settings.get(), args, updater);
+  });
 }

client/platform/desktop/backend/native/common.ts

@@ -38,6 +38,7 @@ import {
   JsonMeta, Settings, JsonMetaCurrentVersion, DesktopMetadata, DesktopJobUpdater,
   RunTraining, ExportDatasetArgs, DesktopMediaImportResponse,
   ExportConfigurationArgs, JobsFolderName, ProjectsFolderName, PipelinesFolderName,
+  ConversionArgs,
 } from 'platform/desktop/constants';
 import {
   cleanString, filterByGlob, makeid, strNumericCompare,
@@ -1084,7 +1085,7 @@ async function _importTrackFile(
 /**
  * After media conversion we need to remove the transcodingKey to signify it is done
  */
-async function completeConversion(settings: Settings, datasetId: string, transcodingJobKey: string, meta: JsonMeta) {
+export async function completeConversion(settings: Settings, datasetId: string, transcodingJobKey: string, meta: JsonMeta) {
   await getValidatedProjectDir(settings, datasetId);
   if (meta.transcodingJobKey === transcodingJobKey) {
     // eslint-disable-next-line no-param-reassign
@@ -1100,7 +1101,7 @@ async function finalizeMediaImport(
   settings: Settings,
   args: DesktopMediaImportResponse,
   updater: DesktopJobUpdater,
-) {
+): Promise<ConversionArgs> {
   const { jsonMeta, globPattern } = args;
   let { mediaConvertList } = args;
   const { type: datasetType } = jsonMeta;
@@ -1135,8 +1136,8 @@ async function finalizeMediaImport(
 
   //Now we will kick off any conversions that are necessary
   let jobBase = null;
+  const srcDstList: [string, string][] = [];
   if (mediaConvertList.length) {
-    const srcDstList: [string, string][] = [];
     const extension = datasetType === 'video' ? '.mp4' : '.png';
     let destAbsPath = '';
     mediaConvertList.forEach((absPath) => {
@@ -1197,7 +1198,10 @@ async function finalizeMediaImport(
   if (args.metaFileAbsPath) {
     await dataFileImport(settings, jsonMeta.id, args.metaFileAbsPath);
   }
-  return finalJsonMeta;
+  return {
+    meta: finalJsonMeta,
+    mediaList: srcDstList,
+  };
 }
 
 async function openLink(url: string) {

client/platform/desktop/backend/native/linux.ts

@@ -14,8 +14,11 @@ import {
   RunTraining,
   DesktopJobUpdater,
   ExportTrainedPipeline,
+  ConversionArgs,
 } from 'platform/desktop/constants';
 import { observeChild } from 'platform/desktop/backend/native/processManager';
+import { convertMedia } from 'platform/desktop/backend/native/mediaJobs';
+import { completeConversion } from 'platform/desktop/backend/native/common';
 import * as viame from './viame';
 
 const DefaultSettings: Settings = {
@@ -111,6 +114,19 @@ async function train(
   });
 }
 
+async function convert(
+  settings: Settings,
+  conversionArgs: ConversionArgs,
+  updater: DesktopJobUpdater
+): Promise<DesktopJob> {
+  return convertMedia(
+    settings,
+    conversionArgs,
+    updater,
+    (jobKey, meta) => completeConversion(settings, conversionArgs.meta.id, jobKey, meta)
+  );
+}
+
 // Based on https://github.com/chrisallenlane/node-nvidia-smi
 async function nvidiaSmi(): Promise<NvidiaSmiReply> {
   return new Promise((resolve) => {
@@ -146,5 +162,6 @@ export default {
   runPipeline,
   exportTrainedPipeline,
   train,
+  convert,
   validateViamePath,
 };

client/platform/desktop/constants.ts

@@ -31,8 +31,6 @@ export interface Settings {
   };
 }
 
-export type JobArgs = RunPipeline | RunTraining | ExportTrainedPipeline;
-
 // Handles Importing and storing of multi camera data
 
 export interface Camera {
@@ -186,11 +184,33 @@ export interface RunTraining {
   };
 }
 
+export type JobArgs = RunPipeline | RunTraining | ExportTrainedPipeline | ConversionArgs;
+
+export function isRunPipeline(spec: JobArgs): spec is RunPipeline {
+  return 'datasetId' in spec && 'pipeline' in spec;
+}
+
+export function isExportTrainedPipeline(spec: JobArgs): spec is ExportTrainedPipeline {
+  return 'path' in spec && 'pipeline' in spec;
+}
+
+export function isRunTraining(spec: JobArgs): spec is RunTraining {
+  return (
+    'datasetIds' in spec
+    && 'pipelineName' in spec
+    && 'trainingConfig' in spec
+  );
+}
+
 export interface ConversionArgs {
   meta: JsonMeta;
   mediaList: [string, string][];
 }
 
+export function isConversion(spec: JobArgs): spec is ConversionArgs {
+  return 'meta' in spec && 'mediaList' in spec;
+}
+
 export interface DesktopJob {
   // key unique identifier for this job
   key: string;

client/platform/desktop/frontend/api.ts

@@ -17,12 +17,12 @@ import {
   inputAnnotationFileTypes, listFileTypes, inputAnnotationTypes,
 } from 'dive-common/constants';
 import {
-  DesktopJob, DesktopMetadata, JsonMeta, NvidiaSmiReply,
+  DesktopJob, DesktopMetadata, NvidiaSmiReply,
   RunPipeline, RunTraining, ExportTrainedPipeline, ExportDatasetArgs, ExportConfigurationArgs,
-  DesktopMediaImportResponse,
+  DesktopMediaImportResponse, ConversionArgs,
 } from 'platform/desktop/constants';
 
-import { pipelineJobQueue } from './store/jobs';
+import { gpuJobQueue } from './store/jobs';
 
 /**
  * Native functions that run entirely in the renderer
@@ -98,14 +98,6 @@ async function runPipeline(itemId: string, pipeline: Pipe): Promise<DesktopJob>
   return ipcRenderer.invoke('run-pipeline', args);
 }
 
-function queuePipeline(itemId: string, pipeline: Pipe): void {
-  const args: RunPipeline = {
-    pipeline,
-    datasetId: itemId,
-  };
-  pipelineJobQueue.enqueue(args);
-}
-
 async function exportTrainedPipeline(path: string, pipeline: Pipe): Promise<DesktopJob> {
   const args: ExportTrainedPipeline = {
     path,
@@ -138,6 +130,38 @@ async function runTraining(
   return ipcRenderer.invoke('run-training', args);
 }
 
+function queueTraining(
+  folderIds: string[],
+  pipelineName: string,
+  config: string,
+  annotatedFramesOnly: boolean,
+  labelText?: string,
+  fineTuneModel?: {
+    name: string;
+    type: string;
+    path?: string;
+    folderId?: string;
+  },
+): void {
+  const args: RunTraining = {
+    datasetIds: folderIds,
+    pipelineName,
+    trainingConfig: config,
+    annotatedFramesOnly,
+    labelText,
+    fineTuneModel,
+  };
+  gpuJobQueue.enqueue(args);
+}
+
+function queuePipeline(itemId: string, pipeline: Pipe): void {
+  const args: RunPipeline = {
+    pipeline,
+    datasetId: itemId,
+  };
+  gpuJobQueue.enqueue(args);
+}
+
 async function deleteTrainedPipeline(pipeline: Pipe): Promise<void> {
   return ipcRenderer.invoke('delete-trained-pipeline', pipeline);
 }
@@ -170,10 +194,15 @@ function importAnnotationFile(id: string, path: string, _htmlFile = undefined, a
   });
 }
 
-function finalizeImport(args: DesktopMediaImportResponse): Promise<JsonMeta> {
+function finalizeImport(args: DesktopMediaImportResponse): Promise<ConversionArgs> {
+  // Have this return JsonMeta as well as everything needed to start a job?
   return ipcRenderer.invoke('finalize-import', args);
 }
 
+async function convert(args: ConversionArgs): Promise<DesktopJob> {
+  return ipcRenderer.invoke('convert', args);
+}
+
 async function exportDataset(id: string, exclude: boolean, typeFilter: readonly string[], type?: 'csv' | 'json'): Promise<string> {
   const location = await dialog.showSaveDialog({
     title: 'Export Dataset',
@@ -262,6 +291,7 @@ export {
   exportDataset,
   exportConfiguration,
   finalizeImport,
+  convert,
   importMedia,
   bulkImportMedia,
   deleteDataset,
@@ -271,4 +301,5 @@ export {
   openLink,
   nvidiaSmi,
   queuePipeline,
+  queueTraining,
 };

client/platform/desktop/frontend/components/Jobs.vue

@@ -8,7 +8,7 @@ import { DesktopJob } from 'platform/desktop/constants';
 import BrowserLink from './BrowserLink.vue';
 import NavigationBar from './NavigationBar.vue';
 import { datasets } from '../store/dataset';
-import { recentHistory, truncateOutputAtLines, pipelineJobQueue } from '../store/jobs';
+import { recentHistory, truncateOutputAtLines, gpuJobQueue } from '../store/jobs';
 
 export default defineComponent({
   components: {
@@ -41,7 +41,7 @@ export default defineComponent({
       clockDriver,
       datasets,
       recentHistory,
-      pipelineJobQueue,
+      gpuJobQueue,
       moment,
       utc,
       visibleOutput,
@@ -232,16 +232,16 @@ export default defineComponent({
               </v-row>
             </v-card>
             <h1
-              v-if="pipelineJobQueue.length() > 0"
+              v-if="gpuJobQueue.length() > 0"
               class="text-h4 mb-4 font-weight-light"
             >
-              Upcoming Jobs ({{ pipelineJobQueue.length() }})
+              Upcoming Jobs ({{ gpuJobQueue.length() }})
             </h1>
             <v-card
-              v-for="jobSpec in pipelineJobQueue.jobSpecs"
-              :key="jobSpec.datasetId"
+              v-for="jobSpec in gpuJobQueue.jobSpecs"
+              :key="jobSpec.datasetId?"
             >
-              {{ jobSpec.datasetId }}
+              {{ jobSpec.datasetId? }}
             </v-card>
           </v-col>
         </v-row>

client/platform/desktop/frontend/components/Recent.vue

@@ -88,8 +88,9 @@ export default defineComponent({
       const imports = await request(async () => Promise.all(argsArray.map((args) => api.finalizeImport(args))));
       pendingImportPayload.value = null;
 
-      imports.forEach(async (jsonMeta) => {
-        const recentsMeta = await api.loadMetadata(jsonMeta.id);
+      imports.forEach(async (conversionArgs) => {
+        // TODO queue conversion job
+        const recentsMeta = await api.loadMetadata(conversionArgs.meta.id);
         setRecents(recentsMeta);
       });
 
@@ -100,16 +101,17 @@ export default defineComponent({
     async function finalizeImport(args: DesktopMediaImportResponse) {
       importing.value = true;
       await request(async () => {
-        const jsonMeta = await api.finalizeImport(args);
+        const conversionArgs = await api.finalizeImport(args);
         pendingImportPayload.value = null; // close dialog
-        if (!jsonMeta.transcodingJobKey) {
+        if (!conversionArgs.meta.transcodingJobKey) {
           router.push({
             name: 'viewer',
-            params: { id: jsonMeta.id },
+            params: { id: conversionArgs.meta.id },
           });
         } else {
           // Display new data and await transcoding to complete
-          const recentsMeta = await api.loadMetadata(jsonMeta.id);
+          // TODO: queue conversion job
+          const recentsMeta = await api.loadMetadata(conversionArgs.meta.id);
           setRecents(recentsMeta);
         }
       });

client/platform/desktop/frontend/store/asyncCpuJobQueue.ts

@@ -0,0 +1,30 @@
+import {
+  ConversionArgs,
+  ExportTrainedPipeline,
+  Settings,
+  DesktopJobUpdater,
+  JsonMeta,
+} from 'platform/desktop/constants';
+import { convertMedia } from 'platform/desktop/backend/native/mediaJobs';
+import AsyncJobQueue from './asyncJobQueue';
+
+export default class AsyncCpuJobQueue extends AsyncJobQueue<ConversionArgs | ExportTrainedPipeline> {
+  async beginConversionJob(
+    spec: ConversionArgs,
+    settings: Settings,
+    updater: DesktopJobUpdater,
+    onComplete?: (jobKey: string, meta: JsonMeta) => void,
+    mediaIndex = 0,
+    key = '',
+    baseWorkDir = '',
+  ): JsonMeta {
+    const result = await this.ipcRenderer.invoke('finalize-import')
+    this.processingJobs.push(result.desk);
+    return
+  }
+
+  async beginJob(spec: ExportTrainedPipeline) {
+    const newJob = await this.ipcRenderer.invoke('export-trained-pipeline', spec);
+    this.processingJobs.push(newJob);
+  }
+}

client/platform/desktop/frontend/store/asyncGpuJobQueue.ts

@@ -0,0 +1,22 @@
+import {
+  RunPipeline,
+  RunTraining,
+  isRunPipeline,
+  isRunTraining,
+  DesktopJob,
+} from 'platform/desktop/constants';
+import AsyncJobQueue from './asyncJobQueue';
+
+export default class AsyncGpuJobQueue extends AsyncJobQueue<RunPipeline | RunTraining> {
+  async beginJob(spec: RunPipeline | RunTraining) {
+    let newJob: DesktopJob;
+    if (isRunPipeline(spec)) {
+      newJob = await this.ipcRenderer.invoke('run-pipeline', spec);
+    } else if (isRunTraining(spec)) {
+      newJob = await this.ipcRenderer.invoke('run-training', spec);
+    } else {
+      throw new Error('Unsupported job arguments provided to beginJob.');
+    }
+    this.processingJobs.push(newJob);
+  }
+}