Improve performance of synchronous publication by not requesting known-completed response

for every completed sent.



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

4 files changed, 25 insertions, 7 deletions

diff --git a/cpp/src/qpid/RangeSet.h b/cpp/src/qpid/RangeSet.h
index af3b2223cd..2a88426f17 100644
--- a/cpp/src/qpid/RangeSet.h
+++ b/cpp/src/qpid/RangeSet.h
@@ -169,6 +169,9 @@ class RangeSet
     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 @@ template <class T> Range<T> RangeSet<T>::rangeContaining(const T& t) const {
     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
 
diff --git a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
index 6d43dd1789..0d1cc57072 100644
--- a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
+++ b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
@@ -240,7 +240,8 @@ void SessionHandler::sendDetach()
 
 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) {
diff --git a/cpp/src/qpid/client/SessionImpl.cpp b/cpp/src/qpid/client/SessionImpl.cpp
index 58f4bc0aa7..66e1b9e40f 100644
--- a/cpp/src/qpid/client/SessionImpl.cpp
+++ b/cpp/src/qpid/client/SessionImpl.cpp
@@ -535,7 +535,7 @@ void SessionImpl::sendFlush()
 
 void SessionImpl::sendCompletionImpl()
 {
-    proxy.completed(completedIn, true);
+    proxy.completed(completedIn, completedIn.span() > 1000);
 }
 
 void SessionImpl::gap(const framing::SequenceSet& /*commands*/)
diff --git a/cpp/src/tests/perftest.cpp b/cpp/src/tests/perftest.cpp
index 91ecd83f50..e0e947eb74 100644
--- a/cpp/src/tests/perftest.cpp
+++ b/cpp/src/tests/perftest.cpp
@@ -84,6 +84,7 @@ struct Opts : public TestOptions {
     bool confirm;
     bool durable;
     bool uniqueData;
+    bool syncPub;
 
     // Subscriber
     size_t subs;
@@ -102,7 +103,7 @@ struct Opts : public TestOptions {
     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 @@ struct Opts : public TestOptions {
             ("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 @@ struct PublishThread : public Client {
                     // 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();