Refactor dispatch and execute_task and add async version

We should be able to directly dispatch from async context using
async-ready functions. Refactoring was done to make it easier to
maintain both sync/async version of the respective functions.

Co-authored-by: Matthias Dellweg <[email protected]>

Author: pedro-psb

pulpcore/app/models/repository.py

@@ -381,6 +381,21 @@ def pull_through_add_content(self, content_artifact):
         body = {"repository_pk": self.pk, "add_content_units": [cpk], "remove_content_units": []}
         return dispatch(add_and_remove, kwargs=body, exclusive_resources=[self], immediate=True)
 
+    async def async_pull_through_add_content(self, content_artifact):
+        cpk = content_artifact.content_id
+        already_present = RepositoryContent.objects.filter(
+            content__pk=cpk, repository=self, version_removed__isnull=True
+        )
+        if not cpk or await already_present.aexists():
+            return None
+
+        from pulpcore.plugin.tasking import adispatch, aadd_and_remove
+
+        body = {"repository_pk": self.pk, "add_content_units": [cpk], "remove_content_units": []}
+        return await adispatch(
+            aadd_and_remove, kwargs=body, exclusive_resources=[self], immediate=True
+        )
+
     @hook(AFTER_UPDATE, when="retain_repo_versions", has_changed=True)
     def _cleanup_old_versions_hook(self):
         # Do not attempt to clean up anything, while there is a transaction involving repo versions

pulpcore/app/models/task.py

@@ -341,8 +341,20 @@ def unblock(self):
             self.unblocked_at = unblocked_at
         else:
             self.refresh_from_db()
-            raise RuntimeError(
-                _("Falied to set task {} unblocked in state '{}'.").format(self.pk, self.state)
+            raise RuntimeError("Failed to set task {} unblocked in state '{}'.").format(
+                self.pk, self.state
+            )
+
+    async def aunblock(self):
+        # This should be safe to be called without holding the lock.
+        unblocked_at = timezone.now()
+        rows = await Task.objects.filter(pk=self.pk).aupdate(unblocked_at=unblocked_at)
+        if rows == 1:
+            self.unblocked_at = unblocked_at
+        else:
+            await self.arefresh_from_db()
+            raise RuntimeError("Failed to set task {} unblocked in state '{}'.").format(
+                self.pk, self.state
             )
 
     class Meta:

pulpcore/app/tasks/repository.py

@@ -236,3 +236,30 @@ def add_and_remove(repository_pk, add_content_units, remove_content_units, base_
     with repository.new_version(base_version=base_version) as new_version:
         new_version.remove_content(models.Content.objects.filter(pk__in=remove_content_units))
         new_version.add_content(models.Content.objects.filter(pk__in=add_content_units))
+
+
+async def aadd_and_remove(
+    repository_pk, add_content_units, remove_content_units, base_version_pk=None
+):
+    """Aynsc version of add_and_remove."""
+    repository = await models.Repository.objects.aget(pk=repository_pk)
+    repository = await repository.acast()
+
+    if base_version_pk:
+        base_version = await models.RepositoryVersion.objects.aget(pk=base_version_pk)
+    else:
+        base_version = None
+
+    if "*" in remove_content_units:
+        latest = await repository.alatest_version()
+        if latest:
+            remove_content_units = latest.content.values_list("pk", flat=True)
+        else:
+            remove_content_units = []
+
+    def add_to_repository():
+        with repository.new_version(base_version=base_version) as new_version:
+            new_version.remove_content(models.Content.objects.filter(pk__in=remove_content_units))
+            new_version.add_content(models.Content.objects.filter(pk__in=add_content_units))
+
+    await sync_to_async(add_to_repository)()

pulpcore/plugin/tasking.py

@@ -1,5 +1,5 @@
 # Support plugins dispatching tasks
-from pulpcore.tasking.tasks import dispatch
+from pulpcore.tasking.tasks import dispatch, adispatch
 
 from pulpcore.app.tasks import (
     ageneral_update,
@@ -13,13 +13,14 @@
     reclaim_space,
 )
 from pulpcore.app.tasks.vulnerability_report import check_content
-from pulpcore.app.tasks.repository import add_and_remove
+from pulpcore.app.tasks.repository import add_and_remove, aadd_and_remove
 
 
 __all__ = [
     "ageneral_update",
     "check_content",
     "dispatch",
+    "adispatch",
     "fs_publication_export",
     "fs_repo_version_export",
     "general_create",
@@ -29,4 +30,5 @@
     "orphan_cleanup",
     "reclaim_space",
     "add_and_remove",
+    "aadd_and_remove",
 ]