Cluster: debug snapshots of queue depth at broker join, help find inconsistencies.

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

4 files changed, 39 insertions, 10 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 7dd8c7e62c..3a6c902d29 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -474,6 +474,7 @@ void Cluster::deliveredFrame(const EventFrame& efConst) {
 
 
 void Cluster::processFrame(const EventFrame& e, Lock& l) {
+    map.incrementFrameSeq();
     if (e.isCluster()) {
         QPID_LOG(trace, *this << " DLVR: " << e);
         ClusterDispatcher dispatch(*this, e.connectionId.getMember(), l);
@@ -481,7 +482,6 @@ void Cluster::processFrame(const EventFrame& e, Lock& l) {
             throw Exception(QPID_MSG("Invalid cluster control"));
     }
     else if (state >= CATCHUP) {
-        map.incrementFrameSeq();
         ConnectionPtr connection = getConnection(e, l);
         if (connection) {
             QPID_LOG(trace, *this << " DLVR " << map.getFrameSeq() << ":  " << e);
@@ -578,7 +578,7 @@ void Cluster::initMapCompleted(Lock& l) {
         elders = initMap.getElders();
         QPID_LOG(debug, *this << " elders: " << elders);
         if (elders.empty())
-            becomeElder();
+            becomeElder(l);
         else {
             broker.getLinks().setPassive(true);
             broker.getQueueEvents().disable();
@@ -635,11 +635,11 @@ void Cluster::configChange(const MemberId&, const std::string& configStr, Lock&
 
     if (state >= CATCHUP && memberChange) {
         memberUpdate(l);
-        if (elders.empty()) becomeElder();
+        if (elders.empty()) becomeElder(l);
     }
 }
 
-void Cluster::becomeElder() {
+void Cluster::becomeElder(Lock&) {
     if (elder) return;          // We were already the elder.
     // We are the oldest, reactive links if necessary
     QPID_LOG(info, *this << " became the elder, active for links.");
@@ -656,6 +656,29 @@ void Cluster::makeOffer(const MemberId& id, Lock& ) {
     }
 }
 
+namespace {
+struct AppendQueue {
+    ostream* os;
+    AppendQueue(ostream& o) : os(&o) {}
+    void operator()(const boost::shared_ptr<broker::Queue>& q) {
+        (*os) << " " << q->getName() << "=" << q->getMessageCount();
+    }
+};
+} // namespace
+
+// Log a snapshot of broker state, used for debugging inconsistency problems.
+// May only be called in deliver thread.
+void Cluster::debugSnapshot(const char* prefix, Connection* connection) {
+    assertClusterSafe();
+    std::ostringstream msg;
+    msg << prefix;
+    if (connection) msg << " " << *connection;
+    msg << " snapshot " << map.getFrameSeq() << ":";
+    AppendQueue append(msg);
+    broker.getQueues().eachQueue(append);
+    QPID_LOG(trace, msg.str());
+}
+
 // Called from Broker::~Broker when broker is shut down.  At this
 // point we know the poller has stopped so no poller callbacks will be
 // invoked. We must ensure that CPG has also shut down so no CPG
@@ -738,6 +761,7 @@ void Cluster::updateOffer(const MemberId& updater, uint64_t updateeInt, Lock& l)
                  << " to " << updatee);
         deliverEventQueue.start(); // Not involved in update.
     }
+    if (updatee != self && url) debugSnapshot("join");
 }
 
 static client::ConnectionSettings connectionSettings(const ClusterSettings& settings) {
@@ -808,6 +832,7 @@ void Cluster::checkUpdateIn(Lock& l) {
         broker.setClusterUpdatee(false);
         discarding = false;     // ok to set, we're stalled for update.
         QPID_LOG(notice, *this << " update complete, starting catch-up.");
+        debugSnapshot("initial");
         deliverEventQueue.start();
     }
     else if (updateRetracted) { // Update was retracted, request another update
diff --git a/cpp/src/qpid/cluster/Cluster.h b/cpp/src/qpid/cluster/Cluster.h
index 977c873e29..ffb870606a 100644
--- a/cpp/src/qpid/cluster/Cluster.h
+++ b/cpp/src/qpid/cluster/Cluster.h
@@ -120,6 +120,9 @@ class Cluster : private Cpg::Handler, public management::Manageable {
 
     bool isElder() const;
 
+    // For debugging only. Can only be called in deliver thread.
+    void debugSnapshot(const char*, Connection* =0);
+
   private:
     typedef sys::Monitor::ScopedLock Lock;
 
@@ -178,10 +181,8 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     void memberUpdate(Lock&);
     void setClusterId(const framing::Uuid&, Lock&);
     void erase(const ConnectionId&, Lock&);       
-
     void initMapCompleted(Lock&);
-
-
+    void becomeElder(Lock&);
 
     // == Called in CPG dispatch thread
     void deliver( // CPG deliver callback. 
@@ -202,8 +203,6 @@ class Cluster : private Cpg::Handler, public management::Manageable {
         const struct cpg_address */*joined*/, int /*nJoined*/
     );
 
-    void becomeElder();
-
     // == Called in management threads.
     virtual qpid::management::ManagementObject* GetManagementObject() const;
     virtual management::Manageable::status_t ManagementMethod (uint32_t methodId, management::Args& args, std::string& text);
diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index c1c9c02611..5faa184e30 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -195,6 +195,11 @@ struct GiveReadCreditOnExit {
     ~GiveReadCreditOnExit() { connection.giveReadCredit(credit); }
 };
 
+void Connection::deliverDoOutput(uint32_t limit) {
+    output.deliverDoOutput(limit);
+    cluster.debugSnapshot("deliver-do-output", this);
+}
+
 // Called in delivery thread, in cluster order.
 void Connection::deliveredFrame(const EventFrame& f) {
     GiveReadCreditOnExit gc(*this, f.readCredit);
diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h
index 7345378130..9a4e52a9d6 100644
--- a/cpp/src/qpid/cluster/Connection.h
+++ b/cpp/src/qpid/cluster/Connection.h
@@ -205,7 +205,7 @@ class Connection :
     
     void init();
     bool checkUnsupported(const framing::AMQBody& body);
-    void deliverDoOutput(uint32_t limit) { output.deliverDoOutput(limit); }
+    void deliverDoOutput(uint32_t limit);
 
     boost::shared_ptr<broker::Queue> findQueue(const std::string& qname);
     broker::SessionState& sessionState();