Dev-iL commented on PR #75396: URL: https://github.com/apache/airflow/pull/75396#issuecomment-6712061255
> the PostgreSQL large-value row reports a median of about 165 ms but a maximum of 553 and 556 ms in the two runs. [...] Do you know what causes it? The harness isn't in the PR, so reviewers can't check. On the ~550 ms PostgreSQL maximum for the large value: I re-measured and couldn't reproduce it, and I can't name its cause. I re-ran it in Breeze, PostgreSQL, Python 3.14.10, SQLAlchemy 2.0.55, alternating base (`f82542a453`) and head (`9679c1a12c`). Cells are median / maximum warm stall in ms: | Revision | Run 1 | Run 2 | | --- | --- | --- | | Base | 1068 / 2116 | 1004 / 1097 | | This PR | 188 / 219 | 193 / 245 | Longer head runs stayed in the same range: 40 warm requests twice, 160 / 234 and 171 / 251, then 150 warm requests, 156 / 259. Across 230 warm requests the largest stall was 259 ms. After the review changes, this PR (`b63da48219`, the current commit before its rebase onto a newer `main`) measured 196 / 217 and 190 / 222 against `main` at `2e9ae2a199` (1010 / 1096 and 974 / 1120). The original runs used base `8cc830d4ef`, Python 3.14.8, SQLAlchemy 2.0.54 and an older Breeze image, so I can't say which difference removed it. What I could rule out, from the probe below, which records garbage-collector pauses and which thread they ran on: - **A full GC on the loop thread during a warm request.** On base, every warm request runs a gen-2 collection of 400–730 ms, which is most of its ~1 s stall. On this PR, gen-2 collections ran on the loop only during the dropped first request (272–389 ms). In the 230 warm requests they ran three times, about every 44 requests, at 398–443 ms. All three ran on the worker thread, and those requests' loop stalls were 152–186 ms. One such collection on the loop on top of the ~165 ms median would give about the observed 550 ms, but I didn't observe that, so it's a guess. - **psycopg auto-prepare.** #74581 turned it off before both bases. On the loop, each large read now spends about 95 ms in psycopg's JSON decoding and about 60 + 57 ms in Cadwyn's decode and re-encode of the `JSONResponse` body (zmievsa/cadwyn#51). That accounts for the remaining median. <details><summary>Probe (disposable, not part of the PR)</summary> Copy it to `dev/xcom_stall_probe.py` in a checkout of this branch, then run: ```bash breeze run --python 3.14 --backend postgres bash -c "python dev/xcom_stall_probe.py --scenario big --requests 11 --label head --out dev/xcom_stall_out.jsonl" ``` ```python #!/usr/bin/env python # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. """ Event-loop stall probe for the Execution API XCom routes (disposable; not part of any PR). Run inside Breeze from the checkout root, e.g.: breeze run --python 3.14 --backend postgres bash -c "python dev/xcom_stall_probe.py \ --scenario big --requests 11 --label head-run1 --out dev/xcom_stall_out.jsonl" The probe is injected into pytest collection of the Execution API XCom test module, so it gets that directory's ``client``, ``dag_maker``, ``session`` and ``exec_app`` fixtures. A 1 ms ticker on the server's event loop records each sleep's own duration; the maximum during a request is its stall. The HTTP request goes through ``client`` from a worker thread, so routing, dependencies, validation and serialization are all measured. Every response's status and size are asserted. Per request it also records: - connections: checkouts from the sync and async pools, and the peak number held at once; - async_idle_ms: time the async connection stayed checked out while a sync connection was also out; - gc: garbage-collector pauses, with the thread they ran on (loop thread or not). """ from __future__ import annotations import argparse import asyncio import gc import json import os import statistics import sys import threading import time import pytest from fastapi import Request TEST_MODULE = "airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py" SCENARIOS = ("big", "small-owner", "small-multiteam", "loop-cold", "loop-warm") def parse_args(): parser = argparse.ArgumentParser() parser.add_argument("--scenario", choices=SCENARIOS, required=True) parser.add_argument("--requests", type=int, default=11) parser.add_argument("--label", required=True) parser.add_argument("--out", required=True) parser.add_argument("--profile", action="store_true", help="profile the loop thread on 1 extra request") parser.add_argument("--loop-tasks", type=int, default=1000) return parser.parse_args() ARGS = parse_args() class PoolTracker: """Track checkouts across the sync and async pools of the metadata DB.""" def __init__(self): self.lock = threading.Lock() self.reset() def reset(self): with self.lock: self.held = {"sync": 0, "async": 0} self.checkouts = {"sync": 0, "async": 0} self.peak = 0 self.async_out_since = None self.overlap_start = None self.overlap_ms = 0.0 def _total(self): return self.held["sync"] + self.held["async"] def _update_overlap(self, now): both = self.held["sync"] > 0 and self.held["async"] > 0 if both and self.overlap_start is None: self.overlap_start = now elif not both and self.overlap_start is not None: self.overlap_ms += (now - self.overlap_start) * 1000 self.overlap_start = None def install(self, sync_engine, async_engine): from sqlalchemy import event for kind, engine in (("sync", sync_engine), ("async", async_engine.sync_engine)): def on_checkout(dbapi_conn, record, proxy, kind=kind): with self.lock: self.held[kind] += 1 self.checkouts[kind] += 1 self.peak = max(self.peak, self._total()) self._update_overlap(time.monotonic()) def on_checkin(dbapi_conn, record, kind=kind): with self.lock: self.held[kind] = max(0, self.held[kind] - 1) self._update_overlap(time.monotonic()) event.listen(engine.pool, "checkout", on_checkout) event.listen(engine.pool, "checkin", on_checkin) def snapshot(self): with self.lock: if self.overlap_start is not None: self._update_overlap(time.monotonic()) return { "checkouts": dict(self.checkouts), "peak_held": self.peak, "both_pools_held_ms": round(self.overlap_ms, 2), } class GcTracker: def __init__(self): self.events = [] self.loop_thread = None self._start = {} def __call__(self, phase, info): tid = threading.get_ident() if phase == "start": self._start[tid] = time.monotonic() elif tid in self._start: ms = (time.monotonic() - self._start.pop(tid)) * 1000 self.events.append( {"gen": info["generation"], "ms": round(ms, 2), "on_loop": tid == self.loop_thread} ) def take(self): events, self.events = self.events, [] return events def seed_big(dag_maker, session): from airflow.models.xcom import XComModel from airflow.providers.standard.operators.empty import EmptyOperator with dag_maker(dag_id="stall_big", serialized=True): EmptyOperator(task_id="producer") >> EmptyOperator(task_id="consumer") dr = dag_maker.create_dagrun() tis = {ti.task_id: ti for ti in dr.task_instances} value = [{"index": i, "name": f"item-{i:06d}"} for i in range(100_000)] XComModel.set_for_attempt( task_instance_id=tis["producer"].id, key="big", value=value, serialize=False, session=session ) session.commit() url = f"/execution/xcoms/{dr.dag_id}/{dr.run_id}/producer/big" return url, tis["consumer"], lambda body: len(body["value"]) == 100_000 def seed_small(dag_maker, session): from airflow.models.xcom import XComModel from airflow.providers.standard.operators.empty import EmptyOperator with dag_maker(dag_id="stall_small", serialized=True): EmptyOperator(task_id="producer") >> EmptyOperator(task_id="consumer") dr = dag_maker.create_dagrun() tis = {ti.task_id: ti for ti in dr.task_instances} XComModel.set_for_attempt( task_instance_id=tis["producer"].id, key="small", value={"a": 1}, serialize=False, session=session ) session.commit() url = f"/execution/xcoms/{dr.dag_id}/{dr.run_id}/producer/small" return url, tis["consumer"], lambda body: body["value"] == {"a": 1} def seed_loop(dag_maker, session): from airflow.models.dynamic_region import DynamicRegion from airflow.models.xcom import XComModel from airflow.providers.standard.operators.empty import EmptyOperator from airflow.sdk import task_group from airflow.sdk.definitions._internal.loop import create_loop @task_group def body(): EmptyOperator(task_id="producer") >> EmptyOperator(task_id="consumer") with dag_maker(dag_id="stall_loop", serialized=True): start = EmptyOperator(task_id="outside") start >> create_loop(body, max_iterations=3) for i in range(ARGS.loop_tasks): start >> EmptyOperator(task_id=f"filler_{i:04d}") dr = dag_maker.create_dagrun() region = DynamicRegion.get_or_create(dag_id=dr.dag_id, run_id=dr.run_id, node_id="body", session=session) session.add(region) session.flush() tis = {ti.task_id: ti for ti in dr.task_instances} producer, consumer = tis["body.producer"], tis["body.consumer"] producer.region_id, producer.region_index = region.id, 1 consumer.region_id, consumer.region_index = region.id, 1 session.flush() XComModel.set_for_attempt( task_instance_id=producer.id, key="key", value="current", serialize=False, session=session ) session.commit() url = f"/execution/xcoms/{dr.dag_id}/{dr.run_id}/{producer.task_id}/key" return url, consumer, lambda body: body["value"] == "current" def probe(client, dag_maker, session, exec_app): from airflow import settings from airflow.api_fastapi.execution_api.datamodels.token import TIClaims, TIToken from airflow.api_fastapi.execution_api.security import require_auth from tests_common.test_utils.config import conf_vars scenario = ARGS.scenario seed = {"big": seed_big, "loop-cold": seed_loop, "loop-warm": seed_loop}.get(scenario, seed_small) url, caller, check = seed(dag_maker, session) session.close() async def authenticated(request: Request) -> TIToken: return TIToken(id=caller.id, claims=TIClaims(scope="execution")) exec_app.dependency_overrides[require_auth] = authenticated pools = PoolTracker() pools.install(settings.engine, settings.async_engine) gc_tracker = GcTracker() gc.callbacks.append(gc_tracker) import sqlalchemy env = { "python": sys.version.split()[0], "sqlalchemy": sqlalchemy.__version__, "async_driver": settings.async_engine.url.drivername, "sync_driver": settings.engine.url.drivername, "loadavg": os.getloadavg(), } def one_request(profile=False): if scenario == "loop-cold": exec_app.state.dag_bag.clear_cache() pools.reset() gc_tracker.take() async def run(): gc_tracker.loop_thread = threading.get_ident() stalls: list[float] = [] done = asyncio.Event() async def ticker(): while not done.is_set(): t0 = time.monotonic() await asyncio.sleep(0.001) stalls.append((time.monotonic() - t0) * 1000) task = asyncio.create_task(ticker()) await asyncio.sleep(0.01) stalls.clear() prof = None if profile: import cProfile prof = cProfile.Profile() prof.enable() t0 = time.monotonic() response = await asyncio.to_thread(client.get, url) elapsed = (time.monotonic() - t0) * 1000 if prof is not None: prof.disable() done.set() await task return response, elapsed, stalls, prof response, elapsed, stalls, prof = client.portal.call(run) assert response.status_code == 200, response.text[:500] assert check(response.json()), response.text[:200] return { "request_ms": round(elapsed, 2), "stall_max_ms": round(max(stalls), 2) if stalls else 0.0, "ticks": len(stalls), "bytes": len(response.content), **pools.snapshot(), "gc": gc_tracker.take(), }, prof ctx = conf_vars({("core", "multi_team"): "True"}) if scenario == "small-multiteam" else None if ctx is not None: ctx.__enter__() try: samples = [one_request()[0] for _ in range(ARGS.requests)] profile_text = None if ARGS.profile: import io import pstats _, prof = one_request(profile=True) buf = io.StringIO() pstats.Stats(prof, stream=buf).sort_stats("tottime").print_stats(25) profile_text = buf.getvalue() finally: if ctx is not None: ctx.__exit__(None, None, None) gc.callbacks.remove(gc_tracker) warm = samples[1:] summary = { "label": ARGS.label, "scenario": scenario, "env": env, "stall_median_ms": round(statistics.median(s["stall_max_ms"] for s in warm), 2), "stall_max_ms": round(max(s["stall_max_ms"] for s in warm), 2), "request_median_ms": round(statistics.median(s["request_ms"] for s in warm), 2), "peak_held_max": max(s["peak_held"] for s in warm), "checkouts_last": warm[-1]["checkouts"], "both_pools_held_median_ms": round(statistics.median(s["both_pools_held_ms"] for s in warm), 2), "samples": samples, "profile": profile_text, } with open(ARGS.out, "a") as f: f.write(json.dumps(summary) + "\n") print( f"RESULT {ARGS.label} {scenario}: stall median/max {summary['stall_median_ms']}/" f"{summary['stall_max_ms']} ms, request median {summary['request_median_ms']} ms, " f"peak held {summary['peak_held_max']}, both-pools median {summary['both_pools_held_median_ms']} ms, " f"per-request stalls {[s['stall_max_ms'] for s in samples]}" ) class _Swap: @pytest.hookimpl(trylast=True) def pytest_collection_modifyitems(self, session, config, items): anchor = next(item for item in items if isinstance(item.parent, pytest.Module)) items[:] = [pytest.Function.from_parent(anchor.parent, name="xcom_stall_probe", callobj=probe)] if __name__ == "__main__": sys.exit(pytest.main([TEST_MODULE, "-x", "-q", "-p", "no:cacheprovider", "-s"], plugins=[_Swap()])) ``` </details> --- Drafted-by: Claude Code (Opus 5.5) (no human review before posting) -- 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]
