function [results,wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_options) % SUBMITJOBS Submits dsp_runid jobs for processing % % Inputs: % run_ids - Single run ID or array of Run IDs for processing % dsp_options - Options for dsp_runid function % submit_mode - Execution mode: 'parallel' or 'linear' % options - Optional parameters % % Outputs: % results - Cell array of results from each job % wh - Updated DataStorage object % USAGE: % % === SETTINGS === % % dsp_options.append_to_db = 1; % dsp_options.max_occurences = 15; % dsp_options.database_path = 'C:\Users\Silas\Documents\MATLAB\Datensätze\sioe_labor\'; % dsp_options.database_name = 'silas_labor_newdsp_newstructure.db'; % dsp_options.storage_path = 'Z:\2024\sioe_labor\'; % dsp_options.parameters = struct(); % dsp_options.parameters.mu_dc = [0.005]; % % % === Get Run ID's === % db = DBHandler("pathToDB", [dsp_options.database_path, dsp_options.database_name], "type", "sqlite"); % fp = QueryFilter(); % % fp.where('Runs', 'run_id','EQUALS', 5108); % fp.where('Runs', 'is_mpi','EQUALS', 0); % fp.where('Runs', 'fiber_length','EQUALS', 1); % fp.where('Runs', 'wavelength','EQUALS', 1310); % fp.where('Runs', 'db_mode','EQUALS', 0); % fp.where('Runs', 'rop_attenuation','EQUALS', 0); % fp.where('Runs', 'pam_level','EQUALS', 4); % fp.where('Runs', 'bitrate','EQUALS', 360e9); % fp.where('Runs', 'power_pd_in','GREATER_THAN', 7); % [dataTable,~] = db.queryDB(fp, db.getTableFieldNames('Runs')); % % === Initialize DataStorage === % wh = DataStorage(dsp_options.parameters); % wh.addStorage("ffe_package"); % wh.addStorage("mlse_package"); % wh.addStorage("vnle_package"); % wh.addStorage("dbtgt_package"); % wh.addStorage("dbenc_package"); % % % === RUN IT === % [results,wh] = submitJobs(dataTable.run_id(:), dsp_options, "parallel", 'wh', wh, 'waitbar', true); % wh.getStoValue('ffe_package',0.005); % wh.getStoValue('mlse_package',0.005); arguments run_ids int32 = 0 dsp_options struct = struct(); submit_mode processingMode = processingMode.serial submit_options.waitbar (1,1) logical = true submit_options.wh = DataStorage(struct()); % Optional DataStorage object end % Ensure run_ids is a row vector run_ids = run_ids(:)'; % Get number of jobs per run_id nJobsPerRunId = submit_options.wh.getLastLinIndice; nRunIds = length(run_ids); totalJobs = nJobsPerRunId * nRunIds; % Initialize results array for all jobs results = cell(nJobsPerRunId, nRunIds); futures = cell(totalJobs, 1); % Initialize waitbar if requested if submit_options.waitbar h = waitbar(0, 'Processing Jobs...'); cleanupObj = onCleanup(@() delete(h)); end % Process jobs based on mode if submit_mode == processingMode.parallel setupParallelPool(); % Submit jobs to parallel pool jobCounter = 0; for r = 1:nRunIds for k = 1:nJobsPerRunId jobCounter = jobCounter + 1; optionalVars = buildOptionalVars(k, submit_options.wh); % Submit job with proper arguments futures{jobCounter} = parfeval(@dsp_runid, 1, run_ids(r), ... "database_name", dsp_options.database_name, ... "append_to_db", dsp_options.append_to_db, ... "database_path", dsp_options.database_path, ... "max_occurences", dsp_options.max_occurences, ... "storage_path", dsp_options.storage_path, ... "parameters", optionalVars); fprintf('[RunID %d, Job %d] Submitted to pool.\n', run_ids(r), k); end end if submit_options.waitbar setupParallelWaitbar(futures, totalJobs, h); end % Fetch results jobCounter = 0; for r = 1:nRunIds for k = 1:nJobsPerRunId jobCounter = jobCounter + 1; try results{k,r} = fetchOutputs(futures{jobCounter}); fprintf('[RunID %d, Job %d] Completed successfully.\n', run_ids(r), k); storeResult(results{k,r}, k, submit_options.wh); catch ME handleError(ME, k, run_ids(r)); results{k,r} = ME; end end end elseif submit_mode == processingMode.serial % Process jobs sequentially jobCounter = 0; for r = 1:nRunIds for k = 1:nJobsPerRunId jobCounter = jobCounter + 1; optionalVars = buildOptionalVars(k, submit_options.wh); try fprintf('[RunID %d, Job %d] Running in linear mode...\n', run_ids(r), k); results{k,r} = dsp_runid(run_ids(r),... "database_name", dsp_options.database_name,... "append_to_db", dsp_options.append_to_db,... "database_path", dsp_options.database_path,... "max_occurences", dsp_options.max_occurences,... "storage_path", dsp_options.storage_path,... "parameters", optionalVars); fprintf('[RunID %d, Job %d] Completed successfully.\n', run_ids(r), k); storeResult(results{k,r}, k, submit_options.wh); catch ME handleError(ME, k, run_ids(r)); results{k,r} = ME; end % Update waitbar if requested if submit_options.waitbar waitbar(jobCounter/totalJobs, h, sprintf('Completed %d/%d jobs', jobCounter, totalJobs)); end end end else error('') end wh = submit_options.wh; % Helper functions remain the same except for handleError % Modified handleError function to include run_id function handleError(ME, jobIndex, run_id) fprintf('[RunID %d, Job %d] ERROR [%s]: %s\n', run_id, jobIndex, ME.identifier, ME.message); for st = ME.stack' fprintf(' %s:%d (%s)\n', st.file, st.line, st.name); end fprintf('Full report:\n%s\n', getReport(ME,'extended')); end % Other helper functions remain unchanged function setupParallelPool() curpool = gcp('nocreate'); if isempty(curpool) parpool; else if ~isempty(curpool.FevalQueue.QueuedFutures) || ~isempty(curpool.FevalQueue.RunningFutures) oldq = length(curpool.FevalQueue.QueuedFutures) + length(curpool.FevalQueue.RunningFutures); curpool.FevalQueue.cancelAll; fprintf('Canceled %d unfetched jobs from old queue.\n', oldq); end end end function optionalVars = buildOptionalVars(jobIndex, wh) optionalVars = struct(); if ~isempty(wh.getDimension) [parametervalues, parameternames] = wh.getPhysIndicesByLinIndex(jobIndex); for pidx = 1:numel(parameternames) optionalVars.(parameternames{pidx}) = parametervalues{pidx}; end end end function setupParallelWaitbar(futures, totalJobs, h) updateWaitbar = @(~) waitbar(sum(cellfun(@(f) strcmp(f.State, 'finished'), futures))/totalJobs, h, ... sprintf('Completed %d/%d jobs', sum(cellfun(@(f) strcmp(f.State, 'finished'), futures)), totalJobs)); futureArray = [futures{:}]; afterEach(futureArray, updateWaitbar, 0); end function storeResult(result, jobIndex, wh) if ~isempty(wh) wh.addValueToStorageByLinIdx(result.ffe_package, 'ffe_package', jobIndex); wh.addValueToStorageByLinIdx(result.mlse_package, 'mlse_package', jobIndex); wh.addValueToStorageByLinIdx(result.vnle_package, 'vnle_package', jobIndex); wh.addValueToStorageByLinIdx(result.dbtgt_package, 'dbtgt_package', jobIndex); wh.addValueToStorageByLinIdx(result.dbenc_package, 'dbenc_package', jobIndex); end end end