qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject svn commit: r733051 - in /qpid/trunk/qpid/cpp/src/qpid/cluster: Cluster.cpp ClusterPlugin.cpp Connection.h FailoverExchange.cpp Multicaster.cpp Multicaster.h Quorum_cman.cpp
Date Fri, 09 Jan 2009 14:47:21 GMT
Author: aconway
Date: Fri Jan  9 06:47:21 2009
New Revision: 733051

URL: http://svn.apache.org/viewvc?rev=733051&view=rev
Log:
Fix --cluster-mast-max: errors in last commit.
Work around problems with CPG flow control.

Modified:
    qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
    qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h
    qpid/trunk/qpid/cpp/src/qpid/cluster/Quorum_cman.cpp

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=733051&r1=733050&r2=733051&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Fri Jan  9 06:47:21 2009
@@ -187,14 +187,15 @@
     Mutex::ScopedLock l(lock);
     MemberId from(nodeid, pid);
     framing::Buffer buf(static_cast<char*>(msg), msg_len);
-    deliver(Event::decode(from, buf), l);
+    Event e(Event::decode(from, buf));
+    if (from == myId) // Record self-deliveries for flow control.
+        mcast.selfDeliver(e);
+    deliver(e, l);
 }
 
 void Cluster::deliver(const Event& e, Lock&) {
     if (state == LEFT) return;
     QPID_LOG(trace, *this << " PUSH: " << e);
-    if (e.getMemberId() == myId)
-        mcast.delivered(e);     // Note delivery for flow control
     deliverQueue.push(e);
 }
 
@@ -215,7 +216,7 @@
     if (e.isCluster())  {
         while (frame.decode(buf)) {
             QPID_LOG(trace, *this << " DLVR: " << e << " " << frame);
-            Mutex::ScopedLock l(lock); // FIXME aconway 2008-12-11: lock scope is too big.
+            Mutex::ScopedLock l(lock); // FIXME aconway 2008-12-11: lock scope too big?
             ClusterDispatcher dispatch(*this, e.getMemberId(), l);
             if (!framing::invoke(dispatch, *frame.getBody()).wasHandled())
                 throw Exception(QPID_MSG("Invalid cluster control"));
@@ -406,8 +407,6 @@
     }
 }
 
-// FIXME aconway 2008-10-15: no longer need a separate control now
-// that the dump control is in the deliver queue.
 void Cluster::dumpStart(const MemberId& dumpee, const Url& url, Lock&) {
     if (state == LEFT) return;
     assert(state == OFFER);

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=733051&r1=733050&r2=733051&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp Fri Jan  9 06:47:21 2009
@@ -43,7 +43,6 @@
     bool quorum;
     size_t readMax, writeEstimate, mcastMax;
 
-    // FIXME aconway 2008-12-09: revisit default.
     ClusterValues() : quorum(false), readMax(10), writeEstimate(64), mcastMax(10) {}
   
     Url getUrl(uint16_t port) const {

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h?rev=733051&r1=733050&r2=733051&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h Fri Jan  9 06:47:21 2009
@@ -167,8 +167,6 @@
     framing::ChannelId currentChannel;
     boost::shared_ptr<broker::TxBuffer> txBuffer;
     
-    int FIXMEcredit;            // FIXME aconway 2008-12-05: remove
-    
   friend std::ostream& operator<<(std::ostream&, const Connection&);
 };
 

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp?rev=733051&r1=733050&r2=733051&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp Fri Jan  9 06:47:21 2009
@@ -79,7 +79,7 @@
 void FailoverExchange::sendUpdate(const Queue::shared_ptr& queue) {
     // Called with lock held.
     if (urls.empty()) return;
-    framing::Array array(0x95); // FIXME aconway 2008-10-06: Array is unusable like this.
Need type constants or better mapping.
+    framing::Array array(0x95);
     for (Urls::const_iterator i = urls.begin(); i != urls.end(); ++i) 
         array.add(boost::shared_ptr<Str16Value>(new Str16Value(i->str())));
     const ProtocolVersion v;

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp?rev=733051&r1=733050&r2=733051&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp Fri Jan  9 06:47:21 2009
@@ -68,21 +68,27 @@
     try {
         PollableEventQueue::Queue::iterator i = values.begin();
         while( i != values.end()) {
-            iovec iov = { const_cast<char*>(i->getStore()), i->getStoreSize()
};
-            if (!cpg.mcast(&iov, 1))
-                break; // cpg.mcast returns false for flow control
-            QPID_LOG(trace, " MCAST " << *i);
-            ++i;
             if (mcastMax) {
                 sys::Mutex::ScopedLock l(lock);
-                assert(pending < mcastMax);
-                if (++pending == mcastMax) {
+                if (pending == mcastMax) {
                     queue.stop();
                     break ;
                 }
+                ++pending;
+            }
+            iovec iov = { const_cast<char*>(i->getStore()), i->getStoreSize()
};
+            if (!cpg.mcast(&iov, 1)) {
+                // cpg didn't send because of CPG flow control.
+                if (mcastMax) {
+                    sys::Mutex::ScopedLock l(lock);
+                    --pending;
+                }
+                break; 
             }
+            QPID_LOG(trace, " MCAST " << *i); 
+            ++i;
         }
-        values.erase(values.begin(), i);
+        values.erase(values.begin(), i); // Erase sent events.
     }
     catch (const std::exception& e) {
         QPID_LOG(critical, "Multicast error: " << e.what());
@@ -98,9 +104,10 @@
     holdingQueue.clear();
 }
 
-void Multicaster::delivered(const Event&) {
+void Multicaster::selfDeliver(const Event&) {
     sys::Mutex::ScopedLock l(lock);
     if (mcastMax) {
+        assert(pending > 0);
         assert(pending <= mcastMax);
         if  (pending == mcastMax) 
             queue.start();

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h?rev=733051&r1=733050&r2=733051&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h Fri Jan  9 06:47:21 2009
@@ -57,7 +57,7 @@
     /** End holding mode, held events are mcast */
     void release();
     /** Call when events are self-delivered to manage flow control. */
-    void delivered(const Event& e);
+    void selfDeliver(const Event&);
     
   private:
     typedef sys::PollableQueue<Event> PollableEventQueue;

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Quorum_cman.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Quorum_cman.cpp?rev=733051&r1=733050&r2=733051&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Quorum_cman.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Quorum_cman.cpp Fri Jan  9 06:47:21 2009
@@ -35,7 +35,7 @@
     enable = true;
     cman = cman_init(0);
     if (cman == 0) throw ErrnoException("Can't connect to cman service");
-    // FIXME aconway 2008-11-13: configure max wait.
+    // FIXME aconway 2008-11-13: configurable max wait.
     for (int retry = 0;  !cman_is_quorate(cman) && retry < 30; retry++) {
         QPID_LOG(info, "Waiting for cluster quorum: " << sys::strError(errno));
         sys::sleep(1);



Mime
View raw message