qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject svn commit: r728072 - in /qpid/trunk/qpid/cpp/src/qpid/cluster: Cluster.cpp Cluster.h ClusterPlugin.cpp OutputInterceptor.cpp OutputInterceptor.h WriteEstimate.cpp WriteEstimate.h
Date Fri, 19 Dec 2008 17:22:54 GMT
Author: aconway
Date: Fri Dec 19 09:22:52 2008
New Revision: 728072

URL: http://svn.apache.org/viewvc?rev=728072&view=rev
Log:
cluster: Increase initial estimate controlling writes.

Modified:
    qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
    qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h
    qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.h

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp?rev=728072&r1=728071&r2=728072&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Fri Dec 19 09:22:52 2008
@@ -83,7 +83,7 @@
     bool invoke(AMQBody& body) { return framing::invoke(*this, body).wasHandled(); }
 };
 
-Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b,
bool quorum_, size_t readMax_) :
+Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b,
bool quorum_, size_t readMax_, size_t writeEstimate_) :
     broker(b),
     poller(b.getPoller()),
     cpg(*this),
@@ -91,6 +91,7 @@
     myUrl(url_),
     myId(cpg.self()),
     readMax(readMax_),
+    writeEstimate(writeEstimate_),
     cpgDispatchHandle(
         cpg,
         boost::bind(&Cluster::dispatch, this, _1), // read

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h?rev=728072&r1=728071&r2=728072&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h Fri Dec 19 09:22:52 2008
@@ -69,7 +69,8 @@
     /**
      * Join a cluster. 
      */
-    Cluster(const std::string& name, const Url& url, broker::Broker&, bool useQuorum,
size_t readMax);
+    Cluster(const std::string& name, const Url& url, broker::Broker&, bool useQuorum,
+            size_t readMax, size_t writeEstimate);
 
     virtual ~Cluster();
 
@@ -95,6 +96,7 @@
     void checkQuorum();         // called in connection threads.
 
     size_t getReadMax() { return readMax; }
+    size_t getWriteEstimate() { return writeEstimate; }
     
   private:
     typedef sys::LockPtr<Cluster,sys::Monitor> LockPtr;
@@ -181,6 +183,7 @@
     const Url myUrl;
     const MemberId myId;
     const size_t readMax;
+    const size_t writeEstimate;
     framing::Uuid clusterId;
     NoOpConnectionOutputHandler shadowOut;
     sys::DispatchHandle cpgDispatchHandle;

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp?rev=728072&r1=728071&r2=728072&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp Fri Dec 19 09:22:52 2008
@@ -41,10 +41,10 @@
     string name;
     string url;
     bool quorum;
-    size_t readMax;
+    size_t readMax, writeEstimate;
 
     // FIXME aconway 2008-12-09: revisit default.
-    ClusterValues() : quorum(false), readMax(0) {}
+    ClusterValues() : quorum(false), readMax(0), writeEstimate(64) {}
   
     Url getUrl(uint16_t port) const {
         if (url.empty()) return Url::getIpAddressesUrl(port);
@@ -68,7 +68,10 @@
 #if HAVE_LIBCMAN
             ("cluster-cman", optValue(values.quorum), "Integrate with Cluster Manager (CMAN)
cluster.")
 #endif
-            ("cluster-read-max", optValue(values.readMax,"N"), "Throttle read rate from client
connections.")
+            ("cluster-read-max", optValue(values.readMax,"N"),
+             "Throttle read rate from client connections.")
+            ("cluster-write-estimate", optValue(values.writeEstimate, "Kb"),
+             "Estimate connection write rate per multicast cycle")
             ;
     }
 };
@@ -88,7 +91,7 @@
         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, values.readMax);
+        cluster = new Cluster(values.name, values.getUrl(broker->getPort(Broker::TCP_TRANSPORT)),
*broker, values.quorum, values.readMax, values.writeEstimate*1024);
         broker->setConnectionFactory(
             boost::shared_ptr<sys::ConnectionCodec::Factory>(
                 new ConnectionCodec::Factory(broker->getConnectionFactory(), *cluster)));

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp?rev=728072&r1=728071&r2=728072&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp Fri Dec 19 09:22:52 2008
@@ -32,8 +32,10 @@
 
 using namespace framing;
 
-OutputInterceptor::OutputInterceptor(cluster::Connection& p, sys::ConnectionOutputHandler&
h)
-    : parent(p), next(&h), sent(), moreOutput(), doingOutput()
+OutputInterceptor::OutputInterceptor(
+    cluster::Connection& p, sys::ConnectionOutputHandler& h)
+    : parent(p), next(&h), sent(), writeEstimate(p.getCluster().getWriteEstimate()),
+      moreOutput(), doingOutput()
 {}
 
 void OutputInterceptor::send(framing::AMQFrame& f) {
@@ -69,7 +71,7 @@
 void OutputInterceptor::deliverDoOutput(size_t requested) {
     size_t buf = next->getBuffered();
     if (parent.isLocal())
-        writeEstimate.delivered(sent, buf); // Update the estimate.
+        writeEstimate.delivered(requested, sent, buf); // Update the estimate.
 
     // Run the real doOutput() till we have added the requested data or there's nothing to
output.
     sent = 0;

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h?rev=728072&r1=728071&r2=728072&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h Fri Dec 19 09:22:52 2008
@@ -64,6 +64,7 @@
     mutable sys::Mutex lock;
     sys::ConnectionOutputHandler* next;
     size_t sent;
+    size_t lastDoOutput;
     WriteEstimate writeEstimate;
     bool moreOutput;
     bool doingOutput;

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.cpp?rev=728072&r1=728071&r2=728072&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.cpp Fri Dec 19 09:22:52 2008
@@ -27,7 +27,7 @@
 namespace cluster {
 
 WriteEstimate::WriteEstimate(size_t initial)
-    : growing(true), estimate(initial) {}
+    : growing(true), estimate(initial), lastEstimate(initial) {}
 
 size_t WriteEstimate::sending(size_t buffered) {
     // We want to send a doOutput request for enough data such
@@ -42,7 +42,8 @@
 
 size_t pad(size_t value) { return value + value/2; }
 
-void WriteEstimate::delivered(size_t sent, size_t buffered) {
+void WriteEstimate::delivered(size_t last, size_t sent, size_t buffered) {
+    lastEstimate = last;
     size_t wrote =  sent > buffered ? sent - buffered : 0;
     if (wrote == 0)             // No change
         return; 

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.h?rev=728072&r1=728071&r2=728072&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/WriteEstimate.h Fri Dec 19 09:22:52 2008
@@ -51,12 +51,17 @@
      * doOutput request just delivered, not yet executed. Update the estimate.
      * and estimate how much data to request in the next onOutput
      * request. 0 means don't send an onOutput request.
+     *
+     * @param delivered value in doOutput control.
      */
-    void delivered(size_t sent, size_t buffered);
+    void delivered(size_t delivered, size_t sent, size_t buffered);
+
+    /** Last estimate delivered, i.e. known to cluster */
+    size_t getLastEstimate() const { return estimate; }
 
   private:
     bool growing;
-    size_t estimate;
+    size_t estimate, lastEstimate;
 };
 
 }} // namespace qpid::cluster



Mime
View raw message