diff --git a/CHANGES/7900.bugfix b/CHANGES/7900.bugfix new file mode 100644 index 0000000000..e04ecaea57 --- /dev/null +++ b/CHANGES/7900.bugfix @@ -0,0 +1 @@ +Fixed head-of-line blocking in RedisWorker fetch_task() when thousands of tasks need the same blocked resource by excluding known-blocked resources from subsequent DB queries. diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index 99ce297413..e1197d5af1 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -599,27 +599,41 @@ 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() + taken_exclusive = set() + taken_shared = 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 +664,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 +700,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..0c1c8cc93a --- /dev/null +++ b/pulpcore/tests/unit/tasking/test_fetch_task_hol_blocking.py @@ -0,0 +1,149 @@ +"""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 django.conf import settings + +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, +) +from pulpcore.tasking.redis_locks import ( + resource_to_lock_key, + safe_release_task_locks, +) +from pulpcore.tasking.redis_worker import RedisWorker + +pytestmark = pytest.mark.skipif( + settings.WORKER_TYPE != "redis", + reason="Only runs with WORKER_TYPE=redis", +) + + +@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()