Repository: keathley/finch
Pull request #24 (2020-04-19 21:26:32)

Open connections lazily, in the client, rather than within the pool
The current strategy works quite well if the server uses keep alive, however if the server does not, we see a significant performance degradation (even more than expected from the lack of keep alive).

If the server does not support keep alive or only allows a few requests per connection, all of our workers repeatedly have their connections closed, and reopened, which slows down our pool, and can also overwhelm the server. The problem gets worse with larger numbers of workers.

Instead, we can open the connections lazily, after giving it to the client. This gives the client better control over retries, and relieves pressure on the pool.

In my benchmarking, connecting lazily barely has an impact on throughput, if any at all, it only occurs when you first start using the pool, but once the connections have been established performance is essentially the same. 

As part of this change, I also removed the reconnections with exponential backoff which was needed for the async connect attempts, but now that we only connect on the caller, where we want to block as little as possible, I think it makes sense to remove, at least for now.

These changes would also correct the behaviour noted in https://github.com/keathley/finch/issues/23

josevalim: Looks great! Don't forget to run `mix deps.unlock --unused` too, if we have no other use for the backoff lib.
QuinnWilton: This is awesome! It also resolves https://github.com/keathley/finch/issues/22
mhanberg: Confirmed that it solves the issue was was seeing in #23 👍

