investigate running some phases of bootstrap in parallel

[dhellmann]

Now that we are using a mostly iterative approach to managing bootstrapping, instead of pure recursion, it should be possible for us to run some operations in parallel. For example, if the stack has 3 work items in the resolve phase, we can run all of those at the same time.

[dhellmann]

Plan: Make Bootstrap Loop Fully Thread-Safe for Future Parallelism

Context

The Bootstrapper.bootstrap() method uses an iterative DFS loop with a LIFO work stack. Items flow through phases: RESOLVE → START → PREPARE_SOURCE → PREPARE_BUILD → BUILD → PROCESS_INSTALL_DEPS → COMPLETE.

The RESOLVE phase calls out to PyPI or the dependency graph — I/O bound work. As a first step, consecutive RESOLVE items at the top of the stack will run in parallel. To make this safe now, and to enable parallelizing additional phases later, all shared mutable state must be properly protected.

The main loop is restructured so ALL work goes through the same uniform cycle: collect a batch of consecutive same-phase items from the top of the stack → decide whether to run in parallel → run → restore results. The only phase-specific logic is whether to parallelize. Everything is inline in bootstrap() — no separate helper method.

Shared State Audit

State	Type	Phases that write	Thread-safe now?	Risk
self.why	list	bootstrap loop, _track_why	No	Corruption if multiple threads share it
self._seen_requirements	set	_phase_start, _handle_phase_error	No	Duplicate processing (TOCTOU check-then-set)
self._build_requirements	set	_add_to_build_order	No	Duplicate builds (TOCTOU check-then-set)
self._build_stack	list	_add_to_build_order	No	Interleaved appends + corrupt JSON file
self.failed_packages	list	_record_test_mode_failure	No	Lost appends
self._failed_versions	list	_handle_phase_error	No	Lost appends
ctx.dependency_graph	DependencyGraph	_add_to_graph, _handle_phase_error	No	No internal locking; dict/list corruption
_resolver._resolved_requirements	dict	cache_resolution	No	TOCTOU, dict corruption
progress bar	tqdm	bootstrap main loop, _handle_build_requirements	No	tqdm not thread-safe; _handle_build_requirements runs in worker threads

Strategy

• self.why: Convert to threading.local() storage via a property+setter. Each thread gets its own why stack. No call sites change. Thread pool workers initialize their thread-local why from item.why_snapshot before dispatching, matching what the sequential path already does.

• Everything else: Protect with a single self._state_lock = threading.Lock(). Covers all fast critical sections (set/dict/list operations + small file writes). Per-package I/O (downloads, builds, subprocess calls) is already isolated by package directory — no lock needed there.

• Progress bar: No lock needed. All progressbar calls happen in the main thread — after parallel futures complete or in the sequential path. The only other call site is _handle_build_requirements() on the git URL path, which is always sequential.

Files to Modify

• src/fromager/bootstrapper.py — self.why property, _state_lock, lock at all write sites, main loop restructure
• src/fromager/bootstrap_requirement_resolver.py — thread-safe cache
• tests/test_bootstrapper.py — new tests

Note: BootstrapPhase (defined in bootstrapper.py) also gains a can_parallelize property — the single place where per-phase parallelism decisions live.

Implementation Steps

1. src/fromager/bootstrap_requirement_resolver.py

Add import threading. Add self._lock = threading.Lock() to __init__. Wrap get_cached_resolution() and cache_resolution() bodies with with self._lock:.

Note: A benign TOCTOU window remains between the two calls (two threads could both miss cache, both resolve, second write overwrites with equivalent data). Acceptable — avoids complex per-key locking.

2. src/fromager/bootstrapper.py — imports

Add to stdlib imports (alphabetical):

import concurrent.futures
import threading

3. src/fromager/bootstrapper.py — convert self.why to threading.local()

In __init__, replace self.why: list[...] = [] with:

self._thread_local = threading.local()
self._thread_local.why = []  # initialize main thread's stack

Add as class-level property (below __init__):

@property
def why(self) -> list[tuple[RequirementType, Requirement, Version]]:
    """Per-thread dependency chain stack. Thread-safe via threading.local()."""
    try:
        return self._thread_local.why
    except AttributeError:
        self._thread_local.why = []
        return self._thread_local.why

