Performance: count_waiting_tasks_for_metric loads all incomplete tasks into Python memory on every heartbeat

[dkliban]

Summary

PR #7938 replaced a simple COUNT(*) SQL query in record_waiting_tasks_metric with count_waiting_tasks_for_metric(), which iterates over every RUNNING/WAITING task in Python to compute resource-aware parallelism for the KEDA waiting_tasks metric. While the metric is now semantically correct (it counts parallelizable lanes rather than raw task count), the implementation has O(n) time and memory cost that is worth optimizing at production scale.

Affected code

pulpcore/tasking/redis_worker.py, function count_waiting_tasks_for_metric() (lines 102-150):

incomplete_tasks = (
    Task.objects.filter(
        state__in=[TASK_STATES.RUNNING, TASK_STATES.WAITING],
        pulp_created__lt=cutoff_time,
    )
    .order_by("pulp_created")
    .only("reserved_resources_record")
)

for task in incomplete_tasks:  # iterates ALL incomplete tasks
    exclusive_resources, shared_resources = extract_task_resources(task)
    # ... resource conflict checking with set operations ...
    parallel_count += 1

Note: this function is called from record_waiting_tasks_metric() which has the @exclusive(TASK_METRICS_LOCK) advisory lock decorator. The function itself is not lock-protected — only the calling method is. The code only runs when self.otel_enabled is True (checked at line 527).

Performance characteristics

What changed from the old code

The old implementation was a single COUNT(*) query — O(1) at the database level using the (state, pulp_created) composite index. The new implementation loads every matching row and iterates in Python with cumulative set operations for resource conflict tracking.

Estimated cost at production scale

At packages.redhat.com production scale (25-75 workers, 18,000+ tasks during peak):

• Memory: .only("reserved_resources_record") creates deferred model instances (~1-1.5 KB each: model overhead + PRN string array + UUID pk). For 18,000 tasks that's approximately 18-27 MB — noticeable but within the 8 GB worker memory limit.
• CPU: Python loop with set operations over 18,000 iterations takes well under 1 second.
• Database: Index scan of (state, pulp_created) returning 18,000+ rows with reserved_resources_record column.
• Frequency: Runs every ~30 seconds (every METRIC_HEARTBEAT_INTERVAL=3 heartbeats, with heartbeat period ~10 seconds). Only one worker runs it at a time due to the advisory lock.

The algorithm cannot short-circuit early because FIFO ordering requires processing all tasks in creation order — even blocked tasks must "reserve" their resources to prevent later tasks from being incorrectly counted as parallelizable (lines 141-144).

Mitigating factors

• The @exclusive(TASK_METRICS_LOCK) advisory lock on the calling method ensures only one worker computes this at a time. Other workers do a trivially cheap non-blocking pg_try_advisory_xact_lock check.
• .only("reserved_resources_record") limits columns fetched.
• The 5-second cutoff_time filter excludes very recent tasks.
• beat() blocking during this computation is unlikely to cause missed heartbeats — the loop completes well under the 10-second heartbeat period.

Suggested improvements (ordered by effort)

1. Increase METRIC_HEARTBEAT_INTERVAL (trivial)

Change from 3 to 6 heartbeats (60-second cycle). KEDA does not need sub-minute granularity for scaling decisions. This halves the computation frequency with no semantic cost.

2. Add .iterator() (easy)

incomplete_tasks = (
    Task.objects.filter(...)
    .order_by("pulp_created")
    .only("reserved_resources_record")
    .iterator()  # Stream rows instead of materializing all at once
)

Reduces peak Python memory from O(n) to O(1) for model instances. The database still scans and transmits all rows, so query time is unchanged — the benefit is purely memory-side.

3. Use values_list to avoid model instantiation (moderate)

for resources in (
    Task.objects.filter(...)
    .order_by("pulp_created")
    .values_list("reserved_resources_record", flat=True)
    .iterator()
):
    exclusive = [r for r in (resources or []) if not r.startswith("shared:")]
    shared = [r[7:] for r in (resources or []) if r.startswith("shared:")]
    # ... conflict checking ...

Avoids constructing Django model instances entirely. Requires inlining the resource extraction logic (currently in extract_task_resources() which expects a Task object), but the logic is simple (split by shared: prefix).

4. Cache result in Redis (moderate)

Cache the parallel_count in Redis with a short TTL (e.g., 30 seconds). The metric reporter reads the cached value instead of recomputing. The expensive computation runs at most once per TTL period regardless of how often the metric is reported.

Related

• PR test: refactor record_waiting_tasks_metric metric #7938 — introduced this change
• The fetch_task() method in the same file uses a similar resource-checking pattern but operates on a bounded FETCH_TASK_LIMIT = 20 tasks per iteration, avoiding the unbounded scan