More CR

Author: ilevkivskyi

mypy/build.py

@@ -959,66 +959,87 @@ def dump_stats(self) -> None:
                 print(f"{key + ':':24}{value}")
 
     def parse_all(self, states: list[State]) -> None:
-        """Parse multiple files in parallel (if possible) and compute dependencies.
+        """Parse multiple files in parallel (if possible) and compute dependencies."""
+        if not self.options.native_parser:
+            # Old parser cannot be parallelized.
+            for state in states:
+                state.parse_file()
+            self.post_parse_all(states)
+            return
+
+        sequential_states = []
+        parallel_states = []
+        for state in states:
+            if state.tree is not None:
+                # The file was already parsed.
+                continue
+            if not self.fscache.exists(state.xpath):
+                # New parser only supports parsing on-disk files.
+                sequential_states.append(state)
+                continue
+            parallel_states.append(state)
+        self.parse_parallel(sequential_states, parallel_states)
+        self.post_parse_all(states)
+
+    def parse_parallel(self, sequential_states: list[State], parallel_states: list[State]) -> None:
+        """Perform parallel parsing of states.
 
         Note: this duplicates a bit of logic from State.parse_file(). This is done
-        as a micro-optimization to parallelize only those parts of the code that
-        can be parallelized efficiently.
+        as an optimization to parallelize only those parts of the code that can be
+        parallelized efficiently.
         """
-        if self.options.native_parser:
-            futures = []
-            parsed_states = {}
-            # 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 states:
-                    state.needs_parse = False
-                    if state.tree is not None:
-                        # The file was already parsed.
-                        continue
-                    # New parser reads source from file directly, we do this only for
-                    # the side effect of parsing inline mypy configurations.
-                    state.get_source()
-                    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, state.source or ""))
-                        parsed_states[state.id] = state
-                    else:
-                        self.log(f"Using cached AST for {state.xpath} ({state.id})")
-                        state.tree, state.early_errors = self.ast_cache[state.id]
-                for fut in wait(futures).done:
-                    state_id, parse_errors = fut.result()
-                    if parse_errors:
-                        state = parsed_states[state_id]
-                        with state.wrap_context():
-                            self.errors.set_file(state.xpath, state.id, options=state.options)
-                            for error in parse_errors:
-                                # New parser reports errors lazily.
-                                report_parse_error(error, self.errors)
-                            if self.errors.is_blockers():
-                                self.log("Bailing due to parse errors")
-                                self.errors.raise_error()
+        futures = []
+        parallel_parsed_states = {}
+        # 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
+                # New parser reads source from file directly, we do this only for
+                # the side effect of parsing inline mypy configurations.
+                state.get_source()
+                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, state.source or ""))
+                    parallel_parsed_states[state.id] = state
+                else:
+                    self.log(f"Using cached AST for {state.xpath} ({state.id})")
+                    state.tree, state.early_errors = self.ast_cache[state.id]
 
-            for state in states:
-                assert state.tree is not None
-                if state.id in parsed_states:
-                    state.early_errors = list(self.errors.error_info_map.get(state.xpath, []))
-                    state.semantic_analysis_pass1()
-                    self.ast_cache[state.id] = (state.tree, state.early_errors)
-                self.modules[state.id] = state.tree
-                state.check_blockers()
-                state.setup_errors()
-        else:
-            # Old parser cannot be parallelized.
-            for state in states:
+            # Parse sequential before waiting on parallel.
+            for state in sequential_states:
                 state.parse_file()
 
+            for fut in wait(futures).done:
+                state_id, parse_errors = fut.result()
+                # New parser reports errors lazily, add them if any.
+                if parse_errors:
+                    state = parallel_parsed_states[state_id]
+                    with state.wrap_context():
+                        self.errors.set_file(state.xpath, state.id, options=state.options)
+                        for error in parse_errors:
+                            report_parse_error(error, self.errors)
+                        if self.errors.is_blockers():
+                            self.log("Bailing due to parse errors")
+                            self.errors.raise_error()
+
+        for state in parallel_states:
+            assert state.tree is not None
+            if state.id in parallel_parsed_states:
+                state.early_errors = list(self.errors.error_info_map.get(state.xpath, []))
+                state.semantic_analysis_pass1()
+                self.ast_cache[state.id] = (state.tree, state.early_errors)
+            self.modules[state.id] = state.tree
+            state.check_blockers()
+            state.setup_errors()
+
+    def post_parse_all(self, states: list[State]) -> None:
         for state in states:
             state.compute_dependencies()
             if self.workers and state.tree:
@@ -1152,7 +1173,8 @@ def parse_file(
         Raise CompileError if there is a parse error.
         """
         imports_only = False
-        if self.workers and self.fscache.exists(path):
+        file_exists = self.fscache.exists(path)
+        if self.workers and file_exists:
             # Currently, we can use the native parser only for actual files.
             imports_only = True
         t0 = time.time()
