Prevent dispatch from non-main thread to run immediately

Immediate/short tasks are now run inside an event loop, but that doesnt
work inside a thread from a thread pool, like it happens with on the
pull-through workflow from the content-app.

This enforces that the short task can't run immediately if the dispatch
is done from inside such a context.

Closes: #6429

Author: pedro-psb

CHANGES/plugin_api/6429.bugfix

@@ -0,0 +1,7 @@
+Re-added* timeout to tasks dispatched as immediate for both deferred and non-deferred runs.
+
+Also, immediate tasks must be coroutines from now on.
+This is to enable immediate tasks to run on the workers foreground without completely blocking heartbeats.
+Support for legacy non-coroutines immediate task will be dropped in pulpcore 3.85.
+
+\* This was added in pulpcore 3.75.0, reverted in 3.75.1 due to a regression and re-applied here with the regression fix.

pulpcore/app/tasks/base.py

@@ -2,6 +2,7 @@
 
 from pulpcore.app.apps import get_plugin_config
 from pulpcore.app.models import CreatedResource
+from pulpcore.app.loggers import deprecation_logger
 from pulpcore.plugin.models import MasterModel
 
 from asgiref.sync import sync_to_async
@@ -65,6 +66,10 @@ def general_update(instance_id, app_label, serializer_name, *args, **kwargs):
             due to validation error. This theoretically should never occur since validation is
             performed before the task is dispatched.
     """
+    deprecation_logger.warning(
+        "`pulpcore.app.tasks.base.general_update` is deprecated and will be removed in Pulp 4. "
+        "Use `pulpcore.app.tasks.base.ageneral_update` instead."
+    )
     data = kwargs.pop("data", None)
     partial = kwargs.pop("partial", False)
     serializer_class = get_plugin_config(app_label).named_serializers[serializer_name]
@@ -87,6 +92,10 @@ def general_delete(instance_id, app_label, serializer_name):
         app_label (str): the Django app label of the plugin that provides the model
         serializer_name (str): name of the serializer class for the model
     """
+    deprecation_logger.warning(
+        "`pulpcore.app.tasks.base.general_delete` is deprecated and will be removed in Pulp 4. "
+        "Use `pulpcore.app.tasks.base.ageneral_delete` instead."
+    )
     serializer_class = get_plugin_config(app_label).named_serializers[serializer_name]
     instance = serializer_class.Meta.model.objects.get(pk=instance_id)
     if isinstance(instance, MasterModel):

pulpcore/app/tasks/test.py

@@ -1,4 +1,6 @@
+import asyncio
 import backoff
+import time
 from pulpcore.app.models import TaskGroup
 from pulpcore.tasking.tasks import dispatch
 
@@ -9,24 +11,18 @@ def dummy_task():
 
 
 def sleep(interval):
-    from time import sleep
-
-    sleep(interval)
+    time.sleep(interval)
 
 
 async def asleep(interval):
     """Async function that sleeps."""
-    import asyncio
-
     await asyncio.sleep(interval)
 
 
 @backoff.on_exception(backoff.expo, BaseException)
 def gooey_task(interval):
     """A sleep task that tries to avoid being killed by ignoring all exceptions."""
-    from time import sleep
-
-    sleep(interval)
+    time.sleep(interval)
 
 
 def dummy_group_task(inbetween=3, intervals=None):
@@ -35,5 +31,5 @@ def dummy_group_task(inbetween=3, intervals=None):
     task_group = TaskGroup.current()
     for interval in intervals:
         dispatch(sleep, args=(interval,), task_group=task_group)
-        sleep(inbetween)
+        time.sleep(inbetween)
     task_group.finish()

pulpcore/tasking/tasks.py

@@ -7,6 +7,7 @@
 import sys
 import traceback
 import tempfile
+import threading
 from asgiref.sync import sync_to_async
 from datetime import timedelta
 from gettext import gettext as _
@@ -130,6 +131,14 @@ def _execute_task(task):
         send_task_notification(task)
 
 
+def running_from_thread_pool() -> bool:
+    # TODO: this needs an alternative approach ASAP!
+    # Currently we rely on the weak fact that ThreadPoolExecutor names threads like:
+    # "ThreadPoolExecutor-0_0"
+    thread_name = threading.current_thread().name
+    return "ThreadPoolExecutor" in thread_name
+
+
 def dispatch(
     func,
     args=None,
@@ -177,6 +186,8 @@ def dispatch(
         ValueError: When `resources` is an unsupported type.
     """
 
+    # Can't run short tasks immediately if running from thread pool
+    immediate = immediate and not running_from_thread_pool()
     assert deferred or immediate, "A task must be at least `deferred` or `immediate`."
 
     if callable(func):