Optimise memory allocation in batchprocessing

[rhaegar325]

Fix OOM and poor parallelism in batch CMORisation of long daily/heavy variables

Summary

Batch CMORisation runs one PBS job per variable. Two problems made the heavy
variables unusable:

1. Jobs used ~1 core of the 12–18 requested. The generated script ran Dask
   as a single process with N threads, so every netCDF4 read serialised on the
   global HDF5 lock.
2. tasmax/tasmin could not complete at all on a long run (5,352 daily
   files, ~446 years). Workers were repeatedly OOM-killed on a
   finalize-hlgfinalizecompute task and the job died.

The root cause of (2) turned out to be a latent eager-compute bug:
np.isclose(dask_array, ...) is not a ufunc, so it does not dispatch lazily
— it materialises the entire array into memory. Applied to the raw daily series
during missing-value normalisation, it loaded ~18 GB into a single worker.

After the fixes, all 22 ILAMB variables complete, with no memory pressure, and
outputs are bit-identical.

Root causes

1. Eager np.isclose on Dask arrays (the OOM)

normalize_dataset_missing_values built its NaN mask with
np.isclose(var, missing_value, ...). Verified behaviour:

np.isclose(dask_backed_dataarray, 1e20)   # -> numpy memoryview, NOT dask

np.isclose is a compound function, not a ufunc, so it bypasses Dask's
dispatch and coerces the array to NumPy. The daily source fld_s03i236 carries
missing_value = _FillValue = 1e+20, so the mask was built over the raw daily
series (~166,000 timesteps ≈ 18.5 GB) inside _normalize_missing_values_early
— during load_dataset, i.e. before the monthly resample and before any write.
That single materialisation is the finalize task that killed the workers.

Log ordering confirmed it: the first event in the failing run was
Could not normalize missing values early: ... finalize ... 4 workers died.

2. Threaded Dask client → ~1 core

processes=False with N threads serialises all netCDF4 reads on the HDF5 lock.
Measured on a 12-core node (1032 cold files, cVeg read+sum):

client	wall	cpu/wall
threads (before)	273.9 s	1.2 (~1 core)
processes (after)	63.1 s	8.4 (~8 cores)

Results were bit-identical (sum=1.19124e+49).

3. Task-graph explosion on daily inputs

Daily files are stored with on-disk HDF5 chunking [1, 145, 192] — one chunk
per day. chunks={} inherits that layout, so a 31-day file became 31 Dask
chunks (~650k tasks at full scale). Monthly variables were unaffected (their
on-disk chunk is the whole field).

4. Monolithic write

The driver defaults enable_chunking=False, so write() materialised the whole
variable in one .compute() rather than streaming per time-chunk.

Changes

file	change
vocabulary_processors.py	Replace np.isclose(var, m, rtol=0, atol=a) with the Dask-native, lazy abs(var - m) <= a (exactly equivalent at rtol=0). This is the OOM fix.
derivations/calc_utils.py	Same fix in _mask_missing_values_for_reduction (same bug, fires during the resample).
base.py	chunks={"time": -1} on the multi-file read — one chunk per file. Collapses the daily task graph ~26x. Monthly inputs unaffected (time size 1).
templates/cmor_python_script.j2	processes=True Dask client, sized per variable after file discovery; enable_chunking=True so writes stream in bounded time-chunks.
executors/dask_config.py	Adaptive, env-driven worker sizing (new module). Reads PBS_NCPUS / PBS_MEM_GB; probes one input file to pick a per-worker memory floor.

Worker sizing

All parallelism comes from processes; threads_per_worker=1. Extra threads
add no read throughput (HDF5 lock + GIL) while doubling a worker's resident
data — the opposite of what memory-limited variables need.

Per-worker memory floors are env-overridable
(MOPPY_WORKER_GB_LIGHT|MEDIUM|HEAVY), calibrated from measured peaks once the
pipeline was fully streaming (~12.5 GB daily, ~10 GB carbon, ~6 GB light):

tier	floor	example	workers (18 CPU / 190 GB)
light	12 GB	tas, pr	14
medium	14 GB	cVeg, cSoil, nbp	12
heavy	16 GB	tasmax, tasmin	10

Validation

Full-scale batch (5,352 daily files, ~446 years) — all 22 variables pass

variable	workers	exit	memory used	pressure warnings	walltime
tasmax	10 x 16 GB	0	106.6 GB / 190 GB	0	25:37
tasmin	10 x 16 GB	0	78.8 GB	0	25:38
cVeg	12 x 14 GB	0	81.9 GB	0	39:25
cSoil	12 x 14 GB	0	81.2 GB	0	36:22
nbp	12 x 14 GB	0	81.4 GB	0	1:22:19
tas	14 x 12 GB	0	82.1 GB	0	23:53

Previously tasmax/tasmin failed outright (workers OOM-killed on
finalize). All other variables complete with zero memory-pressure warnings.

Correctness

• tasmax output is bit-exact to an independently computed monthly maximum
  of the daily source (and clearly not the mean).
• cVeg output is bit-identical with enable_chunking on vs off (data,
  coords and all bounds variables).
• processes=True produces bit-identical results to the threaded client at full
  scale.
• Lazy mask produces the identical NaN set to the old np.isclose.

Test suite

	before (HEAD)	after
failed	37	11
passed	1358	1384

No new failures introduced; 26 previously-failing tests now pass. The 11
remaining failures are a strict subset of the pre-existing ones (test_cmip7_*
CV tests and two test_memory_usage tests that also fail on clean HEAD).

Compliance (IOOS compliance-checker, all 22 outputs)

• CF-1.7: zero high-priority failures (scores 97–98%).
• WCRP CMIP6: 3 high-priority on every file, all from a single cause — the
  outputs were written flat (drs_root unset), so DRS project root 'cmip6' not found in the file path; PATH001/PATH002 merely inherit that failure. Not
  a data problem.
• tasmax/tasmin have 3 extra WCRP items (branded variable not in the
  checker's registry). tsl — a variable this PR does not touch — has the
  identical 3 items, confirming a pre-existing checker-registry gap rather
  than a regression.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 78.1%. Comparing base (ec0c181) to head (96841a2).

Additional details and impacted files

@@          Coverage Diff          @@
##            main    #501   +/-   ##
=====================================
  Coverage   78.1%   78.1%           
=====================================
  Files         33      33           
  Lines       6626    6627    +1     
  Branches    1260    1260           
=====================================
+ Hits        5173    5174    +1     
  Misses      1180    1180           
  Partials     273     273           

Flag	Coverage Δ
unit	78.1% <100.0%> (+<0.1%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.