QPID-2920: Improvements to cluster benchmarks.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid-2920-active@1182605 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 24 insertions, 8 deletions

diff --git a/qpid/cpp/src/tests/qpid-cluster-benchmark b/qpid/cpp/src/tests/qpid-cluster-benchmark
index da6a835a8b..04fa82287b 100755
--- a/qpid/cpp/src/tests/qpid-cluster-benchmark
+++ b/qpid/cpp/src/tests/qpid-cluster-benchmark
@@ -30,7 +30,7 @@ RECEIVERS="-r 3"
 BROKERS=			# Local broker
 CLIENT_HOSTS=			# No ssh, all clients are local
 
-while getopts "m:f:n:b:q:s:r:c:x" opt; do
+while getopts "m:f:n:b:q:s:r:c:x:t" opt; do
     case $opt in
 	m) MESSAGES="-m $OPTARG";;
 	f) FLOW="--flow-control $OPTARG";;
@@ -41,13 +41,14 @@ while getopts "m:f:n:b:q:s:r:c:x" opt; do
 	r) RECEIVERS="-r $OPTARG";;
 	c) CLIENT_HOSTS="-c $OPTARG";;
 	x) SAVE_RECEIVED="--save-received";;
+	t) TCP_NODELAY="--connection-options {tcp-nodelay:true}";;
 	*) echo "Unknown option"; exit 1;;
     esac
 done
 BROKER=$(echo $BROKERS | sed s/,.*//)
 run_test() { echo $*; shift; "$@"; echo; echo; echo; }
 
-run_test "Multiple active brokers:" qpid-cpp-benchmark $REPEAT $BROKERS --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS $SAVE_RECEIVED
-run_test "Single active broker :" qpid-cpp-benchmark $REPEAT $BROKER --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS $SAVE_RECEIVED
-run_test "Latency under low load:" qpid-cpp-benchmark $REPEAT $BROKERS --connection-options "{tcp-nodelay:true}" $MESSAGES $FLOW $CLIENT_HOSTS
+OPTS="$REPEAT $BROKERS --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS $SAVE_RECEIVED $TCP_NODELAY"
+run_test "Queue contention:" qpid-cpp-benchmark $OPTS
+run_test "No queue contention: :" qpid-cpp-benchmark $OPTS --group-receivers
 
diff --git a/qpid/cpp/src/tests/qpid-cpp-benchmark b/qpid/cpp/src/tests/qpid-cpp-benchmark
index c951a02498..5292943e5a 100755
--- a/qpid/cpp/src/tests/qpid-cpp-benchmark
+++ b/qpid/cpp/src/tests/qpid-cpp-benchmark
@@ -69,7 +69,10 @@ op.add_option("--durable", default=False, action="store_true",
               help="Use durable queues and messages")
 op.add_option("--save-received", default=False, action="store_true",
               help="Save received message content to files <queuename>-receiver-<n>.msg")
-
+op.add_option("--group-receivers", default=False, action="store_true",
+              help="Run receivers for the same queue on the same host.")
+op.add_option("--verbose", default=False, action="store_true",
+              help="Show commands executed")
 single_quote_re = re.compile("'")
 def posix_quote(string):
     """ Quote a string for use as an argument in a posix shell"""
@@ -119,6 +122,7 @@ def start_receive(queue, index, opts, ready_queue, broker, host):
     if opts.connection_options:
         command += ["--connection-options",opts.connection_options]
     if host: command = ssh_command(host, command)
+    if opts.verbose: print "Receiver: ", command
     return clients.add(Popen(command, stdout=PIPE, stderr=PIPE))
 
 def start_send(queue, opts, broker, host):
@@ -140,6 +144,7 @@ def start_send(queue, opts, broker, host):
     if opts.connection_options:
         command += ["--connection-options",opts.connection_options]
     if host: command = ssh_command(host, command)
+    if opts.verbose: print "Sender: ", command
     return clients.add(Popen(command, stdout=PIPE, stderr=PIPE))
 
 def error_msg(out, err):
@@ -273,9 +278,19 @@ def main():
         for i in xrange(opts.repeat):
             recreate_queues(queues, opts.broker)
             ready_receiver = ReadyReceiver(ready_queue, opts.broker[0])
-            receivers = [start_receive(q, j, opts, ready_queue,
-                                       brokers.next(), client_hosts.next())
-                         for q in queues for j in xrange(opts.receivers)]
+
+            if opts.group_receivers: # Run receivers for same queue against same broker.
+                receivers = []
+                for q in queues:
+                    b = brokers.next()
+                    for j in xrange(opts.receivers):
+                        receivers.append(
+                            start_receive(q, j, opts, ready_queue, b, client_hosts.next()))
+            else:                       # Don't group receivers
+                receivers = [start_receive(q, j, opts, ready_queue,
+                                           brokers.next(), client_hosts.next())
+                             for q in queues for j in xrange(opts.receivers)]
+
             ready_receiver.wait(filter(None, receivers)) # Wait for receivers to be ready.
             start = time.time()
             senders = [start_send(q, opts,brokers.next(), client_hosts.next())