Always set app_lock when claiming a task.

This removes the need to perform a refresh of a task from the db before executing a task.

This also removes some excessive logging from dispatch method.

Author: dkliban

pulpcore/app/models/task.py

@@ -6,7 +6,6 @@
 import traceback
 from gettext import gettext as _
 
-from django.conf import settings
 from django.contrib.postgres.fields import ArrayField, HStoreField
 from django.contrib.postgres.indexes import GinIndex
 from django.core.serializers.json import DjangoJSONEncoder
@@ -207,18 +206,13 @@ def set_running(self):
         This updates the :attr:`started_at` and sets the :attr:`state` to :attr:`RUNNING`.
         """
         started_at = timezone.now()
-        filter_kwargs = {
-            "pk": self.pk,
-            "state": TASK_STATES.WAITING,
-        }
-        # Only check app_lock for PulpcoreWorker, not RedisWorker
-        if settings.WORKER_TYPE != "redis":
-            filter_kwargs["app_lock"] = AppStatus.objects.current()
-
-        rows = Task.objects.filter(**filter_kwargs).update(
+        rows = Task.objects.filter(
+            pk=self.pk,
+            state=TASK_STATES.WAITING,
+            app_lock=AppStatus.objects.current(),
+        ).update(
             state=TASK_STATES.RUNNING,
             started_at=started_at,
-            app_lock=AppStatus.objects.current(),
         )
         if rows == 1:
             self.state = TASK_STATES.RUNNING

pulpcore/tasking/redis_tasks.py

@@ -440,31 +440,10 @@ def dispatch(
         # are_resources_available() now acquires ALL locks atomically
         if are_resources_available(task):
             # All locks acquired successfully
-            # Reload task state from database to check if it's still WAITING
-            # (a worker might have executed it between creation and lock acquisition)
-            task.refresh_from_db()
-
-            if task.state != TASK_STATES.WAITING:
-                # Task was already executed by a worker, release locks and return
-                _logger.info(
-                    "IMMEDIATE DISPATCH: Task %s already in state "
-                    "'%s', releasing locks without execution",
-                    task.pk,
-                    task.state,
-                )
-                safe_release_task_locks(task)
-                return task
-
-            # Task is still WAITING, proceed with execution
+            # Proceed with execution
             current_app = AppStatus.objects.current()
             lock_owner = current_app.name if current_app else f"immediate-{task.pk}"
             try:
-                _logger.info(
-                    "IMMEDIATE DISPATCH: Task %s acquired all locks,"
-                    " executing in API process (AppStatus=%s)",
-                    task.pk,
-                    lock_owner,
-                )
                 with using_workdir():
                     execute_task(task)
             except Exception:
@@ -474,10 +453,10 @@ def dispatch(
                 raise
         elif deferred:
             # Locks not available, defer to worker
+            # Clear app_lock so workers can pick this up
             # No locks were acquired (atomic operation failed), so nothing to clean up
-            _logger.info(
-                "IMMEDIATE DISPATCH: Task %s could not acquire locks, deferring to worker", task.pk
-            )
+            Task.objects.filter(pk=task.pk).update(app_lock=None)
+            task.app_lock = None
         else:
             # Can't acquire locks and can't be deferred - cancel task
             # No locks were acquired, so just set state
@@ -513,32 +492,10 @@ async def adispatch(
         # async_are_resources_available() now acquires ALL locks atomically
         if await async_are_resources_available(task):
             # All locks acquired successfully
-            # Reload task state from database to check if it's still WAITING
-            # (a worker might have executed it between creation and lock acquisition)
-            await task.arefresh_from_db()
-
-            if task.state != TASK_STATES.WAITING:
-                # Task was already executed by a worker, release locks and return
-                _logger.info(
-                    "IMMEDIATE DISPATCH (async): Task %s already in"
-                    " state '%s', releasing locks without execution",
-                    task.pk,
-                    task.state,
-                )
-                await async_safe_release_task_locks(task)
-                return task
-
-            # Task is still WAITING, proceed with execution
+            # Proceed with execution
             current_app = await sync_to_async(AppStatus.objects.current)()
             lock_owner = current_app.name if current_app else f"immediate-{task.pk}"
             try:
-                _logger.info(
-                    "IMMEDIATE DISPATCH (async): Task %s acquired "
-                    "all locks, executing in API process "
-                    "(AppStatus=%s)",
-                    task.pk,
-                    lock_owner,
-                )
                 with using_workdir():
                     await aexecute_task(task)
             except Exception:
@@ -548,11 +505,10 @@ async def adispatch(
                 raise
         elif deferred:
             # Locks not available, defer to worker
+            # Clear app_lock so workers can pick this up
             # No locks were acquired (atomic operation failed), so nothing to clean up
-            _logger.info(
-                "IMMEDIATE DISPATCH (async): Task %s could not acquire locks, deferring to worker",
-                task.pk,
-            )
+            await Task.objects.filter(pk=task.pk).aupdate(app_lock=None)
+            task.app_lock = None
         else:
             # Can't acquire locks and can't be deferred - cancel task
             # No locks were acquired, so just set state

pulpcore/tasking/redis_worker.py

@@ -506,8 +506,9 @@ def fetch_task(self):
             Task: A task object if one was successfully locked, None otherwise
         """
         # Query waiting tasks, sorted by creation time, limited
