From 5cd2a04d1db0dc95ae0492770e25273bb16d4565 Mon Sep 17 00:00:00 2001 From: Patrick Wyatt Date: Sat, 2 Mar 2013 21:05:56 +0000 Subject: [PATCH 1/4] Make it work on CentOS --- bench.py | 26 +++++++++++++++++++++----- configure-CentOS.sh | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 5 deletions(-) create mode 100755 configure-CentOS.sh diff --git a/bench.py b/bench.py index 14de0ef..38aedba 100755 --- a/bench.py +++ b/bench.py @@ -3,8 +3,24 @@ from os import mkdir from os.path import join from multiprocessing import cpu_count -from subprocess import Popen, check_output, PIPE from sys import stdout +import subprocess + +# Add check_output; CentOS 6.3 uses Python 2.6, which doesn't have this +if "check_output" not in dir( subprocess ): # duck punch it in! + def f(*popenargs, **kwargs): + if 'stdout' in kwargs: + raise ValueError('stdout argument not allowed, it will be overridden.') + process = subprocess.Popen(stdout=subprocess.PIPE, *popenargs, **kwargs) + output, unused_err = process.communicate() + retcode = process.poll() + if retcode: + cmd = kwargs.get("args") + if cmd is None: + cmd = popenargs[0] + raise subprocess.CalledProcessError(retcode, cmd) + return output + subprocess.check_output = f def popen_args(filename, *args): @@ -25,7 +41,7 @@ def run_clients(lang, *args): median messsages per second for each. """ if "--redis" not in args: - broker = Popen(popen_args("run_broker.%s" % lang), stderr=PIPE) + broker = subprocess.Popen(popen_args("run_broker.%s" % lang), stderr=subprocess.PIPE) args = popen_args("test_client.%s" % lang, *args) results = [] num_runs = cpu_count() * 2 @@ -34,7 +50,7 @@ def run_clients(lang, *args): bar = ("#" * clients).ljust(num_runs) stdout.write("\r[%s] %s/%s " % (bar, clients, num_runs)) stdout.flush() - out = check_output(args + ["--num-clients=%s" % clients], stderr=PIPE) + out = subprocess.check_output(args + ["--num-clients=%s" % clients], stderr=subprocess.PIPE) results.append(out.split(" ")[0].strip()) stdout.write("\n") if "--redis" not in args: @@ -82,7 +98,7 @@ def run_clients(lang, *args): # Generate graphs. with open("plot.p", "r") as f: plotfile = f.read() -line = '"%s.dat" using ($0+1):1 with lines title "%s" lw 2 lt rgb "%s"' +line = '"output/%s.dat" using ($0+1):1 with lines title "%s" lw 2 lt rgb "%s"' for name, names in plots.items(): name = output_path(name) with open(output_path(names[0] + ".dat"), "r") as f: @@ -91,4 +107,4 @@ def run_clients(lang, *args): lines = ", ".join([line % (l, l.replace("_", " "), colours[l]) for l in names]) f.write(plotfile % {"name": name, "lines": lines, "clients": clients}) - Popen(["gnuplot", name + ".p"], stderr=PIPE) + subprocess.Popen(["gnuplot", name + ".p"], stderr=subprocess.PIPE) diff --git a/configure-CentOS.sh b/configure-CentOS.sh new file mode 100755 index 0000000..df8d1c9 --- /dev/null +++ b/configure-CentOS.sh @@ -0,0 +1,33 @@ +#!/bin/bash +# install requirements for CentOS +# by Patrick Wyatt +set -e + +sudo yum install -y python python-devel +sudo yum install -y zeromq zeromq-devel +sudo yum install -y redis redis-devel +sudo yum install -y gnuplot + +# Install pip +mkdir -p install-pip +cd install-pip +curl -O https://raw.github.com/pypa/virtualenv/master/virtualenv.py +python virtualenv.py /opt/py_virtual +cd .. +rm -rf install-pip + +source /opt/py_virtual/bin/activate +pip install pyzmq +pip install redis +pip install hiredis +pip install argparse + +# Install go +rm -r /usr/local/go +wget http://go.googlecode.com/files/go1.0.3.linux-amd64.tar.gz +tar -C /usr/local -xzf go1.0.3.linux-amd64.tar.gz +rm go1.0.3.linux-amd64.tar.gz + +go get github.com/garyburd/redigo/redis +go get github.com/alecthomas/gozmq + From c4caeccf72b1afc1b37ab939215f3d7b12344ec9 Mon Sep 17 00:00:00 2001 From: Patrick Wyatt Date: Tue, 5 Mar 2013 04:56:56 +0000 Subject: [PATCH 2/4] added elixir support --- .gitignore | 3 +- bench.py | 34 ++++-- configure-CentOS.sh | 2 + install-elixir.sh | 17 +++ install-erlang.sh | 12 +++ lib/app.ex | 257 ++++++++++++++++++++++++++++++++++++++++++++ mix.exs | 27 +++++ 7 files changed, 340 insertions(+), 12 deletions(-) create mode 100755 install-elixir.sh create mode 100755 install-erlang.sh create mode 100644 lib/app.ex create mode 100644 mix.exs diff --git a/.gitignore b/.gitignore index 97489d3..e0c3ce6 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ *.pyc -output +output/ +ebin/ diff --git a/bench.py b/bench.py index 38aedba..7ab91ad 100755 --- a/bench.py +++ b/bench.py @@ -27,12 +27,16 @@ def popen_args(filename, *args): """ Returns the initial popen args for a given Python or Go file. """ - args = [filename, "--quiet"] + list(args) - if filename.split(".")[-1] == "py": - return ["python"] + args + args = ["--quiet"] + list(args) + lang = filename.split(".")[-1] + if lang == "py": + return ["python", filename] + list(args) + elif lang == "go": + return ["go", "run", filename] + list(args) + elif lang == "exs": + return ["mix", "run"] + list(args) else: - return ["go", "run"] + list(args) - + raise NameError('Unknown extension', lang) def run_clients(lang, *args): """ @@ -40,25 +44,32 @@ def run_clients(lang, *args): from 1 to cpus * 2 as the number of clients, returning the median messsages per second for each. """ - if "--redis" not in args: + cmd = popen_args("test_client.%s" % lang, *args) + print " ".join(cmd) + + broker = None + if lang == "exs": + subprocess.check_output(["mix", "compile"], stderr=subprocess.PIPE) + elif "--redis" not in args: broker = subprocess.Popen(popen_args("run_broker.%s" % lang), stderr=subprocess.PIPE) - args = popen_args("test_client.%s" % lang, *args) + results = [] num_runs = cpu_count() * 2 - print " ".join(args) for clients in range(1, num_runs + 1): bar = ("#" * clients).ljust(num_runs) stdout.write("\r[%s] %s/%s " % (bar, clients, num_runs)) stdout.flush() - out = subprocess.check_output(args + ["--num-clients=%s" % clients], stderr=subprocess.PIPE) + out = subprocess.check_output(cmd + ["--num-clients=%s" % clients] + ["--num-seconds=10"], stderr=subprocess.PIPE) results.append(out.split(" ")[0].strip()) stdout.write("\n") - if "--redis" not in args: + + if broker is not None: broker.kill() return results # All test_client runs and their cli args. runs = { + "elixir": ["exs"], "py_redis": ["py", "--redis", "--unbuffered"], "py_redis_buffered": ["py", "--redis"], "py_zmq": ["py"], @@ -68,6 +79,7 @@ def run_clients(lang, *args): # Consistent graph colours defined for each of the runs. colours = { + "elixir": "cyan", "py_redis": "red", "py_redis_buffered": "green", "py_zmq": "blue", @@ -80,7 +92,7 @@ def run_clients(lang, *args): "two-queues-1": ["py_zmq", "py_redis"], "two-queues-2": ["py_zmq", "py_redis", "py_redis_buffered"], "two-queues-3": ["py_zmq", "py_redis", "py_redis_buffered", - "go_zmq", "go_redis"], + "go_zmq", "go_redis", "elixir"], } # Store all results in an output directory. diff --git a/configure-CentOS.sh b/configure-CentOS.sh index df8d1c9..8100c03 100755 --- a/configure-CentOS.sh +++ b/configure-CentOS.sh @@ -31,3 +31,5 @@ rm go1.0.3.linux-amd64.tar.gz go get github.com/garyburd/redigo/redis go get github.com/alecthomas/gozmq +export PATH=$PATH:/usr/local/go/bin + diff --git a/install-elixir.sh b/install-elixir.sh new file mode 100755 index 0000000..97a6ae2 --- /dev/null +++ b/install-elixir.sh @@ -0,0 +1,17 @@ +#!/bin/bash +# https://github.com/spawngrid/kerl +set -e + +if [ 'which erl' ]; then + echo Erlang must be installed before elixir; use install-erlang.sh + exit 1 +fi + + +git clone https://github.com/elixir-lang/elixir.git /opt/elixir +pushd /opt/elixir +make test +popd + +export PATH=$PATH:/opt/elixir/bin + diff --git a/install-erlang.sh b/install-erlang.sh new file mode 100755 index 0000000..242da34 --- /dev/null +++ b/install-erlang.sh @@ -0,0 +1,12 @@ +#!/bin/bash +# https://github.com/spawngrid/kerl +set -e + +curl -O https://raw.github.com/spawngrid/kerl/master/kerl +chmod +x kerl +mv kerl /opt/ + +kerl build R16A_RELEASE_CANDIDATE r16rc +kerl install r16rc /opt/erlang-r16rc +. /opt/erlang-r16rc/activate + diff --git a/lib/app.ex b/lib/app.ex new file mode 100644 index 0000000..a4062e3 --- /dev/null +++ b/lib/app.ex @@ -0,0 +1,257 @@ +require :timer + +defmodule App do + use Application.Behaviour + + def start() do + start(nil, nil) + end + + def start(_type, _arg) do + args = parse_cmdline + + # Start the publishers and subscribers + num_clients = Helpers.get_int(args, :num_clients) + # TODO: is it really necessary to create a lambda to pass a function pointer? + # ... what is the syntax for just passing a function? + # ... in erlang I think I could use fun Publisher.publisher/0 + {:ok, _pid} = :gen_server.start_link({:local, :metrics}, Metrics, [], []) + publishers = run_workers fn -> Publisher.publisher(args) end, num_clients, [] + run_workers fn -> Subcriber.subscriber(args, publishers) end, num_clients, [] + + # Wait for completion + Metrics.reset_messages() + receive do + after Helpers.get_int(args, :num_seconds) * 1000 -> :ok + end + messages = Metrics.get_messages + + # Get the median messages per second + messages = :lists.sort messages + samples = length(messages) + median = :lists.nth(div(samples, 2) + 1, messages) + # ^^^ one-based; not cheating to get a better sample + + IO.puts "#{median} median msg/sec" + + { :ok, self } + end + + # Start worker tasks and get a list of their pids + defp run_workers(_task, 0, acc) do + acc + end + defp run_workers(task, count, acc) do + pid = task.() + run_workers(task, count - 1, [pid | acc]) + end + + # TODO: make a higher-level abstraction on OptionParser like + # Ruby Trollop gem to avoid all this hardcoded stuff + defp parse_cmdline do + # Print command line + #IO.puts "args:" + #Enum.each System.argv, fn(arg) -> IO.puts " #{arg}" end + + # Parse command line + { args, _ } = OptionParser.parse( + System.argv, + switches: [ + redis: :boolean, + unbuffered: :boolean, + quiet: :boolean, + verbose: :boolean + ] + ) + + # Verify arguments + Enum.each args, fn(arg) -> + { key, val } = arg + case key do + :redis -> :ok + :unbuffered -> :ok + :quiet -> :ok + :host -> :ok + :verbose -> :ok + + # Note the dash-to-underscore conversion by OptionParser: + # ex: --num-clients => :"num_clients" + :"num_clients" -> args = Dict.put args, key, parse_int(key, val) + :"num_seconds" -> args = Dict.put args, key, parse_int(key, val) + :"num_channels" -> args = Dict.put args, key, parse_int(key, val) + :"message_size" -> args = Dict.put args, key, parse_int(key, val) + + key -> bad_arg :"Unknown_argument", key, "" + end + end + + # Set default values + args = Dict.put_new args, :"num_clients", default_num_clients + args = Dict.put_new args, :"num_seconds", 10 + args = Dict.put_new args, :"num_channels", 50 + args = Dict.put_new args, :"message_size", 20 + + # Print arguments + if Dict.get(args, :verbose, false) do + Enum.each args, fn(arg) -> + { key, val } = arg + IO.puts "#{key}: #{val}" + end + end + + args + end + + defp default_num_clients do + [ { :processor, cpus } ] = :erlang.system_info(:cpu_topology) + max(1, div(Kernel.length(cpus), 2)) + end + + defp parse_int(key, val) do + if Regex.match?(%r/\A[1-9][0-9]*\z/, val) do + binary_to_integer(val) + else + bad_arg(:"Invalid number", key, val) + end + end + + def bad_arg(error, key, val) do + IO.puts :stderr, "#{error}: '#{key}' #{val}" + :erlang.error error + end + +end + +defmodule Publisher do + + defrecordp :pubstate, [args: nil, subscribers: nil] + + # Create a publisher to broadcast messages + def publisher(args) do + state = pubstate(args: args, subscribers: []) + {:ok, pid} = :gen_server.start_link(Publisher, state, []) + pid + end + + def init(state) do + {:ok, state} + end + + def handle_info(:timeout, state) do + args = pubstate(state, :args) + msgsize = Helpers.get_int(args, :message_size) + broadcast(pubstate(state, :subscribers), msgsize) + { :noreply, state, 0 } # timeout again immediately + end + + def handle_call({:subscribe}, {pid,_}, state) do + #IO.puts "P#{Kernel.inspect self}: subscribed from #{Kernel.inspect pid}" + subscribers = [pid | pubstate(state, :subscribers)] + { :reply, :ok, pubstate(state, subscribers: subscribers), 0 } # timeout again immediately + end + + defp broadcast([], _msgsize) do + end + + defp broadcast(subscribers, msgsize) do + hd(subscribers) <- { :message, String.duplicate("a", msgsize) } + broadcast tl(subscribers), msgsize + end + +end + +defmodule Subcriber do + + defrecordp :substate, [time: nil, messages: 0, quiet: false] + + # Create a subscriber to get publisher messages + def subscriber(args, publishers) do + {:ok, _pid} = :gen_server.start_link(Subcriber, {args, publishers}, []) + end + + def init({args, publishers}) do + subscribe_to_publishers publishers + {:ok, substate(time: :erlang.now(), quiet: Dict.get(args, :quiet)) } + end + + defp subscribe_to_publishers ([]) do + end + defp subscribe_to_publishers ([pid|tail]) do + #IO.puts "S#{Kernel.inspect self}: subscribing to #{Kernel.inspect pid}" + :gen_server.call pid, {:subscribe} + subscribe_to_publishers tail + end + + def handle_info({:message, _msg}, state) do + # Has one second elapsed? + messages = substate(state, :messages) + now = :erlang.now() + delta_mms = :timer.now_diff(now, substate(state, :time)) + if delta_mms > 1000000 do + if substate(state, :quiet) != true do + IO.puts "#{messages} msgs/sec" + end + Metrics.add_messages messages + state = substate(state, time: now, messages: 0) + else + state = substate(state, messages: messages + 1) + end + + { :noreply, state } + end + +end + +defmodule Metrics do + + def init(_arg) do + {:ok, []} + end + + def reset_messages() do + :ok = :gen_server.call(:metrics, :reset_messages) + end + def get_messages() do + {:ok, messages} = :gen_server.call(:metrics, :get_messages) + messages + end + def add_messages(messages) do + :gen_server.cast(:metrics, {:add_messages, messages}) + end + + def handle_call(:get_messages, _from, state) do + { :reply, {:ok, state}, state } + end + def handle_call(:reset_messages, _from, _state) do + { :reply, :ok, [] } + end + def handle_cast({:add_messages, messages}, state) do + { :noreply, [messages | state] } + end + +end + +defmodule Inspect do + + # Inspect a list + def list([]) do + end + def list([hd|tl]) do + IO.puts Kernel.inspect hd + list tl + end + +end + +defmodule Helpers do + + def get_int(dict, key) do + # TODO: is this the easiest way to do dict[key].to_i ? + value = Dict.get(dict, key) + if is_binary(value) do + value = binary_to_integer(value) + end + value + end + +end diff --git a/mix.exs b/mix.exs new file mode 100644 index 0000000..f9a1bfd --- /dev/null +++ b/mix.exs @@ -0,0 +1,27 @@ +defmodule App.MixFile do + use Mix.Project + + def project do + [ + app: :app, + version: "0.0.1", + deps: deps + ] + end + + # Configuration for the OTP application + def application do + [ + mod: {App, []}, + description: 'minimum app' + ] + end + + defp deps do + [ + # { :some_project, "0.3.0", github: "some_project/other" }, + # { :another_project, "1.0.2", git: "https://example.com/another/repo.git" } + ] + end + +end From 2be106292463ea46e6b33b0ff06f3fa34376cb91 Mon Sep 17 00:00:00 2001 From: Patrick Wyatt Date: Thu, 7 Mar 2013 04:49:30 +0000 Subject: [PATCH 3/4] optimize publish message --- lib/app.ex | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/lib/app.ex b/lib/app.ex index a4062e3..d52c358 100644 --- a/lib/app.ex +++ b/lib/app.ex @@ -140,7 +140,8 @@ defmodule Publisher do def handle_info(:timeout, state) do args = pubstate(state, :args) msgsize = Helpers.get_int(args, :message_size) - broadcast(pubstate(state, :subscribers), msgsize) + message = { :message, String.duplicate("a", msgsize) } + broadcast(pubstate(state, :subscribers), message) { :noreply, state, 0 } # timeout again immediately end @@ -150,12 +151,11 @@ defmodule Publisher do { :reply, :ok, pubstate(state, subscribers: subscribers), 0 } # timeout again immediately end - defp broadcast([], _msgsize) do + defp broadcast([], _message) do end - - defp broadcast(subscribers, msgsize) do - hd(subscribers) <- { :message, String.duplicate("a", msgsize) } - broadcast tl(subscribers), msgsize + defp broadcast(subscribers, message) do + hd(subscribers) <- message + broadcast tl(subscribers), message end end From 8b2f20242fa6b443d01e5cd15eba759ad66f51d0 Mon Sep 17 00:00:00 2001 From: Patrick Wyatt Date: Sat, 9 Mar 2013 06:42:41 +0000 Subject: [PATCH 4/4] Changes to make code more idiomatic; also fix command-line handler --- lib/app.ex | 51 ++++++++------------------------------------------- 1 file changed, 8 insertions(+), 43 deletions(-) diff --git a/lib/app.ex b/lib/app.ex index d52c358..97ab3b5 100644 --- a/lib/app.ex +++ b/lib/app.ex @@ -27,10 +27,9 @@ defmodule App do messages = Metrics.get_messages # Get the median messages per second - messages = :lists.sort messages + messages = Enum.sort messages samples = length(messages) - median = :lists.nth(div(samples, 2) + 1, messages) - # ^^^ one-based; not cheating to get a better sample + median = Enum.at!(messages, div(samples, 2)) IO.puts "#{median} median msg/sec" @@ -46,8 +45,6 @@ defmodule App do run_workers(task, count - 1, [pid | acc]) end - # TODO: make a higher-level abstraction on OptionParser like - # Ruby Trollop gem to avoid all this hardcoded stuff defp parse_cmdline do # Print command line #IO.puts "args:" @@ -64,32 +61,11 @@ defmodule App do ] ) - # Verify arguments - Enum.each args, fn(arg) -> - { key, val } = arg - case key do - :redis -> :ok - :unbuffered -> :ok - :quiet -> :ok - :host -> :ok - :verbose -> :ok - - # Note the dash-to-underscore conversion by OptionParser: - # ex: --num-clients => :"num_clients" - :"num_clients" -> args = Dict.put args, key, parse_int(key, val) - :"num_seconds" -> args = Dict.put args, key, parse_int(key, val) - :"num_channels" -> args = Dict.put args, key, parse_int(key, val) - :"message_size" -> args = Dict.put args, key, parse_int(key, val) - - key -> bad_arg :"Unknown_argument", key, "" - end - end - # Set default values - args = Dict.put_new args, :"num_clients", default_num_clients - args = Dict.put_new args, :"num_seconds", 10 - args = Dict.put_new args, :"num_channels", 50 - args = Dict.put_new args, :"message_size", 20 + args = Dict.put_new args, :num_clients, default_num_clients + args = Dict.put_new args, :num_seconds, 10 + args = Dict.put_new args, :num_channels, 50 + args = Dict.put_new args, :message_size, 20 # Print arguments if Dict.get(args, :verbose, false) do @@ -107,22 +83,10 @@ defmodule App do max(1, div(Kernel.length(cpus), 2)) end - defp parse_int(key, val) do - if Regex.match?(%r/\A[1-9][0-9]*\z/, val) do - binary_to_integer(val) - else - bad_arg(:"Invalid number", key, val) - end - end - - def bad_arg(error, key, val) do - IO.puts :stderr, "#{error}: '#{key}' #{val}" - :erlang.error error - end - end defmodule Publisher do + use GenServer.Behaviour defrecordp :pubstate, [args: nil, subscribers: nil] @@ -161,6 +125,7 @@ defmodule Publisher do end defmodule Subcriber do + use GenServer.Behaviour defrecordp :substate, [time: nil, messages: 0, quiet: false]