Implement resource guard extension point and worker limiter

colcon_parallel_executor/executor/parallel.py

@@ -19,6 +19,14 @@
 from colcon_core.subprocess import new_event_loop
 from colcon_core.subprocess import SIGINT_RESULT
 from colcon_parallel_executor.event.executor import ParallelStatus
+from colcon_parallel_executor.resource_guard import \
+    add_resource_guard_arguments
+from colcon_parallel_executor.resource_guard import \
+    initialize_resource_guard_extensions
+from colcon_parallel_executor.resource_guard import \
+    run_guarded_job
+from colcon_parallel_executor.resource_guard.execution_policy import \
+    ExecutionPolicyGuard
 
 logger = colcon_logger.getChild(__name__)
 
@@ -62,6 +70,8 @@ def add_arguments(self, *, parser):  # noqa: D102
                  "or '0' for no limit "
                  '(default: {max_workers_default})'.format_map(locals()))
 
+        add_resource_guard_arguments(parser=parser)
+
     def execute(self, args, jobs, *, on_error=OnError.interrupt):  # noqa: D102
         # avoid debug message from asyncio when colcon uses debug log level
         asyncio_logger = logging.getLogger('asyncio')
@@ -111,6 +121,11 @@ def execute(self, args, jobs, *, on_error=OnError.interrupt):  # noqa: D102
         return result
 
     async def _execute(self, args, jobs, *, on_error):
+        guards = await initialize_resource_guard_extensions(args)
+
+        policy_guard = ExecutionPolicyGuard(on_error)
+        guards.append(policy_guard)
+
         # count the number of dependent jobs for each job
         # in order to process jobs with more dependent jobs first
         recursive_dependent_counts = {}
@@ -142,19 +157,14 @@ async def _execute(self, args, jobs, *, on_error):
             # take "ready" jobs
             take_jobs = []
             for package_name, job, _ in ready_jobs:
-                # don't schedule more jobs then workers
-                # to prevent starting further jobs when a job fails
-                if args.parallel_workers:
-                    if len(futures) + len(take_jobs) >= args.parallel_workers:
-                        break
                 take_jobs.append((package_name, job))
                 del jobs[package_name]
 
             # pass them to the executor
             for package_name, job in take_jobs:
                 assert iscoroutinefunction(job.__call__), \
                     'Job is not a coroutine'
-                future = asyncio.ensure_future(job())
+                future = asyncio.ensure_future(run_guarded_job(job, guards))
                 futures[future] = job
 
             # wait for futures

colcon_parallel_executor/resource_guard/__init__.py

@@ -0,0 +1,150 @@
+# Copyright 2026 Open Source Robotics Foundation, Inc.
+# Licensed under the Apache License, Version 2.0
+
+import asyncio
+from contextlib import AsyncExitStack
+import traceback
+
+from colcon_core.logging import colcon_logger
+from colcon_core.plugin_system import instantiate_extensions
+from colcon_core.plugin_system import order_extensions_grouped_by_priority
+
+logger = colcon_logger.getChild(__name__)
+
+
+class JobIncompleteError(Exception):
+    """
+    Raised when a job does not finish successfully.
+
+    If `result` is None, the job was cleanly skipped before execution.
+    If `result` is not None, the job failed with a non-zero exit code.
+    """
+
+    def __init__(self, result=None):  # noqa: D107
+        super().__init__()
+        self.result = result
+
+
+class ResourceGuardExtensionPoint:
+    """
+    The interface for parallel execution resource guard extensions.
+
+    A resource guard provider provides asynchronous context managers that
+    act as gatekeepers to throttle job execution.
+    """
+
+    """The version of the resource guard extension interface."""
+    EXTENSION_POINT_VERSION = '1.0'
+
+    """The priority of resource guard extensions."""
+    PRIORITY = 100
+
+    def __init__(self):  # noqa: D107
+        super().__init__()
+
+    def add_arguments(self, *, parser):
+        """
+        Add command line arguments specific to the resource guard.
+
+        :param parser: The argument parser
+        """
+        pass
+
+    async def initialize(self, args):
+        """
+        Initialize the resource guard provider before starting job execution.
+
+        :param args: The parsed command line arguments
+        """
+        pass
+
+    def get_guard(self, job):
+        """
+        Get the asynchronous context manager (guard) for the specified job.
+
+        If this guard does not apply to the job, return None.
+
+        :param job: The job object
+        :returns: An object implementing __aenter__ and __aexit__, or None
+        """
+        return None
+
+
+def get_resource_guard_extensions(*, group_name=None):
+    """
+    Get the available resource guard extensions.
+
+    :rtype: OrderedDict
+    """
+    if group_name is None:
+        group_name = 'colcon_parallel_executor.resource_guard'
+
+    try:
+        extensions = instantiate_extensions(group_name)
+    except Exception as e:  # noqa: B902, F841
+        exc = traceback.format_exc()
+        logger.error(
+            'Exception in resource guard discovery: {e}\n{exc}'
+            .format_map(locals()))
+        return {}
+
+    for name, extension in extensions.items():
+        extension.RESOURCE_GUARD_NAME = name
+    return order_extensions_grouped_by_priority(extensions)
+
+
+def add_resource_guard_arguments(parser, *, extensions=None):
+    """Add command line arguments for the resource guard extensions."""
+    if extensions is None:
+        extensions = get_resource_guard_extensions()
+    for priority in sorted(extensions.keys()):
+        for name, guard in extensions[priority].items():
+            try:
+                retval = guard.add_arguments(parser=parser)
+                assert retval is None, 'add_arguments() should return None'
+            except Exception as e:  # noqa: B902
+                logger.error(
+                    'Failed to add arguments for resource guard: %s', e)
+
+
+async def initialize_resource_guard_extensions(args, *, extensions=None):
+    """
+    Initialize the resource guard extensions.
+
+    :param args: The parsed command line arguments
+    """
+    if extensions is None:
+        extensions = get_resource_guard_extensions()
+    guards = []
+    for priority in sorted(extensions.keys()):
+        for name, guard in extensions[priority].items():
+            try:
+                retval = await guard.initialize(args)
+                assert retval is None, 'initialize() should return None'
+                guards.append(guard)
+            except Exception as e:  # noqa: B902
+                logger.error(
+                    'Failed to initialize resource guard: %s', e)
+    return guards
+
+
+async def run_guarded_job(job, guards):
+    """
+    Acquire resource guards and execute the job inside their context.
+
+    :param job: The job coroutine function
+    :param guards: List of resource guard extension providers
+    """
+    try:
+        async with AsyncExitStack() as stack:
+            for provider in guards:
+                guard = provider.get_guard(job)
+                if guard is not None:
+                    await stack.enter_async_context(guard)
+            await asyncio.sleep(0)
+            result = await job()
+            if result:
+                raise JobIncompleteError(result)
+            return result
+    except JobIncompleteError as e:
+        return e.result

