Fixes deadlock in ProducerPool. Closes celery/celery#1009

2 files changed, 16 insertions, 4 deletions

diff --git a/kombu/connection.py b/kombu/connection.py
index d17a6f6a..e1842d16 100644
--- a/kombu/connection.py
+++ b/kombu/connection.py
@@ -709,7 +709,10 @@ class Resource(object):
                 except Empty:
                     self._add_when_empty()
                 else:
-                    R = self.prepare(R)
+                    try:
+                        R = self.prepare(R)
+                    except BaseException:
+                        self.release_resource(R)
                     self._dirty.add(R)
                     break
         else:
diff --git a/kombu/pools.py b/kombu/pools.py
index f77ed444..c6dcaee7 100644
--- a/kombu/pools.py
+++ b/kombu/pools.py
@@ -39,7 +39,12 @@ class ProducerPool(Resource):
         return self.connections.acquire(block=True)
 
     def create_producer(self):
-        return self.Producer(self._acquire_connection())
+        conn = self._acquire_connection()
+        try:
+            return self.Producer(conn)
+        except BaseException:
+            conn.release()
+            raise
 
     def new(self):
         return lambda: self.create_producer()
@@ -56,8 +61,12 @@ class ProducerPool(Resource):
         if callable(p):
             p = p()
         if not p.channel:
-            connection = self._acquire_connection()
-            p.revive(connection.default_channel)
+            conn = self._acquire_connection()
+            try:
+                p.revive(conn.default_channel)
+            except BaseException:
+                conn.release()
+                raise
         return p
 
     def release(self, resource):