benchmark: add client and refactor a little

Author: tony612

.gitignore

@@ -21,3 +21,5 @@ erl_crash.dump
 /tmp
 
 /log
+
+.elixir_ls

benchmark/.formatter.exs

@@ -1,4 +1,5 @@
 # Used by "mix format"
 [
-  inputs: ["mix.exs", "{config,lib,test}/**/*.{ex,exs}"]
+  inputs: ["mix.exs", "{config,lib,test}/**/*.{ex,exs}"],
+  import_deps: [:protobuf, :grpc]
 ]

benchmark/lib/bench_manager.ex

@@ -0,0 +1,2 @@
+defmodule Benchmark do
+end

benchmark/lib/benchmark.ex

@@ -0,0 +1,62 @@
+defmodule Benchmark.ClientManager do
+  alias Benchmark.ClientWorker
+
+  def start_client(%Grpc.Testing.ClientConfig{} = config) do
+    # get security
+    _payload_type = Benchmark.Manager.payload_type(config.payload_config)
+
+    channels =
+      0..(config.client_channels - 1)
+      |> Enum.zip(config.server_targets)
+      |> Enum.map(fn {_, server} -> new_client(server) end)
+
+    perform_rpcs(config, channels)
+  end
+
+  defp new_client(addr) do
+    {:ok, conn} = GRPC.Stub.connect(addr)
+    conn
+  end
+
+  def perform_rpcs(config, channels) do
+    payload = client_payload(config.payload_config)
+
+    case config.load_params.load do
+      {:closed_loop, _} ->
+        :ok
+
+      _ ->
+        raise GRPC.RPCError, status: :unimplemented
+    end
+
+    rpcs_per_conn = config.outstanding_rpcs_per_channel
+
+    case Grpc.Testing.RpcType.key(config.rpc_type) do
+      :UNARY ->
+        close_loop_unary(channels, rpcs_per_conn, payload)
+
+      _ ->
+        :ok
+    end
+  end
+
+  def client_payload(nil), do: %{}
+
+  def client_payload(config) do
+    case config.payload do
+      {:simple_params, payload} ->
+        %{req_size: payload.req_size, resp_size: payload.resp_size, type: :protobuf}
+
+      _ ->
+        %{}
+    end
+  end
+
+  def close_loop_unary(channels, rpcs_per_conn, payload) do
+    Enum.each(channels, fn conn ->
+      Enum.each(0..(rpcs_per_conn - 1), fn _idx ->
+        Task.async(ClientWorker, :unary_loop, [conn, payload])
+      end)
+    end)
+  end
+end

benchmark/lib/benchmark/client_manager.ex

@@ -0,0 +1,25 @@
+defmodule Benchmark.ClientWorker do
+  require Logger
+
+  def unary_loop(channel, input) do
+    unary_call(channel, input)
+  end
+
+  def unary_call(channel, input) do
+    payload =
+      Grpc.Testing.Payload.new(
+        type: Grpc.Testing.PayloadType.value(:COMPRESSABLE),
+        body: String.duplicate(<<0>>, input.req_size)
+      )
+
+    req =
+      Grpc.Testing.SimpleRequest.new(
+        response_type: payload.type,
+        response_size: input.resp_size,
+        payload: payload
+      )
+
+    Logger.debug("Sending rpc #{req}")
+    Grpc.Testing.BenchmarkService.Stub.unary_call(channel, req)
+  end
+end

benchmark/lib/benchmark/client_worker.ex

@@ -0,0 +1,25 @@
+defmodule Benchmark.Manager do
+  def payload_type(nil), do: :protobuf
+
+  def payload_type(config) do
+    case config.payload do
+      {:bytebuf_params, _} -> :bytebuf
+      {:simple_params, _} -> :protobuf
+      {:complex_params, _} -> :complex
+      _ -> raise GRPC.RPCError, status: :invalid_argument
+    end
+  end
+
+  def get_cores() do
+    :erlang.system_info(:schedulers_online)
+  end
+
+  def set_cores(core_limit) do
+    if core_limit > 0 do
+      :erlang.system_flag(:schedulers_online, core_limit)
+      core_limit
+    else
+      get_cores()
+    end
+  end
+end

benchmark/lib/benchmark/manager.ex

@@ -0,0 +1,14 @@
+defmodule Benchmark.ServerManager do
+  def start_server(%Grpc.Testing.ServerConfig{} = config) do
+    # get security
+    payload_type = Benchmark.Manager.payload_type(config.payload_config)
+    start_server(payload_type, config)
+  end
+
+  def start_server(:protobuf, config) do
+    {:ok, pid, port} = GRPC.Server.start(Grpc.Testing.BenchmarkService.Server, config.port)
+    %{port: port, pid: pid}
+  end
+
+  def start_server(_, _), do: raise(GRPC.RPCError, status: :unimplemented)
+end

benchmark/lib/benchmark/server_manager.ex

@@ -0,0 +1,59 @@
+defmodule Grpc.Testing.WorkerService.Server do
+  use GRPC.Server, service: Grpc.Testing.WorkerService.Service
+  alias GRPC.Server
+
+  alias Benchmark.Manager
+  alias Benchmark.ServerManager
+  alias Benchmark.ClientManager
+
+  require Logger
+
+  def run_server(args_enum, stream) do
+    Enum.each(args_enum, fn args ->
+      Logger.debug("Got args:")
+      Logger.debug(inspect(args))
+
+      status =
+        case args.argtype do
+          {:setup, config} ->
+            cores = Manager.set_cores(config.core_limit)
+            server = ServerManager.start_server(config)
+            Logger.debug("Started server: #{inspect(server)}")
+            Grpc.Testing.ServerStatus.new(stats: nil, port: server[:port], cores: cores)
+
+          {:mark, mark} ->
+            stats = get_stats(mark.reset)
+            Grpc.Testing.ServerStatus.new(stats: stats)
+        end
+
+      Server.send_reply(stream, status)
+    end)
+  end
+
+  def run_client(args_enum, stream) do
+    Enum.each(args_enum, fn args ->
+      Logger.debug("Got args:")
+      Logger.debug(inspect(args))
+
+      status =
+        case args.argtype do
+          {:setup, client_config} ->
+            ClientManager.start_client(client_config)
+            Grpc.Testing.ClientStatus.new()
+
+          {:mark, mark} ->
+            stats = get_stats(mark.reset)
+            Grpc.Testing.ClientStatus.new(stats: stats)
+        end
+
+      Server.send_reply(stream, status)
+    end)
+  end
+
+  def core_count(_, _) do
+    Grpc.Testing.CoreResponse.new(cores: Manager.get_cores())
+  end
+
+  defp get_stats(_reset) do
+  end
+end