From 976252c508b4feebef3b0811f640cd9bf0cfeeaa Mon Sep 17 00:00:00 2001
From: Alan Conway <aconway@apache.org>
Date: Thu, 8 Nov 2007 21:16:02 +0000
Subject: src/tests/perfdist: Use ssh to run distributed perftest.   ./perfdist
 for usage.

src/tests/perftest.cpp:
 --summary gives one-line report, can cut/paste into spreadsheet for multiple runs.
 --purge as initial clean-up step for distributed tests (see perfdist)


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@593321 13f79535-47bb-0310-9956-ffa450edef68
---
 cpp/src/tests/perfdist     | 33 ++++++++++++++++
 cpp/src/tests/perftest.cpp | 94 ++++++++++++++++++++++++++++------------------
 2 files changed, 91 insertions(+), 36 deletions(-)
 create mode 100755 cpp/src/tests/perfdist

(limited to 'cpp/src')

diff --git a/cpp/src/tests/perfdist b/cpp/src/tests/perfdist
new file mode 100755
index 0000000000..07338f6822
--- /dev/null
+++ b/cpp/src/tests/perfdist
@@ -0,0 +1,33 @@
+#!/bin/bash
+#
+echo $_
+usage() {
+cat <<EOF
+Distributed perftest with ssh.
+Arguments before -- are passed to perftest.
+Arguments after -- are treated as host names.
+Publisher runs on first host. One listener runs on each 
+Remaining host. To run multiple liseners on a host, list it more than once.
+Do not pass --consumers, --publish or --listen options, they are computed
+from the host list.
+EOF
+exit 1
+}
+
+while test "$HOSTS" != "$*"; do
+    case $1 in
+	--consumers|--publish|--listen) usage ;;
+	--) shift; PUB="$1" ; shift ; HOSTS="$*" ;;
+	*) ARGS="$ARGS $1"; shift ;;
+    esac
+done
+test -n "$HOSTS" || { echo "No -- found"; usage; }
+N=`echo $HOSTS | wc -w`
+PERFTEST=`PATH=$PWD:$PATH which perftest`
+$PERFTEST --purge $ARGS
+ssh $PUB $PERFTEST $ARGS --publish --consumers $N&
+for h in $HOSTS; do
+    ssh $h $PERFTEST $ARGS --listen&
+done
+wait
+
diff --git a/cpp/src/tests/perftest.cpp b/cpp/src/tests/perftest.cpp
index 4db90fd7be..8d970a0a6f 100644
--- a/cpp/src/tests/perftest.cpp
+++ b/cpp/src/tests/perftest.cpp
@@ -39,26 +39,32 @@ struct Opts : public TestOptions {
 
     bool listen;
     bool publish;
+    bool purge;
     int count;
     int size;
     bool durable;
     int consumers;
     std::string mode;
     int autoAck;
+    bool summary;
     
     Opts() :
-        listen(false), publish(false), count(500000), size(64), consumers(1),
-        mode("shared"), autoAck(100)
+        listen(false), publish(false), purge(false),
+        count(500000), size(64), consumers(1),
+        mode("shared"), autoAck(100),
+        summary(false)
     {
         addOptions() 
             ("listen", optValue(listen), "Consume messages.")
             ("publish", optValue(publish), "Produce messages.")
+            ("purge", optValue(purge), "Purge shared queues.")
             ("count", optValue(count, "N"), "Messages to send.")
             ("size", optValue(size, "BYTES"), "Size of messages.")
             ("durable", optValue(durable, "N"), "Publish messages as durable.")
             ("consumers", optValue(consumers, "N"), "Number of consumers.")
             ("mode", optValue(mode, "shared|fanout|topic"), "consume mode")
-            ("auto-ack", optValue(autoAck, "N"), "ack every N messages.");
+            ("auto-ack", optValue(autoAck, "N"), "ack every N messages.")
+            ("summary,s", optValue(summary), "summary output only");
     }
 };
 
