Make the benchmarks run, move message packing into the Sender

Author: ihumanable

.gitignore

@@ -1,4 +1,5 @@
 /_build
+/bench/snapshots
 /cover
 /deps
 erl_crash.dump

bench/group_by_one.exs renamed to bench/group_by_one_bench.exs

@@ -0,0 +1,29 @@
+defmodule SendVsEncodeBench do
+  use Benchfella
+
+  defmodule Receiver do
+    def loop do
+      receive do
+        _ ->
+          loop()
+      end
+    end
+  end
+
+  setup_all do
+    pid = spawn(Receiver, :loop, [])
+    {:ok, pid}
+  end
+
+  bench "sending message", [message: gen_message()] do
+    send(bench_context, message)
+  end
+
+  bench "encoding message", [message: gen_message()] do
+    :erlang.term_to_binary(message)
+  end
+
+  defp gen_message() do
+    Map.new(1..1_000_000, fn item -> {item, :erlang.unique_integer()} end)
+  end
+end

bench/partitioner_benches.exs renamed to bench/partitioner_bench.exs

@@ -5,6 +5,8 @@ defmodule Manifold do
   alias Manifold.Sender
   alias Manifold.Utils
 
+  @type pack_mode :: :binary | :etf | nil
+
   @max_partitioners 32
   @partitioners min(Application.get_env(:manifold, :partitioners, 1), @max_partitioners)
   @workers_per_partitioner Application.get_env(:manifold, :workers_per_partitioner, System.schedulers_online)
@@ -39,6 +41,8 @@ defmodule Manifold do
   @spec valid_send_options?(Keyword.t()) :: boolean()
   def valid_send_options?(options) when is_list(options) do
     valid_options = [
+      {:pack_mode, :binary},
+      {:pack_mode, :etf},
       {:send_mode, :offload},
     ]
 
@@ -57,13 +61,13 @@ defmodule Manifold do
   def send([pid], message, options), do: __MODULE__.send(pid, message, options)
 
   def send(pids, message, options) when is_list(pids) do
-    message = Utils.pack_message(options[:pack_mode], message)
-
     case options[:send_mode] do
       :offload ->
-        Sender.send(current_sender(), current_partitioner(), pids, message)
+        Sender.send(current_sender(), current_partitioner(), pids, message, options[:pack_mode])
 
       nil ->
+        message = Utils.pack_message(options[:pack_mode], message)
+
         partitioner_name = current_partitioner()
 
         grouped_by =
@@ -85,7 +89,7 @@ defmodule Manifold do
       :offload ->
         # To maintain linearizability guaranteed by send/2, we have to send
         # it to the sender process, even for a single receiving pid.
-        Sender.send(current_sender(), current_partitioner(), [pid], message)
+        Sender.send(current_sender(), current_partitioner(), [pid], message, :etf)
 
       nil ->
         Partitioner.send({current_partitioner(), node(pid)}, [pid], message)

bench/partitioner_benches_one.exs renamed to bench/partitioner_one_bench.exs

@@ -20,9 +20,9 @@ defmodule Manifold.Partitioner do
     GenServer.start_link(__MODULE__, partitions, opts)
   end
 
-  @spec send(pid, [pid], term) :: :ok
-  def send(pid, pids, message) do
-    @gen_module.cast(pid, {:send, pids, message})
+  @spec send(partitioner :: GenServer.server(), pids :: [pid()], message :: term()) :: :ok
+  def send(partitioner, pids, message) do
+    @gen_module.cast(partitioner, {:send, pids, message})
   end
 
   ## Server Callbacks

bench/send_bench_one.exs renamed to bench/send_one_bench.exs

@@ -18,9 +18,9 @@ defmodule Manifold.Sender do
     GenServer.start_link(__MODULE__, :ok, opts)
   end
 
-  @spec send(pid, atom, [pid], term) :: :ok
-  def send(pid, partitioner, pids, message) do
-    @gen_module.cast(pid, {:send, partitioner, pids, message})
+  @spec send(sender :: GenServer.server(), partitioner :: GenServer.server(), pids :: [pid()], message :: term(), pack_mode :: Manifold.pack_mode()) :: :ok
+  def send(sender, partitioner, pids, message, pack_mode) do
+    @gen_module.cast(sender, {:send, partitioner, pids, message, pack_mode})
   end
 
   ## Server Callbacks
@@ -32,7 +32,9 @@ defmodule Manifold.Sender do
     {:ok, nil}
   end
 
-  def handle_cast({:send, partitioner, pids, message}, nil) do
+  def handle_cast({:send, partitioner, pids, message, pack_mode}, nil) do
+    message = Utils.pack_message(pack_mode, message)
+
     grouped_by =
       Utils.group_by(pids, fn
         nil -> nil

bench/send_vs_encode_bench.exs

@@ -45,19 +45,19 @@ defmodule Manifold.Utils do
     value
   end
   def hash(key), do: hash("#{key}")
-  
+
   @doc """
   Gets the next delay at which we should attempt to hibernate a worker or partitioner process.
   """
-  @spec next_hibernate_delay() :: integer 
+  @spec next_hibernate_delay() :: integer
   def next_hibernate_delay() do
     hibernate_delay = Application.get_env(:manifold, :hibernate_delay, 60_000)
     hibernate_jitter = Application.get_env(:manifold, :hibernate_jitter, 30_000)
 
     hibernate_delay + :rand.uniform(hibernate_jitter)
   end
 
-  @spec pack_message(mode :: atom(), message :: term()) :: term()
+  @spec pack_message(mode :: Manifold.pack_mode(), message :: term()) :: term()
   def pack_message(:binary, message), do: {:manifold_binary, :erlang.term_to_binary(message)}
   def pack_message(_mode, message), do: message