Pass blocking IO in content handler to a threadpool executor

The standard async executor is single-threaded and blocking IO shreds
its throughput. Let's create a separate executor for blocking IO.

fixes: #8821
https://pulp.plan.io/issues/8821

CHANGES/8821.misc

@@ -0,0 +1 @@
+Properly utilize async in the content app, improving performance.

pulpcore/content/__init__.py

@@ -18,7 +18,7 @@
 from pulpcore.app.apps import pulp_plugin_configs  # noqa: E402: module level not at top of file
 from pulpcore.app.models import ContentAppStatus  # noqa: E402: module level not at top of file
 
-from .handler import Handler  # noqa: E402: module level not at top of file
+from .handler import Handler, loop  # noqa: E402: module level not at top of file
 
 
 log = logging.getLogger(__name__)
@@ -34,10 +34,18 @@ async def _heartbeat():
     i8ln_msg = _("Content App '{name}' heartbeat written, sleeping for '{interarrival}' seconds")
     msg = i8ln_msg.format(name=name, interarrival=heartbeat_interval)
 
+    def get_status_blocking():
+        return ContentAppStatus.objects.get_or_create(name=name)
+
     while True:
-        content_app_status, created = ContentAppStatus.objects.get_or_create(name=name)
-        if not created:
+        content_app_status, created = await loop.run_in_executor(None, get_status_blocking)
+
+        def save_heartbeat_blocking():
             content_app_status.save_heartbeat()
+
+        if not created:
+            await loop.run_in_executor(None, save_heartbeat_blocking)
+
         log.debug(msg)
         await asyncio.sleep(heartbeat_interval)