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 2009/01/09 15:47:21 UTC
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
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);