@@ -1162,7 +1184,13 @@ def parse_file(
             tree = load_from_raw(path, id, raw_data, self.errors, options)
         else:
             tree, parse_errors = parse(
-                source, path, id, self.errors, options=options, imports_only=imports_only
+                source,
+                path,
+                id,
+                self.errors,
+                options=options,
+                file_exists=file_exists,
+                imports_only=imports_only,
             )
         tree._fullname = id
         if self.stats_enabled:

mypy/checkstrformat.py

@@ -44,7 +44,7 @@
     TempNode,
     TupleExpr,
 )
-from mypy.parse import parse, report_parse_error
+from mypy.parse import parse
 from mypy.subtypes import is_subtype
 from mypy.typeops import custom_special_method
 from mypy.types import (
@@ -581,12 +581,14 @@ def apply_field_accessors(
 
         temp_errors = Errors(self.chk.options)
         dummy = DUMMY_FIELD_NAME + spec.field[len(spec.key) :]
-        temp_ast, parse_errors = parse(
-            dummy, fnam="<format>", module=None, options=self.chk.options, errors=temp_errors
+        temp_ast, _ = parse(
+            dummy,
+            fnam="<format>",
+            module=None,
+            options=self.chk.options,
+            errors=temp_errors,
+            file_exists=False,
         )
-        for error in parse_errors:
-            # New parser reports errors lazily.
-            report_parse_error(error, temp_errors)
         if temp_errors.is_errors():
             self.msg.fail(
                 f'Syntax error in format specifier "{spec.field}"',

mypy/nativeparse.py

@@ -271,8 +271,11 @@ def parse_to_binary_ast(
 ) -> tuple[bytes, list[ParseError], TypeIgnores, bytes, bool, bool]:
     # This is a horrible hack to work around a mypyc bug where imported
     # module may be not ready in a thread sometimes.
+    t0 = time.time()
     while ast_serialize is None:
         time.sleep(0.0001)  # type: ignore[unreachable]
+        if time.time() - t0 > 10.0:
+            raise ImportError("Cannot import ast_serialize")
     ast_bytes, errors, ignores, import_bytes, ast_data = ast_serialize.parse(
         filename,
         skip_function_bodies=skip_function_bodies,

mypy/parse.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import os
 import re
 
 from librt.internal import ReadBuffer
@@ -18,6 +17,7 @@ def parse(
     module: str | None,
     errors: Errors,
     options: Options,
+    file_exists: bool,
     imports_only: bool = False,
 ) -> tuple[MypyFile, list[ParseError]]:
     """Parse a source file, without doing any semantic analysis.
@@ -30,7 +30,7 @@ def parse(
     if options.native_parser:
         # Native parser only works with actual files on disk
         # Fall back to fastparse for in-memory source or non-existent files
-        if os.path.exists(fnam):
+        if file_exists:
             import mypy.nativeparse
 
             ignore_errors = options.ignore_errors or fnam in errors.ignored_files

mypy/stubgen.py

@@ -1745,7 +1745,12 @@ def parse_source_file(mod: StubSource, mypy_options: MypyOptions) -> None:
     source = mypy.util.decode_python_encoding(data)
     errors = Errors(mypy_options)
     mod.ast, errs = mypy.parse.parse(
-        source, fnam=mod.path, module=mod.module, errors=errors, options=mypy_options
+        source,
+        fnam=mod.path,
+        module=mod.module,
+        errors=errors,
+        options=mypy_options,
+        file_exists=True,
     )
     mod.ast._fullname = mod.module
     for err in errs:

mypy/test/testparse.py

@@ -10,7 +10,7 @@
 from mypy.config_parser import parse_mypy_comments
 from mypy.errors import CompileError, Errors
 from mypy.options import Options
-from mypy.parse import parse, report_parse_error
+from mypy.parse import parse
 from mypy.test.data import DataDrivenTestCase, DataSuite
 from mypy.test.helpers import assert_string_arrays_equal, find_test_files, parse_options
 from mypy.util import get_mypy_comments
@@ -60,11 +60,14 @@ def test_parser(testcase: DataDrivenTestCase) -> None:
 
     try:
         errors = Errors(options)
-        n, errs = parse(
-            bytes(source, "ascii"), fnam="main", module="__main__", errors=errors, options=options
+        n, _ = parse(
+            bytes(source, "ascii"),
+            fnam="main",
+            module="__main__",
+            errors=errors,
+            options=options,
+            file_exists=False,
         )
-        for err in errs:
-            report_parse_error(err, errors)
         if errors.is_errors():
             errors.raise_error()
         a = n.str_with_options(options).split("\n")
@@ -98,15 +101,14 @@ def test_parse_error(testcase: DataDrivenTestCase) -> None:
             skip()
         # Compile temporary file. The test file contains non-ASCII characters.
         errors = Errors(options)
-        _, errs = parse(
+        parse(
             bytes("\n".join(testcase.input), "utf-8"),
             INPUT_FILE_NAME,
             "__main__",
             errors=errors,
             options=options,
+            file_exists=False,
         )
-        for err in errs:
-            report_parse_error(err, errors)
         if errors.is_errors():
             errors.raise_error()
         raise AssertionError("No errors reported")