@@ -71,16 +77,19 @@ struct PublishThread : public Runnable { Thread thread; void run(); };
 
 // Create and purge the shared queues 
 void setup() {
-    cout << "Create shared queues" << endl;
     Connection connection;
     opts.open(connection);
     Session_0_10 session = connection.newSession();
     session.setSynchronous(true); // Make sure this is all completed.
     session.queueDeclare(arg::queue="control"); // Control queue
-    session.queuePurge(arg::queue="control");
+    if (opts.purge) {
+        if (!opts.summary) cout << "Purging shared queues" << endl;
+        session.queuePurge(arg::queue="control");
+    }
     if (mode==SHARED) {
         session.queueDeclare(arg::queue="perftest", arg::durable=opts.durable); // Shared data queue
-        session.queuePurge(arg::queue="perftest");
+        if (opts.purge)		
+            session.queuePurge(arg::queue="perftest");
     }
     session.close();
     connection.close();
@@ -93,8 +102,8 @@ int main(int argc, char** argv) {
         else if (opts.mode=="fanout") mode = FANOUT;
         else if (opts.mode=="topic") mode = TOPIC;
         else throw Exception("Invalid mode");
-        if (!opts.listen && !opts.publish)
-            opts.listen = opts.publish = true;
+        if (!opts.listen && !opts.publish && !opts.purge)
+            opts.listen = opts.publish = opts.purge = true;
         setup();
         std::vector<ListenThread> listen(opts.consumers);
         PublishThread publish;
@@ -141,15 +150,15 @@ void PublishThread::run() {
         Session_0_10 session = connection.newSession();
 
         // Wait for consumers.
-        cout << "Publisher wating for consumers " << flush;
+        if (!opts.summary) cout << "Waiting for consumers ready " << flush;
         SubscriptionManager subs(session);
         LocalQueue control;
         subs.subscribe(control, "control");
         for (int i = 0; i < opts.consumers; ++i) {
-            cout << "." << flush;
+            if (!opts.summary) cout << "." << flush;
             expect(control.pop().getData(), "ready");
         }
-        cout << endl;
+        if (!opts.summary) cout << endl;
 
         // Create test message
         size_t msgSize=max(opts.size, 32);
@@ -159,14 +168,14 @@ void PublishThread::run() {
 	    msg.getDeliveryProperties().setDeliveryMode(framing::PERSISTENT);
         // Time sending message.
         AbsTime start=now();
-        cout << "Publishing " << opts.count << " messages " << flush;
+        if (!opts.summary) cout << "Publishing " << opts.count << " messages " << flush;
         for (int i=0; i<opts.count; i++) {
             sprintf(msgBuf, "%d", i);
             session.messageTransfer(arg::destination=exchange(),
                                     arg::content=msg);
-            if ((i%10000)==0) cout << "." << flush;
+            if (!opts.summary && (i%10000)==0) cout << "." << flush;
         }
-        cout << " done." << endl;
+        if (!opts.summary) cout << " done." << endl;
         msg.setData("done");    // Send done messages.
         if (mode==SHARED)
             for (int i = 0; i < opts.consumers; ++i)
@@ -176,18 +185,24 @@ void PublishThread::run() {
         AbsTime end=now();
 
         // Report
-        cout << endl;
-        cout << "publish count:" << opts.count << endl;
-        cout << "publish secs:" << secs(start,end) << endl;
-        cout << "publish rate:" << (opts.count)/secs(start,end) << endl;
+        double publish_rate=(opts.count)/secs(start,end);
+        if (!opts.summary)
+            cout << endl
+                 << "publish count:" << opts.count << endl
+                 << "publish secs:" << secs(start,end) << endl
+                 << "publish rate:" << publish_rate << endl;
 
+        double consume_rate = 0; // Average rate for consumers.
         //  Wait for consumer(s) to finish.
-        cout << "Publisher wating for consumer reports. " << endl;
+        if (!opts.summary) cout << "Waiting for consumers done " << endl;
         for (int i = 0; i < opts.consumers; ++i) {
             string report=control.pop().getData();
-            if (report.find("consume") != 0)
-                throw Exception("Expected consumer report, got: "+report);
-            cout << endl << report;
+            if (!opts.summary)
+                cout << endl << report;
+            else {
+                double rate=boost::lexical_cast<double>(report);
+                consume_rate += rate/opts.consumers;
+            }
         }
         end=now();
 
@@ -197,11 +212,19 @@ void PublishThread::run() {
             transfers=2*opts.count; 
         else                    // sent once, received N times.
             transfers=opts.count*(opts.consumers + 1);
-        
-        cout << endl
-             << "total transfers:" << transfers << endl
-             << "total secs:" << secs(start, end) << endl
-             << "total transfers/sec:" << transfers/secs(start, end) << endl;
+        double total_rate=transfers/secs(start, end);
+        if (opts.summary)
+            cout << opts.mode << '(' << opts.count
+                 << ':' << opts.consumers << ')'
+                 << '\t' << publish_rate
+                 << '\t' << consume_rate
+                 << '\t' << total_rate
+                 << endl;
+        else
+            cout << endl
+                 << "total transfers:" << transfers << endl
+                 << "total secs:" << secs(start, end) << endl
+                 << "total rate:" << total_rate << endl;
 		
         connection.close();
     }
@@ -238,23 +261,22 @@ void ListenThread::run() {
         int consumed=0;
         AbsTime start=now();
         Message msg;
-        if (!opts.publish)
-            cout << "Consuming " << flush;
         while ((msg=consume.pop()).getData() != "done") {
             ++consumed;
-            if (!opts.publish && (consumed%10000) == 0)
-                cout << "." << flush;
         }
-        if (!opts.publish)
-            cout << endl;
         msg.acknowledge();      // Ack all outstanding messages.
         AbsTime end=now();
 
         // Report to publisher.
         ostringstream report;
-        report << "consume count: " << consumed << endl
-               << "consume secs: " << secs(start, end) << endl
-               << "consume rate: " << consumed/secs(start,end) << endl;
+        double consume_rate=consumed/secs(start,end);
+        if (opts.summary)
+            report << consume_rate;
+        else
+            report << "consume count: " << consumed << endl
+                   << "consume secs: " << secs(start, end) << endl
+                   << "consume rate: " << consume_rate << endl;
+        
         session.messageTransfer(arg::content=Message(report.str(), "control"));
         connection.close();
     }
-- 
cgit v1.2.1