summaryrefslogtreecommitdiff
path: root/cpp/src/tests/qpid-perftest.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/tests/qpid-perftest.cpp')
-rw-r--r--cpp/src/tests/qpid-perftest.cpp22
1 files changed, 15 insertions, 7 deletions
diff --git a/cpp/src/tests/qpid-perftest.cpp b/cpp/src/tests/qpid-perftest.cpp
index 4d7b563c8c..664f0cf877 100644
--- a/cpp/src/tests/qpid-perftest.cpp
+++ b/cpp/src/tests/qpid-perftest.cpp
@@ -173,7 +173,7 @@ struct Opts : public TestOptions {
if (count % subs) {
count += subs - (count % subs);
cout << "WARNING: Adjusted --count to " << count
- << " the nearest multiple of --nsubs" << endl;
+ << " the next multiple of --nsubs" << endl;
}
totalPubs = pubs*qt;
totalSubs = subs*qt;
@@ -396,7 +396,7 @@ struct Controller : public Client {
void run() { // Controller
try {
// Wait for subscribers to be ready.
- process(opts.totalSubs, fqn("sub_ready"), bind(expect, _1, "ready"));
+ process(opts.totalSubs, fqn("sub_ready"), boost::bind(expect, _1, "ready"));
LocalQueue pubDone;
LocalQueue subDone;
@@ -413,7 +413,7 @@ struct Controller : public Client {
AbsTime start=now();
send(opts.totalPubs, fqn("pub_start"), "start"); // Start publishers
if (j) {
- send(opts.totalPubs, fqn("sub_iteration"), "next"); // Start subscribers on next iteration
+ send(opts.totalSubs, fqn("sub_iteration"), "next"); // Start subscribers on next iteration
}
Stats pubRates;
@@ -423,8 +423,10 @@ struct Controller : public Client {
process(opts.totalSubs, subDone, fqn("sub_done"), boost::ref(subRates));
AbsTime end=now();
-
double time=secs(start, end);
+ if (time <= 0.0) {
+ throw Exception("ERROR: Test completed in zero seconds. Try again with a larger message count.");
+ }
double txrate=opts.transfers/time;
double mbytes=(txrate*opts.size)/(1024*1024);
@@ -508,10 +510,11 @@ struct PublishThread : public Client {
}
SubscriptionManager subs(session);
LocalQueue lq;
- subs.setFlowControl(1, SubscriptionManager::UNLIMITED, true);
- subs.subscribe(lq, fqn("pub_start"));
+ subs.setFlowControl(0, SubscriptionManager::UNLIMITED, false);
+ Subscription cs = subs.subscribe(lq, fqn("pub_start"));
for (size_t j = 0; j < opts.iterations; ++j) {
+ cs.grantMessageCredit(1);
expect(lq.pop().getData(), "start");
AbsTime start=now();
for (size_t i=0; i<opts.count; i++) {
@@ -543,6 +546,9 @@ struct PublishThread : public Client {
if (opts.confirm) session.sync();
AbsTime end=now();
double time=secs(start,end);
+ if (time <= 0.0) {
+ throw Exception("ERROR: Test completed in zero seconds. Try again with a larger message count.");
+ }
// Send result to controller.
Message report(lexical_cast<string>(opts.count/time), fqn("pub_done"));
@@ -638,7 +644,9 @@ struct SubscribeThread : public Client {
//
// For now verify order only for a single publisher.
size_t offset = opts.uniqueData ? 5 /*marker is 'data:'*/ : 0;
- size_t n = *reinterpret_cast<const size_t*>(msg.getData().data() + offset);
+ size_t n;
+ memcpy (&n, reinterpret_cast<const char*>(msg.getData().data() + offset),
+ sizeof(n));
if (opts.pubs == 1) {
if (opts.subs == 1 || opts.mode == FANOUT) verify(n==expect, "==", expect, n);
else verify(n>=expect, ">=", expect, n);