P5: cloud task for cohort QC validation; configurable gp3 throughput; batched contamination map

- scripts/cloud/tasks/p5.sh: P4 preset re-check, genotype (v5) with QC, validate_p5_large.py
- scripts/validate_p5_large.py: identify samples via het pattern vs 1000G VCF; sex, parent-offspring,
  Mendel, contamination/control distributions
- trial.sh: DISK_MBPS / DISK_IOPS
- qc/contamination: bounded map batches (executor.map submits eagerly)

src/simurg/qc/contamination.py

@@ -26,6 +26,7 @@ import numpy.typing as npt
 
 F32 = np.float32
 CHUNK = 4096
+MAP_BATCH = 32  # chunks in flight per map call
 
 
 @dataclass
@@ -148,10 +149,19 @@ def baf_regress(
     t = float(F32(threshold))  # the reference stores the threshold as float: 0.1f = 0.10000000149…
     keep = np.flatnonzero(np.isfinite(af) & (af >= t) & (af <= 1.0 - t))
     chunks = [keep[s : s + CHUNK] for s in range(0, len(keep), CHUNK)]
-    tasks = ((np.asarray(genotypes[r]), np.asarray(baf[r], dtype=F32), adjust_min, median) for r in chunks)
+
+    def adjustments() -> Iterable[tuple[npt.NDArray[np.float32], npt.NDArray[np.float32]]]:
+        # bounded batches: an executor's map submits its whole input at once
+        for b in range(0, len(chunks), MAP_BATCH):
+            tasks = [
+                (np.asarray(genotypes[r]), np.asarray(baf[r], dtype=F32), adjust_min, median)
+                for r in chunks[b : b + MAP_BATCH]
+            ]
+            yield from map_fn(cluster_adjustments, tasks)
+
     sx2, sxy, sx, sy = (np.zeros(n, F32) for _ in range(4))
     cnt = np.zeros(n, np.int64)
-    for rows, (adj_aa, adj_bb) in zip(chunks, map_fn(cluster_adjustments, tasks), strict=True):
+    for rows, (adj_aa, adj_bb) in zip(chunks, adjustments(), strict=True):
         g = np.asarray(genotypes[rows])
         b = np.asarray(baf[rows], dtype=F32)
         p = af[rows]