fix celery deadlocks by removing all blocking tasks from other (parent) tasks, and moving the logic to inline model methods

i was able to confirm this works by setting a autoscale pool of 10,1 and watching it scale up/down when handling a large number of tasks.

Author: Gabriel Monroy

api/models.py

@@ -185,19 +185,33 @@ def flat(self):
                 'nodes': self.nodes}
 
     def build(self):
-        tasks.build_formation.delay(self).wait()
+        return
 
     def destroy(self, *args, **kwargs):
-        tasks.destroy_formation.delay(self).wait()
+        app_tasks = [tasks.destroy_app.si(a) for a in self.app_set.all()]
+        node_tasks = [tasks.destroy_node.si(n) for n in self.node_set.all()]
+        layer_tasks = [tasks.destroy_layer.si(l) for l in self.layer_set.all()]
+        group(app_tasks + node_tasks).apply_async().join()
+        group(layer_tasks).apply_async().join()
+        CM.purge_formation(self.flat())
+        self.delete()
+        tasks.converge_controller.apply_async().wait()
 
     def publish(self):
         data = self.calculate()
         CM.publish_formation(self.flat(), data)
         return data
 
-    def converge(self, **kwargs):
+    def converge(self, controller=False, **kwargs):
         databag = self.publish()
-        tasks.converge_formation.delay(self).wait()
+        nodes = self.node_set.all()
+        subtasks = []
+        for n in nodes:
+            subtask = tasks.converge_node.si(n)
+            subtasks.append(subtask)
+        if controller is True:
+            subtasks.append(tasks.converge_controller.si())
+        group(*subtasks).apply_async().join()
         return databag
 
     def calculate(self):
@@ -459,10 +473,11 @@ def build(self):
         Release.objects.create(
             version=1, owner=self.owner, app=self, config=config, build=build)
         self.formation.publish()
-        tasks.build_app.delay(self).wait()
 
     def destroy(self):
-        tasks.destroy_app.delay(self).wait()
+        CM.purge_app(self.flat())
+        self.delete()
+        self.formation.publish()
 
     def publish(self):
         """Publish the application to configuration management"""

api/tasks.py

@@ -3,7 +3,6 @@
 import importlib
 
 from celery import task
-from celery.canvas import group
 
 from deis import settings
 from provider import import_provider_module
@@ -58,44 +57,6 @@ def run_node(node, command):
     return output, rc
 
 
-@task
-def build_formation(formation):
-    return
-
-
-@task
-def destroy_formation(formation):
-    app_tasks = [destroy_app.si(a) for a in formation.app_set.all()]
-    node_tasks = [destroy_node.si(n) for n in formation.node_set.all()]
-    layer_tasks = [destroy_layer.si(l) for l in formation.layer_set.all()]
-    group(app_tasks + node_tasks).apply_async().join()
-    group(layer_tasks).apply_async().join()
-    CM.purge_formation(formation.flat())
-    formation.delete()
-
-
-@task
-def converge_formation(formation):
-    nodes = formation.node_set.all()
-    subtasks = []
-    for n in nodes:
-        subtask = converge_node.si(n)
-        subtasks.append(subtask)
-    group(*subtasks).apply_async().join()
-
-
-@task
-def build_app(app):
-    return
-
-
-@task
-def destroy_app(app):
-    CM.purge_app(app.flat())
-    app.delete()
-    app.formation.publish()
-
-
 @task
 def converge_controller():
     CM.converge_controller()

api/views.py

@@ -7,7 +7,6 @@
 import json
 
 from Crypto.PublicKey import RSA
-from celery.canvas import group
 from django.contrib.auth.models import AnonymousUser, User
 from django.utils import timezone
 from rest_framework import permissions, status, viewsets
@@ -18,7 +17,6 @@
 
 from api import models
 from api import serializers
-from api import tasks
 
 
 class AnonymousAuthentication(BaseAuthentication):
@@ -200,7 +198,6 @@ def converge(self, request, **kwargs):
     def destroy(self, request, **kwargs):
         formation = self.get_object()
         formation.destroy()
-        tasks.converge_controller.delay().wait()
         return Response(status=status.HTTP_204_NO_CONTENT)
 
 
@@ -299,8 +296,7 @@ class AppViewSet(OwnerViewSet):
     def post_save(self, app, created=False, **kwargs):
         if created:
             app.build()
-        group(*[tasks.converge_formation.si(app.formation),  # @UndefinedVariable
-                tasks.converge_controller.si()]).apply_async().join()  # @UndefinedVariable
+        app.formation.converge(controller=True)
 
     def pre_save(self, app, created=False, **kwargs):
         if not app.pk and not app.formation.domain and app.formation.app_set.count() > 0:
@@ -372,8 +368,7 @@ def run(self, request, **kwargs):
     def destroy(self, request, **kwargs):
         app = self.get_object()
         app.destroy()
-        group(*[tasks.converge_formation.si(app.formation),  # @UndefinedVariable
-                tasks.converge_controller.si()]).apply_async().join()  # @UndefinedVariable
+        app.formation.converge(controller=True)
         return Response(status=status.HTTP_204_NO_CONTENT)