+        # Filter for app_lock__isnull=True to avoid tasks being executed immediately by API
         waiting_tasks = (
-            Task.objects.filter(state=TASK_STATES.WAITING)
+            Task.objects.filter(state=TASK_STATES.WAITING, app_lock__isnull=True)
             .exclude(pk__in=self.ignored_task_ids)
             .order_by("pulp_created")
             .select_related("pulp_domain")[:FETCH_TASK_LIMIT]
@@ -544,6 +545,19 @@ def fetch_task(self):
                 if should_skip:
                     continue
 
+                # Claim the task by setting app_lock BEFORE acquiring locks
+                rows = Task.objects.filter(
+                    pk=task.pk, state=TASK_STATES.WAITING, app_lock__isnull=True
+                ).update(app_lock=self.app_status)
+
+                if rows == 0:
+                    # Task was already claimed or executed
+                    _logger.debug("WORKER: Task %s already claimed or executed, skipping", task.pk)
+                    continue
+
+                # Update local task object
+                task.app_lock = self.app_status
+
                 # Atomically try to acquire task lock and resource locks in a single operation
                 task_lock_key = get_task_lock_key(task.pk)
 
@@ -730,20 +744,7 @@ def handle_tasks(self):
                     self._maybe_release_locks(task, mark_released=False)
                     break
 
-                # Check if task is still WAITING after acquiring locks
-                # (an API process might have executed it between query and lock acquisition)
-                task.refresh_from_db()
-                if task.state != TASK_STATES.WAITING:
-                    # Task was already executed, release locks and skip
-                    _logger.info(
-                        "Task %s already in state '%s' after acquiring locks, skipping execution",
-                        task.pk,
-                        task.state,
-                    )
-                    self._maybe_release_locks(task)
-                    continue
-
-                # Task is compatible and still waiting, execute it
+                # Task is compatible, execute it
                 if task.immediate:
                     self.supervise_immediate_task(task)
                 else:

pulpcore/tasking/tasks.py

@@ -110,12 +110,14 @@ async def _aexecute_task(task):
 
 def log_task_start(task, domain):
     _logger.info(
-        "Starting task id: %s in domain: %s, task_type: %s, immediate: %s, deferred: %s",
+        "Starting task id: %s in domain: %s, task_type: %s, immediate: %s, deferred: %s, "
+        "worker: %s",
         task.pk,
         domain.name,
         task.name,
         str(task.immediate),
         str(task.deferred),
+        task.app_lock.name,
     )