Cluster: Option --cluster-read-max configures read-to-redeliver flow-control.

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

5 files changed, 18 insertions, 13 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 602933b88b..8d9b5a1864 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -85,7 +85,7 @@ struct ClusterDispatcher : public framing::AMQP_AllOperations::ClusterHandler {
     bool invoke(AMQBody& body) { return framing::invoke(*this, body).wasHandled(); }
 };
 
-Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b, bool useQuorum) :
+Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b, bool quorum_, size_t readMax_) :
     broker(b),
     poller(b.getPoller()),
     cpg(*this),
@@ -104,7 +104,8 @@ Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b, b
     mgmtObject(0),
     state(INIT),
     lastSize(0),
-    lastBroker(false)
+    lastBroker(false),
+    readMax(readMax_)
 {
     ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
     if (agent != 0){
@@ -119,7 +120,7 @@ Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b, b
     deliverQueue.start();
     mcastQueue.start();
     QPID_LOG(notice, *this << " joining cluster " << name << " with url=" << myUrl);
-    if (useQuorum) quorum.init();
+    if (quorum_) quorum.init();
     cpg.join(name);
     broker.addFinalizer(boost::bind(&Cluster::brokerShutdown, this)); // Must be last for exception safety.
 }
diff --git a/cpp/src/qpid/cluster/Cluster.h b/cpp/src/qpid/cluster/Cluster.h
index 2ab2da6fa8..e172a0f180 100644
--- a/cpp/src/qpid/cluster/Cluster.h
+++ b/cpp/src/qpid/cluster/Cluster.h
@@ -64,11 +64,9 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     typedef std::vector<ConnectionPtr> Connections;
     
     /**
-     * Join a cluster.
-     * @param name of the cluster.
-     * @param url of this broker, sent to the cluster.
+     * Join a cluster. 
      */
-    Cluster(const std::string& name, const Url& url, broker::Broker&, bool useQuorum);
+    Cluster(const std::string& name, const Url& url, broker::Broker&, bool useQuorum, size_t readMax);
 
     virtual ~Cluster();
 
@@ -95,6 +93,8 @@ class Cluster : private Cpg::Handler, public management::Manageable {
 
     boost::function<bool ()> isQuorate;
     void checkQuorum();
+
+    size_t getReadMax() { return readMax; }
     
   private:
     typedef sys::LockPtr<Cluster,sys::Monitor> LockPtr;
@@ -215,6 +215,7 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     boost::shared_ptr<FailoverExchange> failoverExchange;
 
     Quorum quorum;
+    size_t readMax;
 
   friend std::ostream& operator<<(std::ostream&, const Cluster&);
   friend class ClusterDispatcher;
diff --git a/cpp/src/qpid/cluster/ClusterPlugin.cpp b/cpp/src/qpid/cluster/ClusterPlugin.cpp
index 02e6fffb71..5e6a5049a5 100644
--- a/cpp/src/qpid/cluster/ClusterPlugin.cpp
+++ b/cpp/src/qpid/cluster/ClusterPlugin.cpp
@@ -41,8 +41,10 @@ struct ClusterValues {
     string name;
     string url;
     bool quorum;
+    size_t readMax;
 
-    ClusterValues() : quorum(false) {}
+    // FIXME aconway 2008-12-09: revisit default.
+    ClusterValues() : quorum(false), readMax(4) {}
   
     Url getUrl(uint16_t port) const {
         if (url.empty()) return Url::getIpAddressesUrl(port);
@@ -66,6 +68,7 @@ struct ClusterOptions : public Options {
 #if HAVE_LIBCMAN
             ("cluster-cman", optValue(values.quorum), "Integrate with Cluster Manager (CMAN) cluster.")
 #endif
+            ("cluster-read-max", optValue(values.readMax,"N"), "Max un-delivered reads per client connection, 0 means unlimited.")
             ;
     }
 };
@@ -85,7 +88,7 @@ struct ClusterPlugin : public Plugin {
         if (values.name.empty()) return; // Only if --cluster-name option was specified.
         Broker* broker = dynamic_cast<Broker*>(&target);
         if (!broker) return;
-        cluster = new Cluster(values.name, values.getUrl(broker->getPort(Broker::TCP_TRANSPORT)), *broker, values.quorum);
+        cluster = new Cluster(values.name, values.getUrl(broker->getPort(Broker::TCP_TRANSPORT)), *broker, values.quorum, values.readMax);
         broker->setConnectionFactory(
             boost::shared_ptr<sys::ConnectionCodec::Factory>(
                 new ConnectionCodec::Factory(broker->getConnectionFactory(), *cluster)));
diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index fae81acf00..a422164c81 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -74,8 +74,7 @@ Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out,
 void Connection::init() {
     QPID_LOG(debug, cluster << " new connection: " << *this);
     if (isLocal() && !isCatchUp()) {
-        // FIXME aconway 2008-12-05: configurable credit limit
-        output.giveReadCredit(10);
+        output.giveReadCredit(cluster.getReadMax());
     }
 }
 
@@ -204,7 +203,8 @@ void Connection::deliverBuffer(Buffer& buf) {
     ++deliverSeq;
     while (mcastDecoder.decode(buf))
         delivered(mcastDecoder.frame);
-    output.giveReadCredit(1);
+    if  (cluster.getReadMax())
+        output.giveReadCredit(1);
 }
 
 broker::SessionState& Connection::sessionState() {
diff --git a/cpp/src/tests/latencytest.cpp b/cpp/src/tests/latencytest.cpp
index 25de361d11..6895964133 100644
--- a/cpp/src/tests/latencytest.cpp
+++ b/cpp/src/tests/latencytest.cpp
@@ -56,7 +56,7 @@ struct Args : public qpid::TestOptions {
     bool durable;
     string base;
 
-    Args() : size(256), count(1000), rate(0), reportFrequency(100),
+    Args() : size(256), count(1000), rate(0), reportFrequency(1000),
     	     timeLimit(0), queues(1), 
              prefetch(100), ack(0),
              durable(false), base("latency-test")