small changes in submission script

This commit is contained in:
Silas Oettinghaus
2025-07-10 10:09:26 +02:00
parent f16475c6ed
commit eed71b5610
3 changed files with 415 additions and 379 deletions

View File

@@ -12,19 +12,33 @@ arguments
options.mode
end
% Initialize output structures
output.ffe_package = {};
output.mlse_package = {};
output.vnle_package = {};
output.dbtgt_package = {};
output.dbenc_package = {};
try
if options.mode == "load_run_id" || options.append_to_db
% Initialize output structures
output.ffe_package = {};
output.mlse_package = {};
output.vnle_package = {};
output.dbtgt_package = {};
output.dbenc_package = {};
if options.mode == "load_run_id" || options.append_to_db
% Initialize database connection
database = DBHandler("dataBase", [options.dataBase], "type", options.database_type );
end
if options.mode == "load_run_id"
% 2. Check if an equalizer configuration with the same hash exists
queryStr = sprintf('SELECT COUNT(DISTINCT eq_id) AS unique_eq_count, COUNT(*) AS entries_for_run FROM `Results` WHERE run_id = %d', run_id);
existing_results = database.fetch(queryStr);
if existing_results.unique_eq_count >= 6
if (existing_results.entries_for_run / existing_results.unique_eq_count) > 8
return
end
end
end
if options.mode == "load_run_id"
dataTable = queryRunid(run_id, database);
fsym = dataTable.symbolrate;
@@ -39,7 +53,7 @@ if options.mode == "load_run_id"
% Load and Sync signal data from DB
[Tx_bits, Symbols, Scpe_cell, ~] = loadAndSyncSignalDataFromDb(dataTable, options);
elseif options.mode == "load_files"
elseif options.mode == "load_files"
Tx_bits = load(options.load_file_path.tx_bits_path);
Symbols = load(options.load_file_path.tx_symbols_path);
@@ -56,53 +70,53 @@ elseif options.mode == "load_files"
Scpe_sig_resampled = Scpe_sig_raw.resample("fs_in", Scpe_sig_raw.fs, "fs_out", 2*fsym);
[~, Scpe_cell, ~, found_sync] = Scpe_sig_resampled.tsynch("reference", Symbols, "fs_ref", fsym, "debug_plots", 1);
else
else
% Run quick Simulation
tx_simulation;
end
end
% Handle Settings and argument replacement
% Handle Settings and argument replacement
len_tr = 4096*2;
len_tr = 4096*2;
ffe_order = [50, 5, 5];
dfe_order = [0, 0, 0];
pf_ncoeffs = 1;
mu_ffe = [0.0001, 0.0008, 0.001];
mu_dfe = 0.0004;
mu_dc = 0.005;
dc_buffer_len = 1;
ffe_order = [50, 5, 5];
dfe_order = [0, 0, 0];
pf_ncoeffs = 1;
mu_ffe = [0.0001, 0.0008, 0.001];
mu_dfe = 0.0004;
mu_dc = 0.005;
dc_buffer_len = 1;
mu_tr = 0;
mu_dd = 0.05;
adaption= 1;
use_dd_mode = 1;
mu_tr = 0;
mu_dd = 0.05;
adaption= 1;
use_dd_mode = 1;
use_ffe = 1;
use_dfe = 1;
use_vnle_mlse = 1;
use_dbtgt = 1;
use_dbenc = 1;
use_ffe = 1;
use_dfe = 1;
use_vnle_mlse = 1;
use_dbtgt = 1;
use_dbenc = 1;
addProcessingResultToDatabase = 0;
addProcessingResultToDatabase = 0;
% Overwrite default parameters if given in options.parameters
paramStruct = options.parameters;
if ~isempty(paramStruct)
% Overwrite default parameters if given in options.parameters
paramStruct = options.parameters;
if ~isempty(paramStruct)
paramNames = fieldnames(paramStruct);
for i = 1:numel(paramNames)
thisName = paramNames{i};
thisValue = paramStruct.(thisName);
eval([thisName ' = thisValue;']);
end
end
end
% Configure equalizers
% Configure equalizers
options.max_occurences = min(options.max_occurences,length(Scpe_cell));
for r = 1:options.max_occurences
options.max_occurences = min(options.max_occurences,length(Scpe_cell));
for r = 1:options.max_occurences
%FFE
% eq_dfe = FFE("epochs_tr",5,"epochs_dd",2,"len_tr",2^13,"mu_dd",mu_dd,"mu_tr",mu_tr,"order",25,"sps",2,"decide",0, "adaption",adaption_method(adaption),"dd_mode",use_dd_mode);
@@ -248,6 +262,11 @@ for r = 1:options.max_occurences
end
end
catch ME
save('workerError.mat','ME');
rethrow(ME);
end

View File

