From 1b5c9ca18694061838f5e57cf7b8ca836cb35a46 Mon Sep 17 00:00:00 2001
From: Alan Conway <aconway@apache.org>
Date: Mon, 28 Mar 2011 23:55:50 +0000
Subject: QPID-2920: minor improvements to qpid-cpp-benchmark

Re-arranged queue creation/deletion to avoid wiring races.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid-2920@1086448 13f79535-47bb-0310-9956-ffa450edef68
---
 qpid/cpp/src/tests/qpid-cluster-benchmark |  3 +--
 qpid/cpp/src/tests/qpid-cpp-benchmark     | 29 ++++++++++++++++++++---------
 2 files changed, 21 insertions(+), 11 deletions(-)

diff --git a/qpid/cpp/src/tests/qpid-cluster-benchmark b/qpid/cpp/src/tests/qpid-cluster-benchmark
index 5f0e020475..9d735c20e3 100755
--- a/qpid/cpp/src/tests/qpid-cluster-benchmark
+++ b/qpid/cpp/src/tests/qpid-cluster-benchmark
@@ -23,6 +23,7 @@
 # Default values
 PORT="5672"
 BROKERS=`echo $HOSTS | sed "s/\>/:$PORT/g;s/ /,/g"` # Broker URL list
+BROKER=`echo $HOSTS | awk '{print $1}'`	# First broker
 COUNT=10000
 FLOW=100	      # Flow control limit on queue depth for latency.
 REPEAT=10
@@ -40,8 +41,6 @@ while getopts "p:c:f:r:t:b:" opt; do
     esac
 done
 
-BROKER=`echo $HOSTS | sed 's/,.*//'` # First broker
-
 run_test() { echo $*; shift; "$@"; echo; echo; echo; }
 
 # Multiple pubs/subs connect via multiple brokers (active-active)
diff --git a/qpid/cpp/src/tests/qpid-cpp-benchmark b/qpid/cpp/src/tests/qpid-cpp-benchmark
index 98794082eb..6a1c187214 100755
--- a/qpid/cpp/src/tests/qpid-cpp-benchmark
+++ b/qpid/cpp/src/tests/qpid-cpp-benchmark
@@ -92,7 +92,7 @@ class Clients:
 clients = Clients()
 
 def start_receive(queue, index, opts, ready_queue, broker, host):
-    address_opts=["create:receiver"] + opts.receive_option
+    address_opts=["create:always"] + opts.receive_option
     if opts.durable: address_opts += ["node:{durable:true}"]
     address="%s;{%s}"%(queue,",".join(address_opts))
     msg_total=opts.senders*opts.messages
@@ -108,7 +108,7 @@ def start_receive(queue, index, opts, ready_queue, broker, host):
                "--receive-rate", str(opts.receive_rate),
                "--report-total",
                "--ack-frequency", str(opts.ack_frequency),
-               "--ready-address", ready_queue,
+               "--ready-address", "%s;{create:always}"%ready_queue,
                "--report-header=no"
                ]
     command += opts.receive_arg
@@ -118,7 +118,7 @@ def start_receive(queue, index, opts, ready_queue, broker, host):
     return clients.add(Popen(command, stdout=PIPE))
 
 def start_send(queue, opts, broker, host):
-    address="%s;{%s}"%(queue,",".join(opts.send_option + ["create:receiver"]))
+    address="%s;{%s}"%(queue,",".join(opts.send_option + ["create:always"]))
     command = ["qpid-send",
                "-b", broker,
                "-a", address,
@@ -138,23 +138,33 @@ def start_send(queue, opts, broker, host):
     if host: command = ssh_command(host, command)
     return clients.add(Popen(command, stdout=PIPE))
 
+def error_msg(out, err): return ("\n".join(filter(None, [out, err]))).strip()
+
 def first_line(p):
     out,err=p.communicate()
-    if p.returncode != 0: raise Exception("Process failed: %s"%(out.strip()))
+    if p.returncode != 0: raise Exception("Process failed: %s"%error_msg(out,err))
     return out.split("\n")[0]
 
 def delete_queues(queues, broker):
     c = qpid.messaging.Connection(broker)
     c.open()
+    s = c.session()
     for q in queues:
         try:
-            s = c.session()
             snd = s.sender("%s;{delete:always}"%(q))
             snd.close()
-            s.sync()
         except qpid.messaging.exceptions.NotFound: pass # Ignore "no such queue"
+    s.sync()
     c.close()
 
+def create_queues(queues, broker):
+    c = qpid.messaging.Connection(broker)
+    c.open()
+    s = c.session()
+    for q in queues:
+        s.sender("%s;{create:always}"%q)
+    s.sync()
+
 def print_header(timestamp):
     if timestamp: latency_header="\tl-min\tl-max\tl-avg"
     else: latency_header=""
@@ -196,7 +206,6 @@ def print_summary(send_stats, recv_stats):
 class ReadyReceiver:
     """A receiver for ready messages"""
     def __init__(self, queue, broker):
-        delete_queues([queue], broker)
         self.connection = qpid.messaging.Connection(broker)
         self.connection.open()
         self.receiver = self.connection.session().receiver(
@@ -212,7 +221,7 @@ class ReadyReceiver:
             for r in receivers:
                 if (r.poll() is not None):
                     out,err=r.communicate()
-                    raise Exception("Receiver error: %s"%(out))
+                    raise Exception("Receiver error: %s"%error_msg(out,err))
             raise Exception("Timed out waiting for receivers to be ready")
 
 def flatten(l): return sum(map(lambda s: s.split(","), l),[])
@@ -242,8 +251,10 @@ def main():
     try:
         for i in xrange(opts.repeat):
             delete_queues(queues, opts.broker[0])
+            time.sleep(.1) # FIXME aconway 2011-03-18: new cluster async wiring
+            create_queues(queues, opts.broker[0])
+            time.sleep(.1) # FIXME aconway 2011-03-18: new cluster async wiring
             ready_receiver = ReadyReceiver(ready_queue, opts.broker[0])
-            time.sleep(.1) # FIXME aconway 2011-03-16: new cluster async wiring
             receivers = [start_receive(q, j, opts, ready_queue,
                                        brokers.next(), client_hosts.next())
                          for q in queues for j in xrange(opts.receivers)]
-- 
cgit v1.2.1