From 58bab116f2452875fb52e0e200dfb5e279b62df0 Mon Sep 17 00:00:00 2001 From: Scott K Logan Date: Fri, 28 Aug 2026 15:49:15 -0500 Subject: [PATCH] Make job completion order deterministic, improve tests When multiple jobs complete in the same "tick" of the parallel execution loop, the `done_futures` set can result in different colcon return codes depending on which one happens to get processed first. A more desirable behavior is to prefer the return code of the job which was started first. While it is currently unlikely to see multiple jobs finish in the same tick due to strict concurrency pre-limiting, this becomes much more common under dynamic scheduling and resource-guarded execution. This change also includes an additional test for the parallel executor, and switches all of the job identifiers to use consistent identifiers, where they were previously using different values in the overall job dictionary from the identifier used on the job objects themselves. Assisted-by: Gemini 3.5 Flash --- colcon_parallel_executor/executor/parallel.py | 4 ++- test/test_executor_parallel.py | 27 ++++++++++++++----- 2 files changed, 23 insertions(+), 8 deletions(-) diff --git a/colcon_parallel_executor/executor/parallel.py b/colcon_parallel_executor/executor/parallel.py index f94a832..5c4cc87 100644 --- a/colcon_parallel_executor/executor/parallel.py +++ b/colcon_parallel_executor/executor/parallel.py @@ -168,7 +168,9 @@ async def _execute(self, args, jobs, *, on_error): f.identifier for f in futures.values()))) # check results of done futures - for done_future in done_futures: + for done_future in [ + f for f in futures.keys() if f in done_futures + ]: job = futures[done_future] del futures[done_future] # get result without raising an exception diff --git a/test/test_executor_parallel.py b/test/test_executor_parallel.py index 47093ae..1c6bc61 100644 --- a/test/test_executor_parallel.py +++ b/test/test_executor_parallel.py @@ -110,7 +110,7 @@ def test_parallel(): args = SimpleNamespace(parallel_workers=2) jobs = OrderedDict() - jobs['one'] = Job1() + jobs['job1'] = Job1() # success rc = extension.execute(args, jobs) @@ -119,8 +119,8 @@ def test_parallel(): ran_jobs.clear() # return error code - jobs['two'] = Job2() - jobs['four'] = Job4() + jobs['job2'] = Job2() + jobs['job4'] = Job4() rc = extension.execute(args, jobs) assert rc == 2 assert ran_jobs == ['job1'] @@ -144,22 +144,22 @@ def test_parallel(): ran_jobs.clear() # continue after error, keeping first error code - jobs['five'] = Job5() + jobs['job5'] = Job5() rc = extension.execute(args, jobs, on_error=OnError.continue_) assert rc == 2 assert ran_jobs == ['job1', 'job4'] ran_jobs.clear() # continue but skip downstream - jobs['six'] = Job6() - jobs['seven'] = Job7() + jobs['job6'] = Job6() + jobs['job7'] = Job7() rc = extension.execute(args, jobs, on_error=OnError.skip_downstream) assert rc == 2 assert ran_jobs == ['job1', 'job7', 'job4'] ran_jobs.clear() # exception - jobs['two'] = Job3() + jobs['job2'] = Job3() rc = extension.execute(args, jobs) assert isinstance(rc, RuntimeError) assert ran_jobs == ['job1'] @@ -400,3 +400,16 @@ def test_parallel_workers_zero(): assert rc == 0 assert set(ran_jobs) == {'job1', 'job2'} ran_jobs.clear() + + +def test_parallel_exception_skip_pending(): + extension = ParallelExecutorExtension() + args = SimpleNamespace(parallel_workers=1) + jobs = OrderedDict() + jobs['job3'] = Job3() + jobs['job4'] = Job4() + + rc = extension.execute(args, jobs, on_error=OnError.skip_pending) + assert isinstance(rc, RuntimeError) + assert ran_jobs == [] + ran_jobs.clear()