From 64db92f9ceba2e8353ca8ca160bd00ef2a7be40f Mon Sep 17 00:00:00 2001 From: "igor.udot" Date: Wed, 3 Jun 2026 17:04:14 +0800 Subject: [PATCH 1/3] feat: diff schedulers --- idf_build_apps/args.py | 81 +++++++++++++++++++++++++++++++++++ idf_build_apps/main.py | 48 +++++++++++++-------- idf_build_apps/scheduler.py | 85 +++++++++++++++++++++++++++++++++++++ 3 files changed, 197 insertions(+), 17 deletions(-) create mode 100644 idf_build_apps/scheduler.py diff --git a/idf_build_apps/args.py b/idf_build_apps/args.py index 393b801..d52f87d 100644 --- a/idf_build_apps/args.py +++ b/idf_build_apps/args.py @@ -830,6 +830,42 @@ class BuildArguments(FindBuildArguments): 'and the current run will build the parallel_index-th part', default=1, ) + parallel_mode: Annotated[ + t.Literal['slice', 'pull'], + CliOption( + choices=['slice', 'pull'], + ), + ] = Field( + description='Parallel scheduling mode. "slice" divides apps by --parallel-count/--parallel-index. ' + '"pull" pulls app indexes from a shared queue.', + default='slice', + ) + pull_queue: Annotated[ + t.Optional[t.Literal['redis']], + CliOption( + choices=['redis'], + ), + ] = Field( + description='Queue backend used when --parallel-mode=pull', + default=None, + ) + pull_queue_url: t.Optional[str] = Field( + description='Queue backend URL used when --parallel-mode=pull. Defaults to REDIS_URL.', + default=None, + exclude=True, # computed field is used + ) + pull_queue_key: t.Optional[str] = Field( + description='Queue key used when --parallel-mode=pull. ' + 'Defaults to job-pull::.', + default=None, + exclude=True, # computed field is used + ) + pull_queue_status_key: t.Optional[str] = Field( + description='Redis hash key for per-app build status in pull mode. ' + 'Defaults to job-app-status::.', + default=None, + exclude=True, # computed field is used + ) dry_run: Annotated[ bool, CliOption( @@ -952,6 +988,20 @@ def model_post_init(self, __context: Any) -> None: App.IGNORE_WARNS_REGEXES = [re.compile(p.strip()) for p in patterns if p.strip()] + if self.parallel_mode == 'pull': + if self.pull_queue != 'redis': + raise InvalidCommand('--pull-queue=redis is required when --parallel-mode=pull.') + if not self.resolved_pull_queue_key: + raise InvalidCommand( + '--pull-queue-key is required when --parallel-mode=pull unless CI_PIPELINE_ID/PARENT_PIPELINE_ID ' + 'and CI_JOB_GROUP_NAME are set.' + ) + if not self.resolved_pull_queue_status_key: + raise InvalidCommand( + '--pull-queue-status-key is required when --parallel-mode=pull unless ' + 'CI_PIPELINE_ID/PARENT_PIPELINE_ID and CI_JOB_NAME are set.' + ) + @computed_field # type: ignore @property def collect_size_info(self) -> t.Optional[str]: @@ -960,6 +1010,37 @@ def collect_size_info(self) -> t.Optional[str]: return None + @computed_field # type: ignore + @property + def resolved_pull_queue_url(self) -> str: + return self.pull_queue_url or os.getenv('REDIS_URL', 'redis://192.168.2.147:16379/0') + + @computed_field # type: ignore + @property + def resolved_pull_queue_key(self) -> t.Optional[str]: + if self.pull_queue_key: + return self.pull_queue_key + + pipeline_id = os.getenv('PARENT_PIPELINE_ID') or os.getenv('CI_PIPELINE_ID') + job_group_name = os.getenv('CI_JOB_GROUP_NAME') + if pipeline_id and job_group_name: + return f'job-pull:{pipeline_id}:{job_group_name}' + + return None + + @computed_field # type: ignore + @property + def resolved_pull_queue_status_key(self) -> t.Optional[str]: + if self.pull_queue_status_key: + return self.pull_queue_status_key + + pipeline_id = os.getenv('PARENT_PIPELINE_ID') or os.getenv('CI_PIPELINE_ID') + job_name = os.getenv('CI_JOB_NAME') + if pipeline_id and job_name: + return f'job-app-status:{pipeline_id}:{job_name}' + + return None + @computed_field # type: ignore @property def collect_app_info(self) -> t.Optional[str]: diff --git a/idf_build_apps/main.py b/idf_build_apps/main.py index 5499a85..e36ce0e 100644 --- a/idf_build_apps/main.py +++ b/idf_build_apps/main.py @@ -33,10 +33,12 @@ from .junit import TestSuite from .manifest.manifest import DEFAULT_BUILD_TARGETS from .manifest.manifest import Manifest +from .scheduler import AppScheduler +from .scheduler import RedisPullScheduler +from .scheduler import SliceScheduler from .utils import AutocompleteActivationError from .utils import InvalidCommand from .utils import drop_none_kwargs -from .utils import get_parallel_start_stop from .utils import to_list LOGGER = logging.getLogger(__name__) @@ -103,6 +105,7 @@ def build_apps( apps: t.Union[t.List[App], App, None] = None, *, build_arguments: t.Optional[BuildArguments] = None, + scheduler: t.Optional[AppScheduler] = None, config_file: t.Optional[str] = None, **kwargs, ) -> int: @@ -138,8 +141,15 @@ def build_apps( test_suite = TestSuite('build_apps') - start, stop = get_parallel_start_stop(len(apps), build_arguments.parallel_count, build_arguments.parallel_index) - LOGGER.info('Processing %d total apps: building apps %d-%d', len(apps), start, stop) + if scheduler is None: + if build_arguments.parallel_mode == 'pull': + scheduler = RedisPullScheduler( + build_arguments.resolved_pull_queue_url, + build_arguments.resolved_pull_queue_key or '', + status_key=build_arguments.resolved_pull_queue_status_key, + ) + else: + scheduler = SliceScheduler(build_arguments.parallel_count, build_arguments.parallel_index) # cleanup collect files if exists at this early-stage for f in (build_arguments.collect_app_info, build_arguments.collect_size_info, build_arguments.junitxml): @@ -157,11 +167,7 @@ def build_apps( LOGGER.debug('Creating empty size info file: %s', build_arguments.collect_size_info) Path(build_arguments.collect_size_info).touch() - for i, app in enumerate(apps): - index = i + 1 # we use 1-based - if index < start or index > stop: - continue - + for index, app in scheduler.iter_apps(apps): # attrs app.dry_run = build_arguments.dry_run app.index = index @@ -170,12 +176,17 @@ def build_apps( LOGGER.info('(%d/%d) Building app: %s', index, len(apps), app) - app.build( - manifest_rootpath=build_arguments.manifest_rootpath, - modified_components=build_arguments.modified_components, - modified_files=build_arguments.modified_files, - check_app_dependencies=build_arguments.dependency_driven_build_enabled, - ) + scheduler.mark_running(index, app) + try: + app.build( + manifest_rootpath=build_arguments.manifest_rootpath, + modified_components=build_arguments.modified_components, + modified_files=build_arguments.modified_files, + check_app_dependencies=build_arguments.dependency_driven_build_enabled, + ) + except Exception: + scheduler.mark_failed(index, app) + raise test_suite.add_test_case(TestCase.from_app(app)) if app.build_comment: @@ -189,14 +200,17 @@ def build_apps( LOGGER.debug('Recorded app info in file: %s', build_arguments.collect_app_info) if app.build_status == BuildStatus.FAILED: + scheduler.mark_failed(index, app) if not build_arguments.keep_going: LOGGER.error('Build failed and keep_going=False, stopping build process') - return 1 + exit_code = 1 + break else: LOGGER.warning('Build failed but keep_going=True, continuing with next app') exit_code = 1 - elif app.build_status == BuildStatus.SUCCESS: - if build_arguments.collect_size_info and app.size_json_path: + else: + scheduler.mark_success(index, app) + if app.build_status == BuildStatus.SUCCESS and build_arguments.collect_size_info and app.size_json_path: if os.path.isfile(app.size_json_path): with open(build_arguments.collect_size_info, 'a') as fw: fw.write( diff --git a/idf_build_apps/scheduler.py b/idf_build_apps/scheduler.py new file mode 100644 index 0000000..03fb617 --- /dev/null +++ b/idf_build_apps/scheduler.py @@ -0,0 +1,85 @@ +# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD +# SPDX-License-Identifier: Apache-2.0 + +import logging +import typing as t + +from .app import App +from .utils import InvalidCommand +from .utils import get_parallel_start_stop + +LOGGER = logging.getLogger(__name__) + + +class AppScheduler: + def iter_apps(self, apps: t.List[App]) -> t.Iterator[t.Tuple[int, App]]: + raise NotImplementedError + + def mark_running(self, index: int, app: App) -> None: # noqa: ARG002 + pass + + def mark_success(self, index: int, app: App) -> None: # noqa: ARG002 + pass + + def mark_failed(self, index: int, app: App) -> None: # noqa: ARG002 + pass + + +class SliceScheduler(AppScheduler): + def __init__(self, parallel_count: int = 1, parallel_index: int = 1) -> None: + self.parallel_count = parallel_count + self.parallel_index = parallel_index + + def iter_apps(self, apps: t.List[App]) -> t.Iterator[t.Tuple[int, App]]: + start, stop = get_parallel_start_stop(len(apps), self.parallel_count, self.parallel_index) + LOGGER.info('Processing %d total apps: building apps %d-%d', len(apps), start, stop) + + for i, app in enumerate(apps): + index = i + 1 # we use 1-based + if index < start or index > stop: + continue + + yield index, app + + +class RedisPullScheduler(AppScheduler): + STATUS_RUNNING = 'running' + STATUS_SUCCESS = 'success' + STATUS_FAILED = 'failed' + + def __init__(self, redis_url: str, queue_key: str, *, status_key: t.Optional[str] = None) -> None: + try: + import redis + except ImportError as e: + raise InvalidCommand('Redis pull mode requires the "redis" Python package to be installed.') from e + + self.redis_client = redis.Redis.from_url(redis_url) + self.queue_key = queue_key + self.status_key = status_key + + def iter_apps(self, apps: t.List[App]) -> t.Iterator[t.Tuple[int, App]]: + LOGGER.info('Processing %d total apps in Redis pull mode from queue %s', len(apps), self.queue_key) + + while True: + app_id = self.redis_client.incr(self.queue_key) - 1 + if app_id >= len(apps): + LOGGER.info('No more apps to build') + return + if app_id < 0: + raise InvalidCommand(f'Redis queue {self.queue_key} returned invalid app id {app_id}.') + + index = app_id + 1 + yield index, apps[app_id] + + def _set_status(self, index: int, status: str) -> None: + if self.status_key: + self.redis_client.hset(self.status_key, index - 1, status) + + def mark_running(self, index: int, app: App) -> None: # noqa: ARG002 + self._set_status(index, self.STATUS_RUNNING) + + def mark_success(self, index: int, app: App) -> None: # noqa: ARG002 + self._set_status(index, self.STATUS_SUCCESS) + + def mark_failed(self, index: int, app: App) -> None: # noqa: ARG002 + self._set_status(index, self.STATUS_FAILED) From 538254b6df6c87c556f68b683896ec811c250919 Mon Sep 17 00:00:00 2001 From: "igor.udot" Date: Wed, 3 Jun 2026 17:05:47 +0800 Subject: [PATCH 2/3] feat: diff schedulers --- idf_build_apps/main.py | 1 + 1 file changed, 1 insertion(+) diff --git a/idf_build_apps/main.py b/idf_build_apps/main.py index e36ce0e..b265420 100644 --- a/idf_build_apps/main.py +++ b/idf_build_apps/main.py @@ -132,6 +132,7 @@ def build_apps( ) apps = to_list(apps) + print("total apps number", len(apps)) if apps is None: apps = find_apps( find_arguments=FindArguments( From 4a28a65af5ee7871b312c80326e9d2b46487abe4 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Wed, 3 Jun 2026 10:07:38 +0000 Subject: [PATCH 3/3] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- idf_build_apps/args.py | 3 +-- idf_build_apps/main.py | 2 +- idf_build_apps/scheduler.py | 6 +++--- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/idf_build_apps/args.py b/idf_build_apps/args.py index d52f87d..732ec9b 100644 --- a/idf_build_apps/args.py +++ b/idf_build_apps/args.py @@ -855,8 +855,7 @@ class BuildArguments(FindBuildArguments): exclude=True, # computed field is used ) pull_queue_key: t.Optional[str] = Field( - description='Queue key used when --parallel-mode=pull. ' - 'Defaults to job-pull::.', + description='Queue key used when --parallel-mode=pull. Defaults to job-pull::.', default=None, exclude=True, # computed field is used ) diff --git a/idf_build_apps/main.py b/idf_build_apps/main.py index b265420..051068b 100644 --- a/idf_build_apps/main.py +++ b/idf_build_apps/main.py @@ -132,7 +132,7 @@ def build_apps( ) apps = to_list(apps) - print("total apps number", len(apps)) + print('total apps number', len(apps)) if apps is None: apps = find_apps( find_arguments=FindArguments( diff --git a/idf_build_apps/scheduler.py b/idf_build_apps/scheduler.py index 03fb617..f0a994c 100644 --- a/idf_build_apps/scheduler.py +++ b/idf_build_apps/scheduler.py @@ -15,13 +15,13 @@ class AppScheduler: def iter_apps(self, apps: t.List[App]) -> t.Iterator[t.Tuple[int, App]]: raise NotImplementedError - def mark_running(self, index: int, app: App) -> None: # noqa: ARG002 + def mark_running(self, index: int, app: App) -> None: pass - def mark_success(self, index: int, app: App) -> None: # noqa: ARG002 + def mark_success(self, index: int, app: App) -> None: pass - def mark_failed(self, index: int, app: App) -> None: # noqa: ARG002 + def mark_failed(self, index: int, app: App) -> None: pass