You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2010/10/04 22:44:41 UTC
svn commit: r1004420 - in /qpid/trunk/qpid/cpp/src/tests: cluster.mk
qpid-cluster-benchmark qpid-cpp-benchmark
Author: aconway
Date: Mon Oct 4 20:44:40 2010
New Revision: 1004420
URL: http://svn.apache.org/viewvc?rev=1004420&view=rev
Log:
Cluster benchmark test script. Improvements to qpid-cpp-bencmhark.
Added:
qpid/trunk/qpid/cpp/src/tests/qpid-cluster-benchmark (with props)
Modified:
qpid/trunk/qpid/cpp/src/tests/cluster.mk
qpid/trunk/qpid/cpp/src/tests/qpid-cpp-benchmark
Modified: qpid/trunk/qpid/cpp/src/tests/cluster.mk
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster.mk?rev=1004420&r1=1004419&r2=1004420&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster.mk (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster.mk Mon Oct 4 20:44:40 2010
@@ -85,7 +85,8 @@ qpidtest_SCRIPTS += run_cluster_tests cl
CLUSTER_TEST_SCRIPTS_LIST= \
allhosts rsynchosts \
qpid-build-rinstall qpid-src-rinstall \
- qpid-test-cluster
+ qpid-test-cluster \
+ qpid-cluster-benchmark
qpidtest_SCRIPTS += $(CLUSTER_TEST_SCRIPTS_LIST)
EXTRA_DIST += $(CLUSTER_TEST_SCRIPTS_LIST)
Added: qpid/trunk/qpid/cpp/src/tests/qpid-cluster-benchmark
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/qpid-cluster-benchmark?rev=1004420&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/qpid-cluster-benchmark (added)
+++ qpid/trunk/qpid/cpp/src/tests/qpid-cluster-benchmark Mon Oct 4 20:44:40 2010
@@ -0,0 +1,20 @@
+#!/bin/sh
+# Benchmark script for comparing cluster performance.
+#PORT=":5555"
+BROKER=`echo $HOSTS | awk '{print $1}'` # Single broker
+BROKERS=`echo $HOSTS | sed "s/\>/$PORT/g;s/ /,/g"` # Broker URL list
+COUNT=100000
+RATE=20000 # Rate to throttle senders for latency results
+run_test() { echo $*; "$@"; echo; echo; echo; }
+
+# Thruput, unshared queue
+run_test qpid-cpp-benchmark --repeat 10 -b $BROKER --no-timestamp -m $COUNT
+
+# Latency
+run_test qpid-cpp-benchmark --repeat 10 -b $BROKER --connection-options '{tcp-nodelay:true}' -m `expr $COUNT / 2` --send-rate $RATE
+
+# Multiple pubs/subs connect via multiple brokers (active-active)
+run_test qpid-cpp-benchmark --repeat 10 -b $BROKERS --no-timestamp --summarize -s10 -r10 -m `expr $COUNT / 10`
+
+# Multiple pubs/subs connect via single broker (active-passive)
+run_test qpid-cpp-benchmark --repeat 10 -b $BROKER --no-timestamp --summarize -s10 -r10 -m `expr $COUNT / 10`
Propchange: qpid/trunk/qpid/cpp/src/tests/qpid-cluster-benchmark
------------------------------------------------------------------------------
svn:executable = *
Modified: qpid/trunk/qpid/cpp/src/tests/qpid-cpp-benchmark
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/qpid-cpp-benchmark?rev=1004420&r1=1004419&r2=1004420&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/qpid-cpp-benchmark (original)
+++ qpid/trunk/qpid/cpp/src/tests/qpid-cpp-benchmark Mon Oct 4 20:44:40 2010
@@ -46,6 +46,8 @@ op.add_option("--ack-frequency", default
help="receiver ack's every N messages, 0 means unconfirmed (default %default)")
op.add_option("--no-report-header", dest="report_header", default=True,
action="store_false", help="don't print header on report")
+op.add_option("--summarize", default=False, action="store_true",
+ help="print summary statistics for multiple senders/receivers: total throughput, average latency")
op.add_option("--repeat", default=1, metavar="N", help="repeat N times", type="int")
op.add_option("--send-option", default=[], action="append", type="str",
help="Additional option for sending addresses")
@@ -100,10 +102,10 @@ def start_send(queue, opts, broker, host
if host: command = ssh_command(host, command)
return Popen(command, stdout=PIPE, stderr=STDOUT)
-def wait_for_output(p):
+def first_line(p):
out,err=p.communicate()
if p.returncode != 0: raise Exception("ERROR:\n%s"%(out))
- return out
+ return out.split("\n")[0]
def delete_queues(queues, broker):
c = qpid.messaging.Connection(broker)
@@ -113,16 +115,43 @@ def delete_queues(queues, broker):
except qpid.messaging.exceptions.NotFound: pass # Ignore "no such queue"
c.close()
-def print_output(senders, receivers, want_header):
- send_stats = sum([wait_for_output(p).split("\n")[:-1] for p in senders],[])
- recv_stats = sum([wait_for_output(p).split("\n")[:-1] for p in receivers],[])
- def empty_if_none(s):
- if s: return s
- else: return ""
- stats = map(lambda s,r: empty_if_none(s)+"\t\t"+empty_if_none(r),
- send_stats, recv_stats)
- if want_header: print "send-tp\t\trecv-tp\tl-min\tl-max\tl-avg"
- for s in stats: print s;
+def print_header(timestamp):
+ if timestamp: latency_header="\tl-min\tl-max\tl-avg"
+ else: latency_header=""
+ print "send-tp\t\trecv-tp%s"%latency_header
+
+def parse(parser, lines): # Parse sender/receiver output
+ for l in lines:
+ fn_val = zip(parser, l)
+
+ return [map(lambda p: p[0](p[1]), zip(parser,line.split())) for line in lines]
+
+def parse_senders(senders):
+ return parse([int],[first_line(p) for p in senders])
+
+def parse_receivers(receivers):
+ return parse([int,float,float,float],[first_line(p) for p in receivers])
+
+def print_data(send_stats, recv_stats):
+ for send,recv in map(None, send_stats, recv_stats):
+ if send: print send[0],
+ if recv:
+ print "\t\t%d"%recv[0],
+ if len(recv) == 4: print "\t%.2f\t%.2f\t%.2f"%tuple(recv[1:]),
+ print
+
+def print_summary(send_stats, recv_stats):
+ def avg(s): sum(s) / len(s)
+ send_tp = sum([l[0] for l in send_stats])
+ recv_tp = sum([l[0] for l in recv_stats])
+ summary = "%d\t\t%d"%(send_tp, recv_tp)
+ if recv_stats and len(recv_stats[0]) == 4:
+ l_min = sum(l[1] for l in recv_stats)/len(recv_stats)
+ l_max = sum(l[2] for l in recv_stats)/len(recv_stats)
+ l_avg = sum(l[3] for l in recv_stats)/len(recv_stats)
+ summary += "\t%.2f\t%.2f\t%.2f"%(l_min, l_max, l_avg)
+ print summary
+
class ReadyReceiver:
"""A receiver for ready messages"""
@@ -176,7 +205,11 @@ def main():
ready_receiver.wait(receivers) # Wait for receivers to be ready.
senders = [start_send(q, opts,brokers.next(), client_hosts.next())
for q in queues for j in xrange(opts.senders)]
- print_output(senders, receivers, opts.report_header and i == 0)
+ if opts.report_header and i == 0: print_header(opts.timestamp)
+ send_stats=parse_senders(senders)
+ recv_stats=parse_receivers(receivers)
+ if opts.summarize: print_summary(send_stats, recv_stats)
+ else: print_data(send_stats, recv_stats)
delete_queues(queues, opts.broker[0])
if __name__ == "__main__": main()
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org