kaxil commented on code in PR #69007:
URL: https://github.com/apache/airflow/pull/69007#discussion_r3484677820
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -347,8 +347,25 @@ def __init__(
if log:
self._log = log
+ dag_cache_size = conf.getint("scheduler", "dag_cache_size",
fallback=1024)
+ dag_cache_ttl_config = conf.getint("scheduler", "dag_cache_ttl",
fallback=10800)
- self.scheduler_dag_bag = DBDagBag(load_op_links=False)
+ if dag_cache_size < 0:
+ self.log.warning("scheduler dag_cache_size must be >= 0, using
unbounded dict")
+ dag_cache_size = 0
+
+ if dag_cache_ttl_config < 0:
+ self.log.warning("scheduler dag_cache_ttl must be >= 0, disabling
TTL")
+ dag_cache_ttl_config = 0
+
+ dag_cache_ttl = dag_cache_ttl_config if dag_cache_ttl_config > 0 else
None
+
+ self.scheduler_dag_bag = DBDagBag(
+ load_op_links=False,
+ cache_size=dag_cache_size,
+ cache_ttl=dag_cache_ttl,
+ stats_prefix="scheduler.dag_bag",
Review Comment:
This enables the bounded LRU/TTL cache for the scheduler, but the
scheduler's access pattern is the one case where a count-based cap backfires,
so I think the approach needs a rethink before merge.
#60804 deliberately left the scheduler on the unbounded dict (its
description: "The scheduler continues using a plain unbounded dict with zero
lock overhead") and enabled the bounded cache for the API server only, because
the access profiles differ:
- `get_dag_for_run()` is called for essentially every running DagRun the
scheduler processes.
- `DagRun.get_running_dag_runs_to_examine()` orders by
`last_scheduling_decision` (least-recently-scheduled first), so across
consecutive loops the scheduler round-robins through all running runs. The
per-loop `lru_cache()` wrappers only dedupe within a single loop; the
persistent cross-loop cache is this `DBDagBag`.
A cyclic sweep over N distinct `dag_version_id`s against an LRU/TTL cache of
`maxsize` M < N is the sequential-flooding case: each key is evicted just
before its next access, so the hit rate collapses toward zero once N > M. Every
miss then pays `session.get(DagVersion, ..., joinedload(serialized_dag))` plus
a full `SerializedDAG` deserialization on the scheduler hot path. The
deployments that hit this OOM are the large ones where active versions exceed
1024, so as written the default puts them straight into that regime. (Same
concern Pierre raised on the default, but it's really about the eviction
mechanism, not just the number.)
The leak here is superseded `dag_version_id`s accumulating: once a version
stops being referenced by running runs, it's never looked up again. TTL
eviction targets exactly those, while the refresh-on-revalidation write-back in
`_get_dag` keeps the hot active set resident regardless of its size. So
TTL-driven eviction (default `dag_cache_size=0`, or a safety-valve cap well
above realistic active-version counts, with the TTL doing the real bounding)
fixes the reported growth without the thrash. Note too that a count cap bounds
cardinality, not bytes -- 1024 large serialized DAGs can still be hundreds of
MB, whereas memray measured bytes retained.
Minor, worth noting in the description: enabling the cache also flips the
scheduler from `nullcontext` to a real `RLock` per `_get_dag`, which #60804
explicitly chose to avoid. Cheap when uncontended, so not blocking, but it
reverses a documented decision.
--
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]
