| #!/usr/bin/env python |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| |
| import optparse, time, qpid.messaging, re |
| from threading import Thread |
| from subprocess import Popen, PIPE, STDOUT |
| |
| op = optparse.OptionParser(usage="usage: %prog [options]", |
| description="simple performance benchmarks") |
| op.add_option("-b", "--broker", default=[], action="append", type="str", |
| help="url of broker(s) to connect to, round robin on multiple brokers") |
| op.add_option("-c", "--client-host", default=[], action="append", type="str", |
| help="host(s) to run clients on via ssh, round robin on mulple hosts") |
| op.add_option("-q", "--queues", default=1, type="int", metavar="N", |
| help="create N queues (default %default)") |
| op.add_option("-s", "--senders", default=1, type="int", metavar="N", |
| help="start N senders per queue (default %default)") |
| op.add_option("-r", "--receivers", default=1, type="int", metavar="N", |
| help="start N receivers per queue (default %default)") |
| op.add_option("-m", "--messages", default=100000, type="int", metavar="N", |
| help="send N messages per sender (default %default)") |
| op.add_option("--queue-name", default="benchmark", metavar="NAME", |
| help="base name for queues (default %default)") |
| op.add_option("--send-rate", default=0, metavar="N", |
| help="send rate limited to N messages/second, 0 means no limit (default %default)") |
| op.add_option("--receive-rate", default=0, metavar="N", |
| help="receive rate limited to N messages/second, 0 means no limit (default %default)") |
| op.add_option("--content-size", default=1024, type="int", metavar="BYTES", |
| help="message size in bytes (default %default)") |
| op.add_option("--ack-frequency", default=100, metavar="N", type="int", |
| 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") |
| op.add_option("--receive-option", default=[], action="append", type="str", |
| help="Additional option for receiving addresses") |
| op.add_option("--create-option", default=[], action="append", type="str", |
| help="Additional option for creating addresses") |
| op.add_option("--send-arg", default=[], action="append", type="str", |
| help="Additional argument for qpid-send") |
| op.add_option("--receive-arg", default=[], action="append", type="str", |
| help="Additional argument for qpid-receive") |
| op.add_option("--no-timestamp", dest="timestamp", default=True, |
| action="store_false", help="don't add a timestamp, no latency results") |
| op.add_option("--sequence", dest="sequence", default=False, |
| action="store_true", help="add a sequence number to each message") |
| op.add_option("--connection-options", type="str", |
| help="Connection options for senders & receivers") |
| 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("--verbose", default=False, action="store_true", |
| help="Show commands executed") |
| op.add_option("--no-delete", default=False, action="store_true", |
| help="Don't delete the test queues.") |
| op.add_option("--fill-drain", default=False, action="store_true", |
| help="First fill the queues, then drain them") |
| |
| single_quote_re = re.compile("'") |
| def posix_quote(string): |
| """ Quote a string for use as an argument in a posix shell""" |
| return "'" + single_quote_re.sub("\\'", string) + "'"; |
| |
| def ssh_command(host, command): |
| """ Convert command into an ssh command on host with quoting""" |
| return ["ssh", host] + [posix_quote(arg) for arg in command] |
| |
| class Clients: |
| def __init__(self): self.clients=[] |
| |
| def add(self, client): |
| self.clients.append(client) |
| return client |
| |
| def kill(self): |
| for c in self.clients: |
| try: c.kill() |
| except: pass |
| |
| class PopenCommand(Popen): |
| """Like Popen but you can query for the command""" |
| def __init__(self, command, *args, **kwargs): |
| self.command = command |
| Popen.__init__(self, command, *args, **kwargs) |
| |
| clients = Clients() |
| |
| def start_receive(queue, index, opts, ready_queue, broker, host): |
| address_opts=opts.receive_option |
| if opts.durable: address_opts += ["node:{durable:true}"] |
| address="%s;{%s}"%(queue,",".join(address_opts)) |
| msg_total=opts.senders*opts.messages |
| messages = msg_total/opts.receivers; |
| if (index < msg_total%opts.receivers): messages += 1 |
| if (messages == 0): return None |
| command = ["qpid-jms-receive", |
| #"-b", broker, |
| "--ready-address", "benchmark-ready;{create:always}", |
| "-a", address, |
| "-m", str(messages), |
| "--forever", |
| "--print-content=no", |
| # "--receive-rate", str(opts.receive_rate), |
| "--report-total", |
| "--ack-frequency", str(opts.ack_frequency), |
| # "--ready-address", "%s;{create:always}"%ready_queue, |
| "--report-header=no -v" |
| ] |
| if opts.save_received: |
| command += ["--save-content=%s-receiver-%s.msg"%(queue,index)] |
| command += opts.receive_arg |
| 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(PopenCommand(command, stdout=PIPE, stderr=PIPE)) |
| |
| def start_send(queue, opts, broker, host): |
| address="%s;{%s}"%(queue,",".join(opts.send_option + ["create:always"])) |
| command = ["qpid-jms-send", |
| #"-b", broker, |
| "-a", address, |
| "--messages", str(opts.messages), |
| "--content-size", str(opts.content_size), |
| "--send-rate", str(opts.send_rate), |
| "--report-total", |
| "--report-header=no", |
| "--timestamp=%s"%(opts.timestamp and "yes" or "no"), |
| "--sequence=%s"%(opts.sequence and "yes" or "no"), |
| "--durable", str(opts.durable) |
| ] |
| command += opts.send_arg |
| 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(PopenCommand(command, stdout=PIPE, stderr=PIPE)) |
| |
| def error_msg(out, err): |
| return ("\n[stdout]\n%s\n[stderr]\n%s[end]"%(out, err)) |
| |
| def first_line(p): |
| out,err=p.communicate() |
| if p.returncode != 0: |
| raise Exception("Process exit %d: %s"%(p.returncode, error_msg(out,err))) |
| |
| print str(out) |
| print str(err) |
| return out.split("\n")[0] |
| |
| def recreate_queues(queues, brokers, no_delete, opts): |
| c = qpid.messaging.Connection(brokers[0]) |
| c.open() |
| s = c.session() |
| for q in queues: |
| if not no_delete: |
| try: s.sender("%s;{delete:always}"%(q)).close() |
| except qpid.messaging.exceptions.NotFound: pass |
| address = "%s;{%s}"%(q, ",".join(opts.create_option + ["create:always"])) |
| if opts.verbose: print "Creating", address |
| s.sender(address) |
| c.close() |
| |
| def print_header(timestamp): |
| if timestamp: latency_header="\tl-min\tl-max\tl-avg\ttotal-tp" |
| else: latency_header="" |
| print "send-tp\trecv-tp%s"%latency_header |
| |
| def parse(parser, lines): # Parse sender/receiver output |
| 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 if p]) |
| |
| def print_data(send_stats, recv_stats, total_tp): |
| for send,recv in map(None, send_stats, recv_stats): |
| line="" |
| if send: line += "%d"%send[0] |
| if recv: |
| line += "\t%d"%recv[0] |
| if len(recv) == 4: line += "\t%.2f\t%.2f\t%.2f"%tuple(recv[1:]) |
| if total_tp is not None: |
| line += "\t%d"%total_tp |
| total_tp = None |
| print line |
| |
| def print_summary(send_stats, recv_stats, total_tp): |
| 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%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) |
| summary += "\t%d"%total_tp |
| print summary |
| |
| |
| class ReadyReceiver: |
| """A receiver for ready messages""" |
| def __init__(self, queue, broker): |
| self.connection = qpid.messaging.Connection(broker) |
| self.connection.open() |
| self.receiver = self.connection.session().receiver( |
| "%s;{create:receiver,delete:receiver,node:{durable:false}}"%(queue)) |
| self.receiver.session.sync() |
| self.timeout=10 |
| |
| def wait(self, receivers): |
| try: |
| for i in receivers: self.receiver.fetch(self.timeout) |
| self.connection.close() |
| except qpid.messaging.Empty: |
| for r in receivers: |
| if (r.poll() is not None): |
| out,err=r.communicate() |
| raise Exception("Receiver error: %s\n%s" % |
| (" ".join(r.command), error_msg(out,err))) |
| raise Exception("Timed out waiting for receivers to be ready") |
| |
| def flatten(l): |
| return sum(map(lambda s: re.split(re.compile("\s*,\s*|\s+"), s), l), []) |
| |
| class RoundRobin: |
| def __init__(self,items): |
| self.items = items |
| self.index = 0 |
| |
| def next(self): |
| if not self.items: return None |
| ret = self.items[self.index] |
| self.index = (self.index+1)%len(self.items) |
| return ret |
| |
| def main(): |
| opts, args = op.parse_args() |
| opts.client_host = flatten(opts.client_host) |
| if not opts.broker: |
| if opts.client_host: |
| raise Exception("--broker must be specified if --client_host is.") |
| opts.broker = ["127.0.0.1"] # Deafult to local broker |
| opts.broker = flatten(opts.broker) |
| brokers = RoundRobin(opts.broker) |
| client_hosts = RoundRobin(opts.client_host) |
| send_out = "" |
| receive_out = "" |
| ready_queue="%s-ready"%(opts.queue_name) |
| queues = ["%s-%s"%(opts.queue_name, i) for i in xrange(opts.queues)] |
| try: |
| for i in xrange(opts.repeat): |
| recreate_queues(queues, opts.broker, opts.no_delete, opts) |
| ready_receiver = ReadyReceiver(ready_queue, opts.broker[0]) |
| |
| def start_receivers(): |
| return [ start_receive(q, j, opts, ready_queue, brokers.next(), client_hosts.next()) |
| for q in queues for j in xrange(opts.receivers) ] |
| |
| |
| def start_senders(): |
| return [ start_send(q, opts,brokers.next(), client_hosts.next()) |
| for q in queues for j in xrange(opts.senders) ] |
| |
| if opts.report_header and i == 0: print_header(opts.timestamp) |
| |
| if opts.fill_drain: |
| # First fill the queues, then drain them |
| start = time.time() |
| senders = start_senders() |
| for p in senders: p.wait() |
| receivers = start_receivers() |
| for p in receivers: p.wait() |
| else: |
| # Run senders and receivers in parallel |
| receivers = start_receivers() |
| ready_receiver.wait(filter(None, receivers)) # Wait for receivers ready |
| start = time.time() |
| senders = start_senders() |
| for p in senders + receivers: p.wait() |
| |
| total_sent = opts.queues * opts.senders * opts.messages |
| total_tp = total_sent / (time.time()-start) |
| #send_stats=parse_senders(senders) |
| recv_stats=parse_receivers(receivers) |
| #if opts.summarize: print_summary(send_stats, recv_stats, total_tp) |
| #else: print_data(send_stats, recv_stats, total_tp) |
| finally: clients.kill() # No strays |
| |
| if __name__ == "__main__": main() |
| |
