From 718392296e47fa32ac9244599a81f7eb7582ea57 Mon Sep 17 00:00:00 2001 From: Dennis Kliban Date: Thu, 13 Aug 2026 23:30:28 -0400 Subject: [PATCH 1/3] fix: eliminate head-of-line blocking in RedisWorker.fetch_task() Track resources that acquire_locks reports as blocked and exclude tasks needing those resources from subsequent DB queries using PostgreSQL's array overlap operator against the existing GIN index. This prevents workers from repeatedly re-scanning thousands of blocked tasks when the queue head is serialized on a few resources. Before: 241 acquire_locks calls for 100 distinct blocked resources After: ~101 calls (each resource tried at most once) Closes: https://github.com/pulp/pulpcore/issues/7900 Co-Authored-By: Claude Opus 4.6 (1M context) --- pulpcore/tasking/redis_worker.py | 42 +++-- .../api/test_fetch_task_hol_blocking.py | 165 ++++++++++++++++++ 2 files changed, 196 insertions(+), 11 deletions(-) create mode 100644 pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index 59ffdca9f95..be6a0f29467 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -477,32 +477,45 @@ def fetch_task(self): Fetch an available waiting task using Redis locks. This method: - 1. Queries waiting tasks (sorted by creation time, limited) - 2. For each task, attempts to acquire Redis distributed locks for exclusive resources - 3. If resource locks acquired, attempts to claim the task - with a Redis task lock (24h expiration) + 1. Queries waiting tasks, excluding tasks that need resources already + known to be blocked (DB-level exclusion via GIN index) + 2. For each task, attempts to acquire Redis distributed locks + 3. If locks fail, adds the blocked resources to an exclusion set so + subsequent queries skip all tasks needing those resources 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. Advances with offset instead of re-scanning from position 0 + + The blocked_resources set is local to this call and resets naturally + when fetch_task() is called again after the current task completes. Returns: Task: A task object if one was successfully locked, None otherwise """ - fetch_limit = FETCH_TASK_LIMIT + blocked_resources = set() + offset = 0 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] + ) + if blocked_resources: + qs = qs.exclude( + reserved_resources_record__overlap=list(blocked_resources) + ) + waiting_tasks = list( + qs.order_by("pulp_created") + .select_related("pulp_domain")[offset:offset + FETCH_TASK_LIMIT] ) if not waiting_tasks: break + new_blocked = False + for task in waiting_tasks: try: exclusive_resources, shared_resources = extract_task_resources(task) @@ -533,6 +546,10 @@ def fetch_task(self): shared_resources, ) if blocked_resource_list: + for resource in blocked_resource_list: + if resource != "__task_lock__": + blocked_resources.add(resource) + new_blocked = True continue rows = Task.objects.filter( @@ -557,10 +574,13 @@ def fetch_task(self): pass continue - if len(waiting_tasks) < fetch_limit: + if len(waiting_tasks) < FETCH_TASK_LIMIT and not new_blocked: break - fetch_limit *= 2 + if new_blocked: + offset = 0 + else: + offset += FETCH_TASK_LIMIT return None diff --git a/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py b/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py new file mode 100644 index 00000000000..5b817d9a6ee --- /dev/null +++ b/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py @@ -0,0 +1,165 @@ +""" +Test for head-of-line blocking in RedisWorker.fetch_task(). + +Reproduces https://github.com/pulp/pulpcore/issues/7900 + +Scenario: Queue has many tasks with different blocked resources at the +head, followed by a task with a free resource. Without the fix, +fetch_task() re-scans from position 0 on each doubling iteration, +calling acquire_locks repeatedly on already-known-blocked resources. +With the fix, blocked resources are excluded at the DB level via the +GIN index, so each resource is tried at most once. +""" + +import time +import uuid +from datetime import timedelta +from unittest.mock import patch as mock_patch + +import pytest +import redis + +from pulpcore.tasking.redis_locks import acquire_locks as real_acquire_locks + + +@pytest.mark.django_db +class TestFetchTaskHeadOfLineBlocking: + @pytest.fixture(autouse=True) + def setup(self): + from pulpcore.app.models import AppStatus + + AppStatus.objects._current_app_status = None + self.app_status = AppStatus.objects.create( + name=f"test-worker-{uuid.uuid4().hex[:8]}", + app_type="worker", + versions={}, + ttl=timedelta(seconds=30), + ) + self.redis_conn = redis.Redis( + host="localhost", port=6379, decode_responses=True + ) + yield + AppStatus.objects._current_app_status = None + + def _make_worker(self): + from pulpcore.tasking.redis_worker import RedisWorker + + worker = object.__new__(RedisWorker) + worker.ignored_task_ids = [] + worker.redis_conn = self.redis_conn + worker.name = self.app_status.name + worker.app_status = self.app_status + return worker + + def _cleanup_task(self, task, worker): + from pulpcore.app.models import Task + from pulpcore.tasking.redis_locks import safe_release_task_locks + + safe_release_task_locks(task, lock_owner=worker.name) + Task.objects.filter(pk=task.pk).update(app_lock=None, state="completed") + + def test_blocked_resource_exclusion_reduces_acquire_locks_calls(self): + """ + With 100 tasks each having a DIFFERENT blocked resource at the + queue head, and 1 task with a free resource behind them, + fetch_task() should call acquire_locks at most once per unique + resource (~101 calls). + + Without DB-level blocked-resource exclusion, the doubling + algorithm re-queries from position 0 and re-calls acquire_locks + on already-tried resources because taken_exclusive/taken_shared + are reset on each iteration. This results in ~241 calls: + batch 1 (20): 20 calls + batch 2 (40 from 0): 40 calls (re-tries first 20) + batch 3 (80 from 0): 80 calls (re-tries first 40) + batch 4 (160 from 0): 101 calls (re-tries first 80) + """ + from pulpcore.app.models import Domain, Task + from pulpcore.app.util import set_domain + + domain = Domain.objects.get(name="default") + set_domain(domain) + + domain_shared = f"shared:prn:core.domain:{domain.pk}" + num_blocked = 100 + blocked_resources = [] + lock_keys = [] + + for i in range(num_blocked): + resource = f"prn:core.repository:{uuid.uuid4()}" + blocked_resources.append(resource) + lock_key = f"pulp:resource_lock:{resource}" + lock_keys.append(lock_key) + self.redis_conn.set(lock_key, "other-worker-holding-lock") + + free_resource = f"prn:core.repository:{uuid.uuid4()}" + + try: + blocked_tasks = [ + Task( + state="waiting", + name="pulpcore.app.tasks.test.sleep", + logging_cid=f"blocked-{i}", + reserved_resources_record=[ + blocked_resources[i], domain_shared + ], + pulp_domain=domain, + enc_args=[0], + enc_kwargs={}, + ) + for i in range(num_blocked) + ] + Task.objects.bulk_create(blocked_tasks) + + time.sleep(0.01) + + free_task = Task.objects.create( + state="waiting", + name="pulpcore.app.tasks.test.sleep", + logging_cid="free-task", + reserved_resources_record=[free_resource, domain_shared], + pulp_domain=domain, + enc_args=[0], + enc_kwargs={}, + ) + + worker = self._make_worker() + + call_count = 0 + + def counting_acquire(*args, **kwargs): + nonlocal call_count + call_count += 1 + return real_acquire_locks(*args, **kwargs) + + with mock_patch( + "pulpcore.tasking.redis_worker.acquire_locks", + side_effect=counting_acquire, + ): + result = worker.fetch_task() + + assert result is not None, "fetch_task() should find the free task" + assert result.pk == free_task.pk, ( + f"Should claim the free task, got {result.logging_cid}" + ) + + # With DB-level exclusion, each blocked resource is tried at + # most once: ~101 calls (100 failures + 1 success). + # Without the fix, doubling + re-scanning causes ~241 calls. + # Threshold at 120 gives margin while catching the regression. + assert call_count <= 120, ( + f"acquire_locks called {call_count} times — expected <= 120. " + f"With blocked-resource DB exclusion each resource is tried " + f"once (~101 calls). The current count indicates the doubling " + f"algorithm is re-scanning already-tried resources." + ) + + self._cleanup_task(result, worker) + + finally: + for key in lock_keys: + self.redis_conn.delete(key) + Task.objects.filter( + logging_cid__startswith="blocked-" + ).delete() + Task.objects.filter(logging_cid="free-task").delete() From 8ea9d161cf5f87910cc748a740492b8a289a498d Mon Sep 17 00:00:00 2001 From: Dennis Kliban Date: Thu, 13 Aug 2026 23:37:31 -0400 Subject: [PATCH 2/3] style: apply ruff formatting Co-Authored-By: Claude Opus 4.6 (1M context) --- pulpcore/tasking/redis_worker.py | 14 ++++++-------- .../functional/api/test_fetch_task_hol_blocking.py | 12 +++--------- 2 files changed, 9 insertions(+), 17 deletions(-) diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index be6a0f29467..548106488c9 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -498,17 +498,15 @@ def fetch_task(self): taken_exclusive = set() taken_shared = set() - qs = ( - Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None) - .exclude(pk__in=self.ignored_task_ids) + qs = Task.objects.filter(state=TASK_STATES.WAITING, app_lock=None).exclude( + pk__in=self.ignored_task_ids ) 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.order_by("pulp_created") - .select_related("pulp_domain")[offset:offset + FETCH_TASK_LIMIT] + qs.order_by("pulp_created").select_related("pulp_domain")[ + offset : offset + FETCH_TASK_LIMIT + ] ) if not waiting_tasks: diff --git a/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py b/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py index 5b817d9a6ee..bb3ff9efd1c 100644 --- a/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py +++ b/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py @@ -35,9 +35,7 @@ def setup(self): versions={}, ttl=timedelta(seconds=30), ) - self.redis_conn = redis.Redis( - host="localhost", port=6379, decode_responses=True - ) + self.redis_conn = redis.Redis(host="localhost", port=6379, decode_responses=True) yield AppStatus.objects._current_app_status = None @@ -100,9 +98,7 @@ def test_blocked_resource_exclusion_reduces_acquire_locks_calls(self): state="waiting", name="pulpcore.app.tasks.test.sleep", logging_cid=f"blocked-{i}", - reserved_resources_record=[ - blocked_resources[i], domain_shared - ], + reserved_resources_record=[blocked_resources[i], domain_shared], pulp_domain=domain, enc_args=[0], enc_kwargs={}, @@ -159,7 +155,5 @@ def counting_acquire(*args, **kwargs): finally: for key in lock_keys: self.redis_conn.delete(key) - Task.objects.filter( - logging_cid__startswith="blocked-" - ).delete() + Task.objects.filter(logging_cid__startswith="blocked-").delete() Task.objects.filter(logging_cid="free-task").delete() From 8fa7768e11fa95b4d8f4c2f568709640ad20f548 Mon Sep 17 00:00:00 2001 From: Dennis Kliban Date: Fri, 14 Aug 2026 06:35:53 -0400 Subject: [PATCH 3/3] fix: move test to unit/ and create Domain in setup The CI functional test environment doesn't have CREATEDB privileges, so @pytest.mark.django_db fails there. Unit tests run with the postgres superuser and have Redis available. Also create the "default" Domain in setup since the unit test database starts empty. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../test_fetch_task_hol_blocking.py | 22 +++++++++++++------ 1 file changed, 15 insertions(+), 7 deletions(-) rename pulpcore/tests/{functional/api => unit}/test_fetch_task_hol_blocking.py (91%) diff --git a/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py b/pulpcore/tests/unit/test_fetch_task_hol_blocking.py similarity index 91% rename from pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py rename to pulpcore/tests/unit/test_fetch_task_hol_blocking.py index bb3ff9efd1c..14debce839f 100644 --- a/pulpcore/tests/functional/api/test_fetch_task_hol_blocking.py +++ b/pulpcore/tests/unit/test_fetch_task_hol_blocking.py @@ -26,7 +26,19 @@ class TestFetchTaskHeadOfLineBlocking: @pytest.fixture(autouse=True) def setup(self): - from pulpcore.app.models import AppStatus + from pulpcore.app.models import AppStatus, Domain + from pulpcore.app.util import set_domain + + try: + self.redis_conn = redis.Redis(host="localhost", port=6379, decode_responses=True) + self.redis_conn.ping() + except redis.ConnectionError: + pytest.skip("Redis not available") + + self.domain, _ = Domain.objects.get_or_create( + name="default", defaults={"storage_class": "pulpcore.app.models.storage.FileSystem"} + ) + set_domain(self.domain) AppStatus.objects._current_app_status = None self.app_status = AppStatus.objects.create( @@ -35,7 +47,6 @@ def setup(self): versions={}, ttl=timedelta(seconds=30), ) - self.redis_conn = redis.Redis(host="localhost", port=6379, decode_responses=True) yield AppStatus.objects._current_app_status = None @@ -72,12 +83,9 @@ def test_blocked_resource_exclusion_reduces_acquire_locks_calls(self): batch 3 (80 from 0): 80 calls (re-tries first 40) batch 4 (160 from 0): 101 calls (re-tries first 80) """ - from pulpcore.app.models import Domain, Task - from pulpcore.app.util import set_domain - - domain = Domain.objects.get(name="default") - set_domain(domain) + from pulpcore.app.models import Task + domain = self.domain domain_shared = f"shared:prn:core.domain:{domain.pk}" num_blocked = 100 blocked_resources = []