236 lines
8.8 KiB
Matlab
236 lines
8.8 KiB
Matlab
function [results, wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_options)
|
|
% SUBMITJOBS Submits dsp_runid jobs for processing (parallel or serial)
|
|
%
|
|
% [results, wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_options)
|
|
%
|
|
% run_ids : scalar or vector of run IDs
|
|
% dsp_options : struct of dsp_runid name/value options
|
|
% submit_mode : processingMode.parallel or .serial
|
|
% submit_options : struct with fields
|
|
% .waitbar (logical)
|
|
% .wh (DataStorage object)
|
|
%
|
|
% results : cell(nJobsPerRunId, nRunIds)
|
|
% wh : updated DataStorage
|
|
|
|
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())
|
|
end
|
|
|
|
% Normalize
|
|
run_ids = run_ids(:)';
|
|
nRunIds = numel(run_ids);
|
|
nJobsPerRunId = submit_options.wh.getLastLinIndice();
|
|
totalJobs = nRunIds * nJobsPerRunId;
|
|
|
|
% Preallocate
|
|
results = cell(nJobsPerRunId, nRunIds);
|
|
futures = parallel.FevalFuture.empty(totalJobs,0);
|
|
jobIndices = zeros(totalJobs,2);
|
|
|
|
% Optional waitbar
|
|
if submit_options.waitbar
|
|
h = waitbar(0, 'Processing Jobs...');
|
|
cleanupObj = onCleanup(@() delete(h));
|
|
end
|
|
|
|
switch submit_mode
|
|
case processingMode.parallel
|
|
%—– SET UP POOL & QUEUE —–
|
|
p = setupParallelPool(11, 300); % 10 workers, 300s idle timeout
|
|
|
|
% === submit all futures ===
|
|
jobCounter = 0;
|
|
for r = 1:nRunIds
|
|
for k = 1:nJobsPerRunId
|
|
jobCounter = jobCounter + 1;
|
|
jobIndices(jobCounter,:) = [r,k];
|
|
|
|
opt = buildOptionalVars(k, submit_options.wh);
|
|
futures(jobCounter) = parfeval( ...
|
|
p, @dsp_runid, 1, ...
|
|
run_ids(r), ...
|
|
"database_type", dsp_options.database_type, ...
|
|
"dataBase", dsp_options.dataBase, ...
|
|
"append_to_db", dsp_options.append_to_db, ...
|
|
"load_file_path", dsp_options.load_file_path, ...
|
|
"max_occurences", dsp_options.max_occurences, ...
|
|
"storage_path", dsp_options.storage_path, ...
|
|
"mode", dsp_options.mode, ...
|
|
"parameters", opt ...
|
|
);
|
|
|
|
fprintf('[RunID %d, Job %d] Submitted to pool.\n', run_ids(r), k);
|
|
end
|
|
end
|
|
|
|
%—– W A I T B A R U P D A T E —–
|
|
if submit_options.waitbar
|
|
futureArray = futures;
|
|
updateWB = @() waitbar( ...
|
|
sum(arrayfun(@(f) strcmp(f.State,'finished'), futureArray))/totalJobs, ...
|
|
h, sprintf('Completed %d/%d jobs', ...
|
|
sum(arrayfun(@(f) strcmp(f.State,'finished'), futureArray)), totalJobs) ...
|
|
);
|
|
afterEach(futureArray, updateWB, 0);
|
|
end
|
|
|
|
% —– before the loop —–
|
|
% Keep track of which futures we've already handled:
|
|
consumedIdx = false(totalJobs,1);
|
|
|
|
% —– fetch in completion order, handling successes and errors —–
|
|
for n = 1:totalJobs
|
|
try
|
|
% This returns the value AND the linear index in 'futures'
|
|
[idx, val] = fetchNext(futures);
|
|
|
|
duration = futures(idx).RunningDuration;
|
|
startDT = futures(idx).StartDateTime;
|
|
finishDT = futures(idx).FinishDateTime;
|
|
duration = finishDT - startDT;
|
|
|
|
% Mark it consumed
|
|
consumedIdx(idx) = true;
|
|
|
|
% Map back to (r,k) and store
|
|
r = jobIndices(idx,1);
|
|
k = jobIndices(idx,2);
|
|
|
|
fprintf('[%s] JobID %d/%d (%.1f%%) — %s — RunID %d — Subjob %d — fetched.\n', ...
|
|
datestr(now,'yyyy-mm-dd HH:MM:SS'), ...
|
|
r, totalJobs, 100*n/totalJobs,char(duration), ...
|
|
run_ids(r), k);
|
|
|
|
% Update waitbar
|
|
if submit_options.waitbar
|
|
waitbar(n/totalJobs, h, ...
|
|
sprintf('Fetched %d/%d (%.1f% Percent)', n, totalJobs, 100*n/totalJobs));
|
|
drawnow; % force the GUI to refresh
|
|
end
|
|
|
|
storeResult(val, k, submit_options.wh);
|
|
results{k,r} = val;
|
|
|
|
catch fetchErr
|
|
% fetchNext has already set Read=true on the errored future.
|
|
% Find the one Read==true that we have _not_ yet consumed.
|
|
readMask = arrayfun(@(f) f.Read, futures);
|
|
idxErr = find(readMask & ~consumedIdx', 1);
|
|
consumedIdx(idxErr) = true;
|
|
|
|
% Pull the _real_ exception out of the future object
|
|
errInfo = futures(idxErr).Error;
|
|
if iscell(errInfo)
|
|
origME = errInfo{1};
|
|
else
|
|
origME = errInfo;
|
|
end
|
|
|
|
% Map back to (r,k) and log
|
|
r = jobIndices(idxErr,1);
|
|
k = jobIndices(idxErr,2);
|
|
handleError(origME, k, run_ids(r));
|
|
results{k,r} = origME;
|
|
end
|
|
end
|
|
|
|
|
|
case processingMode.serial
|
|
%—– SERIAL EXECUTION —–
|
|
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);
|
|
val = dsp_runid( run_ids(r), ...
|
|
"database_type", dsp_options.database_type, ...
|
|
"dataBase", dsp_options.dataBase, ...
|
|
"append_to_db", dsp_options.append_to_db, ...
|
|
"load_file_path", dsp_options.load_file_path, ...
|
|
"max_occurences", dsp_options.max_occurences, ...
|
|
"storage_path", dsp_options.storage_path, ...
|
|
"mode", dsp_options.mode, ...
|
|
"parameters", optionalVars );
|
|
|
|
fprintf('[RunID %d, Job %d] Completed successfully.\n', run_ids(r), k);
|
|
storeResult(val, k, submit_options.wh);
|
|
results{k,r} = val;
|
|
|
|
catch ME
|
|
handleError(ME, k, run_ids(r));
|
|
results{k,r} = ME;
|
|
end
|
|
|
|
if submit_options.waitbar
|
|
waitbar(jobCounter/totalJobs, h, ...
|
|
sprintf('Completed %d/%d jobs', jobCounter, totalJobs));
|
|
end
|
|
end
|
|
end
|
|
|
|
otherwise
|
|
error('Unknown submit_mode "%s".', string(submit_mode))
|
|
end
|
|
|
|
wh = submit_options.wh;
|
|
|
|
%% Local helpers %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
|
|
|
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
|
|
|
|
function optionalVars = buildOptionalVars(jobIndex, wh)
|
|
optionalVars = struct();
|
|
if ~isempty(wh.getDimension())
|
|
[vals, names] = wh.getPhysIndicesByLinIndex(jobIndex);
|
|
for pi = 1:numel(names)
|
|
optionalVars.(names{pi}) = vals{pi};
|
|
end
|
|
end
|
|
end
|
|
|
|
function storeResult(val, jobIndex, wh)
|
|
if ~isempty(wh)
|
|
wh.addValueToStorageByLinIdx(val.ffe_package, 'ffe_package', jobIndex);
|
|
wh.addValueToStorageByLinIdx(val.mlse_package, 'mlse_package', jobIndex);
|
|
wh.addValueToStorageByLinIdx(val.vnle_package, 'vnle_package', jobIndex);
|
|
wh.addValueToStorageByLinIdx(val.dbtgt_package,'dbtgt_package',jobIndex);
|
|
wh.addValueToStorageByLinIdx(val.dbenc_package,'dbenc_package',jobIndex);
|
|
wh.addValueToStorageByLinIdx(val.mlmlse_package,'mlmlse_package',jobIndex);
|
|
end
|
|
end
|
|
function p = setupParallelPool(numWorkers, idleTimeout)
|
|
% Ensure a pool exists at the right size & timeout
|
|
p = gcp('nocreate');
|
|
if isempty(p) || p.NumWorkers~=numWorkers
|
|
if ~isempty(p)
|
|
delete(p);
|
|
end
|
|
p = parpool('local', numWorkers, 'IdleTimeout', idleTimeout);
|
|
end
|
|
|
|
% Cancel anything left in the pool's default queue
|
|
q = p.FevalQueue;
|
|
if ~isempty(q.QueuedFutures) || ~isempty(q.RunningFutures)
|
|
cancelAll(q);
|
|
fprintf('Canceled %d unfetched jobs from old queue.\n', ...
|
|
numel(q.QueuedFutures)+numel(q.RunningFutures));
|
|
end
|
|
end
|
|
|
|
|
|
end
|