QPID-4392 C++ Broker link channel number wrap.

Maintain a pool of available channel numbers.



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

4 files changed, 25 insertions, 7 deletions

diff --git a/qpid/cpp/src/qpid/broker/Bridge.h b/qpid/cpp/src/qpid/broker/Bridge.h
index 2b4d019ccd..da397b8f77 100644
--- a/qpid/cpp/src/qpid/broker/Bridge.h
+++ b/qpid/cpp/src/qpid/broker/Bridge.h
@@ -65,6 +65,7 @@ class Bridge : public PersistableConfig,
 
     QPID_BROKER_EXTERN void close();
     bool isDurable() { return args.i_durable; }
+    framing::ChannelId getChannel() const { return channel; }
     Link *getLink() const { return link; }
     const std::string getSrc() const { return args.i_src; }
     const std::string getDest() const { return args.i_dest; }
diff --git a/qpid/cpp/src/qpid/broker/Link.cpp b/qpid/cpp/src/qpid/broker/Link.cpp
index b014217180..82389a5e3d 100644
--- a/qpid/cpp/src/qpid/broker/Link.cpp
+++ b/qpid/cpp/src/qpid/broker/Link.cpp
@@ -30,6 +30,7 @@
 #include "qpid/log/Statement.h"
 #include "qpid/framing/enum.h"
 #include "qpid/framing/reply_exceptions.h"
+#include "qpid/framing/amqp_types.h"
 #include "qpid/broker/AclModule.h"
 #include "qpid/broker/Exchange.h"
 #include "qpid/UrlArray.h"
@@ -148,7 +149,7 @@ Link::Link(const string&  _name,
       currentInterval(1),
       closing(false),
       reconnectNext(0),         // Index of next address for reconnecting in url.
-      channelCounter(1),
+      freeChannels(1, framing::CHANNEL_MAX),
       connection(0),
       agent(0),
       listener(l),
@@ -542,12 +543,26 @@ bool Link::hideManagement() const {
     return !mgmtObject || ( broker && broker->isInCluster());
 }
 
-uint Link::nextChannel()
+// Allocate channel from link free pool
+framing::ChannelId Link::nextChannel()
 {
     Mutex::ScopedLock mutex(lock);
-    if (channelCounter >= framing::CHANNEL_MAX)
-        channelCounter = 1;
-    return channelCounter++;
+    if (!freeChannels.empty()) {
+        framing::ChannelId c = freeChannels.front();
+        freeChannels -= c;
+        QPID_LOG(debug, "Link " << name << " allocates channel: " << c);
+        return c;
+    } else {
+        throw Exception(Msg() << "Link " << name << " channel pool is empty");
+    }
+}
+
+// Return channel to link free pool
+void Link::returnChannel(framing::ChannelId c)
+{
+    Mutex::ScopedLock mutex(lock);
+    QPID_LOG(debug, "Link " << name << " frees channel: " << c);
+    freeChannels += c;
 }
 
 void Link::notifyConnectionForced(const string text)
diff --git a/qpid/cpp/src/qpid/broker/Link.h b/qpid/cpp/src/qpid/broker/Link.h
index 49ee3ef36a..12bf28423a 100644
--- a/qpid/cpp/src/qpid/broker/Link.h
+++ b/qpid/cpp/src/qpid/broker/Link.h
@@ -82,7 +82,7 @@ class Link : public PersistableConfig, public management::Manageable {
     Bridges created;   // Bridges pending creation
     Bridges active;    // Bridges active
     Bridges cancellations;    // Bridges pending cancellation
-    uint channelCounter;
+    RangeSet<framing::ChannelId> freeChannels;
     Connection* connection;
     management::ManagementAgent* agent;
     boost::function<void(Link*)> listener;
@@ -151,7 +151,8 @@ class Link : public PersistableConfig, public management::Manageable {
 
     bool isDurable() { return durable; }
     void maintenanceVisit ();
-    uint nextChannel();
+    framing::ChannelId nextChannel();        // allocate channel from link free pool
+    void returnChannel(framing::ChannelId);  // return channel to link free pool
     void add(Bridge::shared_ptr);
     void cancel(Bridge::shared_ptr);
 
diff --git a/qpid/cpp/src/qpid/broker/LinkRegistry.cpp b/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
index 4fb9c6cd33..10f689c976 100644
--- a/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
+++ b/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
@@ -254,6 +254,7 @@ void LinkRegistry::destroyBridge(Bridge *bridge)
     Link *link = b->second->getLink();
     if (link) {
         link->cancel(b->second);
+        link->returnChannel( bridge->getChannel() );
     }
     if (b->second->isDurable())
         store->destroy(*(b->second));