Unify batch job processing with a shared runBatch core
This commit is contained in:
Silas Oettinghaus
209
Functions/Job_Processing/runBatch.m
Normal file
209
Functions/Job_Processing/runBatch.m
Normal file
@@ -0,0 +1,209 @@
|
||||
function results = runBatch(workerFcn, jobs, options)
|
||||
%RUNBATCH Execute a list of jobs in serial or parallel.
|
||||
% results = runBatch(workerFcn, jobs) executes each jobs(i).args cell
|
||||
% with workerFcn and returns a cell array aligned with the input jobs.
|
||||
%
|
||||
% Supported job fields:
|
||||
% .args cell array of positional arguments for workerFcn
|
||||
% .label string/char used for progress messages
|
||||
% .meta arbitrary metadata for wrapper-specific bookkeeping
|
||||
|
||||
arguments
|
||||
workerFcn (1,1) function_handle
|
||||
jobs (1,:) struct = struct('args', {})
|
||||
options.mode = processingMode.serial
|
||||
options.waitbar (1,1) logical = true
|
||||
options.waitbarMessage (1,1) string = "Processing Jobs..."
|
||||
options.numWorkers (1,1) double {mustBeNonnegative, mustBeInteger} = 0
|
||||
options.idleTimeout (1,1) double {mustBePositive} = 300
|
||||
options.cancelExistingQueue (1,1) logical = true
|
||||
options.resultHandler = []
|
||||
options.errorHandler = []
|
||||
end
|
||||
|
||||
mode = normalizeProcessingMode(options.mode);
|
||||
jobs = normalizeJobs(jobs);
|
||||
nJobs = numel(jobs);
|
||||
results = cell(1, nJobs);
|
||||
h = [];
|
||||
|
||||
if nJobs == 0
|
||||
return
|
||||
end
|
||||
|
||||
if options.waitbar
|
||||
h = waitbar(0, char(options.waitbarMessage));
|
||||
cleanupWaitbar = onCleanup(@() closeWaitbar(h)); %#ok<NASGU>
|
||||
end
|
||||
|
||||
switch mode
|
||||
case processingMode.parallel
|
||||
pool = setupParallelPool(options.numWorkers, options.idleTimeout, options.cancelExistingQueue);
|
||||
futures = parallel.FevalFuture.empty(nJobs, 0);
|
||||
|
||||
for idx = 1:nJobs
|
||||
fprintf('[%s] Submitted.\n', jobs(idx).label);
|
||||
futures(idx) = parfeval(pool, workerFcn, 1, jobs(idx).args{:});
|
||||
end
|
||||
|
||||
consumedIdx = false(nJobs, 1);
|
||||
for completedCount = 1:nJobs
|
||||
try
|
||||
[idx, value] = fetchNext(futures);
|
||||
consumedIdx(idx) = true;
|
||||
results{idx} = value;
|
||||
fprintf('[%s] Completed (%d/%d).\n', jobs(idx).label, completedCount, nJobs);
|
||||
|
||||
if ~isempty(options.resultHandler)
|
||||
options.resultHandler(value, jobs(idx), idx);
|
||||
end
|
||||
catch fetchErr
|
||||
idxErr = findErroredFuture(futures, consumedIdx);
|
||||
if isempty(idxErr)
|
||||
rethrow(fetchErr);
|
||||
end
|
||||
|
||||
consumedIdx(idxErr) = true;
|
||||
errInfo = extractFutureError(futures(idxErr));
|
||||
results{idxErr} = errInfo;
|
||||
|
||||
if ~isempty(options.errorHandler)
|
||||
options.errorHandler(errInfo, jobs(idxErr), idxErr);
|
||||
else
|
||||
defaultErrorHandler(errInfo, jobs(idxErr), idxErr);
|
||||
end
|
||||
end
|
||||
|
||||
updateWaitbar(options.waitbar, h, completedCount, nJobs);
|
||||
end
|
||||
|
||||
case processingMode.serial
|
||||
for idx = 1:nJobs
|
||||
try
|
||||
fprintf('[%s] Running.\n', jobs(idx).label);
|
||||
value = feval(workerFcn, jobs(idx).args{:});
|
||||
results{idx} = value;
|
||||
fprintf('[%s] Completed (%d/%d).\n', jobs(idx).label, idx, nJobs);
|
||||
|
||||
if ~isempty(options.resultHandler)
|
||||
options.resultHandler(value, jobs(idx), idx);
|
||||
end
|
||||
catch ME
|
||||
results{idx} = ME;
|
||||
|
||||
if ~isempty(options.errorHandler)
|
||||
options.errorHandler(ME, jobs(idx), idx);
|
||||
else
|
||||
defaultErrorHandler(ME, jobs(idx), idx);
|
||||
end
|
||||
end
|
||||
|
||||
updateWaitbar(options.waitbar, h, idx, nJobs);
|
||||
end
|
||||
|
||||
otherwise
|
||||
error('runBatch:UnknownMode', 'Unknown processing mode "%s".', string(mode));
|
||||
end
|
||||
end
|
||||
|
||||
function mode = normalizeProcessingMode(mode)
|
||||
if isa(mode, 'processingMode')
|
||||
return
|
||||
end
|
||||
|
||||
modeString = string(mode);
|
||||
switch modeString
|
||||
case "serial"
|
||||
mode = processingMode.serial;
|
||||
case "parallel"
|
||||
mode = processingMode.parallel;
|
||||
otherwise
|
||||
error('runBatch:InvalidMode', 'Unsupported processing mode "%s".', modeString);
|
||||
end
|
||||
end
|
||||
|
||||
function jobs = normalizeJobs(jobs)
|
||||
if isempty(jobs)
|
||||
jobs = struct('args', {}, 'label', {}, 'meta', {});
|
||||
return
|
||||
end
|
||||
|
||||
for idx = 1:numel(jobs)
|
||||
if ~isfield(jobs, 'args') || isempty(jobs(idx).args)
|
||||
jobs(idx).args = {};
|
||||
end
|
||||
|
||||
if ~iscell(jobs(idx).args)
|
||||
error('runBatch:InvalidArgs', 'jobs(%d).args must be a cell array.', idx);
|
||||
end
|
||||
|
||||
if ~isfield(jobs, 'label') || isempty(jobs(idx).label)
|
||||
jobs(idx).label = sprintf('Job %d', idx);
|
||||
else
|
||||
jobs(idx).label = string(jobs(idx).label);
|
||||
end
|
||||
|
||||
if ~isfield(jobs, 'meta')
|
||||
jobs(idx).meta = struct();
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
function updateWaitbar(useWaitbar, h, completedCount, totalCount)
|
||||
if ~useWaitbar || ~isgraphics(h)
|
||||
return
|
||||
end
|
||||
|
||||
waitbar(completedCount / totalCount, h, ...
|
||||
sprintf('Completed %d/%d jobs', completedCount, totalCount));
|
||||
drawnow;
|
||||
end
|
||||
|
||||
function closeWaitbar(h)
|
||||
if isgraphics(h)
|
||||
delete(h);
|
||||
end
|
||||
end
|
||||
|
||||
function pool = setupParallelPool(numWorkers, idleTimeout, cancelExistingQueue)
|
||||
pool = gcp('nocreate');
|
||||
|
||||
if isempty(pool)
|
||||
if numWorkers > 0
|
||||
pool = parpool('local', numWorkers, 'IdleTimeout', idleTimeout);
|
||||
else
|
||||
pool = parpool('local', 'IdleTimeout', idleTimeout);
|
||||
end
|
||||
elseif numWorkers > 0 && pool.NumWorkers ~= numWorkers
|
||||
delete(pool);
|
||||
pool = parpool('local', numWorkers, 'IdleTimeout', idleTimeout);
|
||||
end
|
||||
|
||||
if cancelExistingQueue
|
||||
queue = pool.FevalQueue;
|
||||
if ~isempty(queue.QueuedFutures) || ~isempty(queue.RunningFutures)
|
||||
queuedCount = numel(queue.QueuedFutures) + numel(queue.RunningFutures);
|
||||
cancelAll(queue);
|
||||
fprintf('Canceled %d unfinished jobs from the current pool queue.\n', queuedCount);
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
function idxErr = findErroredFuture(futures, consumedIdx)
|
||||
readMask = arrayfun(@(future) future.Read, futures).';
|
||||
idxErr = find(readMask & ~consumedIdx, 1);
|
||||
end
|
||||
|
||||
function errInfo = extractFutureError(future)
|
||||
errInfo = future.Error;
|
||||
if iscell(errInfo)
|
||||
errInfo = errInfo{1};
|
||||
end
|
||||
end
|
||||
|
||||
function defaultErrorHandler(ME, job, jobIndex)
|
||||
fprintf('[%s | #%d] ERROR [%s]: %s\n', job.label, jobIndex, ME.identifier, ME.message);
|
||||
for st = ME.stack'
|
||||
fprintf(' %s:%d (%s)\n', st.file, st.line, st.name);
|
||||
end
|
||||
end
|
||||
@@ -27,164 +27,53 @@ 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;
|
||||
results = cell(nJobsPerRunId, nRunIds);
|
||||
jobs = repmat(struct('args', {{}}, 'label', "", 'meta', struct()), 1, totalJobs);
|
||||
|
||||
jobCounter = 0;
|
||||
for r = 1:nRunIds
|
||||
for k = 1:nJobsPerRunId
|
||||
jobCounter = jobCounter + 1;
|
||||
optionalVars = buildOptionalVars(k, submit_options.wh);
|
||||
|
||||
jobs(jobCounter).args = { ...
|
||||
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};
|
||||
jobs(jobCounter).label = sprintf('RunID %d, Job %d', run_ids(r), k);
|
||||
jobs(jobCounter).meta.runIndex = r;
|
||||
jobs(jobCounter).meta.jobIndex = k;
|
||||
jobs(jobCounter).meta.run_id = run_ids(r);
|
||||
end
|
||||
end
|
||||
|
||||
linearResults = runBatch(@dsp_runid, jobs, ...
|
||||
"mode", submit_mode, ...
|
||||
"waitbar", submit_options.waitbar, ...
|
||||
"waitbarMessage", "Processing Jobs...", ...
|
||||
"numWorkers", 11, ...
|
||||
"idleTimeout", 300, ...
|
||||
"cancelExistingQueue", true, ...
|
||||
"resultHandler", @storeResult, ...
|
||||
"errorHandler", @handleError);
|
||||
|
||||
for idx = 1:totalJobs
|
||||
r = jobs(idx).meta.runIndex;
|
||||
k = jobs(idx).meta.jobIndex;
|
||||
results{k, r} = linearResults{idx};
|
||||
end
|
||||
|
||||
%% Local helpers %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
|
||||
function handleError(ME, jobIndex, run_id)
|
||||
fprintf('[RunID %d, Job %d] ERROR [%s]: %s\n', run_id, jobIndex, ...
|
||||
function handleError(ME, job, ~)
|
||||
fprintf('[RunID %d, Job %d] ERROR [%s]: %s\n', job.meta.run_id, job.meta.jobIndex, ...
|
||||
ME.identifier, ME.message);
|
||||
for st = ME.stack'
|
||||
fprintf(' %s:%d (%s)\n', st.file, st.line, st.name);
|
||||
@@ -202,8 +91,9 @@ wh = submit_options.wh;
|
||||
end
|
||||
end
|
||||
|
||||
function storeResult(val, jobIndex, wh)
|
||||
function storeResult(val, job, ~)
|
||||
if ~isempty(wh)
|
||||
jobIndex = job.meta.jobIndex;
|
||||
wh.addValueToStorageByLinIdx(val.ffe_package, 'ffe_package', jobIndex);
|
||||
wh.addValueToStorageByLinIdx(val.mlse_package, 'mlse_package', jobIndex);
|
||||
wh.addValueToStorageByLinIdx(val.vnle_package, 'vnle_package', jobIndex);
|
||||
@@ -212,24 +102,5 @@ wh = submit_options.wh;
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user