refactor(tooling): drop --shard from gemma_naming.py (#833)
Parallelism via two concurrent sr-voice subprocesses does not work on this ROCm + llama-cpp-rs setup — launching a second instance poisons the first one's GPU context (both fall back to 0% GPU / 50% CPU busy-loop and stop making progress). Verified empirically: single shard runs cleanly at ~1.2s/feature, two shards deadlock. Without a working parallel path, --shard is dead weight. Resume semantics were already free: the pipeline skips bodies whose markers.json has non-empty name fields (preserved path), so a killed run re-starts just by re-running the same command. Simplifications: - Remove --shard argument and all slicing logic. - Remove banner_shard / shard_offset / shard_n / shard_m plumbing. - Rename internal total_shard_systems → total_systems. - Default --log path is now .tmp/gemma_naming.log (was conditional on --shard). Pass `--log -` to disable file logging. - Startup banner now prints a one-line resume reminder so the user can see at a glance that a killed run is recoverable.
This commit is contained in:
Binary file not shown.
@@ -1185,22 +1185,12 @@ def main():
|
||||
"'Meridian') may appear across the full run before dedup "
|
||||
"starts rejecting it. 0 = disabled. Default: 20.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--shard",
|
||||
default="0/1",
|
||||
help="Process a slice of the body list for parallel runs. Format "
|
||||
"N/M: worker N of M total. Each worker walks bodies in hop "
|
||||
"order, picking every Mth body starting at offset N. Example: "
|
||||
"run `--shard 0/2` in one terminal and `--shard 1/2` in "
|
||||
"another to split the work in half. Default: 0/1 (all bodies).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--log",
|
||||
default=None,
|
||||
default=str(REPO_ROOT / ".tmp" / "gemma_naming.log"),
|
||||
help="Path to a log file. Every status line is written to both "
|
||||
"stdout and the log. Defaults to "
|
||||
".tmp/gemma_naming.shard{N}of{M}.log when --shard is not "
|
||||
"trivially 0/1; disabled otherwise. Pass '-' to disable.",
|
||||
"stdout and the log. Default: .tmp/gemma_naming.log. "
|
||||
"Pass '-' to disable file logging.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--verbose",
|
||||
@@ -1210,31 +1200,7 @@ def main():
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
# Parse shard spec.
|
||||
try:
|
||||
shard_n_str, shard_m_str = args.shard.split("/", 1)
|
||||
shard_n = int(shard_n_str)
|
||||
shard_m = int(shard_m_str)
|
||||
if shard_m < 1 or not (0 <= shard_n < shard_m):
|
||||
raise ValueError
|
||||
except ValueError:
|
||||
print(
|
||||
f"error: invalid --shard '{args.shard}'. "
|
||||
"Expected N/M with 0 <= N < M, e.g. 0/2.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
# Default log path: enable for non-trivial shards, disable for 0/1.
|
||||
if args.log is None:
|
||||
if shard_m > 1:
|
||||
log_path = REPO_ROOT / ".tmp" / f"gemma_naming.shard{shard_n}of{shard_m}.log"
|
||||
else:
|
||||
log_path = None
|
||||
elif args.log == "-":
|
||||
log_path = None
|
||||
else:
|
||||
log_path = Path(args.log)
|
||||
log_path = None if args.log == "-" else Path(args.log)
|
||||
|
||||
db_path = Path(args.db)
|
||||
if not db_path.exists():
|
||||
@@ -1278,50 +1244,41 @@ def main():
|
||||
ensure_atlas_schema(conn)
|
||||
|
||||
hop_order = load_body_hop_order(conn)
|
||||
all_markers = discover_bodies(args.body, args.limit, hop_order)
|
||||
if not all_markers:
|
||||
markers_paths = discover_bodies(args.body, args.limit, hop_order)
|
||||
if not markers_paths:
|
||||
log(f"error: no markers.json found (body={args.body})")
|
||||
conn.close()
|
||||
sys.exit(1)
|
||||
|
||||
# Apply shard slicing AFTER discovery + sort. Shard 0/2 gets indices
|
||||
# [0, 2, 4, …], shard 1/2 gets [1, 3, 5, …] — interleaved in hop
|
||||
# order so both shards advance core → frontier in parallel.
|
||||
markers_paths = all_markers[shard_n::shard_m]
|
||||
if not markers_paths:
|
||||
log(f"error: shard {shard_n}/{shard_m} contains no bodies "
|
||||
f"(full set = {len(all_markers)}).")
|
||||
conn.close()
|
||||
sys.exit(1)
|
||||
|
||||
# Count distinct systems the shard will touch so the progress lines
|
||||
# can report `systems X/Y done` alongside `bodies X/Y done`.
|
||||
shard_systems = sorted({_body_id_from_path(p)[1] for p in markers_paths})
|
||||
total_shard_systems = len(shard_systems)
|
||||
# Count distinct systems so the progress lines can report
|
||||
# `systems X/Y done` alongside `bodies X/Y done`.
|
||||
total_systems = len({_body_id_from_path(p)[1] for p in markers_paths})
|
||||
seen_systems: set[str] = set()
|
||||
|
||||
first_hop = hop_order.get(_body_id_from_path(markers_paths[0])[0], (99, ""))[0]
|
||||
last_hop = hop_order.get(_body_id_from_path(markers_paths[-1])[0], (99, ""))[0]
|
||||
|
||||
banner_shard = f"{shard_n}/{shard_m}" if shard_m > 1 else "single"
|
||||
log.raw("")
|
||||
log.raw(f" Gemma 2 Batch Naming Pipeline (#833)")
|
||||
log.raw(f" DB: {db_path}")
|
||||
log.raw(f" Mode: {'MOCK' if args.mock else 'REAL'}")
|
||||
log.raw(f" Shard: {banner_shard}")
|
||||
log.raw(f" sr-voice: {MOCK_STDIO if args.mock else sr_voice_bin}")
|
||||
if not args.mock:
|
||||
log.raw(f" model: {model_path}")
|
||||
log.raw(f" seed: {args.seed} refresh: every {args.refresh} requests "
|
||||
f"stem-cap: {args.stem_cap}")
|
||||
log.raw(f" {len(markers_paths)} markers.json files in this shard "
|
||||
f"(full set: {len(all_markers)})")
|
||||
log.raw(f" {len(markers_paths)} markers.json files to process")
|
||||
log.raw(f" hop {first_hop} → hop {last_hop}, core-first ordering")
|
||||
log.raw(f" {total_shard_systems} distinct systems")
|
||||
log.raw(f" {total_systems} distinct systems")
|
||||
log.raw(f" blocklist: {len(blocklist)} Earth-major entries")
|
||||
if log.log_path is not None:
|
||||
log.raw(f" log: {log.log_path}")
|
||||
log.raw("")
|
||||
log.raw(
|
||||
" Resume: re-run this command any time. Bodies whose markers.json "
|
||||
"already has non-empty name fields will be skipped (preserved path)."
|
||||
)
|
||||
log.raw("")
|
||||
|
||||
corpus: dict[tuple[str, str], set[str]] = {}
|
||||
stem_counts: dict[str, int] = {}
|
||||
@@ -1362,7 +1319,7 @@ def main():
|
||||
elapsed = time.time() - t0
|
||||
|
||||
body_progress = f"body {i+1}/{len(markers_paths)}"
|
||||
sys_progress = f"sys {len(seen_systems)}/{total_shard_systems}"
|
||||
sys_progress = f"sys {len(seen_systems)}/{total_systems}"
|
||||
hop = hop_order.get(body_id, (99, ""))[0]
|
||||
progress = f"{body_progress} {sys_progress} hop={hop}"
|
||||
|
||||
@@ -1431,7 +1388,7 @@ def main():
|
||||
eta = "--"
|
||||
log(
|
||||
f" >> CHECKPOINT bodies {i+1}/{len(markers_paths)} "
|
||||
f"systems {len(seen_systems)}/{total_shard_systems} "
|
||||
f"systems {len(seen_systems)}/{total_systems} "
|
||||
f"names {cum} {rate:.1f}/s eta {eta}"
|
||||
)
|
||||
|
||||
@@ -1445,10 +1402,9 @@ def main():
|
||||
elapsed_total = time.time() - t_total
|
||||
log.raw("")
|
||||
log.raw(f" Done: {elapsed_total:.0f}s ({elapsed_total/60:.1f} min)")
|
||||
log.raw(f" shard: {banner_shard}")
|
||||
log.raw(f" bodies processed: {len(markers_paths)}")
|
||||
log.raw(f" bodies touched: {bodies_touched}")
|
||||
log.raw(f" systems seen: {len(seen_systems)}/{total_shard_systems}")
|
||||
log.raw(f" systems seen: {len(seen_systems)}/{total_systems}")
|
||||
log.raw(f" cities named: {totals['cities']}")
|
||||
log.raw(f" rivers named: {totals['rivers']}")
|
||||
log.raw(f" oceans named: {totals['oceans']}")
|
||||
|
||||
Reference in New Issue
Block a user