colcon_parallel_executor/resource_guard/execution_policy.py

@@ -0,0 +1,36 @@
+# Copyright 2026 Open Source Robotics Foundation, Inc.
+# Licensed under the Apache License, Version 2.0
+
+from colcon_core.executor import OnError
+from colcon_parallel_executor.resource_guard import JobIncompleteError
+from colcon_parallel_executor.resource_guard import \
+    ResourceGuardExtensionPoint
+
+
+class ExecutionPolicyGuard(ResourceGuardExtensionPoint):
+    """
+    An implicit resource guard that cleanly skips pending jobs.
+
+    This is activated when a failure has occurred during execution.
+    """
+
+    def __init__(self, on_error):  # noqa: D107
+        super().__init__()
+        self.on_error = on_error
+        self.skip_all = False
+
+    def get_guard(self, job):  # noqa: D102
+        return self
+
+    async def __aenter__(self):  # noqa: D105
+        if self.skip_all:
+            raise JobIncompleteError()
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):  # noqa: D105
+        if (
+            exc_type is not None and
+            self.on_error in (OnError.interrupt, OnError.skip_pending) and
+            (exc_type is not JobIncompleteError or exc_val.result is not None)
+        ):
+            self.skip_all = True
+        return False

colcon_parallel_executor/resource_guard/worker_limiter.py

@@ -0,0 +1,25 @@
+# Copyright 2026 Open Source Robotics Foundation, Inc.
+# Licensed under the Apache License, Version 2.0
+
+import asyncio
+
+from colcon_core.plugin_system import satisfies_version
+from colcon_parallel_executor.resource_guard import ResourceGuardExtensionPoint
+
+
+class WorkerLimiterGuard(ResourceGuardExtensionPoint):
+    """Limits concurrent jobs using a semaphore."""
+
+    PRIORITY = 100
+
+    def __init__(self):  # noqa: D107
+        super().__init__()
+        satisfies_version(
+            ResourceGuardExtensionPoint.EXTENSION_POINT_VERSION, '^1.0')
+
+    async def initialize(self, args):  # noqa: D102
+        workers = getattr(args, 'parallel_workers', 0)
+        self._semaphore = asyncio.Semaphore(workers) if workers > 0 else None
+
+    def get_guard(self, job):  # noqa: D102
+        return self._semaphore

setup.cfg

@@ -65,6 +65,10 @@ colcon_core.event_handler =
     parallel_status = colcon_parallel_executor.event_handler.parallel_status:ParallelStatusEventHandler
 colcon_core.executor =
     parallel = colcon_parallel_executor.executor.parallel:ParallelExecutorExtension
+colcon_core.extension_point =
+    colcon_parallel_executor.resource_guard = colcon_parallel_executor.resource_guard:ResourceGuardExtensionPoint
+colcon_parallel_executor.resource_guard =
+    default_worker_limiter = colcon_parallel_executor.resource_guard.worker_limiter:WorkerLimiterGuard
 
 [flake8]
 import-order-style = google

test/run_until_complete.py

@@ -0,0 +1,16 @@
+# Copyright 2016-2018 Dirk Thomas
+# Licensed under the Apache License, Version 2.0
+
+import asyncio
+
+from colcon_core.subprocess import new_event_loop
+
+
+def run_until_complete(coroutine):
+    loop = new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        return loop.run_until_complete(coroutine)
+    finally:
+        loop.close()
+        assert loop.is_closed()

test/spell_check.words

@@ -1,6 +1,9 @@
+aenter
+aexit
 apache
 argparse
 asyncio
+autouse
 capsys
 colcon
 contextlib
@@ -20,6 +23,8 @@ pydocstyle
 pytest
 readouterr
 returncode
+retval
+rtype
 scspell
 setuptools
 sigint

test/test_executor_parallel.py

@@ -19,11 +19,25 @@
 from colcon_parallel_executor.executor.parallel import counting_number
 from colcon_parallel_executor.executor.parallel \
     import ParallelExecutorExtension
+from colcon_parallel_executor.resource_guard.worker_limiter import \
+    WorkerLimiterGuard
 import pytest
 
 ran_jobs = []
 
 
+@pytest.fixture(autouse=True, scope='module')
+def default_execution_policies():
+    """Mock extension discovery to isolate executor tests."""
+    guard = WorkerLimiterGuard()
+    with patch(
+        'colcon_parallel_executor.resource_guard.'
+        'instantiate_extensions',
+        return_value={'default_worker_limiter': guard}
+    ):
+        yield
+
+
 class Job1(Job):
 
     def __init__(self, identifier='job1'):