Gnat Benchmark Script

benchmark_roundtrips.exs

defmodule LatencyBenchmark do
  @default_settings %{
    num_actors: 1,
    actions_per_actor: 1,
  }

  def benchmark(action_fn, settings) do # = num_actors, actions_per_actor, setup_fn, action_fn) do
    settings = Map.merge(@default_settings, %{setup_fn: fn -> %{} end}) |> Map.merge(settings)
    settings = Map.put(settings, :action_fn, action_fn)
    {:ok, collector_pid} = Agent.start_link(fn -> [] end)
    settings = Map.put(settings, :collector_pid, collector_pid)
    settings = Map.put(settings, :parent_pid, self())
    {micro_seconds, _result} = time_benchmark(settings)
    total_actions = settings.num_actors * settings.actions_per_actor
    throughput = total_actions * 1_000_000.0 / micro_seconds
    IO.puts "It took #{micro_seconds / 1_000_000.0}sec to do #{total_actions} iterations"
    IO.puts "\t#{throughput}it/sec throughput"
    print_statistics(collector_pid, throughput)
    Agent.stop(collector_pid, :normal)
  end

  def record_action_time(collector_pid, micro_seconds) do
    :ok = Agent.update(collector_pid, fn(list) -> [micro_seconds | list] end)
  end

  def print_statistics(collector_pid, throughput) do
    Agent.get(collector_pid, fn(list_of_rpc_times) ->
      tc_l = list_of_rpc_times
      tc_n = Enum.count(list_of_rpc_times)
      tc_min = :lists.min(tc_l)
      tc_max = :lists.max(tc_l)
      sorted = :lists.sort(tc_l)
      tc_med = :lists.nth(round(tc_n * 0.5), sorted)
      tc_90th = :lists.nth(round(tc_n * 0.9), sorted)
      tc_avg = round(Enum.sum(tc_l) / tc_n)
      IO.puts "\tmin: #{tc_min}µs"
      IO.puts "\tmax: #{tc_max}µs"
      IO.puts "\tmedian: #{tc_med}µs"
      IO.puts "\t90th percentile: #{tc_90th}µs"
      IO.puts "\taverage: #{tc_avg}µs"
      IO.puts "\t#{tc_min},#{tc_max},#{tc_med},#{tc_90th},#{tc_avg},#{throughput}"
    end)
  end

  def time_benchmark(settings) do
    :timer.tc(fn() ->
      (1..settings.num_actors) |> Enum.map(fn(_i) ->
        spawn_link(fn() -> run_actor(settings) end)
      end)
      wait_for_times(settings.num_actors)
    end)
  end

  def run_actor(settings) do
    state = settings.setup_fn.()
    (1..settings.actions_per_actor) |> Enum.each(fn(_) ->
      {micros, _} = :timer.tc(fn -> settings.action_fn.(state) end)
      record_action_time(settings.collector_pid, micros)
    end)
    send settings.parent_pid, :ack
  end

  def wait_for_times(0), do: :done
  def wait_for_times(n) do
    receive do
      :ack ->
        wait_for_times(n-1)
    end
  end
end

defmodule EchoServer do
  @ack <<1>>
  @response File.read!("/Users/mmmries/Downloads/RandomNumbers") # 1.3k of random bytes
  def handle_message(%{reply_to: address}) do
    Gnat.pub(:gnat, address, @ack)
    Gnat.pub(:gnat, address, @response)
  end
end

{:ok, gnat} = Gnat.start_link(%{host: '127.0.0.1', port: 4222})
true = Process.register(gnat, :gnat)
{:ok, _consumer_pid} = Gnat.ConsumerSupervisor.start_link(%{
  connection_name: :gnat,
  consuming_function: {EchoServer, :handle_message},
  subscription_topics: [%{topic: "rpc.test.transaction_service.search", queue_group: "rpc.test.transaction_service.search"}],
})

request = "74c93e71c5aa03ad4f0881caa374ba1af08f3e4a04ce5f8bd0b2d82d6d72de6eef3e46ed8d8c3dbe24d0f6109115dcdf13280d1c13c2f6d22d14336b29df8e65"

settings = %{
  num_actors: 64,
  actions_per_actor: 5_000,
  setup_fn: fn ->
    reply_to = "INBOX-#{:crypto.strong_rand_bytes(12) |> Base.encode64}"
    {:ok, _sub} = Gnat.sub(gnat, self(), reply_to)
    %{reply_to: reply_to}
  end,
}
request_fn = fn(%{reply_to: reply_to}) ->
  :ok = Gnat.pub(gnat, "rpc.test.transaction_service.search", request, reply_to: reply_to)
  receive do
    {:msg, %{body: <<1>>, topic: ^reply_to}} ->
      :ack
    after 500 ->
      raise "ACK TIMEOUT"
  end
  receive do
    {:msg, %{topic: ^reply_to, body: _response_bytes}} ->
      :response
    after 5000 ->
      raise "RESPONSE TIMEOUT"
  end
end
# Warmup
LatencyBenchmark.benchmark(request_fn, Map.merge(settings, %{num_actors: 1, actions_per_actor: 5000}))
LatencyBenchmark.benchmark(request_fn, settings)