Add micro-benchmarks for the task SDK's supervisor comms

[dabla]

Draft, opened to start a discussion on the dev list. Based on main; the measurements below come from a fix on #62922, but the benchmarks do not depend on it.

related: #62922, #74579

#74579 asks agents to measure these paths by hand in the meantime; this PR is the repeatable, automated version of that step.

Why

Airflow has no automated check on the performance of its hot paths. A change that makes them slower is only noticed when someone happens to measure it by hand, often after a release: #71721 reports structured task logging in 3.2.0 multiplying high-volume task runtime by 2.64.

A fix on #62922 shows how much is at stake in a single method. CommsDecoder.asend, which every async SDK call of a task goes through (a Variable, a Connection, an XCom, a state store key), made two hops to a worker thread per request; it now makes one. Measured with the benchmarks of this PR, old comms.py against new, on the same machine, the middle value of three runs (pytest-codspeed, walltime mode, best time per round):

Benchmark	Before	After	Change
send: one sync round trip (code unchanged, the control)	11.5 µs	11.4 µs	none
asend: one async round trip	108.7 µs	58.0 µs	−47%
asend: 50 concurrent coroutines on one loop	18.6 ms	11.7 ms	−37%

On main the benchmarks give the "Before" column; the "After" column is the code of #62922.

An iterated async task makes such calls for every item, so the saving repeats thousands of times per task (the supervisor's and API server's own share of each call is not part of this measurement). The opposite change, an extra thread hop added in passing, would cost as much, and nothing in CI would notice today. A separate ad hoc measurement (4000 GetVariable round trips, median of 7 interleaved rounds) agrees: 525 µs to 318 µs per request.

The CI job runs CodSpeed in simulation mode, which counts instructions and leaves system calls (the socket reads and writes, the thread wake-ups) out of the measurement (CodSpeed docs). So the same change was measured that way too, old comms.py and executor.py against new, with CodSpeed's own valgrind (3.26.0.codspeed7) and cycle estimation on as in the action, each benchmark in its own process, the middle value of three runs:

Benchmark	Before	After	Change
send: one sync round trip (the control)	106,961 instr.	107,678 instr.	+0.7%
asend: one async round trip	554,228 instr.	411,956 instr.	−25.7%
asend: 50 concurrent coroutines on one loop	24.54 M instr.	16.22 M instr.	−33.9%

Runs of the same code differ by less than 0.7%. Simulation sees a smaller share of the saving than wall time (−26% against −47% for one round trip), because the kernel's part of a thread hop is not counted, but the change is still nearly forty times the run-to-run spread, so CI in simulation mode would flag the reverse change on the PR that makes it.

What

• task-sdk/benchmarks/: a new directory next to tests/, for benchmarks only. The first module, execution_time/test_comms_benchmark.py, measures one sync round trip, one async round trip and fifty concurrent async round trips against a supervisor thread on a socket pair. The task SDK's pytest configuration only collects tests/, so the benchmarks never run with the unit tests. The fake supervisor closes its socket however it ends, so an unexpected request fails the benchmark instead of hanging it.
• A benchmarks dependency group in task-sdk/pyproject.toml with pytest-codspeed (MIT). Benchmarks are plain pytest functions using its benchmark fixture. Locally they run with uv run --project task-sdk --group benchmarks pytest task-sdk/benchmarks --codspeed.
• .github/workflows/benchmarks.yml: runs them on pushes to main and on PRs touching task-sdk/src/airflow/sdk/execution_time/, the benchmarks, task-sdk/pyproject.toml or uv.lock, through CodSpeedHQ/action in simulation mode. That mode counts CPU instructions instead of measuring time, so the results do not depend on how busy the shared runner is, and CodSpeed comments each PR with the change against main. Every commit on main keeps its own run, as the baseline CodSpeed compares PRs with; the job stops after 15 minutes. The action is already on the ASF allow-list (CodSpeedHQ/action v5.2.1, pinned by SHA in apache/infrastructure-actions). It authenticates through OIDC, so no secret is needed for a public repository, but someone with admin rights has to enable the CodSpeed app for apache/airflow before the workflow can report anything. Until then the job runs the benchmarks and fails at the upload (401 Unauthorized), so the workflow must not be merged before the app is enabled.

Questions for the dev list

1. Do we want micro-benchmarks at all? AIP-59 (Performance tests framework, accepted, still in progress) measures the footprint of whole components: scheduler, workers, triggerer and the database. Micro-benchmarks would be the layer below it, for single methods on hot paths, catching a regression in the PR that introduces it. Is that an extension of AIP-59, or does it need an AIP of its own?
2. Where should they live and how should they be named? This draft uses <distribution>/benchmarks/, next to tests/, with paths that mirror the source module a benchmark covers: src/airflow/sdk/execution_time/comms.py is measured by benchmarks/execution_time/test_comms_benchmark.py. That mapping is what lets CI run only the benchmarks covering the files a PR changes, instead of the hand-kept path filter the workflow uses in this draft. The test_*_benchmark.py names let the existing pytest configuration collect them without changes; a bench_*.py convention would need its own python_files setting.
3. Which tool? pytest-codspeed with the hosted service compares against main on each PR, with results that do not depend on runner load. pytest-benchmark is purely local and compares against a saved baseline with a threshold, but wall-time thresholds on shared CI runners are noisy, so they either fail at random or are set too loose to catch a drift.
4. What should a regression do? Report only (a PR comment), or fail the check above some threshold?
5. Which paths come first? Candidates beyond the comms: XCom serialisation, template rendering, the DAG processor's parsing, and the scheduler's critical section queries.

No newsfragment: nothing here is user-facing.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.