From 928cde3069c84f35c7a3c99649f92d33081286ef Mon Sep 17 00:00:00 2001 From: Jukka Lehtosalo Date: Wed, 16 Sep 2026 09:40:46 +0100 Subject: [PATCH 1/3] Fall back to sequential parsing when threads are unavailable The threaded parsing added in #21823 broke the Pyodide wasm32 wheel builds, since threads aren't available. --- mypy/build.py | 75 +++++++++++++++++++++++++++++++----------- mypy/test/testgraph.py | 40 ++++++++++++++++++++-- mypy/util.py | 13 ++++++++ 3 files changed, 105 insertions(+), 23 deletions(-) diff --git a/mypy/build.py b/mypy/build.py index c616faccadb37..bb0efd6c6d11e 100644 --- a/mypy/build.py +++ b/mypy/build.py @@ -26,7 +26,7 @@ import time import types from collections.abc import Callable, Iterator, Mapping, Sequence, Set as AbstractSet -from concurrent.futures import ThreadPoolExecutor, wait +from concurrent.futures import Future, ThreadPoolExecutor, wait from heapq import heappop, heappush from textwrap import dedent from threading import Lock, Thread @@ -125,6 +125,7 @@ from mypy.semanal_pass1 import SemanticAnalyzerPreAnalysis from mypy.util import ( DecodeError, + can_start_threads, decode_python_encoding, get_available_threads, get_mypy_comments, @@ -1105,34 +1106,53 @@ def parse_files_threaded_raw(self, states: list[State]) -> tuple[list[State], se 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() + if not can_start_threads(): + # Threads are not available (e.g. on WebAssembly platforms), so parse + # files one by one in the current thread. + return self.parse_files_raw(states, None) + # 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.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) + return self.parse_files_raw(states, executor) + + def parse_files_raw( + self, states: list[State], executor: ThreadPoolExecutor | None + ) -> tuple[list[State], set[State]]: + """Parse files, using executor for parallelism if one is provided. + + Without an executor files are parsed in the current thread, but we still + go through futures, so that behavior (such as the order in which errors + are reported) is the same on all platforms. + """ + 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() + for state in 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) + if executor is not None: futures.append(executor.submit(state.parse_file_inner, state.source)) - 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 + futures.append(submit_inline(state.parse_file_inner, state.source)) + 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 - for fut in wait(futures).done: - fut.result() + for fut in wait(futures).done: + fut.result() return parallel_parsed_states, parallel_parsed_states_set @@ -1579,6 +1599,21 @@ def note_multiline( self.note(line, msg_line, code, only_once=only_once) +def submit_inline(fn: Callable[..., Any], *args: Any) -> Future[Any]: + """Call fn in the current thread and return an already-completed future. + + This is a substitute for Executor.submit() on platforms where threads are + not available. Like a thread pool, we only propagate an exception once the + result of the future is collected. + """ + future: Future[Any] = Future() + try: + future.set_result(fn(*args)) + except BaseException as exc: + future.set_exception(exc) + return future + + def deps_to_json(x: dict[str, set[str]]) -> bytes: return json_dumps({k: list(v) for k, v in x.items()}) diff --git a/mypy/test/testgraph.py b/mypy/test/testgraph.py index aec6576189661..3cf72f3faaa6b 100644 --- a/mypy/test/testgraph.py +++ b/mypy/test/testgraph.py @@ -1,11 +1,19 @@ -"""Test cases for graph processing code in build.py.""" +"""Test cases for graph processing and other code in build.py.""" from __future__ import annotations import sys from collections.abc import Set as AbstractSet - -from mypy.build import BuildManager, BuildSourceSet, State, order_ascc, sorted_components +from concurrent.futures import wait + +from mypy.build import ( + BuildManager, + BuildSourceSet, + State, + order_ascc, + sorted_components, + submit_inline, +) from mypy.errors import Errors from mypy.fscache import FileSystemCache from mypy.graph_utils import strongly_connected_components, topsort @@ -136,3 +144,29 @@ def test_order_ascc(self) -> None: ascc = res[1] scc = order_ascc(graph, ascc) assert_equal(scc, ["d", "c", "b", "a"]) + + +class SubmitInlineSuite(Suite): + """Test the fallback used when threads are unavailable (e.g. on wasm).""" + + def test_result(self) -> None: + future = submit_inline(str, 5) + assert future.done() + assert future.result() == "5" + + def test_exception_is_only_raised_when_collected(self) -> None: + def fail() -> None: + raise ValueError("boom") + + future = submit_inline(fail) + assert future.done() + assert isinstance(future.exception(), ValueError) + with self.assertRaises(ValueError): + future.result() + + def test_wait(self) -> None: + # This is how build.py collects results of parse jobs. + futures = [submit_inline(str, i) for i in (1, 2)] + result = wait(futures) + assert result.done == set(futures) + assert not result.not_done diff --git a/mypy/util.py b/mypy/util.py index a48a7ea824daf..d48fd696f28cf 100644 --- a/mypy/util.py +++ b/mypy/util.py @@ -973,12 +973,25 @@ def json_loads(data: bytes) -> Any: _AVAILABLE_THREADS: int | None = None +def can_start_threads() -> bool: + """Can we start new threads? + + WebAssembly builds are typically single-threaded, and starting a thread + fails with RuntimeError there, so we must not even try. + """ + return sys.platform not in ("emscripten", "wasi") + + def get_available_threads() -> int: """Determine number of physical cores that current process can use (best effort).""" global _AVAILABLE_THREADS if _AVAILABLE_THREADS is not None: return _AVAILABLE_THREADS + if not can_start_threads(): + _AVAILABLE_THREADS = 1 + return 1 + # This takes into account -X cpu_count and/or PYTHON_CPU_COUNT, but always # counts virtual cores (which is not what we want for CPU bound tasks). os_cpu_count = os.cpu_count() From db225a49bfd9007cdc070e5a0240159435ce5064 Mon Sep 17 00:00:00 2001 From: Jukka Lehtosalo Date: Wed, 16 Sep 2026 11:50:48 +0100 Subject: [PATCH 2/3] Simplify --- mypy/build.py | 78 ++++++++++++------------------------------ mypy/test/testgraph.py | 40 ++-------------------- mypy/util.py | 4 --- 3 files changed, 25 insertions(+), 97 deletions(-) diff --git a/mypy/build.py b/mypy/build.py index bb0efd6c6d11e..5e26a511194f5 100644 --- a/mypy/build.py +++ b/mypy/build.py @@ -26,7 +26,7 @@ import time import types from collections.abc import Callable, Iterator, Mapping, Sequence, Set as AbstractSet -from concurrent.futures import Future, ThreadPoolExecutor, wait +from concurrent.futures import ThreadPoolExecutor, wait from heapq import heappop, heappush from textwrap import dedent from threading import Lock, Thread @@ -1022,8 +1022,8 @@ def parse_all(self, states: list[State], post_parse: bool = True) -> None: If post_parse is False, skip the last step (used when parsing unchanged files that need to be re-checked due to stale dependencies). """ - if not self.options.native_parser: - # Old parser cannot be parallelized. + if not self.options.native_parser or not can_start_threads(): + # Old parser cannot be parallelized, and some platforms don't support threads. for state in states: state.parse_file() if post_parse: @@ -1106,53 +1106,34 @@ def parse_files_threaded_raw(self, states: list[State]) -> tuple[list[State], se Return (list, set) of states that were actually parsed (not cached). """ - if not can_start_threads(): - # Threads are not available (e.g. on WebAssembly platforms), so parse - # files one by one in the current thread. - return self.parse_files_raw(states, None) - + 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: - return self.parse_files_raw(states, executor) - - def parse_files_raw( - self, states: list[State], executor: ThreadPoolExecutor | None - ) -> tuple[list[State], set[State]]: - """Parse files, using executor for parallelism if one is provided. - - Without an executor files are parsed in the current thread, but we still - go through futures, so that behavior (such as the order in which errors - are reported) is the same on all platforms. - """ - 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() - for state in 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) - if executor is not None: + for state in 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, state.source)) + parallel_parsed_states.append(state) + parallel_parsed_states_set.add(state) else: - futures.append(submit_inline(state.parse_file_inner, state.source)) - 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 + 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 - for fut in wait(futures).done: - fut.result() + for fut in wait(futures).done: + fut.result() return parallel_parsed_states, parallel_parsed_states_set @@ -1599,21 +1580,6 @@ def note_multiline( self.note(line, msg_line, code, only_once=only_once) -def submit_inline(fn: Callable[..., Any], *args: Any) -> Future[Any]: - """Call fn in the current thread and return an already-completed future. - - This is a substitute for Executor.submit() on platforms where threads are - not available. Like a thread pool, we only propagate an exception once the - result of the future is collected. - """ - future: Future[Any] = Future() - try: - future.set_result(fn(*args)) - except BaseException as exc: - future.set_exception(exc) - return future - - def deps_to_json(x: dict[str, set[str]]) -> bytes: return json_dumps({k: list(v) for k, v in x.items()}) diff --git a/mypy/test/testgraph.py b/mypy/test/testgraph.py index 3cf72f3faaa6b..aec6576189661 100644 --- a/mypy/test/testgraph.py +++ b/mypy/test/testgraph.py @@ -1,19 +1,11 @@ -"""Test cases for graph processing and other code in build.py.""" +"""Test cases for graph processing code in build.py.""" from __future__ import annotations import sys from collections.abc import Set as AbstractSet -from concurrent.futures import wait - -from mypy.build import ( - BuildManager, - BuildSourceSet, - State, - order_ascc, - sorted_components, - submit_inline, -) + +from mypy.build import BuildManager, BuildSourceSet, State, order_ascc, sorted_components from mypy.errors import Errors from mypy.fscache import FileSystemCache from mypy.graph_utils import strongly_connected_components, topsort @@ -144,29 +136,3 @@ def test_order_ascc(self) -> None: ascc = res[1] scc = order_ascc(graph, ascc) assert_equal(scc, ["d", "c", "b", "a"]) - - -class SubmitInlineSuite(Suite): - """Test the fallback used when threads are unavailable (e.g. on wasm).""" - - def test_result(self) -> None: - future = submit_inline(str, 5) - assert future.done() - assert future.result() == "5" - - def test_exception_is_only_raised_when_collected(self) -> None: - def fail() -> None: - raise ValueError("boom") - - future = submit_inline(fail) - assert future.done() - assert isinstance(future.exception(), ValueError) - with self.assertRaises(ValueError): - future.result() - - def test_wait(self) -> None: - # This is how build.py collects results of parse jobs. - futures = [submit_inline(str, i) for i in (1, 2)] - result = wait(futures) - assert result.done == set(futures) - assert not result.not_done diff --git a/mypy/util.py b/mypy/util.py index d48fd696f28cf..5ad45af77e329 100644 --- a/mypy/util.py +++ b/mypy/util.py @@ -988,10 +988,6 @@ def get_available_threads() -> int: if _AVAILABLE_THREADS is not None: return _AVAILABLE_THREADS - if not can_start_threads(): - _AVAILABLE_THREADS = 1 - return 1 - # This takes into account -X cpu_count and/or PYTHON_CPU_COUNT, but always # counts virtual cores (which is not what we want for CPU bound tasks). os_cpu_count = os.cpu_count() From 6cc5f5992a49f24dd97b5d87dc0f9799946bbeb0 Mon Sep 17 00:00:00 2001 From: Jukka Lehtosalo Date: Wed, 16 Sep 2026 12:00:45 +0100 Subject: [PATCH 3/3] Skip testDecodeErrorBlockerOnInitialRun-posix for now As suggested by Ivan. --- test-data/unit/fine-grained-blockers.test | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test-data/unit/fine-grained-blockers.test b/test-data/unit/fine-grained-blockers.test index d201e4f07c941..48faa2cf7752b 100644 --- a/test-data/unit/fine-grained-blockers.test +++ b/test-data/unit/fine-grained-blockers.test @@ -522,7 +522,7 @@ a.py: error: Cannot decode file: 'ascii' codec can't decode byte 0xc3 in positio == main:2: error: Argument 1 to "f" has incompatible type "int"; expected "str" -[case testDecodeErrorBlockerOnInitialRun-posix] +[case testDecodeErrorBlockerOnInitialRun-posix-skip] # Note that there's no test variant for Windows, since the above Windows test case is good enough. import a a.f(1)