@why.setter
def why(self, value: list[tuple[RequirementType, Requirement, Version]]) -> None:
    self._thread_local.why = value

No other call sites change — reading and writing self.why is identical at the surface, but each thread has its own stack.

4. src/fromager/bootstrapper.py — add self._state_lock

In __init__, after self.failed_packages:

self._state_lock = threading.Lock()

5. src/fromager/bootstrapper.py — protect shared state at all write sites

_record_test_mode_failure() — wrap the append:

with self._state_lock:
    self.failed_packages.append({...})

Logger calls before the append stay outside the lock (stdlib logger is thread-safe).

_add_to_build_order() — make the entire check-then-set-then-write atomic:

with self._state_lock:
    key = (canonicalize_name(req.name), str(version))
    if key in self._build_requirements:
        return
    logger.info(f"adding {key} to build order")
    self._build_requirements.add(key)
    info = {...}
    if req.url:
        info["source_url"] = req.url
    self._build_stack.append(info)
    with open(self._build_order_filename, "w") as f:
        json.dump(self._build_stack, f, indent=2, default=str)

_add_to_graph() — wrap graph write + file write:

with self._state_lock:
    self.ctx.dependency_graph.add_dependency(...)
    self.ctx.write_to_graph_to_file()

_phase_start() — make seen check-then-set atomic:

with self._state_lock:
    if self._has_been_seen(item.req, item.resolved_version, item.build_sdist_only):
        logger.debug(...)
        return []
    self._mark_as_seen(item.req, item.resolved_version, item.build_sdist_only)

The logger.info after marking stays outside the lock.

_handle_phase_error() — multiple_versions block — wrap all state mutations together:

with self._state_lock:
    self._failed_versions.append((pkg_name, str(item.resolved_version), err))
    self.ctx.dependency_graph.remove_dependency(pkg_name, item.resolved_version)
    self._seen_requirements.discard(self._resolved_key(..., "sdist"))
    self._seen_requirements.discard(self._resolved_key(..., "wheel"))
    self.ctx.write_to_graph_to_file()

Logger warning calls stay outside the lock.

_handle_build_requirements() — this is called from _resolve_version_from_git_url() and updates the progressbar. With all items (including git URL items) running through the ThreadPoolExecutor, these calls can happen from worker threads. Wrap both progressbar calls with self._state_lock:

with self._state_lock:
    self.progressbar.update_total(len(build_dependencies))
# ... loop body ...
with self._state_lock:
    self.progressbar.update()

6. src/fromager/bootstrapper.py — add can_parallelize to BootstrapPhase

Add a property to the existing BootstrapPhase enum (near the existing tracks_why property):

@property
def can_parallelize(self) -> bool:
    """Whether items in this phase can be processed in parallel."""
    return self == BootstrapPhase.RESOLVE

This is the single place where the parallelism decision per phase lives.

7. src/fromager/bootstrapper.py — restructure bootstrap() main loop

Define the run_one worker closure once before the loop, then use a uniform batch collect → run → restore cycle for ALL work. The batch size (1 vs many) determines whether execution is effectively sequential or parallel; the code path is the same either way.

def run_one(item: WorkItem) -> list[WorkItem]:
    # Initialize this thread's why stack from the item's snapshot,
    # matching what the sequential main loop did before each dispatch.
    self.why = list(item.why_snapshot)
    with req_ctxvar_context(item.req), self._track_why(item):
        try:
            return self._dispatch_phase(item)
        except Exception as err:
            return self._handle_phase_error(item, err)

