QPID-1971 - bind in fedOpReorigen mode is not threadsafe for TopicExchange

Applied patch from Ken Giusti.



git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@803996 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 35 insertions, 10 deletions

diff --git a/cpp/src/qpid/broker/DirectExchange.cpp b/cpp/src/qpid/broker/DirectExchange.cpp
index 454b497f11..b9f24dee5f 100644
--- a/cpp/src/qpid/broker/DirectExchange.cpp
+++ b/cpp/src/qpid/broker/DirectExchange.cpp
@@ -92,12 +92,24 @@ bool DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey, con
         if (bk.fedBinding.count() == 0)
             unbind(queue, routingKey, 0);
     } else if (fedOp == fedOpReorigin) {
-        for (std::map<string, BoundKey>::iterator iter = bindings.begin();
-             iter != bindings.end(); iter++) {
-            const BoundKey& bk = iter->second;
-            if (bk.fedBinding.hasLocal()) {
-                propagateFedOp(iter->first, string(), fedOpBind, string());
+        /** gather up all the keys that need rebinding in a local vector
+         * while holding the lock.  Then propagate once the lock is
+         * released
+         */
+        std::vector<std::string> keys2prop;
+        {
+            Mutex::ScopedLock l(lock);    
+            for (Bindings::iterator iter = bindings.begin();
+                 iter != bindings.end(); iter++) {
+                const BoundKey& bk = iter->second;
+                if (bk.fedBinding.hasLocal()) {
+                    keys2prop.push_back(iter->first);
+                }
             }
+        }   /* lock dropped */
+        for (std::vector<std::string>::const_iterator key = keys2prop.begin();
+             key != keys2prop.end(); key++) {
+            propagateFedOp( *key, string(), fedOpBind, string());
         }
     }
 
diff --git a/cpp/src/qpid/broker/TopicExchange.cpp b/cpp/src/qpid/broker/TopicExchange.cpp
index 6c8832179f..6bf0b104ea 100644
--- a/cpp/src/qpid/broker/TopicExchange.cpp
+++ b/cpp/src/qpid/broker/TopicExchange.cpp
@@ -225,12 +225,25 @@ bool TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, cons
         if (reallyUnbind)
             unbind(queue, routingPattern, 0);
     } else if (fedOp == fedOpReorigin) {
-        for (BindingMap::iterator iter = bindings.begin();
-             iter != bindings.end(); iter++) {
-            const BoundKey& bk = iter->second;
-            if (bk.fedBinding.hasLocal()) {
-                propagateFedOp(iter->first, string(), fedOpBind, string());
+        /** gather up all the keys that need rebinding in a local vector
+         * while holding the lock.  Then propagate once the lock is
+         * released
+         */
+        std::vector<std::string> keys2prop;
+        {
+            RWlock::ScopedRlock l(lock);    
+            for (BindingMap::iterator iter = bindings.begin();
+                 iter != bindings.end(); iter++) {
+                const BoundKey& bk = iter->second;
+                
+                if (bk.fedBinding.hasLocal()) {
+                    keys2prop.push_back(iter->first);
+                }
             }
+        }   /* lock dropped */
+        for (std::vector<std::string>::const_iterator key = keys2prop.begin();
+             key != keys2prop.end(); key++) {
+            propagateFedOp( *key, string(), fedOpBind, string());
         }
     }