wip: review iteration

Author: pedro-psb

pulpcore/app/models/repository.py

@@ -25,6 +25,7 @@
     get_domain_pk,
     cache_key,
     reverse,
+    force_deferred_execution,
 )
 from pulpcore.constants import ALL_KNOWN_CONTENT_CHECKSUMS, PROTECTED_REPO_VERSION_MESSAGE
 from pulpcore.download.factory import DownloaderFactory
@@ -379,6 +380,7 @@ def pull_through_add_content(self, content_artifact):
         from pulpcore.plugin.tasking import dispatch, add_and_remove
 
         body = {"repository_pk": self.pk, "add_content_units": [cpk], "remove_content_units": []}
+        force_deferred_execution.set(True)
         return dispatch(add_and_remove, kwargs=body, exclusive_resources=[self], immediate=True)
 
     @hook(AFTER_UPDATE, when="retain_repo_versions", has_changed=True)

pulpcore/app/util.py

@@ -544,6 +544,7 @@ def get_artifact_url(artifact, headers=None, http_method=None):
     return url
 
 
+force_deferred_execution = ContextVar("force_deferred_execution", default=False)
 current_task = ContextVar("current_task", default=None)
 _current_user_func = ContextVar("current_user", default=lambda: None)
 

pulpcore/tasking/tasks.py

@@ -7,7 +7,6 @@
 import sys
 import traceback
 import tempfile
-import threading
 from asgiref.sync import sync_to_async
 from datetime import timedelta
 from gettext import gettext as _
@@ -18,7 +17,13 @@
 from django_guid import get_guid
 from pulpcore.app.apps import MODULE_PLUGIN_VERSIONS
 from pulpcore.app.models import Task, TaskGroup
-from pulpcore.app.util import current_task, get_domain, get_prn, deprecation_logger
+from pulpcore.app.util import (
+    current_task,
+    get_domain,
+    get_prn,
+    deprecation_logger,
+    force_deferred_execution,
+)
 from pulpcore.constants import (
     TASK_FINAL_STATES,
     TASK_INCOMPLETE_STATES,
@@ -126,12 +131,6 @@ def _execute_task(task):
         send_task_notification(task)
 
 
-def running_from_thread_pool() -> bool:
-    thread_name = threading.current_thread().name
-    # ThreadPoolExecutor names threads like: "ThreadPoolExecutor-0_0"
-    return "ThreadPoolExecutor" in thread_name
-
-
 def dispatch(
     func,
     args=None,
@@ -179,13 +178,10 @@ def dispatch(
         ValueError: When `resources` is an unsupported type.
     """
 
+    # Can't run short tasks immediately from certain contexts (e.g, if running from thread pool)
+    immediate = immediate and not force_deferred_execution
     assert deferred or immediate, "A task must be at least `deferred` or `immediate`."
 
-    # Can't run short tasks immediately if running from thread pool
-    force_defer = True if running_from_thread_pool() else False
-    short = immediate
-    immediate = short and not force_defer
-
     if callable(func):
         function_name = f"{func.__module__}.{func.__name__}"
     else: