steveahnahn opened a new pull request, #71914: URL: https://github.com/apache/airflow/pull/71914
`core.parallelism` caps nothing on LocalExecutor. With `parallelism = 4`, admitting `slots_available` workloads per scheduler loop and dispatching them: ``` loop 1: accepted=4 dispatched_total=4 running=0 slots_available=4 loop 2: accepted=4 dispatched_total=8 running=0 slots_available=4 loop 3: accepted=4 dispatched_total=12 running=0 slots_available=4 loop 4: accepted=4 dispatched_total=16 running=0 slots_available=4 loop 5: accepted=4 dispatched_total=20 running=0 slots_available=4 ``` Twenty workloads dispatched against a limit of four, and the executor reports full capacity throughout. `config.yml` documents the option as "the maximum number of task instances that can run concurrently per scheduler in Airflow, regardless of the worker count." ### Root cause `LocalExecutor._process_workloads` puts each workload on the activity queue and pops it from the queued dicts, but never records it in `self.running`. Every other executor does at dispatch — celery, kubernetes, ecs, batch and lambda all call `self.running.add(key)` in their `_process_workloads`. A grep for `running.add` in airflow-core returns nothing. The scheduler enforces parallelism by reading exactly this set: `slots_available` and `slots_occupied` are computed from `len(self.running)` plus the queued dicts, and `heartbeat` computes `open_slots = self.parallelism - len(self.running)`. With `running` permanently empty, every slot reads free the moment work is handed to a worker, and the scheduler keeps admitting more. The only remaining ceiling is pool size, so with the defaults 128 task instances can be outstanding against 32 workers, and the excess accumulates in the `activity_queue` OS pipe. This is a regression from #51009, which removed the Airflow 2 code path in executors. The deleted `BaseExecutor._process_tasks` held core's only `self.running.add(key)`, and LocalExecutor was the one in-tree executor relying on the base class for it. Two second-order effects of the empty set: `has_task()` can never return True, so the "this scheduler has this task already" guard in `process_executor_events` never suppresses a stale event for LocalExecutor; and unbounded admission is what lets the activity-queue pipe fill, which is the precondition for the dispatch deadlock reported in #70526. This PR restores the accounting; it does not claim to fix that deadlock. ### Fix Record the workload in `running` at dispatch. The release path already exists and is what every terminal state flows through: `sync()` drains the result queue into `change_state()`, which removes the key. The new test pins the full cycle — dispatch occupies a slot, `slots_available` reaches zero at parallelism, `has_task()` sees the dispatched workload, and the result releases the slot. related: #51009 related: #70526 --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Fable 5) Generated-by: Claude Code (Fable 5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
