diff options
Diffstat (limited to 'cpp/src')
| -rw-r--r-- | cpp/src/qpid/RangeSet.h | 8 | ||||
| -rw-r--r-- | cpp/src/qpid/amqp_0_10/SessionHandler.cpp | 3 | ||||
| -rw-r--r-- | cpp/src/qpid/client/SessionImpl.cpp | 2 | ||||
| -rw-r--r-- | cpp/src/tests/perftest.cpp | 19 |
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(); |