@@ -1,217 +1,234 @@
function [results,wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_options)
% SUBMITJOBS Submits dsp_runid jobs for processing
function [results, wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_options)
% SUBMITJOBS Submits dsp_runid jobs for processing (parallel or serial)
%
% 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
% [results, wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_options)
%
% Outputs:
% results - Cell array of results from each job
% wh - Updated DataStorage object
% USAGE:
% % === SETTINGS ===
% 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)
%
% 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);
% results : cell(nJobsPerRunId, nRunIds)
% wh : updated DataStorage
arguments
run_ids int32 = 0
dsp_options struct = struct();
dsp_options struct = struct()
submit_mode processingMode = processingMode.serial
submit_options.waitbar (1,1) logical = true
submit_options.wh = DataStorage(struct()); % Optional DataStorage object
submit_options.wh = DataStorage(struct())
end
% Ensure run_ids is a row vector
% Normalize
run_ids = run_ids(:)';
nRunIds = numel(run_ids);
nJobsPerRunId = submit_options.wh.getLastLinIndice();
totalJobs = nRunIds * nJobsPerRunId;
% 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
% Preallocate
results = cell(nJobsPerRunId, nRunIds);
futures = cell(totalJobs, 1);
futures = parallel.FevalFuture.empty(totalJobs,0);
jobIndices = zeros(totalJobs,2);
% Initialize waitbar if requested
% Optional waitbar
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();
switch submit_mode
case processingMode.parallel
% SET UP POOL & QUEUE
p = setupParallelPool(11, 300); % 10 workers, 300s idle timeout
% Submit jobs to parallel pool
% === submit all futures ===
jobCounter = 0;
for r = 1:nRunIds
for k = 1:nJobsPerRunId
jobCounter = jobCounter + 1;
optionalVars = buildOptionalVars(k, submit_options.wh);
jobIndices(jobCounter,:) = [r,k];
% Submit job with proper arguments
futures{jobCounter} = parfeval(@dsp_runid, 1, run_ids(r), ...
"database_type", dsp_options.database_type,...
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", optionalVars);
"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
setupParallelWaitbar(futures, totalJobs, h);
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
% Fetch results
jobCounter = 0;
for r = 1:nRunIds
for k = 1:nJobsPerRunId
jobCounter = jobCounter + 1;
% 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
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;
% 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
elseif submit_mode == processingMode.serial
% Process jobs sequentially
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);
results{k,r} = 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);
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(results{k,r}, k, submit_options.wh);
storeResult(val, k, submit_options.wh);
results{k,r} = val;
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));
waitbar(jobCounter/totalJobs, h, ...
sprintf('Completed %d/%d jobs', jobCounter, totalJobs));
end
end
end
else
error('')
otherwise
error('Unknown submit_mode "%s".', string(submit_mode))
end
wh = submit_options.wh;
% Helper functions remain the same except for handleError
% Modified handleError function to include run_id
%% Local helpers %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function handleError(ME, jobIndex, run_id)
fprintf('[RunID %d, Job %d] ERROR [%s]: %s\n', run_id, jobIndex, ME.identifier, ME.message);
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(10,"IdleTimeout",300);
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};
if ~isempty(wh.getDimension())
[vals, names] = wh.getPhysIndicesByLinIndex(jobIndex);
for pi = 1:numel(names)
optionalVars.(names{pi}) = vals{pi};
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)
function storeResult(val, 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);
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);
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

View File

@@ -1,6 +1,6 @@
% === SETTINGS ===
dsp_options.append_to_db = 1;
dsp_options.max_occurences = 1;
dsp_options.max_occurences = 15;
experiment = "highspeed_2024";
dsp_options.mode = "load_run_id"; % 'simulate' & 'load_files'
@@ -40,14 +40,14 @@ end
fp = QueryFilter();
% fp.where('Runs', 'run_id','EQUALS', 987);
% fp.where('Runs', 'pam_level','EQUALS', 4);
fp.where('Runs', 'bitrate','LESS_THAN', 310e9);
fp.where('Runs', 'fiber_length','EQUALS', 1);
fp.where('Runs', 'pam_level','EQUALS', 4);
% fp.where('Runs', 'bitrate','LESS_THAN', 310e9);
% fp.where('Runs', 'fiber_length','EQUALS', 1);
fp.where('Runs', 'is_mpi','EQUALS', 0);
% fp.where('Runs', 'interference_path_length','EQUALS', 1000);
% fp.where('Runs', 'loop_id','GREATER_THAN', 11);
% fp.where('Runs', 'sir','EQUALS',18);
fp.where('Runs', 'wavelength','EQUALS', 1310);
% fp.where('Runs', 'wavelength','EQUALS', 1310);
% fp.where('Runs', 'db_mode','EQUALS', 1);
fp.where('Runs', 'rop_attenuation','EQUALS', 0);
% fp.where('Runs', 'power_pd_in','GREATER_THAN', 7);
@@ -75,7 +75,7 @@ wh.addStorage("dbenc_package");
% === RUN IT ===
[results,wh] = submitJobs(dataTable.run_id(:), dsp_options, "serial", 'wh', wh, 'waitbar', true);
[results,wh] = submitJobs(dataTable.run_id(:), dsp_options, "parallel", 'wh', wh, 'waitbar', true);
% wh.getStoValue('ffe_package',0.005);