Author: gsim
Date: Mon Jun 2 14:06:36 2008
New Revision: 662561
URL: http://svn.apache.org/viewvc?rev=662561&view=rev
Log:
Improve performance of synchronous publication by not requesting
known-completed response
for every completed sent.
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/RangeSet.h
incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/RangeSet.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/RangeSet.h?rev=662561&r1=662560&r2=662561&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/RangeSet.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/RangeSet.h Mon Jun 2 14:06:36 2008
@@ -169,6 +169,9 @@
RangeIterator rangesEnd() const { return ranges.end(); }
size_t rangesSize() const { return ranges.size(); }
+ // The difference between the start and end of this range set
+ uint32_t span() const;
+
bool empty() const { return ranges.empty(); }
void clear() { ranges.clear(); }
@@ -309,6 +312,11 @@
return (i != ranges.end() && i->contains(t)) ? *i : Range<T>(t,t);
}
+template <class T> uint32_t RangeSet<T>::span() const {
+ if (ranges.empty()) return 0;
+ return ranges.back().last() - ranges.front().first();
+}
+
} // namespace qpid
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp?rev=662561&r1=662560&r2=662561&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp Mon Jun
2 14:06:36 2008
@@ -240,7 +240,8 @@
void SessionHandler::sendCompletion() {
checkAttached();
- peer.completed(getState()->receiverGetUnknownComplete(), true);
+ const SequenceSet& c = getState()->receiverGetUnknownComplete();
+ peer.completed(c, c.span() > 1000);
}
void SessionHandler::sendAttach(bool force) {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp?rev=662561&r1=662560&r2=662561&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp Mon Jun 2
14:06:36 2008
@@ -535,7 +535,7 @@
void SessionImpl::sendCompletionImpl()
{
- proxy.completed(completedIn, true);
+ proxy.completed(completedIn, completedIn.span() > 1000);
}
void SessionImpl::gap(const framing::SequenceSet& /*commands*/)
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp?rev=662561&r1=662560&r2=662561&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp Mon Jun 2 14:06:36
2008
@@ -84,6 +84,7 @@
bool confirm;
bool durable;
bool uniqueData;
+ bool syncPub;
// Subscriber
size_t subs;
@@ -102,7 +103,7 @@
Opts() :
TestOptions(helpText),
setup(false), control(false), publish(false), subscribe(false),
- pubs(1), count(500000), size(1024), confirm(true), durable(false),
uniqueData(false),
+ pubs(1), count(500000), size(1024), confirm(true), durable(false),
uniqueData(false), syncPub(false),
subs(1), ack(0),
qt(1), iterations(1), mode(SHARED), summary(false),
intervalSub(0), intervalPub(0)
@@ -124,6 +125,7 @@
("pub-confirm", optValue(confirm, "yes|no"), "Publisher use
confirm-mode.")
("durable", optValue(durable, "yes|no"), "Publish messages as
durable.")
("unique-data", optValue(uniqueData, "yes|no"), "Make data for
each message unique.")
+ ("sync-publish", optValue(syncPub, "yes|no"), "Wait for
confirmation of each message before sending the next one.")
("nsubs", optValue(subs, "N"), "Create N subscribers.")
("sub-ack", optValue(ack, "N"), "N>0: Subscriber acks batches of
N.\n"
@@ -461,10 +463,17 @@
// any heap allocation.
const_cast<std::string&>(msg.getData()).replace(offset,
sizeof(uint32_t),
reinterpret_cast<const char*>(&i), sizeof(uint32_t));
- session.messageTransfer(
- arg::destination=destination,
- arg::content=msg,
- arg::acceptMode=1);
+ if (opts.syncPub) {
+ sync(session).messageTransfer(
+ arg::destination=destination,
+ arg::content=msg,
+ arg::acceptMode=1);
+ } else {
+ session.messageTransfer(
+ arg::destination=destination,
+ arg::content=msg,
+ arg::acceptMode=1);
+ }
if (opts.intervalPub) ::usleep(opts.intervalPub*1000);
}
if (opts.confirm) session.sync();
