diff --git a/resources/tester/test_612_precompute.py b/resources/tester/test_612_precompute.py index 32b06e9d0..92cc16e2b 100644 --- a/resources/tester/test_612_precompute.py +++ b/resources/tester/test_612_precompute.py @@ -70,7 +70,7 @@ def fake_store(tmp_path, monkeypatch): calls: list[tuple[str, str]] = [] fail = {"36901002"} - def fake_run(step: str, no: str): + def fake_run(step: str, no: str, part=None): calls.append((step, no)) if no in fail: return step, no, {"ok": False, "error": "RuntimeError: 받기 실패"} @@ -85,8 +85,9 @@ def test_resume_skips_done_and_retries_failed(fake_store) -> None: sheets = ["36901001", "36901002", "36901003"] progress = tool.load_progress() tool.run(sheets, ["e07", "dem"], 1, progress, say=lambda _: None) - assert [s for s, _ in calls] == ["dem"] * 3 + ["e07"] * 3 # 차례 고정 - saved = json.loads((root / tool.PROGRESS_NAME).read_text(encoding="utf-8")) + # 차례 고정 · 지형 단계는 0차곡 이웃(3x3)까지 셈(618-1) + assert [s for s, n in calls if n in sheets] == ["dem"] * 3 + ["e07"] * 3 + saved = json.loads(tool.progress_path().read_text(encoding="utf-8")) assert saved["steps"]["dem"]["36901001"]["ok"] is True assert saved["steps"]["dem"]["36901002"]["ok"] is False diff --git a/resources/tester/test_618_precompute_split.py b/resources/tester/test_618_precompute_split.py new file mode 100644 index 000000000..b8cf989ab --- /dev/null +++ b/resources/tester/test_618_precompute_split.py @@ -0,0 +1,195 @@ +"""PLAN 618-1 — 전국 미리 계산 두 PC 나눔(`tools/precompute_terrain.py`) 지킴 시험. + +나눔 겹침 0 · 합치면 전체 · 이웃 일감 한 PC 만 · PC 별 기록 이어 받기 · CPU 상한(가짜 값) · +0차곡 남의 몫 이웃 없으면 미룸 · 지정 코어 인자 풀기 · +셈 함수는 가짜로 바꿔 서버 · 전국 파일 없이 돎. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) + +from B04_PreProcess.B04_PreProcess_Engine_MapSheet import neighbors_3x3 # noqa: E402 +from tools import precompute_terrain as tool # noqa: E402 + +SHEETS = [f"36901{i:03d}" for i in range(1, 101)] + [f"35810{i:03d}" for i in range(1, 101)] + + +def test_parse_part_and_cpus() -> None: + assert tool.parse_part("") is None + assert tool.parse_part("0-69/100") == (0, 69, 100) + assert tool.parse_part("70-99/100") == (70, 99, 100) + for bad in ("50-10/100", "0-100/100"): + with pytest.raises(ValueError): + tool.parse_part(bad) + assert tool.parse_cpus("12-19") == list(range(12, 20)) + assert tool.parse_cpus("1, 3-4") == [1, 3, 4] + assert tool.parse_cpus("") == [] + + +def test_split_no_overlap_and_union_is_all() -> None: + for spans in (("0-69/100", "70-99/100"), ("0-49/100", "50-99/100")): + a, b = (tool.parse_part(s) for s in spans) + mine = {no for no in SHEETS if tool.in_part(no, a)} + other = {no for no in SHEETS if tool.in_part(no, b)} + assert not mine & other and mine | other == set(SHEETS) + ja, jb = set(tool.dem_jobs(SHEETS, a)), set(tool.dem_jobs(SHEETS, b)) + assert not ja & jb and ja | jb == set(tool.dem_jobs(SHEETS, None)) + big = {no for no in SHEETS if tool.in_part(no, (0, 69, 100))} + assert 0.6 < len(big) / len(SHEETS) < 0.8 + assert tool.in_part("36901001@5186", (0, 99, 100)) + + +def test_dem_jobs_cover_neighbors_once() -> None: + jobs = tool.dem_jobs(["36901055"], None) + assert jobs[0] == "36901055" and set(jobs) == set(neighbors_3x3("36901055")) + # 띠 경계(128° · 5186 | 5187) 넘는 이웃 = 가운데 좌표계로 옮겨 담는 일감 + west = [no for no in tool.dem_jobs(["36809001"], None) if "@" in no] + assert west and all(j.endswith(f"@{tool.epsg_of('36809001')}") for j in west) + + +@pytest.fixture() +def fake_store(tmp_path, monkeypatch): + monkeypatch.setattr(tool.store, "TERRAIN_ROOT", tmp_path) + monkeypatch.setattr(tool.store, "_gpkg_stamp", lambda: "stamp-1") + monkeypatch.setattr(tool, "PC_NAME", "LAPTOP") + monkeypatch.setattr(tool, "CPU_SAMPLE_S", 0.0) + log: list = [] + + def fake_run(step: str, job: str, part=None): + log.append(("run", step, job)) + return step, job, {"ok": True, "s": 0.01, "fresh": True, "empty": False} + + monkeypatch.setattr(tool, "run_step", fake_run) + return tmp_path, log + + +def test_progress_per_pc_and_legacy_resume(fake_store, monkeypatch) -> None: + root, log = fake_store + legacy = {"gpkg": "stamp-1", "cell_m": 2.0, "steps": {"dem": {"36901001": {"ok": True}}}} + (root / tool.PROGRESS_NAME).write_text(json.dumps(legacy), encoding="utf-8") + tool.run(["36901001", "36901002"], ["e07"], 1, tool.load_progress(), say=lambda _: None) + assert (root / "_precompute_progress_LAPTOP.json").is_file() + assert json.loads((root / tool.PROGRESS_NAME).read_text(encoding="utf-8")) == legacy # 안 씀 + log.clear() + tool.run(["36901001", "36901002"], ["dem"], 1, tool.load_progress(), say=lambda _: None) + assert ("run", "dem", "36901001") not in log # 옛 기록 성공 = 이어 받음 + + monkeypatch.setattr(tool, "PC_NAME", "DESKTOP") + desk = tool.load_progress() + assert desk["steps"]["e07"] == {} # 다른 PC 기록은 run 에선 안 섞음 + desk["steps"]["e07"]["36901003"] = {"ok": True, "s": 1.0, "fresh": True} + tool.save_progress(desk) + merged = tool.load_progress(every_pc=True)["steps"]["e07"] + assert {"36901001", "36901002", "36901003"} <= set(merged) + + +def test_cpu_cap_holds_new_work(fake_store) -> None: + _, log = fake_store + values = iter([95, 90, 60, 80, 40]) + + def cpu() -> float: + value = next(values) + log.append(("cpu", value)) + return value + + tool.run( + ["36901001", "36901002"], + ["e07"], + 1, + tool.load_progress(), + cpu_cap=70, + cpu=cpu, + say=lambda _: None, + ) + assert [x[1] if x[0] == "cpu" else x[2] for x in log] == [ + 95, + 90, + 60, + "36901001", + 80, + 40, + "36901002", + ] + + +def test_zero_defers_when_foreign_neighbor_missing(tmp_path, monkeypatch) -> None: + monkeypatch.setattr(tool.store, "TERRAIN_ROOT", tmp_path) + center = "36901055" + part = (0, 49, 100) + foreign = [no for no in neighbors_3x3(center) if not tool.in_part(no, part)] + assert foreign # 고른 가운데는 남의 몫 이웃이 있음 + stored: set[str] = set() + monkeypatch.setattr( + tool.store, "_stored_ok", lambda no, e, c: {"k": 1} if no in stored else None + ) + ensured: list[str] = [] + monkeypatch.setattr(tool.store, "ensure_sheet_terrain", lambda no, e, c: ensured.append(no)) + _, _, info = tool.run_step("zero", center, part) + assert info["ok"] is False and info["wait"] and not ensured # 셈 안 함 + + stored.update(foreign) # 다른 PC 가 셈 · 동기 옴 + from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero + + monkeypatch.setattr(zero, "sheet_accumulation", lambda no, e, c: tmp_path / "acc.npz") + _, _, info = tool.run_step("zero", center, part) + assert info["ok"] is True and len(ensured) == 9 + + +def test_deferred_rounds_retry_until_done(fake_store, monkeypatch) -> None: + root, log = fake_store + tries: dict[str, int] = {} + + def fake_run(step: str, job: str, part=None): + tries[job] = tries.get(job, 0) + 1 + if job == "36901002" and tries[job] < 3: + return step, job, {"ok": False, "wait": True, "error": "이웃 지형 없음"} + return step, job, {"ok": True, "s": 0.01, "fresh": True, "empty": False} + + monkeypatch.setattr(tool, "run_step", fake_run) + monkeypatch.setattr(tool.time, "sleep", lambda s: None) + progress = tool.load_progress() + tool.run(["36901001", "36901002"], ["zero"], 1, progress, say=lambda _: None, defer_wait=1) + assert tries == {"36901001": 1, "36901002": 3} + out = tool.summary(["36901001", "36901002"], progress)["zero"] + assert out["ok"] == 2 and out["failed"] == {} and out["waiting"] == 0 + + +def test_clean_dead_only_own_part(tmp_path) -> None: + part = (0, 49, 100) + mine = next(no for no in SHEETS if tool.in_part(no, part)) + other = next(no for no in SHEETS if not tool.in_part(no, part)) + for no in (mine, other): + (tmp_path / no).mkdir() + (tmp_path / no / ".dem_5187_2m.lock").write_text("999999999 0", encoding="utf-8") + assert tool.clean_dead(tmp_path, part) == 1 + assert (tmp_path / other / ".dem_5187_2m.lock").exists() + + +def test_worker_init_pins_first_e_workers(monkeypatch, tmp_path) -> None: + import multiprocessing + + import psutil + + pinned: list[list[int]] = [] + + class FakeProc: + def nice(self, _): + pass + + def cpu_affinity(self, cpus): + pinned.append(cpus) + + monkeypatch.setattr(psutil, "Process", FakeProc) + monkeypatch.setattr(tool.store, "TERRAIN_ROOT", tool.store.TERRAIN_ROOT) # 끝나면 되돌림 + counter = multiprocessing.Value("i", 0) + for _ in range(5): + tool.init_worker(str(tmp_path), counter, [12, 13], 3) + assert pinned == [[12, 13]] * 3 and tool.store.TERRAIN_ROOT == tmp_path diff --git a/tools/precompute_terrain.py b/tools/precompute_terrain.py index 5dbfc3e6b..b5a78e4a3 100644 --- a/tools/precompute_terrain.py +++ b/tools/precompute_terrain.py @@ -6,16 +6,23 @@ --sheets 36901,36712089 도엽 번호 또는 앞자리(1:50,000 묶음) · 주면 높이 차 거름 안 씀 --box lon0,lat0,lon1,lat1 도엽 가운데가 이 경위도 상자 안 --min-relief 150 도엽 안 높이 차(m) 이상만(산지) · --limit N 앞 N 장만 - --workers 4 일꾼 수 · --status 진행 기록 요약만 · --rebuild-index 도엽 목록 다시 + --workers 4 일꾼 수 · --status 진행 기록 요약만(모든 PC) · --rebuild-index 도엽 목록 다시 + --part 0-69/100 두 PC 나눔(도엽 번호 crc32 % 100 이 0~69 인 것만 · PLAN 618) + --cpu-cap 70 전체 CPU % 가 넘으면 새 도엽 안 줌 · --e-cores 12-19 --e-workers 3 + 일꾼 셋만 그 논리 CPU 로 지정 · --root 창고 자리(시험용 임시 폴더) - 도엽 목록 = 전국 등고선 파일 꼭짓점이 닿는 도엽(0.025°) — `_precompute_sheets.json`(전국 파일 지문이 같으면 다시 안 셈) · 차례 = 등고선 길이 큰 것(산지)부터 · 같으면 높이 차. - 좌표계 = 도엽 가운데 경도의 TM 띠(`tm_epsg_for_lon` · 5185~5187) · 격자 2 m. - 셈 = 공용 함수만 부름(창고 지형 · D 0차곡 누적 · E07 도엽) — 저장 열쇠가 같으면 함수가 건너뜀 · 쓰기 = 함수 쪽 원자 쓰기(끊겨도 반쯤 쓴 파일 없음). -- 이어 하기 = `_precompute_progress.json`(단계 · 도엽마다 성공 · 시간) — 성공한 도엽은 - 다시 안 부름 · 실패는 다음에 다시 · 전국 파일이 바뀌면 기록을 새로 시작. -- 시작 때 죽은 셈의 잠금(주인 프로세스 없음) · 오래된 임시 파일을 치움 · +- 이어 하기 = PC 별 `_precompute_progress_<컴퓨터 이름>.json`(단계 · 도엽마다 성공 · 시간 · + 시놀로지가 두 PC 사이 옮김 → 한 파일은 한 PC 만 씀) · 옛 `_precompute_progress.json` 의 성공은 + 읽어서 이어 받음(안 씀) · 성공한 도엽은 다시 안 부름 · 실패는 다음에 다시 · 전국 파일이 바뀌면 + 기록을 새로 시작. +- 나눔 = 지형 단계가 목록 도엽 + 0차곡이 쓸 이웃(목록 밖 · 띠 넘음 포함)을 해시 몫대로 셈 · + 0차곡은 남의 몫 이웃 지형이 창고에 없으면 미뤘다 다시(다른 PC 셈 · 시놀로지 동기 기다림). +- 시작 때 자기 몫 도엽의 죽은 셈 잠금(주인 프로세스 없음) · 오래된 임시 파일을 치움 · 낮은 우선순위(BELOW_NORMAL). """ @@ -24,11 +31,15 @@ from __future__ import annotations import argparse import json import math +import multiprocessing import os +import platform import sqlite3 import sys +import threading import time -from concurrent.futures import ProcessPoolExecutor, as_completed +import zlib +from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait from pathlib import Path from typing import Any @@ -50,7 +61,10 @@ from common_util.common_util_json import atomic_write_json # noqa: E402 STEPS = ("dem", "zero", "e07") CELL_M = 2.0 -PROGRESS_NAME = "_precompute_progress.json" +PROGRESS_NAME = "_precompute_progress.json" # 옛 한 벌 기록(읽기만) +PC_NAME = platform.node() or "pc" +CPU_SAMPLE_S = 3.0 # 전체 CPU 평균 잼 · 상한 쓸 때 새 도엽 주는 간격 +DEFER_WAIT_S = 600.0 # 0차곡 이웃 지형 기다림 한 바퀴 INDEX_NAME = "_precompute_sheets.json" INDEX_CHUNK = 50_000 # 가닥 묶음(약 0.5 GB 읽기) INDEX_STRIDE = 8 # 꼭짓점 솎음(도엽 2 km 에 비해 넉넉) @@ -68,8 +82,77 @@ def below_normal() -> None: psutil.Process().nice(psutil.BELOW_NORMAL_PRIORITY_CLASS) -def epsg_of(sheet_no: str) -> int: - return tm_epsg_for_lon(sheet5k_center(sheet_no)[1]) +def epsg_of(job: str) -> int: + """도엽(또는 `도엽@좌표계` 일감)의 좌표계 — 기본 = 도엽 가운데 경도 띠.""" + no, _, epsg = job.partition("@") + return int(epsg) if epsg else tm_epsg_for_lon(sheet5k_center(no)[1]) + + +# ── 나눔 · CPU ─────────────────────────────────────────────────────────────── + +Part = tuple[int, int, int] + + +def parse_part(text: str) -> Part | None: + """`A-B/N` → (A, B, N) · 빈 글 = 나눔 없음.""" + if not text: + return None + span, _, n = text.partition("/") + a, _, b = span.partition("-") + part = (int(a), int(b or a), int(n or 100)) + if not 0 <= part[0] <= part[1] < part[2]: + raise ValueError(f"--part = A-B/N (0 ≤ A ≤ B < N): {text}") + return part + + +def in_part(job: str, part: Part | None) -> bool: + """도엽 번호 crc32 % N 이 A~B — PC 와 실행에 상관없이 같은 몫(겹침 0).""" + if part is None: + return True + value = zlib.crc32(job.partition("@")[0].encode()) % part[2] + return part[0] <= value <= part[1] + + +def parse_cpus(text: str) -> list[int]: + """`12-19` · `12,14,16-17` → 논리 CPU 번호.""" + cpus: list[int] = [] + for piece in filter(None, (p.strip() for p in text.split(","))): + a, _, b = piece.partition("-") + cpus.extend(range(int(a), int(b or a) + 1)) + return cpus + + +class CpuMeter: + """전체 CPU %(시스템 · 다른 일 포함) — 뒤 실이 `CPU_SAMPLE_S` 평균을 잼 · 부르면 마지막 값.""" + + def __init__(self) -> None: + import psutil + + self.value = psutil.cpu_percent(interval=CPU_SAMPLE_S) + + def loop() -> None: + while True: + self.value = psutil.cpu_percent(interval=CPU_SAMPLE_S) + + threading.Thread(target=loop, daemon=True).start() + + def __call__(self) -> float: + return self.value + + +def init_worker(root: str, counter: Any, e_cpus: list[int], e_workers: int) -> None: + """일꾼 시작 — 창고 자리 · 낮은 우선순위 · 앞선 `e_workers` 명은 `e_cpus` 에만(저전력 코어).""" + store.TERRAIN_ROOT = Path(root) + below_normal() + if counter is None: + return + with counter.get_lock(): + slot = counter.value + counter.value += 1 + if slot < e_workers: + import psutil + + psutil.Process().cpu_affinity(e_cpus) # ── 도엽 목록 ──────────────────────────────────────────────────────────────── @@ -203,9 +286,24 @@ def _mtime(path: Path) -> int | None: return None -def run_step(step: str, sheet_no: str) -> tuple[str, str, dict[str, Any]]: - """도엽 하나 · 단계 하나 — 공용 함수만 부름 · fresh = 이번에 새로 셈.""" - epsg = epsg_of(sheet_no) +def dem_jobs(sheets: list[str], part: Part | None) -> list[str]: + """지형 단계 일감 — 목록 도엽 뒤 0차곡이 쓸 3x3 이웃(그 가운데 도엽 좌표계) · 자기 몫만. + + 이웃이 가운데와 띠가 다르면 `도엽@좌표계`(옮겨 담기) — 목록 밖 이웃도 한 PC 만 셈. + """ + jobs: dict[str, None] = {no: None for no in sheets if in_part(no, part)} + for center in sheets: + epsg = epsg_of(center) + for no in neighbors_3x3(center): + if in_part(no, part): + jobs[no if epsg == epsg_of(no) else f"{no}@{epsg}"] = None + return list(jobs) + + +def run_step(step: str, job: str, part: Part | None = None) -> tuple[str, str, dict[str, Any]]: + """일감 하나 · 단계 하나 — 공용 함수만 부름 · fresh = 이번에 새로 셈 · wait = 이웃 기다림.""" + sheet_no = job.partition("@")[0] + epsg = epsg_of(job) started = time.monotonic() try: if step == "dem": @@ -214,7 +312,11 @@ def run_step(step: str, sheet_no: str) -> tuple[str, str, dict[str, Any]]: elif step == "zero": from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero - for no in neighbors_3x3(sheet_no): # 여유 200 m 이웃 지형(누적 열쇠가 나중에 안 바뀌게) + neighbors = neighbors_3x3(sheet_no) + for no in neighbors: # 남의 몫 이웃 지형이 아직 없음 = 미룸(누적 열쇠가 나중에 바뀜) + if not in_part(no, part) and store._stored_ok(no, epsg, CELL_M) is None: + return step, job, {"ok": False, "wait": True, "error": f"이웃 지형 없음 {no}"} + for no in neighbors: # 여유 200 m 이웃 지형(누적 열쇠가 나중에 안 바뀌게) store.ensure_sheet_terrain(no, epsg, CELL_M) target = zero._acc_names(sheet_no, epsg, CELL_M)[0] before = _mtime(target) @@ -231,30 +333,55 @@ def run_step(step: str, sheet_no: str) -> tuple[str, str, dict[str, Any]]: else: raise ValueError(f"모르는 단계: {step}") except Exception as exc: # noqa: BLE001 — 한 장이 죽어도 나머지는 감 · 기록에 까닭 - return step, sheet_no, {"ok": False, "error": f"{type(exc).__name__}: {exc}"[:300]} + return step, job, {"ok": False, "error": f"{type(exc).__name__}: {exc}"[:300]} seconds = round(time.monotonic() - started, 2) - return step, sheet_no, {"ok": True, "s": seconds, "fresh": fresh, "empty": empty} + return step, job, {"ok": True, "s": seconds, "fresh": fresh, "empty": empty} # ── 진행 기록 · 정리 ───────────────────────────────────────────────────────── -def load_progress() -> dict[str, Any]: - path = store.TERRAIN_ROOT / PROGRESS_NAME - stamp = store._gpkg_stamp() +def progress_path() -> Path: + return store.TERRAIN_ROOT / f"_precompute_progress_{PC_NAME}.json" + + +def progress_files() -> list[Path]: + """모든 PC 기록(옛 한 벌 · 시놀로지 충돌 사본 포함).""" + return sorted(store.TERRAIN_ROOT.glob("_precompute_progress*.json")) + + +def _read_progress(path: Path, stamp: str | None) -> dict[str, Any] | None: try: saved = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): - saved = None - if not saved or saved.get("gpkg") != stamp or saved.get("cell_m") != CELL_M: - saved = {"gpkg": stamp, "cell_m": CELL_M, "steps": {}} + return None + if not isinstance(saved, dict) or saved.get("gpkg") != stamp or saved.get("cell_m") != CELL_M: + return None + return saved + + +def load_progress(every_pc: bool = False) -> dict[str, Any]: + """이 PC 기록 + 옛 한 벌의 성공(every_pc = 모든 PC 기록의 성공) — 쓰기는 이 PC 파일만.""" + stamp = store._gpkg_stamp() + own_path = progress_path() + saved = _read_progress(own_path, stamp) or {"gpkg": stamp, "cell_m": CELL_M, "steps": {}} for step in STEPS: saved["steps"].setdefault(step, {}) + others = progress_files() if every_pc else [store.TERRAIN_ROOT / PROGRESS_NAME] + for path in others: + other = None if path == own_path else _read_progress(path, stamp) + for step, records in (other or {}).get("steps", {}).items(): + mine = saved["steps"].get(step) + if mine is None: + continue + for job, info in records.items(): + if info.get("ok") and not mine.get(job, {}).get("ok"): + mine[job] = info return saved def save_progress(progress: dict[str, Any]) -> None: - atomic_write_json(store.TERRAIN_ROOT / PROGRESS_NAME, progress) + atomic_write_json(progress_path(), progress) def _alive(pid: int) -> bool: @@ -263,15 +390,18 @@ def _alive(pid: int) -> bool: return psutil.pid_exists(pid) -def clean_dead(root: Path | None = None) -> int: - """주인이 죽은 잠금 · 오래된 임시 파일을 지움(끊긴 일괄 셈이 남긴 것) → 지운 수.""" +def clean_dead(root: Path | None = None, part: Part | None = None) -> int: + """자기 몫 도엽의 주인이 죽은 잠금 · 오래된 임시 파일을 지움(끊긴 일괄 셈이 남긴 것) → 지운 수. + + 남의 몫 도엽 = 다른 PC 잠금(이 PC 에선 주인 번호가 없어 보임)이라 안 건드림. + """ root = root or store.TERRAIN_ROOT removed = 0 if not root.is_dir(): return 0 now = time.time() for folder in root.iterdir(): - if not folder.is_dir(): + if not folder.is_dir() or not in_part(folder.name, part): continue for path in folder.glob(".*"): try: @@ -288,47 +418,102 @@ def clean_dead(root: Path | None = None) -> int: return removed +def _execute(step: str, todo: list[str], workers: int, record, opts: dict[str, Any]) -> None: + """일감 차례대로 일꾼에 줌 — 상한 쓰면 전체 CPU 가 넘는 동안 새 일 안 줌(도는 것은 그대로).""" + part, cap, cpu = opts["part"], opts["cpu_cap"], opts["cpu"] + + def over() -> bool: + return cap is not None and cpu() > cap + + if workers <= 1: + for job in todo: + while over(): + time.sleep(CPU_SAMPLE_S) + record(run_step(step, job, part)) + return + e_cpus, e_workers = opts["e_cpus"], opts["e_workers"] + counter = multiprocessing.Value("i", 0) if e_cpus and e_workers else None + initargs = (str(store.TERRAIN_ROOT), counter, e_cpus, e_workers) + with ProcessPoolExecutor(workers, initializer=init_worker, initargs=initargs) as pool: + queue = list(reversed(todo)) + running: set = set() + try: + while queue or running: + while queue and len(running) < workers and not over(): + running.add(pool.submit(run_step, step, queue.pop(), part)) + if cap is not None: + break # 상한 쓸 때 한 번에 하나 — 다음 잼을 보고 또 + if not running: + time.sleep(CPU_SAMPLE_S) + continue + timeout = CPU_SAMPLE_S if cap is not None and queue else None + finished, running = wait(running, timeout=timeout, return_when=FIRST_COMPLETED) + for future in finished: + record(future.result()) + except BaseException: + pool.shutdown(wait=False, cancel_futures=True) + raise + + def run( sheets: list[str], steps: list[str], workers: int, progress: dict[str, Any], say=print, + *, + part: Part | None = None, + cpu_cap: float | None = None, + cpu=None, + e_cpus: list[int] | None = None, + e_workers: int = 0, + defer_wait: float = 0.0, ) -> dict[str, Any]: - """단계 차례대로 · 단계 안은 일꾼 여럿 · 성공 기록된 도엽은 건너뜀 → 단계별 벽시계 초.""" + """단계 차례대로 · 단계 안은 일꾼 여럿 · 성공 기록된 일감은 건너뜀 → 단계별 벽시계 초. + + sheets = 나누기 전 목록(이웃 몫에 씀) · 미룬 0차곡 = defer_wait 초 뒤 다시(0 = 안 기다림). + """ + if cpu_cap is not None and cpu is None: + cpu = CpuMeter() + opts = {"part": part, "cpu_cap": cpu_cap, "cpu": cpu, "e_cpus": e_cpus or []} + opts["e_workers"] = e_workers walls: dict[str, float] = {} for step in [s for s in STEPS if s in steps]: done = progress["steps"][step] - todo = [no for no in sheets if not done.get(no, {}).get("ok")] - say(f"[{step}] {len(todo)} 장 (건너뜀 {len(sheets) - len(todo)})") + jobs = dem_jobs(sheets, part) if step == "dem" else [n for n in sheets if in_part(n, part)] + todo = [job for job in jobs if not done.get(job, {}).get("ok")] + say(f"[{step}] {len(todo)} 장 (건너뜀 {len(jobs) - len(todo)} · 몫 {len(jobs)})") started = time.monotonic() saved_at = started count = 0 + waiting: list[str] = [] def record(result: tuple[str, str, dict[str, Any]]) -> None: nonlocal count, saved_at - _, no, info = result - done[no] = info + _, job, info = result + done[job] = info count += 1 - state = f"{info['s']} s" if info["ok"] else f"실패 {info['error']}" - say(f"[{step}] {count}/{len(todo)} {no} {state}") + if info.get("wait"): + waiting.append(job) + state = ( + f"{info['s']} s" + if info["ok"] + else f"{'기다림' if info.get('wait') else '실패'} {info['error']}" + ) + say(f"[{step}] {count}/{len(todo)} {job} {state}") if time.monotonic() - saved_at > SAVE_EVERY_S: save_progress(progress) saved_at = time.monotonic() try: - if workers <= 1: - for no in todo: - record(run_step(step, no)) - else: - with ProcessPoolExecutor(workers, initializer=below_normal) as pool: - futures = [pool.submit(run_step, step, no) for no in todo] - try: - for future in as_completed(futures): - record(future.result()) - except BaseException: - pool.shutdown(wait=False, cancel_futures=True) - raise + while True: + _execute(step, todo, workers, record, opts) + if not waiting or defer_wait <= 0: + break + save_progress(progress) + say(f"[{step}] 이웃 지형 기다림 {len(waiting)} 장 · {defer_wait:g} s 뒤 다시") + time.sleep(defer_wait) + todo, waiting[:], count = list(waiting), [], 0 finally: save_progress(progress) walls[step] = round(time.monotonic() - started, 1) @@ -341,12 +526,14 @@ def summary(sheets: list[str], progress: dict[str, Any]) -> dict[str, Any]: for step in STEPS: done = progress["steps"][step] infos = [done[no] for no in sheets if no in done] - failed = {no: done[no]["error"] for no in sheets if done.get(no, {}).get("ok") is False} + failed = { + no: done[no]["error"] + for no in sheets + if done.get(no, {}).get("ok") is False and not done[no].get("wait") + } fresh = [i["s"] for i in infos if i.get("ok") and i.get("fresh")] - sizes = [] + sizes = [] # 창고 실물(다른 PC 가 셈 · 시놀로지로 옴 포함) for no in sheets: - if not done.get(no, {}).get("ok"): - continue stem = f"{epsg_of(no)}_{CELL_M:g}m" folder = store.TERRAIN_ROOT / no total = 0 @@ -358,6 +545,8 @@ def summary(sheets: list[str], progress: dict[str, Any]) -> dict[str, Any]: "ok": sum(1 for i in infos if i.get("ok")), "fresh": len(fresh), "failed": failed, + "waiting": sum(1 for no in sheets if done.get(no, {}).get("wait")), + "stored": len(sizes), "s_median": round(float(np.median(fresh)), 1) if fresh else None, "s_mean": round(float(np.mean(fresh)), 1) if fresh else None, "s_max": round(float(np.max(fresh)), 1) if fresh else None, @@ -377,7 +566,22 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--workers", type=int, default=4) parser.add_argument("--status", action="store_true") parser.add_argument("--rebuild-index", action="store_true") + parser.add_argument("--part", default="", help="A-B/N — 도엽 번호 crc32 %% N 이 A~B 인 것만") + parser.add_argument("--cpu-cap", type=float, default=None, help="전체 CPU %% 상한") + parser.add_argument("--e-cores", default="", help="저전력 논리 CPU(예 12-19)") + parser.add_argument("--e-workers", type=int, default=0, help="그 CPU 로 지정할 일꾼 수") + parser.add_argument("--defer-wait", type=float, default=DEFER_WAIT_S) + parser.add_argument("--root", default="", help="창고 자리(시험용 임시 폴더)") args = parser.parse_args(argv) + try: + part = parse_part(args.part) + e_cpus = parse_cpus(args.e_cores) + except ValueError as exc: + parser.error(str(exc)) + if (args.e_workers and not e_cpus) or args.e_workers > args.workers: + parser.error("--e-workers 는 --e-cores 와 함께 · --workers 이하") + if args.root: + store.TERRAIN_ROOT = Path(args.root) steps = [s.strip() for s in args.steps.split(",") if s.strip()] unknown = [s for s in steps if s not in STEPS] if unknown: @@ -392,14 +596,39 @@ def main(argv: list[str] | None = None) -> int: sheets = pick_sheets( index, sheets=wanted or None, box=box, min_relief=args.min_relief, limit=args.limit or None ) - print(f"도엽 {len(sheets)} 장 (목록 {len(index)} 장) · 단계 {steps} · 일꾼 {args.workers}") - progress = load_progress() - if not args.status: - print(f"치운 잠금 · 임시 파일 {clean_dead()}") + print( + f"도엽 {len(sheets)} 장 (목록 {len(index)} 장) · 단계 {steps} · 일꾼 {args.workers}" + f" · 몫 {args.part or '전부'} · PC {PC_NAME} · 상한 {args.cpu_cap}" + f" · 지정 CPU {e_cpus} × {args.e_workers}" + ) + mine = [no for no in sheets if in_part(no, part)] + if args.status: + out = summary(mine, load_progress(every_pc=True)) + stamp = store._gpkg_stamp() + out["pc"] = {} + for path in progress_files(): + recs = (_read_progress(path, stamp) or {}).get("steps", {}) + out["pc"][path.name] = { + step: sum(1 for i in recs.get(step, {}).values() if i.get("ok")) for step in STEPS + } + else: + progress = load_progress() + print(f"치운 잠금 · 임시 파일 {clean_dead(part=part)}") started = time.monotonic() - walls = run(sheets, steps, args.workers, progress) + walls = run( + sheets, + steps, + args.workers, + progress, + part=part, + cpu_cap=args.cpu_cap, + e_cpus=e_cpus, + e_workers=args.e_workers, + defer_wait=args.defer_wait, + ) print(f"벽시계 {walls} · 전체 {time.monotonic() - started:.1f} s") - print(json.dumps(summary(sheets, progress), ensure_ascii=False, indent=1)) + out = summary(mine, progress) + print(json.dumps(out, ensure_ascii=False, indent=1)) return 0