Feat/schedulers

idf_build_apps/args.py

@@ -830,6 +830,41 @@ 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:<pipeline-id>:<job-group-name>.',
+        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:<pipeline-id>:<job-name>.',
+        default=None,
+        exclude=True,  # computed field is used
+    )
     dry_run: Annotated[
         bool,
         CliOption(
@@ -952,6 +987,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 +1009,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]:

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:
@@ -129,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(
@@ -138,8 +142,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 +168,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 +177,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 +201,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(

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:
+        pass
+
+    def mark_success(self, index: int, app: App) -> None:
+        pass
+
+    def mark_failed(self, index: int, app: App) -> None:
+        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)