feat(도구): 전국 미리 계산 두 PC 나눔 — 몫 · PC 별 기록 · CPU 상한 · 저전력 코어 지정 (PLAN 618-1 · 요구 507~509)

- --part A-B/N = 도엽 번호 crc32 % N 몫 · 지형 단계가 0차곡 이웃(목록 밖 · 띠 넘음 포함)까지 몫대로 셈
- 0차곡 = 남의 몫 이웃 지형이 창고에 없으면 미룸 · --defer-wait 초 뒤 다시
- 진행 기록 = _precompute_progress_<컴퓨터 이름>.json · 옛 한 벌은 성공만 읽어 이어 받음 · --status 모든 PC 합침 · 창고 실물 수
- --cpu-cap = 전체 CPU 평균이 넘으면 새 도엽 안 줌 · --e-cores --e-workers = 일꾼 앞선 몇 명만 그 논리 CPU 지정
- 죽은 잠금 치움 = 자기 몫 도엽만 · --root 시험용 창고 자리
- test_612: 지형 단계 이웃 일감 덧 · 기록 파일 이름 바뀜에 맞춤

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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  # 차례 고정
+    # 차례 고정 · 지형 단계는 0차곡 이웃(3x3)까지 셈(618-1)
-    saved = json.loads((root / tool.PROGRESS_NAME).read_text(encoding="utf-8"))
+    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
 

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

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:
+def epsg_of(job: str) -> int:
-    return tm_epsg_for_lon(sheet5k_center(sheet_no)[1])
+    """도엽(또는 `도엽@좌표계` 일감)의 좌표계 — 기본 = 도엽 가운데 경도 띠."""
+    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]]:
+def dem_jobs(sheets: list[str], part: Part | None) -> list[str]:
-    """도엽 하나 · 단계 하나 — 공용 함수만 부름 · fresh = 이번에 새로 셈."""
+    """지형 단계 일감 — 목록 도엽 뒤 0차곡이 쓸 3x3 이웃(그 가운데 도엽 좌표계) · 자기 몫만.
-    epsg = epsg_of(sheet_no)
+
+    이웃이 가운데와 띠가 다르면 `도엽@좌표계`(옮겨 담기) — 목록 밖 이웃도 한 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]:
+def progress_path() -> Path:
-    path = store.TERRAIN_ROOT / PROGRESS_NAME
+    return store.TERRAIN_ROOT / f"_precompute_progress_{PC_NAME}.json"
-    stamp = store._gpkg_stamp()
+
+
+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
+        return None
-    if not saved or saved.get("gpkg") != stamp or saved.get("cell_m") != CELL_M:
+    if not isinstance(saved, dict) or saved.get("gpkg") != stamp or saved.get("cell_m") != CELL_M:
-        saved = {"gpkg": stamp, "cell_m": CELL_M, "steps": {}}
+        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")]
+        jobs = dem_jobs(sheets, part) if step == "dem" else [n for n in sheets if in_part(n, part)]
-        say(f"[{step}] {len(todo)} 장 (건너뜀 {len(sheets) - len(todo)})")
+        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
+            _, job, info = result
-            done[no] = info
+            done[job] = info
             count += 1
-            state = f"{info['s']} s" if info["ok"] else f"실패 {info['error']}"
+            if info.get("wait"):
-            say(f"[{step}] {count}/{len(todo)} {no} {state}")
+                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:
+            while True:
-                for no in todo:
+                _execute(step, todo, workers, record, opts)
-                    record(run_step(step, no))
+                if not waiting or defer_wait <= 0:
-            else:
+                    break
-                with ProcessPoolExecutor(workers, initializer=below_normal) as pool:
+                save_progress(progress)
-                    futures = [pool.submit(run_step, step, no) for no in todo]
+                say(f"[{step}] 이웃 지형 기다림 {len(waiting)} 장 · {defer_wait:g} s 뒤 다시")
-                    try:
+                time.sleep(defer_wait)
-                        for future in as_completed(futures):
+                todo, waiting[:], count = list(waiting), [], 0
-                            record(future.result())
-                    except BaseException:
-                        pool.shutdown(wait=False, cancel_futures=True)
-                        raise
         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}")
+    print(
-    progress = load_progress()
+        f"도엽 {len(sheets)} 장 (목록 {len(index)} 장) · 단계 {steps} · 일꾼 {args.workers}"
-    if not args.status:
+        f" · 몫 {args.part or '전부'} · PC {PC_NAME} · 상한 {args.cpu_cap}"
-        print(f"치운 잠금 · 임시 파일 {clean_dead()}")
+        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