From ef4f7a8ba4cdd59dcf13639636dd801cfab7d891 Mon Sep 17 00:00:00 2001 From: Dennis Kliban Date: Sat, 15 Aug 2026 22:08:24 -0400 Subject: [PATCH 1/3] fix: prevent head-of-line blocking in RedisWorker fetch_task() Track blocked resources across fetch_task() iterations and exclude them at the DB level using reserved_resources_record__overlap, leveraging the existing partial GIN index. This prevents workers from re-scanning thousands of blocked tasks when free resources exist further in the queue. Closes: #7900 Co-Authored-By: Claude Opus 4.6 (1M context) --- pulpcore/tasking/redis_worker.py | 39 +++- .../tasking/test_fetch_task_hol_blocking.py | 145 ++++++++++++ .../unit/tasking/test_fetch_task_scale.py | 214 ++++++++++++++++++ 3 files changed, 395 insertions(+), 3 deletions(-) create mode 100644 pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py create mode 100644 pulpcore/tests/unit/tasking/test_fetch_task_scale.py diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index 99ce297413..11005d8110 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -599,27 +599,43 @@ def fetch_task(self): 3. If resource locks acquired, attempts to claim the task with a Redis task lock (24h expiration) 4. Returns the first task for which both locks can be acquired - 5. If no task found in the batch, doubles the fetch limit and retries from the oldest + 5. Tracks blocked resources across iterations and excludes them at the DB level + using the ``reserved_resources_record__overlap`` filter, leveraging the partial + GIN index ``pulp_task_resources_index`` + + The ``blocked_resources`` set is local to each ``fetch_task()`` call — it resets + between calls so that resources freed while the worker was busy are retried. + FIFO ordering within a resource is preserved because all tasks needing a blocked + resource are excluded together. Returns: Task: A task object if one was successfully locked, None otherwise """ fetch_limit = FETCH_TASK_LIMIT + blocked_resources = set() while True: taken_exclusive = set() taken_shared = set() - waiting_tasks = list( + qs = ( Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None) .exclude(pk__in=self.ignored_task_ids) .order_by("pulp_created") - .select_related("pulp_domain")[:fetch_limit] + .select_related("pulp_domain") ) + if blocked_resources: + qs = qs.exclude( + reserved_resources_record__overlap=list(blocked_resources) + ) + + waiting_tasks = list(qs[:fetch_limit]) if not waiting_tasks: break + prev_blocked_count = len(blocked_resources) + for task in waiting_tasks: try: exclusive_resources, shared_resources = extract_task_resources(task) @@ -650,6 +666,18 @@ def fetch_task(self): shared_resources, ) if blocked_resource_list: + for resource_name in blocked_resource_list: + # Redis returns bytes; decode for ORM compatibility. + if isinstance(resource_name, bytes): + resource_name = resource_name.decode() + if resource_name == "__task_lock__": + continue + # Add the raw resource name (matches exclusive entries + # in reserved_resources_record). + blocked_resources.add(resource_name) + # Also add the shared: variant so tasks referencing + # this resource as shared are excluded too. + blocked_resources.add(f"shared:{resource_name}") continue rows = Task.objects.filter( @@ -674,6 +702,11 @@ def fetch_task(self): pass continue + # If we learned new blocked resources, re-query with updated exclusions + # (same fetch_limit — the exclusion yields different tasks). + if len(blocked_resources) > prev_blocked_count: + continue + if len(waiting_tasks) < fetch_limit: break diff --git a/pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py b/pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py new file mode 100644 index 0000000000..38bebf73bc --- /dev/null +++ b/pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py @@ -0,0 +1,145 @@ +"""Reproduction test for RedisWorker fetch_task() head-of-line blocking (issue #7900). + +When many waiting tasks need the same blocked exclusive resource, fetch_task() +should skip them at the DB level and find tasks for free resources without +excessive acquire_locks calls. +""" + +from datetime import timedelta +from unittest.mock import patch as mock_patch +from uuid import uuid4 + +import pytest + +from pulpcore.app.models import AppStatus, Domain, Task +from pulpcore.app.redis_connection import get_redis_connection +from pulpcore.constants import TASK_STATES +from pulpcore.tasking.redis_locks import ( + acquire_locks as real_acquire, + resource_to_lock_key, + safe_release_task_locks, +) +from pulpcore.tasking.redis_worker import RedisWorker + + +@pytest.mark.django_db +def test_fetch_task_skips_blocked_resources(): + """fetch_task() must skip tasks for blocked resources and find free ones. + + Reproduces issue #7900: when many tasks need a blocked exclusive resource, + fetch_task() should use DB-level exclusion to skip them and find tasks + for free resources, without excessive acquire_locks calls. + + With the bug (no DB-level exclusion): + - The doubling algorithm (20->40->80->160->320) re-scans from position 0 + each iteration, calling acquire_locks once per iteration for the first + blocked task. Total: ~6 acquire_locks calls for 200 blocked tasks. + + With the fix (DB-level exclusion via __overlap): + - After the first acquire_locks failure, the blocked resource is excluded + from subsequent DB queries. The free-resource task is found directly. + Total: 2 acquire_locks calls. + """ + redis_conn = get_redis_connection() + domain = Domain.objects.get(name="default") + domain_shared = f"shared:prn:core.domain:{domain.pk}" + test_id = uuid4().hex[:8] + redis_keys = [] + + AppStatus.objects._current_app_status = None + app_status = AppStatus.objects.create( + name=f"test-hol-{test_id}", + app_type="worker", + versions={}, + ttl=timedelta(seconds=30), + ) + + worker = object.__new__(RedisWorker) + worker.ignored_task_ids = list( + Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None).values_list( + "pk", flat=True + ) + ) + worker.redis_conn = redis_conn + worker.name = app_status.name + worker.app_status = app_status + + # Block one resource in Redis (simulates another worker holding it) + blocked_resource = f"prn:test.hol-{test_id}:blocked" + blocked_key = resource_to_lock_key(blocked_resource) + redis_conn.set(blocked_key, "other-worker") + redis_keys.append(blocked_key) + + free_resource = f"prn:test.hol-{test_id}:free" + + result = None + try: + # Create 200 tasks needing the blocked resource (fill the queue head) + Task.objects.bulk_create( + [ + Task( + state=TASK_STATES.WAITING, + name="pulpcore.app.tasks.test.sleep", + logging_cid=f"hol-{test_id}-blocked-{i}", + reserved_resources_record=[blocked_resource, domain_shared], + pulp_domain=domain, + ) + for i in range(200) + ] + ) + + # Create 1 task needing a free resource (last in FIFO order) + Task.objects.bulk_create( + [ + Task( + state=TASK_STATES.WAITING, + name="pulpcore.app.tasks.test.sleep", + logging_cid=f"hol-{test_id}-free", + reserved_resources_record=[free_resource, domain_shared], + pulp_domain=domain, + ) + ] + ) + + # Count acquire_locks calls during fetch_task + acquire_count = 0 + + def counting_acquire(*args, **kwargs): + nonlocal acquire_count + acquire_count += 1 + return real_acquire(*args, **kwargs) + + with mock_patch( + "pulpcore.tasking.redis_worker.acquire_locks", + side_effect=counting_acquire, + ): + result = worker.fetch_task() + + # The free-resource task must be found + assert result is not None, ( + "fetch_task() returned None -- failed to find the free-resource task " + "behind 200 blocked tasks" + ) + assert f"hol-{test_id}-free" in result.logging_cid, ( + f"fetch_task() returned wrong task: {result.logging_cid}" + ) + + # With DB-level exclusion, acquire_locks should be called at most 3 times + # (1 for the blocked resource + 1 for the free resource + margin). + # Without the fix, the doubling algorithm calls it ~6 times. + assert acquire_count <= 3, ( + f"acquire_locks called {acquire_count} times -- fetch_task() is " + f"re-scanning blocked resources instead of excluding them at the DB level" + ) + + finally: + for key in redis_keys: + redis_conn.delete(key) + if result: + safe_release_task_locks(result, lock_owner=worker.name) + Task.objects.filter(pk=result.pk).update( + app_lock=None, state=TASK_STATES.COMPLETED + ) + Task.objects.filter(logging_cid__startswith=f"hol-{test_id}").delete() + AppStatus.objects._current_app_status = None + app_status.delete() diff --git a/pulpcore/tests/unit/tasking/test_fetch_task_scale.py b/pulpcore/tests/unit/tasking/test_fetch_task_scale.py new file mode 100644 index 0000000000..b732a64717 --- /dev/null +++ b/pulpcore/tests/unit/tasking/test_fetch_task_scale.py @@ -0,0 +1,214 @@ +"""Scale test for fetch_task head-of-line blocking fix (issue #7900). + +Asserts on actual database behavior to prove the fix is efficient: +1. The overlap operator (&&) appears in captured SQL +2. auto_explain PostgreSQL logs show Index Scan or Bitmap usage +3. Query count is bounded +4. acquire_locks calls are proportional to distinct blocked resources +""" + +import subprocess +from datetime import timedelta +from unittest.mock import patch as mock_patch +from uuid import uuid4 + +import pytest +from django.db import connection + +from pulpcore.app.models import AppStatus, Domain, Task +from pulpcore.app.redis_connection import get_redis_connection +from pulpcore.constants import TASK_STATES +from pulpcore.tasking.redis_locks import ( + acquire_locks as real_acquire, + resource_to_lock_key, + safe_release_task_locks, +) +from pulpcore.tasking.redis_worker import RedisWorker + + +NUM_BLOCKED_RESOURCES = 50 +NUM_BLOCKED_TASKS = 2000 + + +@pytest.mark.django_db +def test_fetch_task_uses_db_exclusion_at_scale(): + """fetch_task() must use DB-level overlap exclusion at scale. + + Creates 2000 tasks across 50 blocked resources, then verifies: + 1. The && operator appears in captured SQL (DB-level exclusion) + 2. auto_explain shows Index Scan or Bitmap usage (not Seq Scan) + 3. Total DB query count is small (not O(log n) doubling) + 4. acquire_locks calls are proportional to distinct blocked resources + """ + redis_conn = get_redis_connection() + domain = Domain.objects.get(name="default") + domain_shared = f"shared:prn:core.domain:{domain.pk}" + test_id = uuid4().hex[:8] + redis_keys = [] + + AppStatus.objects._current_app_status = None + app_status = AppStatus.objects.create( + name=f"scale-{test_id}", + app_type="worker", + versions={}, + ttl=timedelta(seconds=30), + ) + + worker = object.__new__(RedisWorker) + worker.ignored_task_ids = list( + Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None).values_list( + "pk", flat=True + ) + ) + worker.redis_conn = redis_conn + worker.name = app_status.name + worker.app_status = app_status + + # Block 50 distinct resources in Redis + blocked_resources = [ + f"prn:test.scale-{test_id}.r:{i}" for i in range(NUM_BLOCKED_RESOURCES) + ] + for res in blocked_resources: + key = resource_to_lock_key(res) + redis_conn.set(key, "other-worker") + redis_keys.append(key) + + result = None + try: + # Create 2000 tasks distributed across the 50 blocked resources + Task.objects.bulk_create( + [ + Task( + state=TASK_STATES.WAITING, + name="pulpcore.app.tasks.test.sleep", + logging_cid=f"scale-{test_id}-{i}", + reserved_resources_record=[ + blocked_resources[i % NUM_BLOCKED_RESOURCES], + domain_shared, + ], + pulp_domain=domain, + ) + for i in range(NUM_BLOCKED_TASKS) + ] + ) + + # Force statistics update so the query planner has accurate row counts + cursor = connection.cursor() + cursor.execute("ANALYZE core_task") + + # Count acquire_locks calls + acquire_count = 0 + + def counting_acquire(*args, **kwargs): + nonlocal acquire_count + acquire_count += 1 + return real_acquire(*args, **kwargs) + + # Mark log position BEFORE test so we only check new auto_explain entries + subprocess.run( + [ + "bash", + "-c", + "TODAY=$(date +%a); " + "wc -l < /data/pgsql/log/postgresql-${TODAY}.log " + "> /tmp/.scale_test_log_pos", + ], + capture_output=True, + ) + + # Capture DB queries during fetch_task + connection.force_debug_cursor = True + connection.queries_log.clear() + + with mock_patch( + "pulpcore.tasking.redis_worker.acquire_locks", + side_effect=counting_acquire, + ): + result = worker.fetch_task() + + captured_queries = list(connection.queries) + connection.force_debug_cursor = False + + # === DB Query Assertions === + + # 1. The overlap operator (&&) must appear in at least one query + overlap_queries = [ + q + for q in captured_queries + if "&&" in q["sql"] and "core_task" in q["sql"] + ] + assert len(overlap_queries) > 0, ( + "NO OVERLAP QUERIES: fetch_task() did not use the && operator " + "to exclude blocked resources at the DB level. " + "Captured queries:\n" + + "\n".join( + q["sql"][:200] + for q in captured_queries + if "core_task" in q["sql"] + ) + ) + + # 2. Total task queries should be small (not O(log n) doubling) + task_select_queries = [ + q + for q in captured_queries + if "core_task" in q["sql"] and "SELECT" in q["sql"].upper() + ] + assert len(task_select_queries) <= 10, ( + f"TOO MANY DB QUERIES: {len(task_select_queries)} SELECT queries " + f"on core_task. With DB-level exclusion, fetch_task() should need " + f"a small number of queries, not {len(task_select_queries)}." + ) + + # 3. acquire_locks calls proportional to distinct resources + assert acquire_count <= NUM_BLOCKED_RESOURCES * 2, ( + f"acquire_locks called {acquire_count} times for " + f"{NUM_BLOCKED_RESOURCES} distinct resources -- re-scanning" + ) + + # 4. auto_explain log: verify an index was used during actual execution + log_after = subprocess.run( + [ + "bash", + "-c", + "TODAY=$(date +%a); " + "tail -n +$(($(cat /tmp/.scale_test_log_pos 2>/dev/null || echo 1))) " + "/data/pgsql/log/postgresql-${TODAY}.log", + ], + capture_output=True, + text=True, + ).stdout + + plan_lines = [ + line.strip() + for line in log_after.split("\n") + if "Scan" in line or "Index" in line or "Bitmap" in line + ] + index_used = any( + "Index Scan" in line or "Bitmap" in line for line in plan_lines + ) + seq_scan_only = ( + all("Seq Scan" in line for line in plan_lines if "Scan" in line) + if plan_lines + else True + ) + + assert index_used and not seq_scan_only, ( + "NO INDEX USED: auto_explain shows the overlap query used " + "Seq Scan instead of an index during actual execution.\n" + "Plan lines from PostgreSQL log:\n" + "\n".join(plan_lines[:10]) + ) + + finally: + for key in redis_keys: + redis_conn.delete(key) + if result: + safe_release_task_locks(result, lock_owner=worker.name) + Task.objects.filter(pk=result.pk).update( + app_lock=None, state=TASK_STATES.COMPLETED + ) + Task.objects.filter( + logging_cid__startswith=f"scale-{test_id}" + ).delete() + AppStatus.objects._current_app_status = None + app_status.delete() From 2d9dc1d175f8e01ac3d34992789403effd3314f3 Mon Sep 17 00:00:00 2001 From: Dennis Kliban Date: Sat, 15 Aug 2026 22:09:30 -0400 Subject: [PATCH 2/3] style: fix import sorting in test files Co-Authored-By: Claude Opus 4.6 (1M context) --- .../tasking/test_fetch_task_hol_blocking.py | 10 ++--- .../unit/tasking/test_fetch_task_scale.py | 41 +++++-------------- 2 files changed, 15 insertions(+), 36 deletions(-) diff --git a/pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py b/pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py index 38bebf73bc..617c82de8a 100644 --- a/pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py +++ b/pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py @@ -16,6 +16,8 @@ from pulpcore.constants import TASK_STATES from pulpcore.tasking.redis_locks import ( acquire_locks as real_acquire, +) +from pulpcore.tasking.redis_locks import ( resource_to_lock_key, safe_release_task_locks, ) @@ -56,9 +58,7 @@ def test_fetch_task_skips_blocked_resources(): worker = object.__new__(RedisWorker) worker.ignored_task_ids = list( - Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None).values_list( - "pk", flat=True - ) + Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None).values_list("pk", flat=True) ) worker.redis_conn = redis_conn worker.name = app_status.name @@ -137,9 +137,7 @@ def counting_acquire(*args, **kwargs): redis_conn.delete(key) if result: safe_release_task_locks(result, lock_owner=worker.name) - Task.objects.filter(pk=result.pk).update( - app_lock=None, state=TASK_STATES.COMPLETED - ) + Task.objects.filter(pk=result.pk).update(app_lock=None, state=TASK_STATES.COMPLETED) Task.objects.filter(logging_cid__startswith=f"hol-{test_id}").delete() AppStatus.objects._current_app_status = None app_status.delete() diff --git a/pulpcore/tests/unit/tasking/test_fetch_task_scale.py b/pulpcore/tests/unit/tasking/test_fetch_task_scale.py index b732a64717..41cd910594 100644 --- a/pulpcore/tests/unit/tasking/test_fetch_task_scale.py +++ b/pulpcore/tests/unit/tasking/test_fetch_task_scale.py @@ -20,12 +20,13 @@ from pulpcore.constants import TASK_STATES from pulpcore.tasking.redis_locks import ( acquire_locks as real_acquire, +) +from pulpcore.tasking.redis_locks import ( resource_to_lock_key, safe_release_task_locks, ) from pulpcore.tasking.redis_worker import RedisWorker - NUM_BLOCKED_RESOURCES = 50 NUM_BLOCKED_TASKS = 2000 @@ -56,18 +57,14 @@ def test_fetch_task_uses_db_exclusion_at_scale(): worker = object.__new__(RedisWorker) worker.ignored_task_ids = list( - Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None).values_list( - "pk", flat=True - ) + Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None).values_list("pk", flat=True) ) worker.redis_conn = redis_conn worker.name = app_status.name worker.app_status = app_status # Block 50 distinct resources in Redis - blocked_resources = [ - f"prn:test.scale-{test_id}.r:{i}" for i in range(NUM_BLOCKED_RESOURCES) - ] + blocked_resources = [f"prn:test.scale-{test_id}.r:{i}" for i in range(NUM_BLOCKED_RESOURCES)] for res in blocked_resources: key = resource_to_lock_key(res) redis_conn.set(key, "other-worker") @@ -133,26 +130,18 @@ def counting_acquire(*args, **kwargs): # 1. The overlap operator (&&) must appear in at least one query overlap_queries = [ - q - for q in captured_queries - if "&&" in q["sql"] and "core_task" in q["sql"] + q for q in captured_queries if "&&" in q["sql"] and "core_task" in q["sql"] ] assert len(overlap_queries) > 0, ( "NO OVERLAP QUERIES: fetch_task() did not use the && operator " "to exclude blocked resources at the DB level. " "Captured queries:\n" - + "\n".join( - q["sql"][:200] - for q in captured_queries - if "core_task" in q["sql"] - ) + + "\n".join(q["sql"][:200] for q in captured_queries if "core_task" in q["sql"]) ) # 2. Total task queries should be small (not O(log n) doubling) task_select_queries = [ - q - for q in captured_queries - if "core_task" in q["sql"] and "SELECT" in q["sql"].upper() + q for q in captured_queries if "core_task" in q["sql"] and "SELECT" in q["sql"].upper() ] assert len(task_select_queries) <= 10, ( f"TOO MANY DB QUERIES: {len(task_select_queries)} SELECT queries " @@ -184,13 +173,9 @@ def counting_acquire(*args, **kwargs): for line in log_after.split("\n") if "Scan" in line or "Index" in line or "Bitmap" in line ] - index_used = any( - "Index Scan" in line or "Bitmap" in line for line in plan_lines - ) + index_used = any("Index Scan" in line or "Bitmap" in line for line in plan_lines) seq_scan_only = ( - all("Seq Scan" in line for line in plan_lines if "Scan" in line) - if plan_lines - else True + all("Seq Scan" in line for line in plan_lines if "Scan" in line) if plan_lines else True ) assert index_used and not seq_scan_only, ( @@ -204,11 +189,7 @@ def counting_acquire(*args, **kwargs): redis_conn.delete(key) if result: safe_release_task_locks(result, lock_owner=worker.name) - Task.objects.filter(pk=result.pk).update( - app_lock=None, state=TASK_STATES.COMPLETED - ) - Task.objects.filter( - logging_cid__startswith=f"scale-{test_id}" - ).delete() + Task.objects.filter(pk=result.pk).update(app_lock=None, state=TASK_STATES.COMPLETED) + Task.objects.filter(logging_cid__startswith=f"scale-{test_id}").delete() AppStatus.objects._current_app_status = None app_status.delete() From 37d4c8beccec3d78761e221d7d2696684b58af2e Mon Sep 17 00:00:00 2001 From: Dennis Kliban Date: Sat, 15 Aug 2026 22:20:53 -0400 Subject: [PATCH 3/3] style: fix formatting --- pulpcore/tasking/redis_worker.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index 11005d8110..fc9482670d 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -625,9 +625,7 @@ def fetch_task(self): .select_related("pulp_domain") ) if blocked_resources: - qs = qs.exclude( - reserved_resources_record__overlap=list(blocked_resources) - ) + qs = qs.exclude(reserved_resources_record__overlap=list(blocked_resources)) waiting_tasks = list(qs[:fetch_limit])