From eed71b561087fc7a9c31441542d3b18aa03e79f9 Mon Sep 17 00:00:00 2001 From: Silas Oettinghaus Date: Thu, 10 Jul 2025 10:09:26 +0200 Subject: [PATCH] small changes in submission script --- Functions/EQ_structures/dsp_runid.m | 471 +++++++++--------- Functions/Job_Processing/submitJobs.m | 311 ++++++------ .../Auswertung_JLT/run_dsp_from_db.m | 12 +- 3 files changed, 415 insertions(+), 379 deletions(-) diff --git a/Functions/EQ_structures/dsp_runid.m b/Functions/EQ_structures/dsp_runid.m index 1dc618d..50dfc46 100644 --- a/Functions/EQ_structures/dsp_runid.m +++ b/Functions/EQ_structures/dsp_runid.m @@ -5,249 +5,268 @@ arguments options.append_to_db = 0; options.max_occurences = 4; options.parameters = struct(); - options.database_type + options.database_type options.dataBase options.load_file_path = struct(); options.storage_path options.mode end -% 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" - - dataTable = queryRunid(run_id, database); - fsym = dataTable.symbolrate; - M = double(dataTable.pam_level); - duob_mode = db_mode(strrep(dataTable.db_mode,'"','')); - - % if database.checkIfRunExists('Results','run_id',run_id) - % disp(['Already got at least one reulst for run id: ',num2str(run_id),' ']) - % return - % end - - % Load and Sync signal data from DB - [Tx_bits, Symbols, Scpe_cell, ~] = loadAndSyncSignalDataFromDb(dataTable, options); - -elseif options.mode == "load_files" - - Tx_bits = load(options.load_file_path.tx_bits_path); - Symbols = load(options.load_file_path.tx_symbols_path); - Scpe_sig_raw = load(options.load_file_path.rx_raw_path); - - Tx_bits = Tx_bits.Bits; - Symbols = Symbols.Symbols; - Scpe_sig_raw = Scpe_sig_raw.Scpe_sig_raw; - - fsym = Symbols.fs; - M = Symbols.logbook.ModifierCopy{1}.M; - duob_mode = Symbols.logbook.ModifierCopy{1}.duobinary_mode; - - 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 - - % Run quick Simulation - tx_simulation; - -end - -% Handle Settings and argument replacement - -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; - -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; - -addProcessingResultToDatabase = 0; - -% 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 - -% Configure equalizers - -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); +try - % - eq_ = EQ("Ne",ffe_order,"Nb",dfe_order,"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",1); - pf_ = Postfilter("ncoeff",pf_ncoeffs,"useBurg",1); - % mlse_ = MLSE_viterbi("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); - mlse_ = MLSE("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); - mlse_db_ = MLSE("DIR",[1,1],"duobinary_output",0,"M",M,"trellis_states",PAMmapper(M,0).levels); - eq_post = FFE("epochs_tr",5,"epochs_dd",5,"len_tr",4096*2,"mu_dd",1e-4,"mu_tr",0,"order",2001,"sps",1,"decide",0,"adaption_technique","lms"); + % Initialize output structures + output.ffe_package = {}; + output.mlse_package = {}; + output.vnle_package = {}; + output.dbtgt_package = {}; + output.dbenc_package = {}; - % Duobinary signaling (db encoded) - mlse_db_enc = MLSE("DIR", [1,1], "duobinary_output", 0, "M", M, "trellis_states", PAMmapper(M,0).levels); - eq_db_enc = EQ("Ne", ffe_order, "Nb", dfe_order, "training_length", len_tr, ... - "training_loops", 5, "dd_loops", 5, "K", 2, "DCmu", mu_dc, ... - "DDmu", [mu_ffe mu_dfe], "DFEmu", 0.005, "FFEmu", 0, "plotfinal", 0, "ideal_dfe", 1); + if options.mode == "load_run_id" || options.append_to_db + % Initialize database connection + database = DBHandler("dataBase", [options.dataBase], "type", options.database_type ); - % Preprocess signal - Scpe_sig = preprocessSignal(Scpe_cell{r}, Symbols, fsym); + % 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 duob_mode ~= db_mode.db_encoded - - if use_ffe - - ffe_order = [50, 0, 0]; - eq_dfe = EQ("Ne",ffe_order,"Nb",[0,0,0],"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",0); - - dfe_results = ffe(eq_dfe,M,Scpe_sig,Symbols,Tx_bits,... - "precode_mode",duob_mode,... - 'showAnalysis',0,... - "postFFE",[],... - "eth_style_symbol_mapping",0); - - output.ffe_package{r} = dfe_results; - - dfe_results.metrics.print; - dfe_results.config.equalizer_structure = "ffe"; - - if options.append_to_db - database.addProcessingResult(run_id, dfe_results.metrics, dfe_results.config); + if existing_results.unique_eq_count >= 6 + if (existing_results.entries_for_run / existing_results.unique_eq_count) > 8 + return end - - end - - if use_dfe - - ffe_order = [50, 5, 5]; - eq_dfe = EQ("Ne",ffe_order,"Nb",[2,0,0],"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",0); - - dfe_results = ffe(eq_dfe,M,Scpe_sig,Symbols,Tx_bits,... - "precode_mode",duob_mode,... - 'showAnalysis',0,... - "postFFE",[],... - "eth_style_symbol_mapping",0); - - output.ffe_package{r} = dfe_results; - dfe_results.config.equalizer_structure = "dfe"; - - dfe_results.metrics.print; - - if options.append_to_db - database.addProcessingResult(run_id, dfe_results.metrics, dfe_results.config); - end - - end - - if use_vnle_mlse - - pf_ncoeffs = 1; - eq_ = EQ("Ne",ffe_order,"Nb",dfe_order,"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",1); - pf_ = Postfilter("ncoeff",pf_ncoeffs,"useBurg",1); - % mlse_ = MLSE_viterbi("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); - mlse_ = MLSE("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); - - [ffe_results, mlse_results] = vnle_postfilter_mlse(eq_, pf_, mlse_, M, Scpe_sig, Symbols, Tx_bits, ... - "precode_mode", duob_mode,... - 'showAnalysis', 1, ... - "postFFE", [],... - "eth_style_symbol_mapping", 0); - - ffe_results.metrics.print; - ffe_results.config.equalizer_structure = "vnle"; - mlse_results.metrics.print; - - output.mlse_package{r} = mlse_results; - output.vnle_package{r} = ffe_results; - - if options.append_to_db - database.addProcessingResult(run_id, mlse_results.metrics, mlse_results.config); - database.addProcessingResult(run_id, ffe_results.metrics, ffe_results.config); - end - - pf_ncoeffs = 2; - eq_ = EQ("Ne",ffe_order,"Nb",dfe_order,"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",1); - pf_ = Postfilter("ncoeff",pf_ncoeffs,"useBurg",1); - % mlse_ = MLSE_viterbi("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); - mlse_ = MLSE("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); - - [ffe_results, mlse_results] = vnle_postfilter_mlse(eq_, pf_, mlse_, M, Scpe_sig, Symbols, Tx_bits, ... - "precode_mode", duob_mode,... - 'showAnalysis', 1, ... - "postFFE", [],... - "eth_style_symbol_mapping", 0); - - ffe_results.metrics.print; - mlse_results.metrics.print; - - output.mlse_package{r} = mlse_results; - output.vnle_package{r} = ffe_results; - - if options.append_to_db - database.addProcessingResult(run_id, mlse_results.metrics, mlse_results.config); - database.addProcessingResult(run_id, ffe_results.metrics, ffe_results.config); - end - end - - if use_dbtgt - - dbt_results = duobinary_target(eq_, mlse_db_, M, Scpe_sig, Symbols, Tx_bits, ... - "precode_mode", duob_mode, ... - 'showAnalysis', 0,... - "postFFE", []); - - output.dbtgt_package{r} = dbt_results; - - if options.append_to_db - database.addProcessingResult(run_id, dbt_results.metrics, dbt_results.config); - end - - end - end - - if duob_mode == db_mode.db_encoded - - db_results = duobinary_signaling(eq_db_enc, mlse_db_enc, M, Scpe_sig, Symbols, Tx_bits, "precode_mode",duob_mode, "showAnalysis",0,"postFFE",[]); - output.dbenc_package{r} = db_results; - if options.append_to_db - database.addProcessingResult(run_id, db_results.metrics, db_results.config); end end + if options.mode == "load_run_id" + dataTable = queryRunid(run_id, database); + fsym = dataTable.symbolrate; + M = double(dataTable.pam_level); + duob_mode = db_mode(strrep(dataTable.db_mode,'"','')); + + % if database.checkIfRunExists('Results','run_id',run_id) + % disp(['Already got at least one reulst for run id: ',num2str(run_id),' ']) + % return + % end + + % Load and Sync signal data from DB + [Tx_bits, Symbols, Scpe_cell, ~] = loadAndSyncSignalDataFromDb(dataTable, options); + + elseif options.mode == "load_files" + + Tx_bits = load(options.load_file_path.tx_bits_path); + Symbols = load(options.load_file_path.tx_symbols_path); + Scpe_sig_raw = load(options.load_file_path.rx_raw_path); + + Tx_bits = Tx_bits.Bits; + Symbols = Symbols.Symbols; + Scpe_sig_raw = Scpe_sig_raw.Scpe_sig_raw; + + fsym = Symbols.fs; + M = Symbols.logbook.ModifierCopy{1}.M; + duob_mode = Symbols.logbook.ModifierCopy{1}.duobinary_mode; + + 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 + + % Run quick Simulation + tx_simulation; + + end + + % Handle Settings and argument replacement + + 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; + + 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; + + addProcessingResultToDatabase = 0; + + % 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 + + % Configure equalizers + + 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); + + + % + eq_ = EQ("Ne",ffe_order,"Nb",dfe_order,"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",1); + pf_ = Postfilter("ncoeff",pf_ncoeffs,"useBurg",1); + % mlse_ = MLSE_viterbi("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); + mlse_ = MLSE("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); + mlse_db_ = MLSE("DIR",[1,1],"duobinary_output",0,"M",M,"trellis_states",PAMmapper(M,0).levels); + eq_post = FFE("epochs_tr",5,"epochs_dd",5,"len_tr",4096*2,"mu_dd",1e-4,"mu_tr",0,"order",2001,"sps",1,"decide",0,"adaption_technique","lms"); + + % Duobinary signaling (db encoded) + mlse_db_enc = MLSE("DIR", [1,1], "duobinary_output", 0, "M", M, "trellis_states", PAMmapper(M,0).levels); + eq_db_enc = EQ("Ne", ffe_order, "Nb", dfe_order, "training_length", len_tr, ... + "training_loops", 5, "dd_loops", 5, "K", 2, "DCmu", mu_dc, ... + "DDmu", [mu_ffe mu_dfe], "DFEmu", 0.005, "FFEmu", 0, "plotfinal", 0, "ideal_dfe", 1); + + % Preprocess signal + Scpe_sig = preprocessSignal(Scpe_cell{r}, Symbols, fsym); + + if duob_mode ~= db_mode.db_encoded + + if use_ffe + + ffe_order = [50, 0, 0]; + eq_dfe = EQ("Ne",ffe_order,"Nb",[0,0,0],"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",0); + + dfe_results = ffe(eq_dfe,M,Scpe_sig,Symbols,Tx_bits,... + "precode_mode",duob_mode,... + 'showAnalysis',0,... + "postFFE",[],... + "eth_style_symbol_mapping",0); + + output.ffe_package{r} = dfe_results; + + dfe_results.metrics.print; + dfe_results.config.equalizer_structure = "ffe"; + + if options.append_to_db + database.addProcessingResult(run_id, dfe_results.metrics, dfe_results.config); + end + + end + + if use_dfe + + ffe_order = [50, 5, 5]; + eq_dfe = EQ("Ne",ffe_order,"Nb",[2,0,0],"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",0); + + dfe_results = ffe(eq_dfe,M,Scpe_sig,Symbols,Tx_bits,... + "precode_mode",duob_mode,... + 'showAnalysis',0,... + "postFFE",[],... + "eth_style_symbol_mapping",0); + + output.ffe_package{r} = dfe_results; + dfe_results.config.equalizer_structure = "dfe"; + + dfe_results.metrics.print; + + if options.append_to_db + database.addProcessingResult(run_id, dfe_results.metrics, dfe_results.config); + end + + end + + if use_vnle_mlse + + pf_ncoeffs = 1; + eq_ = EQ("Ne",ffe_order,"Nb",dfe_order,"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",1); + pf_ = Postfilter("ncoeff",pf_ncoeffs,"useBurg",1); + % mlse_ = MLSE_viterbi("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); + mlse_ = MLSE("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); + + [ffe_results, mlse_results] = vnle_postfilter_mlse(eq_, pf_, mlse_, M, Scpe_sig, Symbols, Tx_bits, ... + "precode_mode", duob_mode,... + 'showAnalysis', 1, ... + "postFFE", [],... + "eth_style_symbol_mapping", 0); + + ffe_results.metrics.print; + ffe_results.config.equalizer_structure = "vnle"; + mlse_results.metrics.print; + + output.mlse_package{r} = mlse_results; + output.vnle_package{r} = ffe_results; + + if options.append_to_db + database.addProcessingResult(run_id, mlse_results.metrics, mlse_results.config); + database.addProcessingResult(run_id, ffe_results.metrics, ffe_results.config); + end + + pf_ncoeffs = 2; + eq_ = EQ("Ne",ffe_order,"Nb",dfe_order,"training_length",len_tr,"training_loops",5,"dd_loops",5,"K",2,"DCmu",mu_dc,"DDmu",[mu_ffe mu_dfe],"DFEmu",0.005,"FFEmu",0,"plotfinal",0,"ideal_dfe",1); + pf_ = Postfilter("ncoeff",pf_ncoeffs,"useBurg",1); + % mlse_ = MLSE_viterbi("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); + mlse_ = MLSE("duobinary_output",0,'M',M,'trellis_states',PAMmapper(M,0).levels); + + [ffe_results, mlse_results] = vnle_postfilter_mlse(eq_, pf_, mlse_, M, Scpe_sig, Symbols, Tx_bits, ... + "precode_mode", duob_mode,... + 'showAnalysis', 1, ... + "postFFE", [],... + "eth_style_symbol_mapping", 0); + + ffe_results.metrics.print; + mlse_results.metrics.print; + + output.mlse_package{r} = mlse_results; + output.vnle_package{r} = ffe_results; + + if options.append_to_db + database.addProcessingResult(run_id, mlse_results.metrics, mlse_results.config); + database.addProcessingResult(run_id, ffe_results.metrics, ffe_results.config); + end + end + + if use_dbtgt + + dbt_results = duobinary_target(eq_, mlse_db_, M, Scpe_sig, Symbols, Tx_bits, ... + "precode_mode", duob_mode, ... + 'showAnalysis', 0,... + "postFFE", []); + + output.dbtgt_package{r} = dbt_results; + + if options.append_to_db + database.addProcessingResult(run_id, dbt_results.metrics, dbt_results.config); + end + + end + end + + if duob_mode == db_mode.db_encoded + + db_results = duobinary_signaling(eq_db_enc, mlse_db_enc, M, Scpe_sig, Symbols, Tx_bits, "precode_mode",duob_mode, "showAnalysis",0,"postFFE",[]); + output.dbenc_package{r} = db_results; + if options.append_to_db + database.addProcessingResult(run_id, db_results.metrics, db_results.config); + end + + end + + + end + +catch ME + save('workerError.mat','ME'); + rethrow(ME); end diff --git a/Functions/Job_Processing/submitJobs.m b/Functions/Job_Processing/submitJobs.m index 6cc834f..d667cc8 100644 --- a/Functions/Job_Processing/submitJobs.m +++ b/Functions/Job_Processing/submitJobs.m @@ -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 === -% -% 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); - +% 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 + 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 + 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(); - - % Submit jobs to parallel pool +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; - optionalVars = buildOptionalVars(k, submit_options.wh); - - % Submit job with proper arguments - futures{jobCounter} = parfeval(@dsp_runid, 1, run_ids(r), ... - "database_type", dsp_options.database_type,... - "dataBase", dsp_options.dataBase, ... - "append_to_db", dsp_options.append_to_db, ... + 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", optionalVars); - + "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 - 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; - 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 + % —– 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 - -elseif submit_mode == processingMode.serial - % Process jobs sequentially + 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); - - 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 \ No newline at end of file + +end diff --git a/projects/HighSpeedExperiment_2024/Auswertung_JLT/run_dsp_from_db.m b/projects/HighSpeedExperiment_2024/Auswertung_JLT/run_dsp_from_db.m index c73a50b..d34ec2c 100644 --- a/projects/HighSpeedExperiment_2024/Auswertung_JLT/run_dsp_from_db.m +++ b/projects/HighSpeedExperiment_2024/Auswertung_JLT/run_dsp_from_db.m @@ -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);