Skip to content

Implement job queues for desktop dive #1534

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 10 commits into
base: main
Choose a base branch
from
Draft

Implement job queues for desktop dive #1534

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
a26ed0a
WIP create job queues
naglepuff Oct 20, 2025
df10132
Show upcoming jobs in jobs tab
naglepuff Oct 21, 2025
fb4e83b
Allow concurrent jobs with async job queue
naglepuff Oct 21, 2025
78ea7d0
Send backend most recent settings
naglepuff Oct 27, 2025
ff5041f
WIP add gpu and cpu job queues
naglepuff Oct 28, 2025
52c692b
Queue CPU jobs
naglepuff Oct 30, 2025
55c3626
Clean up PR
naglepuff Oct 30, 2025
b9ce509
Fix linting/syntax issues
naglepuff Oct 31, 2025
ae365a1
WIP show queued pipeline jobs
naglepuff Oct 31, 2025
0747d5c
Update test of video that needs conversion
naglepuff Oct 31, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions client/platform/desktop/backend/ipcService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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) => {
Expand Down
38 changes: 22 additions & 16 deletions client/platform/desktop/backend/native/common.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand All @@ -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',
Expand Down Expand Up @@ -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);

Expand All @@ -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);
});
Expand All @@ -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);
});
Expand All @@ -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');
Expand All @@ -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');
Expand All @@ -640,16 +643,18 @@ 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(payload.jsonMeta.transcodingJobKey).toBe('jobKey');
// expect(payload.jsonMeta.type).toBe('video');
expect(conversionArgs.mediaList.length).toBeGreaterThan(0);
});

it('check Dastset existence', async () => {
Expand Down Expand Up @@ -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);
Expand Down
33 changes: 13 additions & 20 deletions client/platform/desktop/backend/native/common.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,14 +30,15 @@ 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,
} from 'platform/desktop/constants';
import {
cleanString, filterByGlob, makeid, strNumericCompare,
Expand Down Expand Up @@ -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
Expand All @@ -1099,8 +1100,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;
Expand Down Expand Up @@ -1133,10 +1133,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) => {
Expand All @@ -1158,16 +1158,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;
Comment on lines -1161 to -1170
Copy link
Collaborator Author

@naglepuff naglepuff Oct 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@BryonLewis this feels like the most controversial part of the change. Previously the conversion job was kicked off deep within the finalizeMediaImport code path. With my proposed change, I'm no longer starting a job here.

Consuming code that needs to finalize an import is responsible for starting the conversion job (which it can do by inspecting the mediaList to see if there are any files that need to be converted). To me it looked like Recent.vue was the place where this is happening, so there are changes in that file that queue conversion jobs based on the result of the finalization process.

I'd love to hear your thoughts on this. If we don't want to make a big structural change like this I can revert the associated changes and keep the status quo for conversion jobs.

Copy link
Collaborator

@BryonLewis BryonLewis Nov 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've been looking at the PR and I think I'm fine with that. Both the BatchImport and others use Recent.vue for the finalizeImport and having that be responsible for job creation seems to be fine.

}

//We need to create datasets for each of the multiCam folders as well
Expand Down Expand Up @@ -1197,7 +1187,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) {
Expand Down
8 changes: 6 additions & 2 deletions client/platform/desktop/backend/native/mediaJobs.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,7 @@ async function convertMedia(
args: ConversionArgs,
updater: DesktopJobUpdater,
onComplete?: (jobKey: string, meta: JsonMeta) => void,
setTranscodingKey = false,
mediaIndex = 0,
key = '',
baseWorkDir = '',
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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);
}
}
});
Expand Down
22 changes: 22 additions & 0 deletions client/platform/desktop/constants.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,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
);
}
Comment on lines 186 to +203
Copy link
Collaborator

@BryonLewis BryonLewis Nov 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just wondering if for simplicity sake we make a base Type of JobArgs that the other items extend and that base type has a 'type' that is a string value and each subsequent type implements a type as 'pipeline' | 'training' | 'conversion' | 'exportTrained' ?

Might make it easier to add new types and no have these functions that are relying on multiple arguments to determine the type?


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;
Expand Down
Loading
Loading