P9 complete: compiled kernels validated (byte-identical, 78-80 GSA samples/min/core), scaling report
Graviton3 c7g.8xlarge, 360 GSE165845 arrays: 360/360 GTCs identical between the Rust kernels and the pure-Python paths; 1 worker 78-80 samples/min (pure Python 29.6). Scaling 1->8 workers efficiency 0.87-0.90; above that the 125 MB/s gp3 disk caps the run (~620 samples/min) and, on a RAM disk, a fixed per-worker start-up cost limits a 360-sample batch to efficiency 0.48 at 32 workers (K-025). Worker state is pickled once. docs/validation/P9.md, result JSONs, guide/README/CHANGELOG updates. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01814GKVzZVL3FuF3DeiZXhv
This commit is contained in:
1 parent
bd19450364
commit
ed4895516d
8 files changed
+263
-9
No files matched your search
@@ -9,6 +9,8 @@ import json
|
||||
import math
|
||||
import multiprocessing as mp
|
||||
import os
|
||||
import pickle
|
||||
import tempfile
|
||||
from concurrent.futures import ProcessPoolExecutor
|
||||
from dataclasses import dataclass, field, replace
|
||||
from pathlib import Path
|
||||
@@ -147,6 +149,13 @@ def _init(
|
||||
)
|
||||
|
||||
|
||||
def _init_from_file(path: str) -> None:
|
||||
"""Worker initializer: the shared state is pickled once by the parent (not once per worker, which serialised
|
||||
worker start-up: ~0.25 s per worker for a GSA manifest context) and read by every worker in parallel."""
|
||||
with open(path, "rb") as fh:
|
||||
_init(*pickle.load(fh))
|
||||
|
||||
|
||||
def _process(job: Job) -> dict[str, Any]:
|
||||
ctx: ManifestContext = _W["ctx"]
|
||||
params: GenotypeParams = _W["params"]
|
||||
@@ -338,13 +347,17 @@ def run_genotype(opts: GenotypeOptions, rep: Reporter) -> int:
|
||||
_init(ctx, params, auto_1_22, probes, started, beads, th.bg_offset)
|
||||
results = [_process(j) for j in jobs]
|
||||
else:
|
||||
with ProcessPoolExecutor(
|
||||
workers,
|
||||
mp_context=mp.get_context("spawn"),
|
||||
initializer=_init,
|
||||
initargs=(ctx, params, auto_1_22, probes, started, beads, th.bg_offset),
|
||||
) as ex:
|
||||
results = list(ex.map(_process, jobs))
|
||||
opts.out.mkdir(parents=True, exist_ok=True)
|
||||
fd, state = tempfile.mkstemp(prefix=".simurg-workers-", suffix=".pkl", dir=opts.out)
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as fh:
|
||||
pickle.dump((ctx, params, auto_1_22, probes, started, beads, th.bg_offset), fh, protocol=5)
|
||||
with ProcessPoolExecutor(
|
||||
workers, mp_context=mp.get_context("spawn"), initializer=_init_from_file, initargs=(state,)
|
||||
) as ex:
|
||||
results = list(ex.map(_process, jobs))
|
||||
finally:
|
||||
Path(state).unlink(missing_ok=True)
|
||||
except KeyboardInterrupt:
|
||||
rm.status = "interrupted"
|
||||
rm.write(opts.out / "run_manifest.json")
|
||||
|
||||
Reference in new issue
Block a user