88 lines
2.9 KiB
Matlab
88 lines
2.9 KiB
Matlab
function [results, wh] = submitMpiSimulationJobs(wh, simulation_config, options)
|
|
%SUBMITMPISIMULATIONJOBS Execute MPI simulation warehouse points via runBatch.
|
|
|
|
arguments
|
|
wh DataStorage
|
|
simulation_config struct
|
|
options.mode = processingMode.serial
|
|
options.waitbar (1,1) logical = true
|
|
options.numWorkers (1,1) double {mustBeNonnegative, mustBeInteger} = 0
|
|
options.idleTimeout (1,1) double {mustBePositive} = 300
|
|
options.cancelExistingQueue (1,1) logical = true
|
|
end
|
|
|
|
nJobs = wh.getLastLinIndice();
|
|
jobs = repmat(struct("args", {{}}, "label", "", "meta", struct()), 1, nJobs);
|
|
|
|
for linIdx = 1:nJobs
|
|
userParameters = buildUserParameters(wh, linIdx);
|
|
jobs(linIdx).args = {userParameters, simulation_config};
|
|
jobs(linIdx).label = buildJobLabel(userParameters, linIdx);
|
|
jobs(linIdx).meta.lin_idx = linIdx;
|
|
jobs(linIdx).meta.userParameters = userParameters;
|
|
end
|
|
|
|
results = runBatch(@mpi_simulation_worker, jobs, ...
|
|
"mode", options.mode, ...
|
|
"waitbar", options.waitbar, ...
|
|
"waitbarMessage", "Processing MPI simulations...", ...
|
|
"numWorkers", options.numWorkers, ...
|
|
"idleTimeout", options.idleTimeout, ...
|
|
"cancelExistingQueue", options.cancelExistingQueue, ...
|
|
"resultHandler", @storeResult, ...
|
|
"errorHandler", @handleError);
|
|
|
|
function userParameters = buildUserParameters(storageWh, linIdx)
|
|
userParameters = struct();
|
|
if isempty(storageWh.getDimension())
|
|
return
|
|
end
|
|
|
|
[values, names] = storageWh.getPhysIndicesByLinIndex(linIdx);
|
|
for paramIdx = 1:numel(names)
|
|
userParameters.(char(names{paramIdx})) = values{paramIdx};
|
|
end
|
|
end
|
|
|
|
function label = buildJobLabel(userParameters, linIdx)
|
|
label = sprintf("MPI sim job %d", linIdx);
|
|
if isfield(userParameters, "sir")
|
|
label = sprintf("%s, SIR %g dB", label, userParameters.sir);
|
|
end
|
|
if isfield(userParameters, "block_update")
|
|
label = sprintf("%s, block %g", label, userParameters.block_update);
|
|
end
|
|
end
|
|
|
|
function storeResult(val, job, ~)
|
|
if isempty(val) || ~isstruct(val)
|
|
return
|
|
end
|
|
|
|
storageNames = fieldnames(val);
|
|
for storageIdx = 1:numel(storageNames)
|
|
storageName = storageNames{storageIdx};
|
|
if isempty(val.(storageName))
|
|
continue
|
|
end
|
|
|
|
ensureStorage(storageName);
|
|
wh.addValueToStorageByLinIdx(val.(storageName), storageName, job.meta.lin_idx);
|
|
end
|
|
end
|
|
|
|
function ensureStorage(storageName)
|
|
if ~isfield(wh.sto, storageName)
|
|
wh.addStorage(storageName);
|
|
end
|
|
end
|
|
|
|
function handleError(ME, job, ~)
|
|
fprintf("[%s] ERROR [%s]: %s\n", job.label, 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
|
|
end
|