Fixes deadlock when producer_pool inherits connection pool from parent.

celery/app/amqp.py

@@ -282,6 +282,10 @@ class AMQP(object):
     #: Cached and prepared routing table.
     _rtable = None
 
+    #: Underlying producer pool instance automatically
+    #: set by the :attr:`producer_pool`.
+    _producer_pool = None
+
     def __init__(self, app):
         self.app = app
 
@@ -358,10 +362,14 @@ class AMQP(object):
     def router(self):
         return self.Router()
 
-    @cached_property
+    @property
     def producer_pool(self):
-        return ProducerPool(self.app.pool, limit=self.app.pool.limit,
-                            Producer=self.TaskProducer)
+        if self._producer_pool is None:
+            self._producer_pool = ProducerPool(self.app.pool,
+                limit=self.app.pool.limit,
+                Producer=self.TaskProducer,
+            )
+        return self._producer_pool
     publisher_pool = producer_pool  # compat alias
 
     @cached_property

celery/app/base.py

@@ -336,6 +336,10 @@ class Celery(object):
         if self._pool:
             self._pool.force_close_all()
             self._pool = None
+            amqp = self.amqp
+            if amqp._producer_pool:
+                amqp._producer_pool.force_close_all()
+                amqp._producer_pool = None
 
     def create_task_cls(self):
         """Creates a base task class using default configuration