You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2010/02/02 16:47:00 UTC
svn commit: r905674 - in /qpid/trunk/qpid/cpp/src/qpid/cluster: Cluster.cpp
Cluster.h Connection.cpp Connection.h
Author: aconway
Date: Tue Feb 2 15:46:46 2010
New Revision: 905674
URL: http://svn.apache.org/viewvc?rev=905674&view=rev
Log:
Cluster: debug snapshots of queue depth at broker join, help find inconsistencies.
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/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.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=905674&r1=905673&r2=905674&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Tue Feb 2 15:46:46 2010
@@ -474,6 +474,7 @@
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 @@
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 @@
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 @@
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 @@
}
}
+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 @@
<< " 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 @@
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
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=905674&r1=905673&r2=905674&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h Tue Feb 2 15:46:46 2010
@@ -120,6 +120,9 @@
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 @@
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 @@
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);
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=905674&r1=905673&r2=905674&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Tue Feb 2 15:46:46 2010
@@ -195,6 +195,11 @@
~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);
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=905674&r1=905673&r2=905674&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h Tue Feb 2 15:46:46 2010
@@ -205,7 +205,7 @@
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();
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org