diff --git a/src/icmtoolchain/concurrent_build.py b/src/icmtoolchain/concurrent_build.py index 1971229..4aa71de 100644 --- a/src/icmtoolchain/concurrent_build.py +++ b/src/icmtoolchain/concurrent_build.py @@ -204,10 +204,13 @@ class ConcurrentBuilderApp: self.app.exit() -async def build_executor_loop(app: ConcurrentBuilderApp, scheduler: ConcurrentScheduler, task_names: List[str], max_workers: int, queue: Queue, all_logs: list, status_obj: BuildStatus): - from concurrent.futures import ProcessPoolExecutor +async def build_executor_loop(app: ConcurrentBuilderApp, scheduler: ConcurrentScheduler, task_names: List[str], max_workers: int, queue: Any, all_logs: list, status_obj: BuildStatus, use_processes: bool = False): + if use_processes: + from concurrent.futures import ProcessPoolExecutor as Executor + else: + from concurrent.futures import ThreadPoolExecutor as Executor loop = asyncio.get_event_loop() - with ProcessPoolExecutor(max_workers=max_workers) as executor: + with Executor(max_workers=max_workers) as executor: pending_futures: Dict[Any, Tuple[ProjectEdge, Optional[WorkerStatePane]]] = {} while scheduler.has_unfinished_tasks(): @@ -265,16 +268,24 @@ async def build_executor_loop(app: ConcurrentBuilderApp, scheduler: ConcurrentSc app.exit() async def run_concurrent_build(graph: 'ProjectGraph', task_names: List[str]): - import multiprocessing - if multiprocessing.get_start_method(allow_none=True) != "spawn": - preferred_method = "spawn" - if "forkserver" in multiprocessing.get_all_start_methods(): - preferred_method = "forkserver" - try: - multiprocessing.set_start_method(preferred_method, force=True) - except RuntimeError: - pass + use_processes = GLOBALS.TOOLCHAIN_CONFIG.get_value("concurrentProcesses", False) + if use_processes: + import multiprocessing + if multiprocessing.get_start_method(allow_none=True) != "spawn": + preferred_method = "spawn" + if "forkserver" in multiprocessing.get_all_start_methods(): + preferred_method = "forkserver" + try: + multiprocessing.set_start_method(preferred_method, force=True) + except RuntimeError: + pass + manager = multiprocessing.Manager() + queue = manager.Queue() + else: + queue = Queue() + + import sys if sys.version_info < (3, 13): from os import cpu_count as _cpu_count cpu_count = _cpu_count() @@ -285,14 +296,12 @@ async def run_concurrent_build(graph: 'ProjectGraph', task_names: List[str]): max_workers = max(1, cpu_count // 2 if cpu_count else 1) scheduler = ConcurrentScheduler(graph) - manager = multiprocessing.Manager() - queue = manager.Queue() app = ConcurrentBuilderApp(len(scheduler.pending), max_workers) all_logs = [] build_status = BuildStatus() - executor_task = asyncio.create_task(build_executor_loop(app, scheduler, task_names, max_workers, queue, all_logs, build_status)) + executor_task = asyncio.create_task(build_executor_loop(app, scheduler, task_names, max_workers, queue, all_logs, build_status, use_processes)) try: await app.run_async() diff --git a/src/icmtoolchain/context.py b/src/icmtoolchain/context.py index cc84cdd..31ea390 100644 --- a/src/icmtoolchain/context.py +++ b/src/icmtoolchain/context.py @@ -1,4 +1,5 @@ import os +import threading from copy import deepcopy from os.path import abspath, dirname, isfile, join from typing import TYPE_CHECKING, Iterable, Optional @@ -51,7 +52,7 @@ def get_current_directory() -> str: return project_path return os.getcwd() -class Globals: +class Globals(threading.local): @property def ADB_COMMAND(self): if not hasattr(self, "adb_command"): diff --git a/src/icmtoolchain/task.py b/src/icmtoolchain/task.py index 3702cb2..879eca8 100644 --- a/src/icmtoolchain/task.py +++ b/src/icmtoolchain/task.py @@ -1,3 +1,4 @@ +import threading from os.path import join from typing import Any, Callable, Dict, Final, List, Optional @@ -8,10 +9,24 @@ from .output_directory import get_temporary_directory, lock_file, unlock_file class Task: name: Final[str] description: str = "" - _status: Optional[str] = None callable: Callable locks: Optional[List[str]] = None - on_status_changed: Optional[Callable[[str], None]] = None + + @property + def _status(self) -> Optional[str]: + return getattr(self._local, "status", None) + + @_status.setter + def _status(self, value: Optional[str]) -> None: + self._local.status = value + + @property + def on_status_changed(self) -> Optional[Callable[[str], None]]: + return getattr(self._local, "on_status_changed", None) + + @on_status_changed.setter + def on_status_changed(self, value: Optional[Callable[[str], None]]) -> None: + self._local.on_status_changed = value @property def status(self) -> str: @@ -43,6 +58,7 @@ class Task: else: raise ValueError(f"Task {name!r} is already exists.") self.name = name + self._local = threading.local() if description: self.description = description if status: