From 05a69ae4dc27724d122cd7cd7b03e4730c45298b Mon Sep 17 00:00:00 2001 From: buzzer-re <22428720+buzzer-re@users.noreply.github.com> Date: Wed, 14 Oct 2026 11:24:19 -0300 Subject: [PATCH] Export folder binaries in parallel - --parallel-files N (-pf) runs N binaries at once, each in its own process with -j workers. Auto: one per CPU (at most 16), capped by free memory; with -j auto each file's workers are capped at its CPU share. - NativeRunner schedules the parallel exports: largest first, a new one only when a slot is free and memory allows it on top of exports still ramping up. An export killed by a signal while others ran is retried once, alone, from its checkpoint. APK/.NET native libraries still run one at a time. - Folder state saves are locked; Ctrl-C kills every running export and leaves it pending, so a rerun resumes it. - The folder display shows one row per running binary. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 5 +- README.md | 3 +- src/tocode/apk_native.py | 227 +++++++++++++++++++---- src/tocode/cli.py | 13 +- src/tocode/folder.py | 247 +++++++++++++++++-------- src/tocode/parallel.py | 43 +++++ tests/test_folder.py | 5 +- tests/test_folder_parallel.py | 334 ++++++++++++++++++++++++++++++++++ 8 files changed, 763 insertions(+), 114 deletions(-) create mode 100644 tests/test_folder_parallel.py diff --git a/AGENTS.md b/AGENTS.md index 6feb38b..a77cb75 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,7 @@ This repository contains ToCode, a Python-only binary exporter. ToCode takes one - `src/tocode/dotnet.py`: .NET export pipeline; `dotnet_metadata.py`: .NET JSON documents and the generated `AGENTS.md`. - `src/tocode/apk.py`: APK export pipeline (extraction, decompile pool, resource decoding, metadata); `apk_metadata.py`: manifest parsing and Android JSON documents; `apk_native.py`: extraction and background native export of every `.so`. - `src/tocode/links/`: pure-Python link facts for folder exports (no backend involved): `sniff.py` file kinds; `binfmt_elf.py`, `binfmt_pe.py`, `binfmt_macho.py` (thin/fat, chained fixups, bind opcodes, export trie), `binfmt_dotnet.py` (dnlib-free metadata tables, signatures, P/Invoke); `inventory.py` folder scan; `resolve.py` loader rules per platform; `bind.py` symbol binding; `names.py` backend-name matching keys. -- `src/tocode/folder.py`: folder export orchestration (plan, job keys and resume state in `.tocode-folder/`, one isolated process per binary via `NativeRunner`, the two-bar progress display); `folder_links.py`: joins link facts to the exports (`mappings.json`, per-file enrichment); `folder_annotate.py`: line-stable `tocode-link:` source comments; `folder_docs.py`: master `AGENTS.md`/JSON. +- `src/tocode/folder.py`: folder export orchestration (plan, job keys and resume state in `.tocode-folder/`, isolated processes per binary run in parallel by `NativeRunner`, the overall bar plus one row per running binary); `folder_links.py`: joins link facts to the exports (`mappings.json`, per-file enrichment); `folder_annotate.py`: line-stable `tocode-link:` source comments; `folder_docs.py`: master `AGENTS.md`/JSON. - `src/tocode/exporter.py`: project writer, function rendering, worker-session rendering, generated export `AGENTS.md`. - `src/tocode/metadata.py`: JSON metadata and triage documents. - `src/tocode/cluster.py`: call-graph clustering. @@ -100,7 +100,8 @@ Framework/runtime assemblies and native runtime libraries inside bundles and pac - Do not commit generated export directories such as `here/`, `ls/`, `*_decompiler/`, or `*_decompiled/`. - Folder links must come only from import/load tables and loader rules; never add name-similarity or string-based (`dlopen`/`LoadLibrary`) links. `links/names.py` keys only locate a backend's function for a symbol the tables already bound. - Source annotations must never add or remove lines (`folder_annotate` asserts it); line ranges in the JSON depend on it. -- Folder runs print one overall bar and one current-binary bar; nested export logs go to `tocode.log` only. Keep per-file terminal output to failures. +- Folder runs export `--parallel-files` binaries at once (auto: one per CPU, capped by free memory at `TOCODE_NATIVE_MIN_FREE_MB` each), largest first, each with `-j` workers (auto: capped at its CPU share through `TOCODE_MAX_DECOMPILE_WORKERS` in the child). `NativeRunner` starts a file only when a slot is free and memory allows it on top of files still ramping up; an export killed by a signal while others ran is retried once, alone. `save_state` is locked (exports finish on their own threads); Ctrl-C kills every running export and leaves it pending, never failed, so a rerun resumes it from its own checkpoint. APK/.NET native libraries still run one at a time (`parallel=1`). +- Folder runs print one overall bar and one row per running binary; nested export logs go to `tocode.log` only. Keep per-file terminal output to failures. ## Dependencies diff --git a/README.md b/README.md index c1963e7..9de6e14 100644 --- a/README.md +++ b/README.md @@ -159,13 +159,14 @@ Game_decompiled/ Inside each export, `imports.json` and `functions.json` carry the same links, and calls into other files get a trailing comment in the decompiled source: `/* tocode-link: foo -> lib/libfoo.so/src/raw/app/x.c:23-45 */`. The comments never add lines, so all line numbers stay valid. -Binaries are exported one at a time, executables first and then the libraries they load. Each runs in its own process, so a crash or out-of-memory kill only fails that file. Re-running the same command skips finished binaries and resumes the rest. Bundled runtimes (VC++/MinGW/libstdc++ copies, the .NET framework) are linked but only decompiled with `--include-framework`. Universal Mach-O files use the main executable's architecture (`--arch` to override). +Several binaries are exported at once, each in its own process, so a crash or out-of-memory kill only fails that file. `--parallel-files N` sets how many run at once and `-j N` the workers inside each (default: from your CPUs and free memory). Re-running the same command skips finished binaries and resumes the rest. Bundled runtimes (VC++/MinGW/libstdc++ copies, the .NET framework) are linked but only decompiled with `--include-framework`. Universal Mach-O files use the main executable's architecture (`--arch` to override). ```bash tocode --plan Game/ # list binaries, export order, and links; decompile nothing tocode Game/ --only 'bin/*' --only '**/libengine.so' # export a subset; the rest stays linked tocode Game/ --exclude '*.dll' # skip matches tocode Game/ --links-only # recompute links over existing exports +tocode Game/ -pf 8 -j 2 # 8 binaries at once (--parallel-files), 2 workers each ``` ### Using diff --git a/src/tocode/apk_native.py b/src/tocode/apk_native.py index c612920..f607603 100644 --- a/src/tocode/apk_native.py +++ b/src/tocode/apk_native.py @@ -265,6 +265,10 @@ def run() -> None: threading.Thread(target=run, name="tocode-parent-watch", daemon=True).start() +class ExportKilledError(ToCodeError): + """An isolated export process was killed by a signal (usually out of memory).""" + + _active_children: set[int] = set() _active_lock = threading.Lock() @@ -319,7 +323,7 @@ def collect_child_result( kill_process_group(process.pid) code = process.exitcode if code is not None and code < 0: - raise ToCodeError( + raise ExportKilledError( f"export process was killed by signal {-code} " "(out of memory?); try -j 1 or --no-native" ) @@ -491,9 +495,29 @@ def status(self) -> str: return self.last_line +# A file that just started has not loaded its binary yet: count this much free +# memory as already taken by it for this long when deciding to start another. +RAMP_SECONDS = 20.0 + + +@dataclass(slots=True) +class RunningExport: + lib: NativeLib + started: float + tail: NestedLogTail | None = None + + @dataclass class NativeRunner: - """Runs the native exports on a background thread.""" + """Runs the native exports on a background thread. + + Up to ``parallel`` exports run at once, each in its own process (the + exporter blocks a thread per export). A new export starts only when a slot + is free and the host has ``min_free_mb`` of free memory for it on top of + what recently started exports have not claimed yet. An export killed by a + signal (the OOM killer) while others ran beside it is retried once, alone; + it resumes from its own checkpoint. + """ libs: list[NativeLib] root: Path @@ -510,30 +534,49 @@ class NativeRunner: # Folder exports draw their own progress display: per-library lines and # the nested logs then go to the log file only. quiet: bool = False + parallel: int = 1 + ramp_seconds: float = RAMP_SECONDS current: NativeLib | None = None waiting: tuple[NativeLib, int] | None = None current_started: float = 0.0 current_tail: NestedLogTail | None = None + _running: dict[str, RunningExport] = field(default_factory=dict) + _lock: threading.Lock = field(default_factory=threading.Lock) + _changed: threading.Event = field(default_factory=threading.Event) + _overlapped: set[str] = field(default_factory=set) + _retries: set[str] = field(default_factory=set) + _alone: set[str] = field(default_factory=set) + _decisions: dict[str, bool] = field(default_factory=dict) + + def running(self) -> list[RunningExport]: + """The exports running now, oldest first.""" + with self._lock: + return sorted(self._running.values(), key=lambda item: item.started) def describe_current(self) -> str | None: """One line about what the native thread is doing right now.""" waiting = self.waiting - if waiting is not None: + running = self.running() + if waiting is not None and not running: pending, available = waiting return ( f"native: waiting for memory before {pending.name} ({pending.abi}): " f"{available} MB free, need {self.min_free_mb} MB " "(set TOCODE_NATIVE_MIN_FREE_MB to change)" ) - lib = self.current - if lib is None: + if not running: return None - elapsed = time.monotonic() - self.current_started - detail = self.current_tail.status() if self.current_tail else "running" done = sum(1 for item in self.libs if item.status == "done") finished = sum(1 for item in self.libs if item.status not in {"pending"}) + now = time.monotonic() + parts = [ + f"{item.lib.name} ({item.lib.abi}) " + f"{item.tail.status() if item.tail else 'running'}, " + f"{now - item.started:.0f}s elapsed" + for item in running + ] return ( - f"native: {lib.name} ({lib.abi}) {detail}, {elapsed:.0f}s elapsed; " + f"native: {'; '.join(parts)}; " f"{finished}/{len(self.libs)} libraries finished, {done} exported" ) @@ -549,18 +592,136 @@ def join(self) -> None: def stop(self) -> None: self.stop_event.set() + self._changed.set() terminate_active_native_children() + def claim_retry(self, lib: NativeLib, exc: BaseException) -> bool: + """Whether a failed export is retried alone (decided once per attempt). + + Only a kill by a signal while other exports ran beside it qualifies, + and only once: alone, the export has the memory the others were holding. + """ + with self._lock: + decided = self._decisions.get(lib.entry) + if decided is not None: + return decided + retry = ( + not self.stop_event.is_set() + and isinstance(exc, ExportKilledError) + and lib.entry in self._overlapped + and lib.entry not in self._retries + ) + if retry: + self._retries.add(lib.entry) + self._alone.add(lib.entry) + self._decisions[lib.entry] = retry + return retry + def _run(self) -> None: - for lib in self.libs: + queue = list(self.libs) + threads: list[threading.Thread] = [] + while True: + threads = [thread for thread in threads if thread.is_alive()] if self.stop_event.is_set(): - lib.status = "skipped: interrupted" + break + with self._lock: + running = len(self._running) + queue[:0] = [lib for lib in self._requeued() if lib not in queue] + if not queue: + if running == 0: + break + self._wait_for_change() + continue + lib = queue[0] + with self._lock: + alone_running = any(entry in self._alone for entry in self._running) + if ( + running >= max(1, self.parallel) + or alone_running + or (lib.entry in self._alone and running > 0) + ): + self._wait_for_change() + continue + if not self._memory_allows(lib, running): continue - self._wait_for_memory(lib) if self.stop_event.is_set(): + break + queue.pop(0) + self._launch(lib, threads) + for thread in threads: + thread.join() + for lib in self.libs: + if lib.status == "pending": lib.status = "skipped: interrupted" - continue - self._export_one(lib) + + def _requeued(self) -> list[NativeLib]: + """Exports that failed and are claimed for a retry (caller holds the lock).""" + return [ + lib + for lib in self.libs + if lib.entry in self._alone + and lib.status == "pending" + and lib.entry not in self._running + and lib.seconds is not None + ] + + def _wait_for_change(self) -> None: + self._changed.wait(self.poll_seconds) + self._changed.clear() + + def _memory_allows(self, lib: NativeLib, running: int) -> bool: + """True when ``lib`` may start now; otherwise waits a little, returns False.""" + if self.min_free_mb <= 0: + return True + if running == 0: + # Nothing else runs: wait for memory up to the limit, then go anyway. + self._wait_for_memory(lib) + return True + now = time.monotonic() + with self._lock: + ramping = sum( + 1 + for item in self._running.values() + if now - item.started < self.ramp_seconds + ) + available = self.available_memory() + if available is None or available >= self.min_free_mb * (1 + ramping): + self.waiting = None + return True + self.waiting = (lib, available) + self._wait_for_change() + return False + + def _launch(self, lib: NativeLib, threads: list[threading.Thread]) -> None: + self.waiting = None + lib.status = "pending" + out_dir = lib.export_dir or native_export_dir(self.root, lib) + lib.export_dir = out_dir + tail = ( + NestedLogTail( + out_dir / "tocode.log", lib, self.progress, echo=not self.quiet + ) + if self.mirror_logs + else None + ) + item = RunningExport(lib=lib, started=time.monotonic(), tail=tail) + with self._lock: + if self._running: + self._overlapped.add(lib.entry) + self._overlapped.update(self._running) + self._running[lib.entry] = item + self._decisions.pop(lib.entry, None) + self.current = lib + self.current_started = item.started + self.current_tail = tail + thread = threading.Thread( + target=self._export_one, + args=(item,), + name=f"tocode-native-{lib.stem}", + daemon=True, + ) + threads.append(thread) + thread.start() def _wait_for_memory(self, lib: NativeLib) -> None: if self.min_free_mb <= 0: @@ -586,35 +747,28 @@ def _wait_for_memory(self, lib: NativeLib) -> None: f"native: waited {waited:.0f}s for memory before {lib.name}" ) - def _export_one(self, lib: NativeLib) -> None: + def _export_one(self, item: RunningExport) -> None: # Folder exports place each library themselves; APK/.NET use native/. + lib = item.lib out_dir = lib.export_dir or native_export_dir(self.root, lib) - lib.export_dir = out_dir - log_path = out_dir / "tocode.log" child = Progress(enabled=False) - child.set_log_path(log_path) + child.set_log_path(out_dir / "tocode.log") say = self.progress.file_log if self.quiet else self.progress.log say( f"native: {lib.name} ({lib.abi}) export started -> " f"{_rel(out_dir, self.root)}" ) - started = time.monotonic() - tail = ( - NestedLogTail(log_path, lib, self.progress, echo=not self.quiet) - if self.mirror_logs - else None - ) - self.current = lib - self.current_started = started - self.current_tail = tail + tail = item.tail if tail is not None: tail.start() + retry = False try: backend, decompiler, functions, failures = self.exporter( lib, out_dir, child ) except ToCodeError as exc: - lib.status = f"failed: {exc}" + retry = self.claim_retry(lib, exc) + lib.status = "pending" if retry else f"failed: {exc}" except Exception as exc: # pragma: no cover - backend crash isolation lib.status = f"failed: {type(exc).__name__}: {exc}" else: @@ -626,10 +780,21 @@ def _export_one(self, lib: NativeLib) -> None: finally: if tail is not None: tail.stop() - self.current = None - self.current_tail = None - lib.seconds = time.monotonic() - started - if lib.status == "done": + lib.seconds = time.monotonic() - item.started + with self._lock: + self._running.pop(lib.entry, None) + if self.current is lib: + self.current = None + self.current_tail = None + if not retry: + self._alone.discard(lib.entry) + self._changed.set() + if retry: + say( + f"native: {lib.name} ({lib.abi}) was killed while other exports ran; " + "retrying it alone" + ) + elif lib.status == "done": say( f"native: {lib.name} ({lib.abi}) exported with {lib.backend} " f"in {lib.seconds:.1f}s: functions={lib.function_count} " diff --git a/src/tocode/cli.py b/src/tocode/cli.py index 9754e40..d67769f 100644 --- a/src/tocode/cli.py +++ b/src/tocode/cli.py @@ -224,7 +224,15 @@ def build_parser() -> argparse.ArgumentParser: "--jobs", type=parse_jobs, default=None, - help="Worker sessions for decompilation: positive integer or 'auto' (default: auto).", + help="Worker sessions for decompilation: positive integer or 'auto' (default: auto). In a folder export, the workers of each binary.", + ) + parser.add_argument( + "-pf", + "--parallel-files", + type=parse_jobs, + default=None, + metavar="N", + help="Folder input: binaries exported at once, each in its own process with -j workers: positive integer or 'auto' (default: auto, from the CPU count and free memory).", ) parser.add_argument( "--tree", @@ -343,6 +351,8 @@ def main(argv: list[str] | None = None) -> int: ) if binary.is_dir(): return _run_folder(binary, args=args, progress=progress) + if args.parallel_files is not None: + parser.error("--parallel-files applies to folder input") if is_apk_input(binary): return _run_apk(binary, args=args, progress=progress, parser=parser) if not args.as_native and binary.is_file() and is_dotnet_input(binary): @@ -517,6 +527,7 @@ def ensure_dotnet() -> bool: links_only=args.links_only, restart=args.restart, jobs=args.jobs, + parallel_files=args.parallel_files, native_options=_native_options(args), ) try: diff --git a/src/tocode/folder.py b/src/tocode/folder.py index a190330..853f9a8 100644 --- a/src/tocode/folder.py +++ b/src/tocode/folder.py @@ -25,10 +25,12 @@ import re import shutil import sys +import threading import time from typing import Any, Callable from .apk_native import ( + ExportKilledError, NativeExporter, NativeLib, NativeOptions, @@ -46,7 +48,7 @@ from .links.inventory import JOB_APK, JOB_DOTNET, JOB_NATIVE, Entry, Inventory, scan from .metadata import write_json from .naming import clean_path_component -from .parallel import available_memory_mb +from .parallel import available_memory_mb, choose_parallel_files, configured_job_limit from .progress import Progress STATE_DIR = ".tocode-folder" @@ -80,6 +82,8 @@ class FolderOptions: links_only: bool = False restart: bool = False jobs: int | None = None + # Binaries exported at once (None: auto, from CPUs and free memory). + parallel_files: int | None = None native_options: NativeOptions = field(default_factory=NativeOptions) @@ -303,20 +307,27 @@ def load_state(master: Path) -> dict[str, Any]: return payload +# Parallel exports finish on their own threads; each saves the state. +_state_lock = threading.Lock() + + def save_state(master: Path, jobs: list[FolderJob], state: dict[str, Any]) -> None: - recorded = state.setdefault("jobs", {}) - for job in jobs: - if job.status == DONE or job.status.startswith("failed"): - recorded[job.relpath] = { - "key": job.key, - "status": job.status, - "backend": job.backend, - "decompiler": job.decompiler, - "function_count": job.function_count, - "failure_count": job.failure_count, - "seconds": round(job.seconds, 1) if job.seconds is not None else None, - } - write_json(state_path(master), state) + with _state_lock: + recorded = state.setdefault("jobs", {}) + for job in jobs: + if job.status == DONE or job.status.startswith("failed"): + recorded[job.relpath] = { + "key": job.key, + "status": job.status, + "backend": job.backend, + "decompiler": job.decompiler, + "function_count": job.function_count, + "failure_count": job.failure_count, + "seconds": round(job.seconds, 1) + if job.seconds is not None + else None, + } + write_json(state_path(master), state) def apply_state(jobs: list[FolderJob], state: dict[str, Any]) -> None: @@ -357,6 +368,9 @@ class JobSettings: native: bool origin: str restart: bool + # Decompile workers this file may use (its share of the CPUs when several + # files run at once and -j is auto); None leaves the export's own choice. + max_workers: int | None = None def run_folder_job( @@ -421,6 +435,10 @@ def _folder_job_child( progress = Progress(enabled=False) progress.set_log_path(log_path) _redirect_output(log_path) + if settings.max_workers is not None: + os.environ["TOCODE_MAX_DECOMPILE_WORKERS"] = str( + min(settings.max_workers, configured_job_limit()) + ) try: result = run_folder_job(lib, out_dir, settings, progress) except ToCodeError as exc: @@ -640,8 +658,25 @@ def _run_jobs( progress.log(f"{reused} binaries already exported and unchanged; skipping them") if not pending: return + min_free = native_min_free_mb() + files, max_workers = choose_parallel_files( + pending=len(pending), + requested_files=options.parallel_files, + requested_jobs=options.jobs, + available_memory_mb=available_memory_mb(), + file_memory_mb=min_free, + ) + if files == 1: + how = "one at a time, in load order" + else: + workers = ( + f"-j {options.jobs}" + if options.jobs is not None + else f"up to {max_workers} workers" + ) + how = f"{files} at a time ({workers} each), largest first" progress.log( - f"Exporting {len(pending)} binaries one at a time, in load order " + f"Exporting {len(pending)} binaries {how} " f"(backend: {options.native_options.backend}; per-binary logs in tocode.log)" ) by_entry: dict[str, FolderJob] = {} @@ -661,6 +696,10 @@ def _run_jobs( ) by_entry[job.relpath] = job libs.append(lib) + if files > 1: + # The biggest binaries take longest: starting them first keeps one + # from running alone at the end while every other slot sits idle. + libs.sort(key=lambda item: item.size, reverse=True) native = options.native_options native_options = NativeOptions( backend=native.backend, @@ -684,6 +723,7 @@ def settings_for(lib: NativeLib) -> JobSettings: native=options.native, origin=lib.origin or "", restart=options.restart, + max_workers=max_workers, ) run_one = exporter or isolated_exporter(settings_for) @@ -697,8 +737,16 @@ def tracked( try: result = run_one(lib, out_dir, child) except BaseException as exc: - # Killed by Ctrl-C: stays pending and resumes from its checkpoint. - if not (holder and holder[0].stop_event.is_set()): + # Killed by Ctrl-C, or killed beside other exports and retried + # alone: stays pending and resumes from its checkpoint. + runner = holder[0] if holder else None + interrupted = runner is not None and runner.stop_event.is_set() + retried = ( + runner is not None + and isinstance(exc, ExportKilledError) + and runner.claim_retry(lib, exc) + ) + if not (interrupted or retried): job.status = f"failed: {exc}" job.seconds = time.monotonic() - started save_state(master, jobs, state) @@ -714,12 +762,13 @@ def tracked( root=master, progress=progress, exporter=tracked, - min_free_mb=native_min_free_mb(), + min_free_mb=min_free, available_memory=available_memory_mb, quiet=True, + parallel=files, ) holder.append(runner) - display = FolderDisplay(progress, pending) + display = FolderDisplay(progress, pending, files) runner.start() thread = runner.thread try: @@ -779,17 +828,80 @@ def _duration(seconds: float) -> str: return f"{hours}h{minutes:02d}m" +# Running binaries shown with a row each; more are counted on the overall bar. +MAX_ROWS = 8 + + +class _Row: + """One running binary's bar: its phase, then functions rendered.""" + + def __init__(self, progress: Progress, position: int) -> None: + self.progress = progress + self.position = position + self.name: str | None = None + self.bar: Any = None + self.total: int | None = None + + def show(self, name: str | None, tail: Any, phase: str | None) -> None: + if name != self.name: + self.close() + self.name = name + if name is None: + return + self._open(None) + if self.bar is None: + return + if tail is not None and tail.total and self.total is None: + self.close() + self.name = name + self._open(tail.total) + if self.total is not None and tail is not None: + self.bar.n = min(tail.rendered, self.total) + self.bar.refresh() + return + if phase is None: + phase = tail.last_line if tail is not None else "starting" + self.bar.set_postfix_str(_short(phase, 60), refresh=False) + self.bar.refresh() + + def _open(self, total: int | None) -> None: + label = " ↳ " + _short(self.name or "", NAME_WIDTH) + self.total = total + if total is None: + fmt = "{desc}{postfix} [{elapsed}]" # " ↳ name, [time]" + else: + label = f"{label:<{LABEL_WIDTH}}" + fmt = BAR_FORMAT.replace("{unit_label}", "functions") + self.bar = self.progress.new_bar( + total=total, + desc=label, + position=self.position, + disable=None, + bar_format=fmt, + ) + + def close(self) -> None: + if self.bar is not None: + self.bar.close() + self.bar = None + self.total = None + self.name = None + + class FolderDisplay: """Progress for many binaries without flooding the terminal. On a terminal: one bar for the folder (binaries done, failures, ETA) and - one for the binary being exported (its phase, then functions rendered). - Nested export logs only go to ``tocode.log``; failures are printed above - the bars. When stderr is not a terminal: one line per finished binary and - a status line every minute. + one row per binary being exported (its phase, then functions rendered). + Rows keep their place while other exports come and go. Nested export logs + only go to ``tocode.log``; failures are printed above the bars. When stderr + is not a terminal: one line per finished binary and a status line every + minute. """ - def __init__(self, progress: Progress, jobs: list[FolderJob]) -> None: + def __init__( + self, progress: Progress, jobs: list[FolderJob], parallel: int = 1 + ) -> None: self.progress = progress self.jobs = jobs self.total = len(jobs) @@ -809,9 +921,8 @@ def __init__(self, progress: Progress, jobs: list[FolderJob]) -> None: if self.overall is not None and not self.tty: self.overall.close() self.overall = None - self.current: Any = None - self.current_name: str | None = None - self.current_total: int | None = None + rows = max(1, min(parallel, MAX_ROWS)) + self.rows = [_Row(progress, position) for position in range(1, rows + 1)] self.last_text = self.started def _finished(self) -> list[FolderJob]: @@ -825,13 +936,15 @@ def refresh(self, runner: NativeRunner) -> None: self.reported.add(job.relpath) self._report(job, len(self.reported)) if self.tty and self.overall is not None: + running = runner.running() failed = sum(1 for job in finished if job.status.startswith("failed")) + extra = [f"{failed} failed"] if failed else [] + if len(running) > len(self.rows): + extra.append(f"{len(running)} running") self.overall.n = len(finished) - self.overall.set_postfix_str( - f"{failed} failed" if failed else "", refresh=False - ) + self.overall.set_postfix_str(", ".join(extra), refresh=False) self.overall.refresh() - self._refresh_current(runner) + self._refresh_rows(runner, running) elif time.monotonic() - self.last_text >= TEXT_HEARTBEAT_SECONDS: self.last_text = time.monotonic() status = runner.describe_current() @@ -851,55 +964,35 @@ def _report(self, job: FolderJob, number: int) -> None: f"[{number}/{self.total}] ✓ {job.relpath}: {job.function_count} functions in {took}" ) - def _refresh_current(self, runner: NativeRunner) -> None: - lib = runner.current + def _refresh_rows(self, runner: NativeRunner, running: list[Any]) -> None: + wanted: dict[str, tuple[Any, str | None]] = { + item.lib.name: (item.tail, None) for item in running[: len(self.rows)] + } waiting = runner.waiting - name = lib.name if lib is not None else (waiting[0].name if waiting else None) - if name != self.current_name: - self._close_current() - self.current_name = name - if name is None: - return - self._open_current(None) - tail = runner.current_tail - if self.current is None: - return - if tail is not None and tail.total and self.current_total is None: - self._close_current() - self._open_current(tail.total) - if self.current_total is not None and tail is not None: - self.current.n = min(tail.rendered, self.current_total) - self.current.refresh() - return - if waiting is not None: - phase = f"waiting for memory ({waiting[1]} MB free, need {runner.min_free_mb} MB)" - elif tail is not None: - phase = tail.last_line - else: - phase = "starting" - self.current.set_postfix_str(_short(phase, 60), refresh=False) - self.current.refresh() - - def _open_current(self, total: int | None) -> None: - label = " ↳ " + _short(self.current_name or "", NAME_WIDTH) - self.current_total = total - if total is None: - fmt = "{desc}{postfix} [{elapsed}]" # " ↳ name, [time]" - else: - label = f"{label:<{LABEL_WIDTH}}" - fmt = BAR_FORMAT.replace("{unit_label}", "functions") - self.current = self.progress.new_bar( - total=total, desc=label, position=1, disable=None, bar_format=fmt - ) - - def _close_current(self) -> None: - if self.current is not None: - self.current.close() - self.current = None - self.current_total = None + if waiting is not None and len(wanted) < len(self.rows): + lib, available = waiting + wanted.setdefault( + lib.name, + ( + None, + f"waiting for memory ({available} MB free, need " + f"{runner.min_free_mb} MB)", + ), + ) + for row in self.rows: + if row.name is not None and row.name not in wanted: + row.close() + # A binary keeps the row it started on; new ones take a free row. + for name, (tail, phase) in wanted.items(): + slot = next((r for r in self.rows if r.name == name), None) or next( + (r for r in self.rows if r.name is None), None + ) + if slot is not None: + slot.show(name, tail, phase) def close(self) -> None: - self._close_current() + for row in self.rows: + row.close() if self.overall is not None: self.overall.close() self.overall = None diff --git a/src/tocode/parallel.py b/src/tocode/parallel.py index df473bc..a7f3687 100644 --- a/src/tocode/parallel.py +++ b/src/tocode/parallel.py @@ -156,6 +156,49 @@ def describe_jobs( return f"Workers: {selected} after {analysis_seconds:.2f}s analysis" +# Folder exports: binaries exported at once when --parallel-files is auto. +MAX_AUTO_PARALLEL_FILES = 16 +# With -j auto, each of the parallel files may use at least this many workers. +# Small binaries (most of a folder) render with one worker anyway; a big one +# gets a second instead of rendering alone on one CPU at the end. +MIN_AUTO_WORKERS_PER_FILE = 2 + + +def choose_parallel_files( + *, + pending: int, + requested_files: int | None, + requested_jobs: int | None, + cpu_count: int | None = None, + available_memory_mb: int | None = None, + file_memory_mb: int = 0, +) -> tuple[int, int | None]: + """How many binaries a folder export runs at once, and the worker cap per file. + + Explicit values are honored (only capped by the number of pending files). + Auto runs one file per CPU (one per ``-j`` workers' worth of CPUs when + ``-j`` is given), at most ``MAX_AUTO_PARALLEL_FILES``, and no more than + free memory holds at ``file_memory_mb`` each. With ``-j`` auto, each file's + workers are capped at its share of the CPUs (at least + ``MIN_AUTO_WORKERS_PER_FILE``); the cap is None when one file runs at a + time, which keeps the export's own choice. + """ + if pending <= 0: + return 1, None + cpus = cpu_count if cpu_count is not None else (os.cpu_count() or 1) + if requested_files is not None: + files = requested_files + else: + files = max(1, cpus // (requested_jobs or 1)) + files = min(files, MAX_AUTO_PARALLEL_FILES) + if available_memory_mb is not None and file_memory_mb > 0: + files = min(files, max(1, available_memory_mb // file_memory_mb)) + files = max(1, min(files, pending)) + if files == 1 or requested_jobs is not None: + return files, None + return files, max(MIN_AUTO_WORKERS_PER_FILE, cpus // files) + + def configured_job_limit() -> int: raw = os.environ.get("TOCODE_MAX_DECOMPILE_WORKERS", str(DEFAULT_JOB_LIMIT)).strip() try: diff --git a/tests/test_folder.py b/tests/test_folder.py index 1ffe448..f3b7c0f 100644 --- a/tests/test_folder.py +++ b/tests/test_folder.py @@ -154,9 +154,10 @@ def test_folder_export_links_exports_and_annotates(tmp_path: Path) -> None: root = _app(tmp_path / "app") exporter = FakeExporter() - summary = _export(root, tmp_path, exporter) + summary = _export(root, tmp_path, exporter, parallel_files=1) - # Executable first, then what it loads; the bundled libstdc++ is skipped. + # One at a time: executable first, then what it loads; the bundled + # libstdc++ is skipped. assert exporter.calls == ["bin/app", "lib/libfoo.so.1", "lib/libbar.so"] statuses = {job.relpath: job.status for job in summary.jobs} assert statuses["lib/libstdc++.so.6"] == RUNTIME diff --git a/tests/test_folder_parallel.py b/tests/test_folder_parallel.py new file mode 100644 index 0000000..ddd45bd --- /dev/null +++ b/tests/test_folder_parallel.py @@ -0,0 +1,334 @@ +"""Folder exports with several binaries at once (--parallel-files).""" + +from __future__ import annotations + +import json +from pathlib import Path +import threading +import time +from typing import Any + +import pytest + +from test_folder import FakeExporter, _app +from tocode import apk_native, cli +from tocode import folder as folder_module +from tocode.apk_native import ExportKilledError, NativeLib, NativeRunner +from tocode.errors import ToCodeError +from tocode.folder import DONE, PENDING, FolderOptions, export_folder +from tocode.parallel import choose_parallel_files +from tocode.progress import Progress + + +# -- sizing ------------------------------------------------------------------ + + +def test_explicit_values_are_honored() -> None: + assert choose_parallel_files( + pending=10, requested_files=3, requested_jobs=4, cpu_count=2 + ) == (3, None) + # Never more files at once than are left to export. + assert choose_parallel_files( + pending=2, requested_files=8, requested_jobs=None, cpu_count=16 + ) == (2, 8) + + +def test_auto_splits_cpus_between_files() -> None: + # -j auto: one file per CPU, each allowed two workers for a big binary. + assert choose_parallel_files( + pending=50, requested_files=None, requested_jobs=None, cpu_count=8 + ) == (8, 2) + # At most MAX_AUTO_PARALLEL_FILES; the spare CPUs go to each file's workers. + assert choose_parallel_files( + pending=50, requested_files=None, requested_jobs=None, cpu_count=64 + ) == (16, 4) + # -j 4: one file per four CPUs, -j is left alone. + assert choose_parallel_files( + pending=50, requested_files=None, requested_jobs=4, cpu_count=16 + ) == (4, None) + # One file at a time keeps the export's own worker choice. + assert choose_parallel_files( + pending=50, requested_files=None, requested_jobs=None, cpu_count=1 + ) == (1, None) + + +def test_auto_is_capped_by_free_memory() -> None: + assert choose_parallel_files( + pending=50, + requested_files=None, + requested_jobs=None, + cpu_count=64, + available_memory_mb=3000, + file_memory_mb=1024, + ) == (2, 32) + + +# -- runner ------------------------------------------------------------------- + + +def _libs(tmp_path: Path, count: int) -> list[NativeLib]: + return [ + NativeLib( + abi="x", + name=f"lib{index}.so", + entry=f"lib{index}.so", + source_apk="folder", + size=index, + sha256="", + path=tmp_path / f"lib{index}.so", + export_dir=tmp_path / "out" / f"lib{index}", + ) + for index in range(count) + ] + + +class SlowExporter: + """Records how many exports run at once; ``kill`` names fail like an OOM kill.""" + + def __init__(self, seconds: float = 0.05, kill: dict[str, int] | None = None): + self.seconds = seconds + self.kill = dict(kill or {}) + self.lock = threading.Lock() + self.active: set[str] = set() + self.peak = 0 + self.calls: list[str] = [] + + def __call__(self, lib: NativeLib, out_dir: Path, progress: Progress): + with self.lock: + self.calls.append(lib.entry) + self.active.add(lib.entry) + self.peak = max(self.peak, len(self.active)) + try: + time.sleep(self.seconds) + with self.lock: + left = self.kill.get(lib.entry, 0) + if left: + self.kill[lib.entry] = left - 1 + if left: + raise ExportKilledError("export process was killed by signal 9") + return ("fake", "fake", 1, 0) + finally: + with self.lock: + self.active.discard(lib.entry) + + +def _run( + tmp_path: Path, libs: list[NativeLib], exporter, **kwargs: Any +) -> NativeRunner: + runner = NativeRunner( + libs=libs, + root=tmp_path, + progress=Progress(enabled=False), + exporter=exporter, + mirror_logs=False, + quiet=True, + poll_seconds=0.01, + **kwargs, + ) + runner.start() + runner.join() + return runner + + +def test_runner_keeps_the_slots_full(tmp_path: Path) -> None: + libs = _libs(tmp_path, 7) + exporter = SlowExporter() + _run(tmp_path, libs, exporter, parallel=3) + assert exporter.peak == 3 + assert all(lib.status == "done" for lib in libs) + assert sorted(exporter.calls) == sorted(lib.entry for lib in libs) + + +def test_runner_holds_back_while_memory_is_short(tmp_path: Path) -> None: + libs = _libs(tmp_path, 4) + exporter = SlowExporter() + # 150 MB free: one export (100 MB) fits, a second beside a starting one + # (2 x 100 MB) does not. + _run( + tmp_path, + libs, + exporter, + parallel=4, + min_free_mb=100, + available_memory=lambda: 150, + ) + assert exporter.peak == 1 + assert all(lib.status == "done" for lib in libs) + + +def test_killed_export_is_retried_alone_once(tmp_path: Path) -> None: + libs = _libs(tmp_path, 4) + exporter = SlowExporter(kill={"lib0.so": 1}) + _run(tmp_path, libs, exporter, parallel=4) + assert exporter.calls.count("lib0.so") == 2 + assert all(lib.status == "done" for lib in libs) + + # Killed again alone: a real failure, not another retry. + libs = _libs(tmp_path, 3) + exporter = SlowExporter(kill={"lib0.so": 5}) + runner = _run(tmp_path, libs, exporter, parallel=3) + assert exporter.calls.count("lib0.so") == 2 + assert libs[0].status.startswith("failed: export process was killed") + assert runner.errors + + +def test_retry_waits_for_the_others_and_runs_alone(tmp_path: Path) -> None: + libs = _libs(tmp_path, 3) + seen: list[tuple[str, int]] = [] + exporter = SlowExporter(kill={"lib1.so": 1}) + original = exporter.__call__ + + def recording(lib: NativeLib, out_dir: Path, progress: Progress): + with exporter.lock: + seen.append((lib.entry, len(exporter.active))) + return original(lib, out_dir, progress) + + _run(tmp_path, libs, recording, parallel=3) + retry = [active for entry, active in seen if entry == "lib1.so"][-1] + assert retry == 0 # nothing else was running when the retry started + assert all(lib.status == "done" for lib in libs) + + +def test_sequential_runner_never_retries(tmp_path: Path) -> None: + libs = _libs(tmp_path, 2) + exporter = SlowExporter(kill={"lib0.so": 1}) + _run(tmp_path, libs, exporter, parallel=1) + assert exporter.calls.count("lib0.so") == 1 + assert libs[0].status.startswith("failed") + + +# -- folder ------------------------------------------------------------------- + + +class SlowFakeExporter(FakeExporter): + def __init__(self, seconds: float = 0.05) -> None: + super().__init__() + self.seconds = seconds + self.lock = threading.Lock() + self.active = 0 + self.peak = 0 + + def __call__(self, lib: NativeLib, out_dir: Path, progress: Progress): + with self.lock: + self.active += 1 + self.peak = max(self.peak, self.active) + try: + time.sleep(self.seconds) + return super().__call__(lib, out_dir, progress) + finally: + with self.lock: + self.active -= 1 + + +def _export(root: Path, tmp_path: Path, exporter, **kwargs: Any): + options = FolderOptions(out_dir=tmp_path / "out", **kwargs) + return export_folder( + root, options=options, progress=Progress(enabled=False), exporter=exporter + ) + + +def test_folder_exports_files_in_parallel_biggest_first( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = _app(tmp_path / "app") + sizes = { + relpath: (root / relpath).stat().st_size + for relpath in ("bin/app", "lib/libfoo.so.1", "lib/libbar.so") + } + queued: list[list[str]] = [] + + class Recording(NativeRunner): + def start(self) -> None: + queued.append([lib.entry for lib in self.libs]) + super().start() + + monkeypatch.setattr(folder_module, "NativeRunner", Recording) + exporter = SlowFakeExporter() + summary = _export(root, tmp_path, exporter, parallel_files=3) + + assert exporter.peak > 1 + assert all(job.status == DONE for job in summary.jobs if job.relpath in sizes) + order = [entry for entry in queued[0] if entry in sizes] + assert order == sorted(sizes, key=lambda relpath: -sizes[relpath]) + state = json.loads((tmp_path / "out/.tocode-folder/state.json").read_text()) + assert {relpath for relpath in sizes} <= set(state["jobs"]) + assert all(state["jobs"][relpath]["status"] == DONE for relpath in sizes) + # The links are joined after every export finished, as before. + mappings = json.loads((tmp_path / "out/mappings.json").read_text()) + assert any(row["symbol"] == "foo_add" for row in mappings["links"]) + + +def test_one_file_at_a_time_keeps_load_order(tmp_path: Path) -> None: + root = _app(tmp_path / "app") + exporter = FakeExporter() + _export(root, tmp_path, exporter, parallel_files=1) + assert exporter.calls[0] == "bin/app" # executables first, as before + + +def test_interrupt_keeps_running_exports_pending_and_resumes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = _app(tmp_path / "app") + killed = threading.Event() + started = threading.Event() + + class Blocking(FakeExporter): + def __call__(self, lib: NativeLib, out_dir: Path, progress: Progress): + if lib.entry == "bin/app": + return super().__call__(lib, out_dir, progress) + started.set() + # Runs until Ctrl-C kills it, like a real export process. + killed.wait(5) + raise ToCodeError("export process was killed by signal 9") + + monkeypatch.setattr(apk_native, "terminate_active_native_children", killed.set) + real_refresh = folder_module.FolderDisplay.refresh + + def refresh(self, runner) -> None: + real_refresh(self, runner) + if started.is_set() and any(job.status == DONE for job in self.jobs): + raise KeyboardInterrupt + + monkeypatch.setattr(folder_module.FolderDisplay, "refresh", refresh) + monkeypatch.setattr(folder_module, "DISPLAY_REFRESH_SECONDS", 0.01) + summary = _export(root, tmp_path, Blocking(), parallel_files=3) + + assert summary.interrupted + statuses = {job.relpath: job.status for job in summary.jobs} + assert statuses["bin/app"] == DONE + # Killed by the interrupt: pending, never recorded as failed. + assert statuses["lib/libfoo.so.1"] == PENDING + assert statuses["lib/libbar.so"] == PENDING + state = json.loads((tmp_path / "out/.tocode-folder/state.json").read_text()) + assert set(state["jobs"]) == {"bin/app"} + + monkeypatch.setattr(folder_module.FolderDisplay, "refresh", real_refresh) + again = FakeExporter() + summary = _export(root, tmp_path, again, parallel_files=3) + assert sorted(again.calls) == ["lib/libbar.so", "lib/libfoo.so.1"] + assert not summary.interrupted + + +def test_cli_passes_parallel_files( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + seen: list[FolderOptions] = [] + + def fake_export_folder(folder, *, options, progress): + seen.append(options) + raise ToCodeError("stop here") + + monkeypatch.setattr(folder_module, "export_folder", fake_export_folder) + root = _app(tmp_path / "app") + assert cli.main([str(root), "--parallel-files", "4", "-j", "2", "-q"]) == 1 + assert seen[0].parallel_files == 4 and seen[0].jobs == 2 + assert cli.main([str(root), "-pf", "3", "-q"]) == 1 + assert seen[1].parallel_files == 3 + + +def test_cli_rejects_parallel_files_for_one_file(tmp_path: Path, capsys) -> None: + binary = tmp_path / "a.bin" + binary.write_bytes(b"\x7fELF" + bytes(64)) + with pytest.raises(SystemExit): + cli.main([str(binary), "--parallel-files", "2"]) + assert "--parallel-files applies to folder input" in capsys.readouterr().err