Implement job queues for desktop dive (#1534)

* WIP create job queues

* Show upcoming jobs in jobs tab

* Allow concurrent jobs with async job queue

* Send backend most recent settings

* WIP add gpu and cpu job queues

* Queue CPU jobs

* Clean up PR

* Fix linting/syntax issues

* WIP show queued pipeline jobs

* Update test of video that needs conversion

* Show pending conversions on recents tab

* Move job queues to subdirectory

* Add component to show queued jobs

* Use queues exclusively for desktop jobs

* Remove upcoming jobs from jobs history component

* Refactor job types

* Watch queue directly for "awaiting conversion"

Author: naglepuff

client/platform/desktop/backend/ipcService.ts

@@ -7,7 +7,10 @@ import {
   DesktopJobUpdate, RunPipeline, RunTraining, Settings, ExportDatasetArgs,
   DesktopMediaImportResponse,
   ExportTrainedPipeline,
+  ConversionArgs,
+  DesktopJob,
 } from 'platform/desktop/constants';
+import { convertMedia } from 'platform/desktop/backend/native/mediaJobs';
 
 import linux from './native/linux';
 import win32 from './native/windows';
@@ -110,12 +113,21 @@ export default function register() {
     return ret;
   });
 
-  ipcMain.handle('finalize-import', async (event, args: DesktopMediaImportResponse) => {
+  ipcMain.handle('finalize-import', async (event, args: DesktopMediaImportResponse) => common.finalizeMediaImport(settings.get(), args));
+
+  ipcMain.handle('convert', async (event, args: ConversionArgs) => {
     const updater = (update: DesktopJobUpdate) => {
       event.sender.send('job-update', update);
     };
-    const ret = await common.finalizeMediaImport(settings.get(), args, updater);
-    return ret;
+    const currentSettings: Settings = settings.get();
+    const job: DesktopJob = await convertMedia(
+      currentSettings,
+      args,
+      updater,
+      (jobKey, meta) => common.completeConversion(currentSettings, args.meta.id, jobKey, meta),
+      true,
+    );
+    return job;
   });
 
   ipcMain.handle('validate-settings', async (_, s: Settings) => {

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

@@ -6,7 +6,7 @@ import { Console } from 'console';
 
 import {
   AnnotationsCurrentVersion, DesktopJob,
-  DesktopJobUpdate, JsonMeta, RunTraining, Settings,
+  DesktopJobUpdate, JobType, JsonMeta, RunTraining, Settings,
 } from 'platform/desktop/constants';
 import { makeEmptyAnnotationFile } from 'platform/desktop/backend/serializers/dive';
 
@@ -484,7 +484,8 @@ describe('native.common', () => {
       '/home/user/data/imageLists/success/image1.png',
     ]);
     expect(payload.jsonMeta.name).toBe('success');
-    const final = await common.finalizeMediaImport(settings, payload, updater);
+    const res = await common.finalizeMediaImport(settings, payload);
+    const final = res.meta;
     expect(final.originalImageFiles.length).toBe(4);
     expect(final.name).toBe('success');
     expect(final.imageListPath).toBe('/home/user/data/imageLists/success/image_list.txt');
@@ -497,7 +498,8 @@ describe('native.common', () => {
     );
     expect(payload.jsonMeta.originalBasePath).toBe('');
     payload.globPattern = '2018*';
-    const final = await common.finalizeMediaImport(settings, payload, updater);
+    const res = await common.finalizeMediaImport(settings, payload);
+    const final = res.meta;
     const expectedImageFiles = [
       '/home/user/data/imageLists/successGlob/2018-image2.png',
       '/home/user/data/imageLists/successGlob/nested/2018-image1.png',
@@ -537,7 +539,8 @@ describe('native.common', () => {
     const payload = await common.beginMediaImport(
       '/home/user/data/imageLists/success/image_list.txt',
     );
-    const final = await common.finalizeMediaImport(settings, payload, updater);
+    const res = await common.finalizeMediaImport(settings, payload);
+    const final = res.meta;
     const annotations = await common.loadDetections(settings, final.id);
     expect(Object.keys(annotations.tracks)).toHaveLength(0);
 
@@ -562,7 +565,7 @@ describe('native.common', () => {
     const payload = await common.beginMediaImport('/home/user/data/imageSuccessWithAnnotations');
     payload.trackFileAbsPath = ''; //It returns null be default but users change it.
     payload.jsonMeta.fps = 12; // simulate user specify FPS action
-    await common.finalizeMediaImport(settings, payload, updater);
+    await common.finalizeMediaImport(settings, payload);
     const meta = await common.loadMetadata(settings, payload.jsonMeta.id, urlMapper);
     expect(meta.fps).toBe(12);
   });
@@ -571,20 +574,20 @@ describe('native.common', () => {
     const payload = await common.beginMediaImport('/home/user/data/imageSuccessWithAnnotations');
     payload.trackFileAbsPath = '/home/user/data/imageSuccessWithAnnotations/file1.csv';
     payload.jsonMeta.fps = 12; // simulate user specify FPS action
-    await common.finalizeMediaImport(settings, payload, updater);
+    await common.finalizeMediaImport(settings, payload);
     const meta = await common.loadMetadata(settings, payload.jsonMeta.id, urlMapper);
     expect(meta.fps).toBe(32);
   });
 
   it('import with user selected FPS > originalFPS', async () => {
     const payload = await common.beginMediaImport('/home/user/data/videoSuccess/video1.mp4');
     payload.jsonMeta.fps = 50; // above 30
-    await common.finalizeMediaImport(settings, payload, updater);
+    await common.finalizeMediaImport(settings, payload);
     const meta1 = await common.loadMetadata(settings, payload.jsonMeta.id, urlMapper);
     expect(meta1.fps).toBe(30);
 
     payload.jsonMeta.fps = -1; // above 30
-    await common.finalizeMediaImport(settings, payload, updater);
+    await common.finalizeMediaImport(settings, payload);
     const meta2 = await common.loadMetadata(settings, payload.jsonMeta.id, urlMapper);
     expect(meta2.fps).toBe(1);
   });
@@ -600,15 +603,15 @@ describe('native.common', () => {
 
   it('importMedia empty json file success', async () => {
     const payload = await common.beginMediaImport('/home/user/data/annotationEmptySuccess/video1.mp4');
-    await common.finalizeMediaImport(settings, payload, updater);
+    await common.finalizeMediaImport(settings, payload);
     const annotations = await common.loadDetections(settings, payload.jsonMeta.id);
     expect(annotations).toEqual(makeEmptyAnnotationFile());
   });
 
   it('importMedia include meta.json file ', async () => {
     const payload = await common.beginMediaImport('/home/user/data/metaJsonIncluded/video1.mp4');
     expect(payload.metaFileAbsPath).toBe('/home/user/data/metaJsonIncluded/meta.json');
-    await common.finalizeMediaImport(settings, payload, updater);
+    await common.finalizeMediaImport(settings, payload);
     const tracks = await common.loadDetections(settings, payload.jsonMeta.id);
     const meta = await common.loadMetadata(settings, payload.jsonMeta.id, urlMapper);
     expect(meta?.customTypeStyling?.other.color).toBe('blue');
@@ -618,7 +621,7 @@ describe('native.common', () => {
   it('Export  meta.json file ', async () => {
     const payload = await common.beginMediaImport('/home/user/data/metaJsonIncluded/video1.mp4');
     expect(payload.metaFileAbsPath).toBe('/home/user/data/metaJsonIncluded/meta.json');
-    await common.finalizeMediaImport(settings, payload, updater);
+    await common.finalizeMediaImport(settings, payload);
     const tracks = await common.loadDetections(settings, payload.jsonMeta.id);
     const meta = await common.loadMetadata(settings, payload.jsonMeta.id, urlMapper);
     expect(meta?.customTypeStyling?.other.color).toBe('blue');
@@ -640,16 +643,15 @@ describe('native.common', () => {
   });
   it('import first CSV in list', async () => {
     const payload = await common.beginMediaImport('/home/user/data/multiCSV/video1.mp4');
-    await common.finalizeMediaImport(settings, payload, updater);
+    await common.finalizeMediaImport(settings, payload);
     const tracks = await common.loadDetections(settings, payload.jsonMeta.id);
     expect(tracks).toEqual(makeEmptyAnnotationFile());
   });
 
-  it('importMedia video, start conversion', async () => {
+  it('importMedia video, has conversion file list', async () => {
     const payload = await common.beginMediaImport('/home/user/data/videoSuccess/video1.avi');
-    await common.finalizeMediaImport(settings, payload, updater);
-    expect(payload.jsonMeta.transcodingJobKey).toBe('jobKey');
-    expect(payload.jsonMeta.type).toBe('video');
+    const conversionArgs = await common.finalizeMediaImport(settings, payload);
+    expect(conversionArgs.mediaList.length).toBeGreaterThan(0);
   });
 
   it('check Dastset existence', async () => {
@@ -693,6 +695,7 @@ describe('native.common', () => {
 
   it('processing good Trained Pipeline folder', async () => {
     const trainingArgs: RunTraining = {
+      type: JobType.RunTraining,
       datasetIds: ['randomID'],
       pipelineName: 'trainedPipelineName',
       trainingConfig: 'trainingConfig',
@@ -713,6 +716,7 @@ describe('native.common', () => {
 
   it('processing bad Trained Pipeline folders', async () => {
     const trainingArgs: RunTraining = {
+      type: JobType.RunTraining,
       datasetIds: ['randomID'],
       pipelineName: 'trainedBadPipelineName',
       trainingConfig: 'trainingConfig',
@@ -728,6 +732,7 @@ describe('native.common', () => {
 
   it('getPipelineList lists pipelines with Trained pipelines', async () => {
     const trainingArgs: RunTraining = {
+      type: JobType.RunTraining,
       datasetIds: ['randomID'],
       pipelineName: 'trainedPipelineName',
       trainingConfig: 'trainingConfig',
@@ -771,7 +776,8 @@ describe('native.common', () => {
         '9.png',
       ]);
       // eslint-disable-next-line no-await-in-loop
-      const final = await common.finalizeMediaImport(settings, payload, updater);
+      const res = await common.finalizeMediaImport(settings, payload);
+      const final = res.meta;
       expect(final.attributes).toEqual(testData[num][2]);
       // eslint-disable-next-line no-await-in-loop
       const tracks = await common.loadDetections(settings, final.id);

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

@@ -30,14 +30,16 @@ import * as dive from 'platform/desktop/backend/serializers/dive';
 import kpf from 'platform/desktop/backend/serializers/kpf';
 // TODO:  Check to Refactor this
 // eslint-disable-next-line import/no-cycle
-import { checkMedia, convertMedia } from 'platform/desktop/backend/native/mediaJobs';
+import { checkMedia } from 'platform/desktop/backend/native/mediaJobs';
 import {
   websafeImageTypes, websafeVideoTypes, otherImageTypes, otherVideoTypes, MultiType, JsonMetaRegEx,
 } from 'dive-common/constants';
 import {
-  JsonMeta, Settings, JsonMetaCurrentVersion, DesktopMetadata, DesktopJobUpdater,
+  JsonMeta, Settings, JsonMetaCurrentVersion, DesktopMetadata,
   RunTraining, ExportDatasetArgs, DesktopMediaImportResponse,
-  ExportConfigurationArgs, JobsFolderName, ProjectsFolderName, PipelinesFolderName,
+  ExportConfigurationArgs, JobsFolderName, ProjectsFolderName,
+  PipelinesFolderName, ConversionArgs,
+  JobType,
 } from 'platform/desktop/constants';
 import {
   cleanString, filterByGlob, makeid, strNumericCompare,
@@ -1084,7 +1086,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
@@ -1099,8 +1101,7 @@ async function completeConversion(settings: Settings, datasetId: string, transco
 async function finalizeMediaImport(
   settings: Settings,
   args: DesktopMediaImportResponse,
-  updater: DesktopJobUpdater,
-) {
+): Promise<ConversionArgs> {
   const { jsonMeta, globPattern } = args;
   let { mediaConvertList } = args;
   const { type: datasetType } = jsonMeta;
@@ -1133,10 +1134,10 @@ async function finalizeMediaImport(
     }
   }
 
-  //Now we will kick off any conversions that are necessary
-  let jobBase = null;
+  // Determine which files, if any, need to be queued for conversion. Consumers
+  // of this function are responsible for starting the conversion.
+  const srcDstList: [string, string][] = [];
   if (mediaConvertList.length) {
-    const srcDstList: [string, string][] = [];
     const extension = datasetType === 'video' ? '.mp4' : '.png';
     let destAbsPath = '';
     mediaConvertList.forEach((absPath) => {
@@ -1158,16 +1159,6 @@ async function finalizeMediaImport(
       }
       srcDstList.push([absPath, destAbsPath]);
     });
-    jobBase = await convertMedia(
-      settings,
-      {
-        meta: jsonMeta,
-        mediaList: srcDstList,
-      },
-      updater,
-      (jobKey, meta) => completeConversion(settings, jsonMeta.id, jobKey, meta),
-    );
-    jsonMeta.transcodingJobKey = jobBase.key;
   }
 
   //We need to create datasets for each of the multiCam folders as well
@@ -1197,7 +1188,12 @@ async function finalizeMediaImport(
   if (args.metaFileAbsPath) {
     await dataFileImport(settings, jsonMeta.id, args.metaFileAbsPath);
   }
-  return finalJsonMeta;
+  const conversionJobArgs: ConversionArgs = {
+    type: JobType.Conversion,
+    meta: finalJsonMeta,
+    mediaList: srcDstList,
+  };
+  return conversionJobArgs;
 }
 
 async function openLink(url: string) {

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

@@ -169,6 +169,7 @@ async function convertMedia(
   args: ConversionArgs,
   updater: DesktopJobUpdater,
   onComplete?: (jobKey: string, meta: JsonMeta) => void,
+  setTranscodingKey = false,
   mediaIndex = 0,
   key = '',
   baseWorkDir = '',
@@ -205,7 +206,10 @@ async function convertMedia(
     exitCode: job.exitCode,
     startTime: new Date(),
   };
-
+  if (setTranscodingKey) {
+    // eslint-disable-next-line no-param-reassign
+    args.meta.transcodingJobKey = jobBase.key;
+  }
   fs.writeFile(npath.join(jobWorkDir, DiveJobManifestName), JSON.stringify(jobBase, null, 2));
 
   job.stdout.on('data', jobFileEchoMiddleware(jobBase, updater, joblog));
@@ -247,7 +251,7 @@ async function convertMedia(
           ...jobBase,
           body: [`Conversion ${mediaIndex + 1} of ${args.mediaList.length} Complete`],
         });
-        convertMedia(settings, args, updater, onComplete, mediaIndex + 1, jobKey, jobWorkDir);
+        convertMedia(settings, args, updater, onComplete, setTranscodingKey, mediaIndex + 1, jobKey, jobWorkDir);
       }
     }
   });

client/platform/desktop/constants.ts

@@ -152,19 +152,33 @@ export interface NvidiaSmiReply {
   error: string;
 }
 
+export enum JobType {
+  Conversion,
+  ExportTrainedPipeline,
+  RunPipeline,
+  RunTraining,
+}
+
+export interface JobArgs {
+  type: JobType;
+}
+
 /** TODO promote to apispec */
-export interface RunPipeline {
+export interface RunPipeline extends JobArgs {
+  type: JobType.RunPipeline;
   datasetId: string;
   pipeline: Pipe;
 }
 
-export interface ExportTrainedPipeline {
+export interface ExportTrainedPipeline extends JobArgs {
+  type: JobType.ExportTrainedPipeline;
   path: string;
   pipeline: Pipe;
 }
 
 /** TODO promote to apispec */
-export interface RunTraining {
+export interface RunTraining extends JobArgs {
+  type: JobType.RunTraining;
   // datasets to run training on
   datasetIds: string[];
   // new pipeline name to be created
@@ -184,11 +198,14 @@ export interface RunTraining {
   };
 }
 
-export interface ConversionArgs {
+export interface ConversionArgs extends JobArgs {
+  type: JobType.Conversion;
   meta: JsonMeta;
   mediaList: [string, string][];
 }
 
+export type Job = ConversionArgs | RunPipeline | RunTraining | ExportTrainedPipeline;
+
 export interface DesktopJob {
   // key unique identifier for this job
   key: string;