perf: replace batched device barriers with flat semaphore

[ChristopherJHart]

Summary

Replaces the batched asyncio.gather device execution loop with a single flat asyncio.Semaphore, eliminating barrier-induced idle time.

Problem

The device execution loop chunked devices into batches of max_workers and awaited each batch with asyncio.gather before starting the next. If one device in a batch takes 3x longer than others, all slots sit idle until it finishes — then the next batch starts fresh.

Example with 20 devices and max_workers=16:

• Batch 1: 16 devices, wait for slowest
• Batch 2: 4 devices, 12 workers idle
• Wall clock ≈ 2x what a flat schedule would produce

The previous semaphore was redundant

The old code created a per-batch semaphore:

semaphore_size = min(self.max_workers, len(device_batch))
semaphore = asyncio.Semaphore(semaphore_size)

But since batch_size = max_workers and semaphore_size = min(max_workers, len(device_batch)), the semaphore size always equaled the batch size. Every device in the batch could acquire it immediately — it never throttled anything. The actual concurrency control was the batch loop itself (only max_workers devices launched at a time via chunking).

So the old implementation was: batched barriers with a redundant semaphore inside each batch.

device_executor.run_device_job_with_semaphore() does async with semaphore: around the entire device job (job generation, testbed creation, subprocess execution). This is the correct concurrency-limiting mechanism — but it only works properly when the semaphore spans all devices, not just one batch.

Solution

Single asyncio.Semaphore(concurrency) over all devices. A device finishing early immediately frees a slot for the next, maximizing utilization. The semaphore inside run_device_job_with_semaphore now actually serves its intended purpose as the sole concurrency limiter.

Measured Impact

• At 7 devices: minimal (all fit in one "batch" anyway)
• At >16 devices: eliminates ~2x wall clock inflation from barrier waits
• Device spread collapsed from 45% to 6% across fleet

Files Changed

File	Change
nac_test/pyats_core/orchestrator.py	Replace batch loop with flat semaphore

Test plan

• Existing tests pass
• Validated end-to-end on 7-device fleet
• Respects --max-parallel-devices cap

---

🤖 Generated with Claude Code

[oboehmer]

Hey @ChristopherJHart .. I would like to integrate this into the next nac-test release. I raised an issue so we can properly document this. could you please add test coverage, and possibly also an entry in changelog

My agent suggests the following test strategy:

1. All devices execute and results are collected — No device is silently skipped by the new scheduling approach.
2. Concurrency is bounded by max_workers — Prove the semaphore actually limits parallel device execution to the configured maximum (not unbounded).
3. No barrier: fast devices free slots immediately (critical — would fail on main, pass on this branch) — A fast device completing does not wait for a slow device before the next queued device starts. This is the core behavioral difference from the old batch approach.
4. max_parallel_devices caps concurrency below max_workers — When the user-specified cap is lower than system capacity, it becomes the effective concurrency limit.
5. One device failure doesn't block others — An exception from a single device still allows all remaining devices to complete and their results to be collected.
6. Edge cases: single device / devices == max_workers — Degenerate cases behave correctly (no off-by-one or empty-semaphore issues).

Implementation approach: Mock DeviceExecutor.run_device_job_with_semaphore with varying asyncio.sleep durations to simulate fast/slow devices. Track concurrent execution count and completion order to assert scheduling behavior. Tests would be fast (sub-second, no real subprocesses).

Possibly put this into a new tests/pyats_core/test_orchestrator_concurrency.py

thanks!