while stack:
    top_phase = stack[-1].phase

    # Collect batch. For phases that support parallelism, gather all
    # consecutive same-phase items from the top of the stack.
    # Stop at git URL items (RESOLVE only): they require sequential
    # orchestration (filesystem access + recursive bootstrap calls).
    # For phases that do not support parallelism, take only the top item.
    if top_phase.can_parallelize:
        batch: list[WorkItem] = []
        i = len(stack) - 1
        while i >= 0 and stack[i].phase == top_phase:
            batch.append(stack[i])
            i -= 1
        del stack[i + 1:]
    else:
        batch = [stack.pop()]

    # Run batch through ThreadPoolExecutor. A batch of 1 item is effectively
    # sequential; a batch of N > 1 runs in parallel.
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = [executor.submit(run_one, item) for item in batch]
        all_new_items: list[list[WorkItem]] = []
        for future in futures:
            try:
                all_new_items.append(future.result())
            except Exception:
                for f in futures:
                    f.cancel()
                raise

    # Restore results to stack in original order.
    # batch[0] was topmost; extending in reverse puts its results back on top.
    for result_items in reversed(all_new_items):
        stack.extend(result_items)

    # Progress bar: count new RESOLVE items as new dependencies to track
    new_dep_count = sum(
        sum(1 for it in result_items if it.phase == BootstrapPhase.RESOLVE)
        for result_items in all_new_items
    )
    if new_dep_count > 0:
        self.progressbar.update_total(new_dep_count)
    # Advance bar for each item that produced no further work
    for result_items in all_new_items:
        if not result_items:
            self.progressbar.update()

Batch semantics: Collecting all consecutive same-phase items (for parallelizable phases) means siblings at the same phase all dispatch before any of their children. This changes DFS within-phase ordering to BFS. This is intentional — it is the prerequisite for parallelizing those siblings.

Stack ordering: batch[0] was topmost. Iterating reversed(all_new_items) when extending puts batch[0]'s results back on top.

Single-item batches (non-parallelizable phases): The item runs in a ThreadPoolExecutor worker thread. self.why is thread-local so each worker initializes its own stack cleanly. Overhead is small relative to the I/O-heavy work each item performs.

Git URL RESOLVE items: No longer excluded from batch collection. They run through the same ThreadPoolExecutor path. Recursive self.bootstrap() calls inside _resolve_version_from_git_url() are safe because each recursion creates its own per-batch executor (no deadlock). Progressbar calls in _handle_build_requirements() are protected by self._state_lock.

7. tests/test_bootstrapper.py — new tests

Add at end of file:

• test_bootstrap_batch_parallel_resolve_returns_ordered_results — two RESOLVE items resolve in parallel, results returned in original stack order
• test_bootstrap_batch_sequential_single_item — single item in batch processes correctly through sequential path
• test_bootstrap_batch_test_mode_failure_records_and_continues — failed RESOLVE in test mode recorded, sibling succeeds, returns empty inner list
• test_bootstrap_loop_uses_parallel_for_multiple_resolve_items — integration: ThreadPoolExecutor is used when 2+ RESOLVE items at stack top

Verification

hatch run test:test tests/test_bootstrapper.py
hatch run mypy:check src/fromager/bootstrapper.py src/fromager/bootstrap_requirement_resolver.py
hatch run lint:fix src/fromager/bootstrapper.py src/fromager/bootstrap_requirement_resolver.py tests/test_bootstrapper.py
hatch run lint:check src/fromager/bootstrapper.py src/fromager/bootstrap_requirement_resolver.py tests/test_bootstrapper.py

TODO

• bootstrap_requirement_resolver.py: add threading.Lock in __init__, wrap get_cached_resolution and cache_resolution
• bootstrapper.py: add imports (concurrent.futures, threading)
• bootstrapper.py: convert self.why to threading.local() property + setter
• bootstrapper.py: add self._state_lock in __init__
• bootstrapper.py: add lock to _record_test_mode_failure() (failed_packages append)
• bootstrapper.py: add lock to _add_to_build_order() (entire check-then-set-then-write)
• bootstrapper.py: add lock to _add_to_graph() (graph write + file write)
• bootstrapper.py: add lock to _phase_start() (seen check-then-set)
• bootstrapper.py: add lock to _handle_phase_error() multiple_versions block
• bootstrapper.py: add lock to progressbar calls in _handle_build_requirements()
• bootstrapper.py: add can_parallelize property to BootstrapPhase
• bootstrapper.py: restructure bootstrap() main loop (uniform batch cycle)
• tests/test_bootstrapper.py: add four new tests
• Run verification commands