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/08/14 14:56:31 UTC
svn commit: r804206 - in /qpid/trunk/qpid/cpp/src/qpid/cluster: Cluster.cpp
Event.cpp Multicaster.cpp Multicaster.h
Author: aconway
Date: Fri Aug 14 12:56:31 2009
New Revision: 804206
URL: http://svn.apache.org/viewvc?rev=804206&view=rev
Log:
Revert "Batch multiple events into a single CPG multicast."
This reverts svn revision 803713: Batch multiple events into a single CPG multicast.
Modified:
qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Event.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.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=804206&r1=804205&r2=804206&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Fri Aug 14 12:56:31 2009
@@ -323,11 +323,9 @@
{
MemberId from(nodeid, pid);
framing::Buffer buf(static_cast<char*>(msg), msg_len);
- while (buf.available()) {
- Event e(Event::decodeCopy(from, buf));
- LATENCY_TRACK(if (e.getConnectionId().getMember() == self) mcast.cpgLatency.finish());
- deliverEvent(e);
- }
+ Event e(Event::decodeCopy(from, buf));
+ LATENCY_TRACK(if (e.getConnectionId().getMember() == self) mcast.cpgLatency.finish());
+ deliverEvent(e);
}
LATENCY_TRACK(sys::LatencyTracker<const char*> eventQueueLatencyTracker("EventQueue");)
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Event.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Event.cpp?rev=804206&r1=804205&r2=804206&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Event.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Event.cpp Fri Aug 14 12:56:31 2009
@@ -72,7 +72,7 @@
if (buf.available() < e.size)
throw Exception("Not enough data for multicast event");
e.store = RefCountedBuffer::create(e.size + HEADER_SIZE);
- buf.getRawData((uint8_t*)(e.getData()), e.size);
+ memcpy(e.getData(), buf.getPointer() + buf.getPosition(), e.size);
return e;
}
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=804206&r1=804205&r2=804206&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp Fri Aug 14 12:56:31 2009
@@ -24,14 +24,10 @@
#include "qpid/log/Statement.h"
#include "qpid/framing/AMQBody.h"
#include "qpid/framing/AMQFrame.h"
-#include <boost/bind.hpp>
-#include <algorithm>
namespace qpid {
namespace cluster {
-static const int MCAST_IOV_MAX=63; // Limit imposed by CPG
-
Multicaster::Multicaster(Cpg& cpg_,
const boost::shared_ptr<sys::Poller>& poller,
boost::function<void()> onError_) :
@@ -40,8 +36,7 @@
#endif
onError(onError_), cpg(cpg_),
queue(boost::bind(&Multicaster::sendMcast, this, _1), poller),
- holding(true),
- ioVector(MCAST_IOV_MAX)
+ holding(true)
{
queue.start();
}
@@ -75,29 +70,26 @@
queue.push(e);
}
-Multicaster::PollableEventQueue::Batch::const_iterator Multicaster::sendMcast(
- const PollableEventQueue::Batch& events)
-{
- PollableEventQueue::Batch::const_iterator i = events.begin();
+
+Multicaster::PollableEventQueue::Batch::const_iterator Multicaster::sendMcast(const PollableEventQueue::Batch& values) {
try {
- while (i < events.end()) {
- size_t count = std::min(MCAST_IOV_MAX, int(events.end() - i));
- std::transform(i, i+count, ioVector.begin(),
- boost::bind(&Event::toIovec, _1));
- if (!cpg.mcast(&ioVector.front(), count)) {
- QPID_LOG(trace, "CPG flow control, will resend "
- << events.end() - i << " events");
- break;
+ PollableEventQueue::Batch::const_iterator i = values.begin();
+ while( i != values.end()) {
+ iovec iov = i->toIovec();
+ if (!cpg.mcast(&iov, 1)) {
+ // cpg didn't send because of CPG flow control.
+ break;
}
- i += count;
+ ++i;
}
+ return i;
}
catch (const std::exception& e) {
QPID_LOG(critical, "Multicast error: " << e.what());
queue.stop();
onError();
+ return values.end();
}
- return i;
}
void Multicaster::release() {
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=804206&r1=804205&r2=804206&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h Fri Aug 14 12:56:31 2009
@@ -29,7 +29,6 @@
#include "qpid/sys/LatencyTracker.h"
#include <boost/shared_ptr.hpp>
#include <deque>
-#include <vector>
namespace qpid {
@@ -73,7 +72,7 @@
PollableEventQueue queue;
bool holding;
PlainEventQueue holdingQueue;
- std::vector< ::iovec> ioVector;
+ std::vector<struct ::iovec> ioVector;
};
}} // namespace qpid::cluster
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org