start with 400G analysis work

2026-07-20 10:22:11 +02:00
parent cae81c0dae
commit 125d8508ca
43 changed files with 4089 additions and 123 deletions
--- a/projects/Diss/MPI_revisit/simulation/submitMpiSimulationJobs.m
+++ b/projects/Diss/MPI_revisit/simulation/submitMpiSimulationJobs.m
@@ -0,0 +1,87 @@
+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