Method parse_files_threaded_raw

mypy/build.py:1100–1143

Parse files using a thread pool. Also parse sequential states while waiting for the parallel results. Trees from the new parser are left in raw (serialized) form. Return (list, set) of states that were actually parsed (not cached).

(
        self, sequential_states: list[State], parallel_states: list[State]
    )

            state.setup_errors()

    def parse_files_threaded_raw(
        self, sequential_states: list[State], parallel_states: list[State]
    ) -> tuple[list[State], set[State]]:
        """Parse files using a thread pool.

        Also parse sequential states while waiting for the parallel results.
        Trees from the new parser are left in raw (serialized) form.

        Return (list, set) of states that were actually parsed (not cached).
        """
        futures = []
        # Use both list and a set to have more predictable order of errors,
        # while also not sacrificing performance.
        parallel_parsed_states: list[State] = []
        parallel_parsed_states_set: set[State] = set()
        # Use at least --num-workers if specified by user.
        available_threads = max(get_available_threads(), self.options.num_workers)
        # Overhead from trying to parallelize (small) blocking portion of
        # parse_file_inner() results in no visible improvement with more than 8 threads.
        # TODO: reuse thread pool and/or batch small files in single submit() call.
        with ThreadPoolExecutor(max_workers=min(available_threads, 8)) as executor:
            for state in parallel_states:
                state.needs_parse = False
                if state.id not in self.ast_cache:
                    self.log(f"Parsing {state.xpath} ({state.id})")
                    ignore_errors = state.ignore_all or state.options.ignore_errors
                    if ignore_errors:
                        self.errors.ignored_files.add(state.xpath)
                    futures.append(executor.submit(state.parse_file_inner, ""))
                    parallel_parsed_states.append(state)
                    parallel_parsed_states_set.add(state)
                else:
                    self.log(f"Using cached AST for {state.xpath} ({state.id})")
                    state.tree, state.early_errors, source_hash = self.ast_cache[state.id]
                    state.source_hash = source_hash

            # Parse sequential before waiting on parallel.
            for state in sequential_states:
                state.parse_file()

            for fut in wait(futures).done:
                fut.result()

        return parallel_parsed_states, parallel_parsed_states_set

    def post_parse_all(self, states: list[State]) -> None:
        for state in states: