chirped WDM analsyis
minor stuff in FSO link with Magnus
This commit is contained in:
silas (home)
@@ -6,7 +6,7 @@ function WDM_model_10km_queue(options)
|
||||
|
||||
%%% Run parameters
|
||||
arguments
|
||||
options.num_channels = 16;
|
||||
options.num_channels = 8;
|
||||
options.channel_spacing = 400e9;
|
||||
options.fiber_length_km = 10;
|
||||
options.rand_key = 1;
|
||||
@@ -26,7 +26,7 @@ warning('off','MATLAB:datetime:AmbiguousTimeZone');
|
||||
|
||||
%% How many workers?
|
||||
cpus = str2double(getenv('SLURM_CPUS_PER_TASK'));
|
||||
if ~isfinite(cpus) || cpus < 1, cpus = max(1, feature('numcores')); end
|
||||
if ~isfinite(cpus) || cpus < 1, cpus = max(1, feature('numcores'))-1; end
|
||||
|
||||
% Use a per-job, node-local JobStorageLocation to avoid stale locks on $HOME
|
||||
tmpbase = getenv('TMPDIR');
|
||||
@@ -43,7 +43,7 @@ c.JobStorageLocation = jsl;
|
||||
p = gcp('nocreate');
|
||||
if isempty(p) || p.NumWorkers ~= cpus
|
||||
if ~isempty(p), delete(p); end
|
||||
p = parpool(c, cpus);
|
||||
p = parpool(c, cpus,"IdleTimeout",300);
|
||||
end
|
||||
fprintf('parpool up with %d workers; JobStorage=%s\n', p.NumWorkers, c.JobStorageLocation);
|
||||
|
||||
@@ -119,7 +119,7 @@ upsample_pow = 2^nextpow2(upsample_required);
|
||||
s.f_opt = fdac*kover*upsample_pow;
|
||||
s.f_opt_nyq = s.f_opt/2;
|
||||
|
||||
s.rop = -10:0.5:0;
|
||||
s.rop = -10:1:0;
|
||||
|
||||
%% ---------- Intermediate evaluation points (distance dimension) ----------
|
||||
segment_length = 1; % km (must match fiber loop below)
|
||||
@@ -184,7 +184,7 @@ for realiz = 1:s.num_realiz
|
||||
for l = 1:N
|
||||
|
||||
[Digi_sig,Symbols{l},Tx_bits{l}] = PAMsource( ...
|
||||
"fsym",fsym,"M",s.M,"order",17,"useprbs",0, ...
|
||||
"fsym",fsym,"M",s.M,"order",15,"useprbs",0, ...
|
||||
"fs_out",fdac, ...
|
||||
"applyclipping",0,"clipfactor",1.5, ...
|
||||
"applypulseform",apply_pulsef,"pulseformer",Pform, ...
|
||||
@@ -282,20 +282,21 @@ for realiz = 1:s.num_realiz
|
||||
[F, meta, output_ffe, output_dfe, output_vnle, output_mlse, output_dbt] = ...
|
||||
throttle_inflight(F, meta, output_ffe, output_dfe, output_vnle, output_mlse, output_dbt, maxInFlight);
|
||||
|
||||
% Enqueue one job per (l,ri)
|
||||
F(end+1,1) = parfeval(p, @rx_job, 5, ...
|
||||
m = struct('l',l,'ri',ri,'realiz',realiz,'eval_ptr',eval_ptr);
|
||||
|
||||
[F, meta] = enqueue_atomic(F, meta, p, @rx_job, 5, ...
|
||||
Opt_sig_wdm_demux{l}, ...
|
||||
s.rop(ri), ...
|
||||
Symbols{l}, Tx_bits{l}, ...
|
||||
s, l, ri, realiz, eval_ptr, ...
|
||||
fdac, kover, fadc, fsym, ...
|
||||
len_tr, mu_dc, mu_ffe, mu_dfe, dfe_order, duob_mode, ...
|
||||
memlog_dir);
|
||||
memlog_dir, ...
|
||||
'META', m);
|
||||
|
||||
fprintf('Enqueued job %d/%d for realiz %d/%d, l=%d/%d, ri=%d/%d at 0 km (inflight=%d/%d)\n', ...
|
||||
cnt, total, realiz, s.num_realiz, l, N, ri, length(s.rop), numel(F), maxInFlight);
|
||||
|
||||
meta(end+1) = struct('l',l,'ri',ri,'realiz',realiz,'eval_ptr',eval_ptr);
|
||||
end
|
||||
end
|
||||
|
||||
@@ -304,11 +305,6 @@ for realiz = 1:s.num_realiz
|
||||
end
|
||||
% -----------------------------------------------------
|
||||
|
||||
save_results_per_realization( ...
|
||||
s, eval_dist_km, ...
|
||||
output_ffe, output_dfe, output_vnle, output_mlse, output_dbt, ...
|
||||
output_root, fname);
|
||||
|
||||
for seg = 1:nSegments
|
||||
|
||||
fprintf('Realiz %d/%d: Segment %d/%d \n', realiz, s.num_realiz, seg, nSegments);
|
||||
@@ -336,19 +332,21 @@ for realiz = 1:s.num_realiz
|
||||
[F, meta, output_ffe, output_dfe, output_vnle, output_mlse, output_dbt] = ...
|
||||
throttle_inflight(F, meta, output_ffe, output_dfe, output_vnle, output_mlse, output_dbt, maxInFlight);
|
||||
|
||||
F(end+1,1) = parfeval(p, @rx_job, 5, ...
|
||||
m = struct('l',l,'ri',ri,'realiz',realiz,'eval_ptr',eval_ptr);
|
||||
|
||||
[F, meta] = enqueue_atomic(F, meta, p, @rx_job, 5, ...
|
||||
Opt_sig_wdm_demux{l}, ...
|
||||
s.rop(ri), ...
|
||||
Symbols{l}, Tx_bits{l}, ...
|
||||
s, l, ri, realiz, eval_ptr, ...
|
||||
fdac, kover, fadc, fsym, ...
|
||||
len_tr, mu_dc, mu_ffe, mu_dfe, dfe_order, duob_mode, ...
|
||||
memlog_dir);
|
||||
memlog_dir, ...
|
||||
'META', m);
|
||||
|
||||
fprintf('Seg: %d - Enqueued job %d/%d for realiz %d/%d, l=%d/%d, ri=%d/%d (inflight=%d/%d)\n', ...
|
||||
seg, cnt, total, realiz, s.num_realiz, l, N, ri, length(s.rop), numel(F), maxInFlight);
|
||||
|
||||
meta(end+1) = struct('l',l,'ri',ri,'realiz',realiz,'eval_ptr',eval_ptr);
|
||||
end
|
||||
end
|
||||
|
||||
@@ -546,15 +544,21 @@ end
|
||||
first = true;
|
||||
while ~isempty(F)
|
||||
|
||||
assert(numel(F)==numel(meta))
|
||||
idxUn = strcmp({F.State}, 'unavailable');
|
||||
if any(idxUn)
|
||||
for ii = find(idxUn)
|
||||
m = meta(ii);
|
||||
fprintf('[UNAVAILABLE] realiz=%d eval=%d l=%d ri=%d\n', ...
|
||||
m.realiz, m.eval_ptr, m.l, m.ri);
|
||||
end
|
||||
warning('collect_done: dropping %d unavailable futures.', sum(idxUn));
|
||||
F(idxUn) = [];
|
||||
meta(idxUn) = [];
|
||||
|
||||
if isempty(F), break; end
|
||||
end
|
||||
|
||||
|
||||
if first
|
||||
timeout = timeout_first;
|
||||
first = false;
|
||||
@@ -703,65 +707,146 @@ catch
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
function [F, meta, output_ffe, output_dfe, output_vnle, output_mlse, output_dbt] = ...
|
||||
throttle_inflight(F, meta, output_ffe, output_dfe, output_vnle, output_mlse, output_dbt, maxInFlight)
|
||||
%THROTTLE_INFLIGHT
|
||||
% Enforces a HARD cap on outstanding futures (running + queued + finished-not-yet-fetched).
|
||||
% Blocks until numel(F) < maxInFlight, then returns.
|
||||
% Hard cap on outstanding futures. Robust against:
|
||||
% - 'unavailable' futures (ID=-1 or State unavailable)
|
||||
% - finished-with-error futures (State may be 'finished' with non-empty Error)
|
||||
% - rare F/meta desync
|
||||
% - race-y property access on FevalFuture arrays (use snapshot Fs)
|
||||
|
||||
while numel(F) >= maxInFlight
|
||||
|
||||
% Wait until at least one finishes OR fails
|
||||
% (wait('finished',1) only wakes for finished, so we also handle failures via periodic checks)
|
||||
% Wait for at least one to finish (or time out quickly)
|
||||
try
|
||||
wait(F, 'finished', 1);
|
||||
catch
|
||||
% If wait fails for some reason, fall back to a short pause
|
||||
pause(0.1);
|
||||
end
|
||||
|
||||
% Harvest finished and failed futures (robust snapshot)
|
||||
n = numel(F);
|
||||
% ---- Ensure F/meta alignment (best-effort repair) ----
|
||||
if numel(F) ~= numel(meta)
|
||||
warning('throttle_inflight: F/meta desync (F=%d meta=%d). Attempting repair.', numel(F), numel(meta));
|
||||
|
||||
% Safety: meta must match F length
|
||||
if numel(meta) ~= n
|
||||
% Keep them aligned (rare; indicates earlier mismatch)
|
||||
m = min(numel(meta), n);
|
||||
F = F(1:m);
|
||||
% Prefer dropping unusable futures first (likely the unmatched ones)
|
||||
Fs = F; % snapshot
|
||||
n = numel(Fs);
|
||||
|
||||
st = cell(1,n);
|
||||
idbad = false(1,n);
|
||||
for k = 1:n
|
||||
try, st{k} = Fs(k).State; catch, st{k} = ''; end
|
||||
try, idbad(k) = (Fs(k).ID < 0); catch, idbad(k) = false; end
|
||||
end
|
||||
idxBad = find(strcmp(st,'unavailable') | idbad);
|
||||
|
||||
if numel(F) > numel(meta) && ~isempty(idxBad)
|
||||
need = numel(F) - numel(meta);
|
||||
idxBad = idxBad(1:min(numel(idxBad), need));
|
||||
idxBad = idxBad(idxBad >= 1 & idxBad <= numel(F));
|
||||
if ~isempty(idxBad)
|
||||
warning('throttle_inflight: dropping %d bad futures to restore alignment.', numel(idxBad));
|
||||
F(idxBad) = [];
|
||||
end
|
||||
end
|
||||
|
||||
% Final fallback: truncate both to min length
|
||||
m = min(numel(F), numel(meta));
|
||||
F = F(1:m);
|
||||
meta = meta(1:m);
|
||||
n = m;
|
||||
end
|
||||
|
||||
states = cell(1,n);
|
||||
if isempty(F)
|
||||
break;
|
||||
end
|
||||
|
||||
% ===================== IMPORTANT: snapshot =====================
|
||||
Fs = F; % stable snapshot for property access
|
||||
n = numel(Fs);
|
||||
% ===============================================================
|
||||
|
||||
% ---- Compute unavailable indices from snapshot ONLY ----
|
||||
st = cell(1,n);
|
||||
idbad = false(1,n);
|
||||
for k = 1:n
|
||||
states{k} = F(k).State;
|
||||
try, st{k} = Fs(k).State; catch, st{k} = ''; end
|
||||
try, idbad(k) = (Fs(k).ID < 0); catch, idbad(k) = false; end
|
||||
end
|
||||
|
||||
doneFinished = strcmp(states, 'finished');
|
||||
doneFailed = strcmp(states, 'failed');
|
||||
done = doneFinished | doneFailed;
|
||||
idxUn = find(strcmp(st,'unavailable') | idbad);
|
||||
|
||||
% Sanitize indices (prevents out-of-range deletions even if something weird happens)
|
||||
idxUn = idxUn(idxUn >= 1 & idxUn <= numel(F));
|
||||
|
||||
if ~isempty(idxUn)
|
||||
for ii = idxUn(:).'
|
||||
if ii <= numel(meta)
|
||||
mm = meta(ii);
|
||||
fprintf('[UNAVAILABLE@throttle] realiz=%d eval=%d l=%d ri=%d\n', ...
|
||||
mm.realiz, mm.eval_ptr, mm.l, mm.ri);
|
||||
else
|
||||
fprintf('[UNAVAILABLE@throttle] meta_missing for ii=%d\n', ii);
|
||||
end
|
||||
end
|
||||
warning('throttle_inflight: dropping %d unavailable/bad futures.', numel(idxUn));
|
||||
|
||||
% Delete in descending order is safest for structs (not strictly necessary, but robust)
|
||||
idxUn = sort(idxUn, 'descend');
|
||||
F(idxUn) = [];
|
||||
meta(idxUn) = [];
|
||||
continue; % re-check cap after shrinking
|
||||
end
|
||||
|
||||
% ---- Determine "done" (finished or finished-with-error) ----
|
||||
isFinished = strcmp(st, 'finished');
|
||||
|
||||
hasErr = false(1,n);
|
||||
for k = 1:n
|
||||
try
|
||||
hasErr(k) = ~isempty(Fs(k).Error); % use snapshot object
|
||||
catch
|
||||
hasErr(k) = false;
|
||||
end
|
||||
end
|
||||
|
||||
done = isFinished | hasErr;
|
||||
idxDone = find(done);
|
||||
idxDone = idxDone(idxDone >= 1 & idxDone <= numel(F)); % sanitize
|
||||
|
||||
if isempty(idxDone)
|
||||
pause(0.05);
|
||||
continue;
|
||||
end
|
||||
|
||||
% Fetch outputs for finished; mark failed as empty results
|
||||
% ---- Harvest done futures ----
|
||||
for jj = 1:numel(idxDone)
|
||||
ii = idxDone(jj);
|
||||
m = meta(ii);
|
||||
|
||||
if doneFinished(ii)
|
||||
[ffe_r, dfe_r, vnle_r, mlse_r, dbt_r] = fetchOutputs(F(ii));
|
||||
if isFinished(ii) && ~hasErr(ii)
|
||||
try
|
||||
[ffe_r, dfe_r, vnle_r, mlse_r, dbt_r] = fetchOutputs(F(ii));
|
||||
catch ME
|
||||
warning('throttle_inflight: fetchOutputs failed: realiz=%d eval=%d l=%d ri=%d | %s', ...
|
||||
m.realiz, m.eval_ptr, m.l, m.ri, ME.message);
|
||||
ffe_r = []; dfe_r = []; vnle_r = []; mlse_r = []; dbt_r = [];
|
||||
end
|
||||
else
|
||||
% Failed
|
||||
% finished-with-error
|
||||
try
|
||||
err = F(ii).Error;
|
||||
warning('rx_job failed: realiz=%d eval=%d l=%d ri=%d | %s', ...
|
||||
m.realiz, m.eval_ptr, m.l, m.ri, err.message);
|
||||
if ~isempty(err)
|
||||
warning('rx_job error: realiz=%d eval=%d l=%d ri=%d | %s', ...
|
||||
m.realiz, m.eval_ptr, m.l, m.ri, err.message);
|
||||
else
|
||||
warning('rx_job error: realiz=%d eval=%d l=%d ri=%d | (unknown error)', ...
|
||||
m.realiz, m.eval_ptr, m.l, m.ri);
|
||||
end
|
||||
catch
|
||||
warning('rx_job error: realiz=%d eval=%d l=%d ri=%d | (error unreadable)', ...
|
||||
m.realiz, m.eval_ptr, m.l, m.ri);
|
||||
end
|
||||
ffe_r = []; dfe_r = []; vnle_r = []; mlse_r = []; dbt_r = [];
|
||||
end
|
||||
@@ -773,9 +858,43 @@ while numel(F) >= maxInFlight
|
||||
output_dbt{m.l, m.ri, m.realiz, m.eval_ptr} = dbt_r;
|
||||
end
|
||||
|
||||
% Remove harvested futures/meta entries using numeric indices (safe)
|
||||
% Remove harvested entries (descending for safety)
|
||||
idxDone = sort(idxDone, 'descend');
|
||||
F(idxDone) = [];
|
||||
meta(idxDone) = [];
|
||||
|
||||
end
|
||||
|
||||
end
|
||||
|
||||
|
||||
|
||||
|
||||
function [F, meta] = enqueue_atomic(F, meta, p, fun, nOut, varargin)
|
||||
%ENQUEUE_ATOMIC Enqueue a parfeval job and append matching meta atomically.
|
||||
% Usage:
|
||||
% m = struct('l',l,'ri',ri,'realiz',realiz,'eval_ptr',eval_ptr);
|
||||
% [F, meta, f] = enqueue_atomic(F, meta, p, @rx_job, 5, args..., 'META', m);
|
||||
|
||||
% Split varargin into args + meta struct
|
||||
idx = find(strcmp(varargin, 'META'), 1, 'last');
|
||||
if isempty(idx) || idx == numel(varargin)
|
||||
error('enqueue_atomic: META struct must be provided as last named argument.');
|
||||
end
|
||||
args = varargin(1:idx-1);
|
||||
m = varargin{idx+1};
|
||||
|
||||
% Create future first (local variable), but don't mutate F/meta yet
|
||||
f = parfeval(p, fun, nOut, args{:});
|
||||
|
||||
% If future is unusable, do not append (prevents meta shift)
|
||||
if f.ID < 0 || strcmp(f.State, 'unavailable')
|
||||
warning('enqueue_atomic: got unusable future (ID=%d, State=%s). Dropping enqueue: realiz=%d eval=%d l=%d ri=%d', ...
|
||||
f.ID, string(f.State), m.realiz, m.eval_ptr, m.l, m.ri);
|
||||
return;
|
||||
end
|
||||
|
||||
% Now append both together (atomic w.r.t. fprintf etc.)
|
||||
F(end+1,1) = f;
|
||||
meta(end+1,1) = m;
|
||||
end
|
||||
|
||||
Reference in New Issue
Block a user