Using threads for multiprocessing by default, but it could be changed via concurrentProcesses property in 'toolchain.json'
This commit is contained in:
@@ -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()
|
||||
|
||||
@@ -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"):
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user