Patch:
diff --git a/lib/finch.ex b/lib/finch.ex
index 1e3adc5..3f78d0d 100644
--- a/lib/finch.ex
+++ b/lib/finch.ex
@@ -41,7 +41,7 @@ defmodule Finch do
      name: MyConfiguredFinch,
      pools: %{
        :default => [size: 10],
-       "https://hex.pm" => [size: 32, count: 8, backoff: [initial: 1, max: 30_000]]
+       "https://hex.pm" => [size: 32, count: 8]
      }}
   ]
   ```
@@ -91,24 +91,6 @@ defmodule Finch do
       type: :pos_integer,
       doc: "Number of pools to start.",
       default: 1
-    ],
-    backoff: [
-      type: :non_empty_keyword_list,
-      default: [initial: 1, max: :timer.minutes(1)],
-      doc:
-        "Failed connection attempts will be retried using an exponential backoff with jitter. Values are in milliseconds.",
-      keys: [
-        initial: [
-          type: :pos_integer,
-          doc: "Backoff will begin at this value, and increase exponentially.",
-          default: 1
-        ],
-        max: [
-          type: :pos_integer,
-          doc: "Backoff will be capped to this value.",
-          default: :timer.minutes(1)
-        ]
-      ]
     ]
   ]
 
@@ -252,7 +234,6 @@ defmodule Finch do
     %{
       size: valid[:size],
       count: valid[:count],
-      backoff: Map.new(valid[:backoff]),
       conn_opts: []
     }
   end
diff --git a/lib/finch/conn.ex b/lib/finch/conn.ex
index 821f06b..ee4eb62 100644
--- a/lib/finch/conn.ex
+++ b/lib/finch/conn.ex
@@ -12,7 +12,6 @@ defmodule Finch.Conn do
       port: port,
       opts: opts.conn_opts,
       parent: parent,
-      backoff: opts.backoff,
       mint: nil
     }
   end
@@ -25,12 +24,10 @@ defmodule Finch.Conn do
     }
 
     Telemetry.event(:reused_connection, %{}, meta)
-    conn
+    {:ok, conn}
   end
 
-  def connect(conn), do: connect(conn, conn.backoff.initial)
-
-  def connect(conn, prev_backoff) do
+  def connect(conn) do
     meta = %{
       scheme: conn.scheme,
       host: conn.host,
@@ -38,19 +35,17 @@ defmodule Finch.Conn do
     }
 
     start_time = Telemetry.start(:connect, meta)
+    conn_opts = Keyword.merge([mode: :passive], conn.opts)
+
+    case HTTP.connect(conn.scheme, conn.host, conn.port, conn_opts) do
+      {:ok, mint} ->
+        Telemetry.stop(:connect, start_time, meta)
+        {:ok, %{conn | mint: mint}}
 
-    with {:ok, mint} <- HTTP.connect(conn.scheme, conn.host, conn.port, conn.opts),
-         {:ok, mint} <- HTTP.controlling_process(mint, conn.parent) do
-      Telemetry.stop(:connect, start_time, meta)
-      %{conn | mint: mint}
-    else
       {:error, error} ->
         meta = Map.put(meta, :error, error)
         Telemetry.stop(:connect, start_time, meta)
-
-        backoff = :backoff.rand_increment(prev_backoff, conn.backoff.max)
-        Process.sleep(backoff)
-        connect(conn, backoff)
+        {:error, conn, error}
     end
   end
 
@@ -69,14 +64,15 @@ defmodule Finch.Conn do
   def stream(%{mint: nil}, _), do: {:error, "Connection is dead"}
 
   def stream(conn, message) do
-    HTTP.stream(conn.mint, message)
+    with {:ok, mint, responses} <- HTTP.stream(conn.mint, message) do
+      {:ok, %{conn | mint: mint}, responses}
+    end
   end
 
   def transfer(%{mint: nil}, _), do: {:error, "Connection is dead"}
 
   def transfer(conn, pid) do
-    with {:ok, mint} <- HTTP.set_mode(conn.mint, :passive),
-         {:ok, mint} <- HTTP.controlling_process(mint, pid) do
+    with {:ok, mint} <- HTTP.controlling_process(conn.mint, pid) do
       {:ok, %{conn | mint: mint}}
     end
   end
diff --git a/lib/finch/pool.ex b/lib/finch/pool.ex
index 24b36a3..201404b 100644
--- a/lib/finch/pool.ex
+++ b/lib/finch/pool.ex
@@ -36,14 +36,13 @@ defmodule Finch.Pool do
         :checkout,
         fn {conn, pool} ->
           Telemetry.stop(:queue, start_time, metadata)
-          conn = Conn.connect(conn)
-
-          case Conn.request(conn, req, receive_timeout) do
-            {:ok, conn, response} ->
-              {{:ok, response}, transfer_conn(conn, pool)}
 
+          with {:ok, conn} <- Conn.connect(conn),
+               {:ok, conn, response} <- Conn.request(conn, req, receive_timeout) do
+            {{:ok, response}, transfer_if_open(conn, pool)}
+          else
             {:error, conn, error} ->
-              {{:error, error}, conn}
+              {{:error, error}, transfer_if_open(conn, pool)}
           end
         end,
         pool_timeout
@@ -63,41 +62,42 @@ defmodule Finch.Pool do
 
   @impl NimblePool
   def init_worker({_, {scheme, host, port}, opts} = pool_state) do
-    parent = self()
-
-    async = fn ->
-      conn = Conn.new(scheme, host, port, opts, parent)
-      Conn.connect(conn)
-    end
-
-    {:async, async, pool_state}
+    {:ok, Conn.new(scheme, host, port, opts, self()), pool_state}
   end
 
   @impl NimblePool
-  # Transfer the conn to the caller.
-  # If we lost the connection, then we remove it to try again.
+  def handle_checkout(:checkout, _, %{mint: nil} = conn) do
+    {:ok, {conn, self()}, conn}
+  end
+
   def handle_checkout(:checkout, {pid, _}, conn) do
-    case Conn.transfer(conn, pid) do
-      {:ok, conn} -> {:ok, {conn, self()}, conn}
-      _ -> {:remove, :closed}
+    with {:ok, conn} <- Conn.set_mode(conn, :passive),
+         {:ok, conn} <- Conn.transfer(conn, pid) do
+      {:ok, {conn, self()}, conn}
+    else
+      {:error, _error} ->
+        {:remove, :closed}
     end
   end
 
   @impl NimblePool
   def handle_checkin(conn, _from, _old_conn) do
-    case Conn.set_mode(conn, :active) do
-      {:ok, conn} -> {:ok, conn}
-      {:error, _} -> {:remove, :closed}
+    with true <- Conn.open?(conn),
+         {:ok, conn} <- Conn.set_mode(conn, :active) do
+      {:ok, conn}
+    else
+      _ ->
+        {:remove, :closed}
     end
   end
 
   @impl NimblePool
   def handle_info(message, conn) do
     case Conn.stream(conn, message) do
-      {:ok, _, _} -> {:ok, conn}
-      {:error, _, _, _} -> {:remove, :closed}
-      {:error, _} -> {:remove, :closed}
+      {:ok, conn, _} -> {:ok, conn}
       :unknown -> {:ok, conn}
+      {:error, _mint, _error, _responses} -> {:remove, :closed}
+      {:error, _error} -> {:remove, :closed}
     end
   end
 
@@ -109,7 +109,7 @@ defmodule Finch.Pool do
     {:ok, pool_state}
   end
 
-  defp transfer_conn(conn, pid) do
+  defp transfer_if_open(conn, pid) do
     if Conn.open?(conn) do
       {:ok, conn} = Conn.transfer(conn, pid)
       conn
diff --git a/lib/finch/telemetry.ex b/lib/finch/telemetry.ex
index 6ae1c9a..dd04ca4 100644
--- a/lib/finch/telemetry.ex
+++ b/lib/finch/telemetry.ex
@@ -114,6 +114,7 @@ defmodule Finch.Telemetry do
     * `:error` - This value is optional. It includes any errors that occured while receiving the response.
 
   * `[:finch, :reused_connection]` - Executed if an existing connection is reused. There are no measurements provided with this event.
+
     #### Metadata:
     * `:scheme` - The scheme used in the connection. either `http` or `https`
     * `:host` - The host address
diff --git a/mix.exs b/mix.exs
index cbdfdd1..8cc735a 100644
--- a/mix.exs
+++ b/mix.exs
@@ -24,7 +24,6 @@ defmodule Finch.MixProject do
       {:nimble_pool, github: "dashbitco/nimble_pool"},
       {:nimble_options, github: "dashbitco/nimble_options"},
       {:telemetry, "~> 0.4"},
-      {:backoff, "~> 1.1"},
       {:ex_doc, "~> 0.21", only: :dev, runtime: false},
       {:credo, "~> 1.3", only: [:dev, :test]},
       {:bypass, "~> 1.0", only: :test}
diff --git a/mix.lock b/mix.lock
index e780a8c..373cc5e 100644
--- a/mix.lock
+++ b/mix.lock
@@ -1,5 +1,4 @@
 %{
-  "backoff": {:hex, :backoff, "1.1.6", "83b72ed2108ba1ee8f7d1c22e0b4a00cfe3593a67dbc792799e8cce9f42f796b", [:rebar3], [], "hexpm", "cf0cfff8995fb20562f822e5cc47d8ccf664c5ecdc26a684cbe85c225f9d7c39"},
   "bunt": {:hex, :bunt, "0.2.0", "951c6e801e8b1d2cbe58ebbd3e616a869061ddadcc4863d0a2182541acae9a38", [:mix], [], "hexpm", "7af5c7e09fe1d40f76c8e4f9dd2be7cebd83909f31fee7cd0e9eadc567da8353"},
   "bypass": {:hex, :bypass, "1.0.0", "b78b3dcb832a71aca5259c1a704b2e14b55fd4e1327ff942598b4e7d1a7ad83d", [:mix], [{:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 1.0 or ~> 2.0", [hex: :plug_cowboy, repo: "hexpm", optional: false]}], "hexpm", "5a1dc855dfcc86160458c7a70d25f65d498bd8012bd4c06a8d3baa368dda3c45"},
   "castore": {:hex, :castore, "0.1.5", "591c763a637af2cc468a72f006878584bc6c306f8d111ef8ba1d4c10e0684010", [:mix], [], "hexpm", "6db356b2bc6cc22561e051ff545c20ad064af57647e436650aa24d7d06cd941a"},


Test patch:
diff --git a/test/finch_test.exs b/test/finch_test.exs
index aeb103d..01021ff 100644
--- a/test/finch_test.exs
+++ b/test/finch_test.exs
@@ -30,7 +30,7 @@ defmodule FinchTest do
       {:ok, _} =
         Finch.start_link(
           name: MyFinch,
-          pools: %{default: [count: 5, size: 5, backoff: [initial: 2, max: 4]]}
+          pools: %{default: [count: 5, size: 5]}
         )
 
       expect_any(bypass)
@@ -235,6 +235,13 @@ defmodule FinchTest do
                )
     end
 
+    test "returns error when requesting bad address" do
+      start_supervised({Finch, name: MyFinch})
+
+      assert {:error, %{reason: :nxdomain}} =
+               Finch.request(MyFinch, :get, "http://idontexist.wat")
+    end
+
     test "worker exits when pool times out", %{bypass: bypass} do
       start_supervised({Finch, name: MyFinch})
       expect_any(bypass)
@@ -254,76 +261,6 @@ defmodule FinchTest do
       Bypass.up(bypass)
       assert {:ok, %Response{}} = Finch.request(MyFinch, :get, endpoint(bypass))
     end
-
-    test "will exponentially backoff until connection succeeds", %{bypass: bypass} do
-      {test_name, _arity} = __ENV__.function
-
-      Bypass.expect_once(bypass, fn conn ->
-        Plug.Conn.send_resp(conn, 200, "OK")
-      end)
-
-      initial_backoff = Enum.random(10..50)
-      max_backoff = Enum.random(100..250)
-      backoff = [initial: initial_backoff, max: max_backoff]
-      start_supervised({Finch, name: MyFinch, pools: %{default: [size: 1, backoff: backoff]}})
-
-      parent = self()
-      ref = make_ref()
-
-      handler = fn event, measurements, meta, _ ->
-        send(parent, {ref, event, measurements, meta})
-      end
-
-      :telemetry.attach_many(
-        to_string(test_name),
-        [
-          [:finch, :connect, :start],
-          [:finch, :connect, :stop]
-        ],
-        handler,
-        nil
-      )
-
-      Bypass.down(bypass)
-
-      spawn_link(fn ->
-        Finch.request(MyFinch, :get, endpoint(bypass), [], nil, pool_timeout: 5_000)
-        send(parent, {ref, :request_sent})
-      end)
-
-      Process.sleep(max_backoff * 10)
-
-      Bypass.up(bypass)
-
-      Enum.reduce_while(1..999, [], fn _, start_times ->
-        assert_receive {^ref, [:finch, :connect, :start], %{system_time: start_time}, _},
-                       max_backoff
-
-        assert_receive {^ref, [:finch, :connect, :stop], _, meta}, max_backoff
-
-        case meta do
-          %{error: _} ->
-            {:cont, [start_time | start_times]}
-
-          _ ->
-            {:halt, start_times}
-        end
-      end)
-      |> Enum.reverse()
-      |> Enum.map(&System.convert_time_unit(&1, :native, :millisecond))
-      |> Enum.chunk_every(2)
-      |> Enum.each(fn
-        [a, b] ->
-          interval = b - a
-          assert interval >= initial_backoff
-          assert interval <= max_backoff
-
-        _ ->
-          :ok
-      end)
-
-      assert_receive {^ref, :request_sent}
-    end
   end
 
   describe "telemetry" do