Merged out from trunkQPID-2519

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-2519@1187375 13f79535-47bb-0310-9956-ffa450edef68

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);