Add built in connection pool (#108)

First step towards a faster/ singe pool written specifically for DBConnection. This pool uses ETS tables to store connection state and moves ownership of the connection around to avoid tracking state internally. The pool is faster because interaction between the clients and the pool requires only a single message roundtrip to checkout, and a single cast to return. Thus beating or matching the efficiency of our existing pools in each case.

Author: fishcakez

integration_test/cases/idle_test.exs

@@ -28,7 +28,7 @@ defmodule TestIdle do
       end]
     {:ok, agent} = A.start_link(stack)
 
-    opts = [agent: agent, parent: self(), idle_timeout: 50]
+    opts = [agent: agent, parent: self(), idle_timeout: 50, idle_interval: 50]
     {:ok, pool} = P.start_link(opts)
     assert_receive {:hi, conn}
     assert_receive {:pong, ^conn}

integration_test/cases/queue_test.exs

@@ -42,7 +42,8 @@ defmodule QueueTest do
     stack = [{:ok, :state}]
     {:ok, agent} = A.start_link(stack)
 
-    opts = [agent: agent, parent: self(), queue_timeout: 50]
+    opts = [agent: agent, parent: self(), queue_timeout: 50, queue_target: 50,
+      queue_interval: 50]
     {:ok, pool} = P.start_link(opts)
 
     parent = self()
@@ -147,7 +148,8 @@ defmodule QueueTest do
     {:ok, agent} = A.start_link(stack)
 
     opts = [agent: agent, parent: self(), backoff_start: 30_000,
-      queue_timeout: 10, pool_timeout: 10]
+      queue_timeout: 10, pool_timeout: 10, queue_target: 10,
+      queue_interval: 10]
     {:ok, pool} = P.start_link(opts)
 
     P.run(pool, fn(_) ->
@@ -165,7 +167,7 @@ defmodule QueueTest do
     {:ok, agent} = A.start_link(stack)
 
     opts = [agent: agent, parent: self(), backoff_start: 30_000,
-      queue_timeout: 10, pool_timeout: 10]
+      queue_timeout: 10, pool_timeout: 10, queue_target: 10, queue_interval: 10]
     {:ok, pool} = P.start_link(opts)
 
     P.run(pool, fn(_) ->

integration_test/connection_pool/all_test.exs

@@ -0,0 +1 @@
+Code.require_file "../tests.exs", __DIR__

integration_test/connection_pool/test_helper.exs

@@ -0,0 +1,11 @@
+ExUnit.start([capture_log: true, assert_receive_timeout: 500,
+              exclude: [:pool_overflow, :enqueue_disconnected,
+                        :queue_timeout_exit]])
+
+Code.require_file "../../test/test_support.exs", __DIR__
+
+defmodule TestPool do
+  use TestConnection, [pool: DBConnection.ConnectionPool, pool_size: 1]
+end
+
+{:ok, _} = TestPool.ensure_all_started()

lib/db_connection/app.ex

@@ -9,6 +9,7 @@ defmodule DBConnection.App do
       supervisor(DBConnection.Task, []),
       supervisor(DBConnection.Sojourn.Supervisor, []),
       supervisor(DBConnection.Ownership.PoolSupervisor, []),
+      supervisor(DBConnection.ConnectionPool.PoolSupervisor, []),
       worker(DBConnection.Watcher, [])
     ]
     Supervisor.start_link(children, strategy: :one_for_all, name: __MODULE__)

lib/db_connection/connection.ex

@@ -17,6 +17,7 @@ defmodule DBConnection.Connection do
   use Connection
   require Logger
   alias DBConnection.Backoff
+  alias DBConnection.ConnectionPool
 
   @pool_timeout  5_000
   @timeout       15_000
@@ -83,6 +84,11 @@ defmodule DBConnection.Connection do
     end
   end
 
+  @doc false
+  def ping({pid, ref}, state) do
+    Connection.cast(pid, {:ping, ref, state})
+  end
+
   ## Internal API
 
   @doc false
@@ -96,19 +102,26 @@ defmodule DBConnection.Connection do
     Supervisor.Spec.worker(__MODULE__, [mod, opts, mode], child_opts)
   end
 
+  @doc false
+  def child_spec(mod, opts, mode, info, child_opts) do
+    Supervisor.Spec.worker(__MODULE__, [mod, opts, mode, info], child_opts)
+  end
+
   ## Connection API
 
   @doc false
   def init({mod, opts, mode, info}) do
-    queue         = if mode == :sojourn, do: :broker, else: :queue.new()
+    queue         = if mode in [:connection, :poolboy], do: :queue.new(), else: mode
+    idle          = if mode in [:connection, :poolboy], do: get_idle(opts), else: :passive
     broker        = if mode == :sojourn, do: elem(info, 0)
     regulator     = if mode == :sojourn, do: elem(info, 1)
-    idle          = if mode == :sojourn, do: :passive, else: get_idle(opts)
     after_timeout = if mode == :poolboy, do: :stop, else: :backoff
+    pool          = if mode == :connection_pool, do: elem(info, 0)
+    tag           = if mode == :connection_pool, do: elem(info, 1)
 
     s = %{mod: mod, opts: opts, state: nil, client: :closed, broker: broker,
-          regulator: regulator, lock: nil, queue: queue, timer: nil,
-          backoff: Backoff.new(opts),
+          regulator: regulator, lock: nil, pool: pool, tag: tag, queue: queue,
+          timer: nil, backoff: Backoff.new(opts),
           after_connect: Keyword.get(opts, :after_connect),
           after_connect_timeout: Keyword.get(opts, :after_connect_timeout,
                                              @timeout), idle: idle,
@@ -206,7 +219,7 @@ defmodule DBConnection.Connection do
   @doc false
   def handle_call({:checkout, ref, queue?, timeout}, {pid, _} = from, s) do
     case s do
-      %{queue: :broker} ->
+      %{queue: mode} when is_atom(mode) ->
         exit(:bad_checkout)
       %{client: nil, idle: :passive, mod: mod, state: state} ->
         Connection.reply(from, {:ok, {self(), ref}, mod, state})
@@ -244,6 +257,16 @@ defmodule DBConnection.Connection do
   end
 
   @doc false
+  def handle_cast({:ping, ref, state}, %{client: {ref, :pool}} = s) do
+    %{mod: mod} = s
+    case apply(mod, :ping, [state]) do
+      {:ok, state} ->
+        pool_update(state, s)
+      {:disconnect, err, state} ->
+        {:disconnect, {:log, err}, %{s | state: state}}
+    end
+  end
+
   def handle_cast({:checkin, ref, state}, %{client: {ref, _}} = s) do
     handle_next(state, s)
   end
@@ -259,7 +282,7 @@ defmodule DBConnection.Connection do
     {:stop, {err, stack}, %{s | state: state}}
   end
 
-  def handle_cast({:cancel, _}, %{queue: :broker}) do
+  def handle_cast({:cancel, _}, %{queue: mode}) when is_atom(mode) do
     exit(:bad_cancel)
   end
   def handle_cast({:cancel, ref}, %{client: {ref, _}, state: state} = s) do
@@ -308,7 +331,9 @@ defmodule DBConnection.Connection do
   def handle_cast({:connected, ref}, %{client: {ref, :connect}} = s) do
     %{mod: mod, state: state, queue: queue, broker: broker} = s
     case apply(mod, :checkout, [state]) do
-      {:ok, state} when queue == :broker ->
+      {:ok, state} when queue == :connection_pool ->
+        pool_update(state, s)
+      {:ok, state} when queue == :sojourn ->
         info = {self(), mod, state}
         {:await, ^ref, _} = :sbroker.async_ask_r(broker, info, {self(), ref})
         {:noreply,  %{s | client: {ref, :broker}, state: state}}
@@ -335,7 +360,8 @@ defmodule DBConnection.Connection do
     err = DBConnection.ConnectionError.exception(message)
     {:disconnect, {down_log(reason), err}, %{s | client: {ref, nil}}}
   end
-  def handle_info({:DOWN, _, :process, _, _} = msg, %{queue: :broker} = s) do
+  def handle_info({:DOWN, _, :process, _, _} = msg, %{queue: mode} = s)
+      when is_atom(mode) do
     do_handle_info(msg, s)
   end
   def handle_info({:DOWN, ref, :process, _, _} = msg, %{queue: queue} = s) do
@@ -425,7 +451,7 @@ defmodule DBConnection.Connection do
   defp start_opts(:connection, opts) do
     Keyword.take(opts, [:debug, :name, :timeout, :spawn_opt])
   end
-  defp start_opts(mode, opts) when mode in [:poolboy, :sojourn] do
+  defp start_opts(mode, opts) when mode in [:poolboy, :sojourn, :connection_pool] do
     Keyword.take(opts, [:debug, :spawn_opt])
   end
 
@@ -486,7 +512,10 @@ defmodule DBConnection.Connection do
     demonitor(client)
     handle_next(state, %{s | client: nil, backoff: backoff})
   end
-  defp handle_next(state, %{queue: :broker} = s) do
+  defp handle_next(state, %{queue: :connection_pool} = s) do
+    pool_update(state, s)
+  end
+  defp handle_next(state, %{queue: :sojourn} = s) do
     %{client: client, timer: timer, mod: mod, broker: broker} = s
     demonitor(client)
     cancel_timer(timer)
@@ -626,7 +655,9 @@ defmodule DBConnection.Connection do
     end
   end
 
-  defp clear_queue(:broker), do: :broker
+  defp clear_queue(queue) when is_atom(queue) do
+    queue
+  end
   defp clear_queue(queue) do
     clear =
       fn({{_, mon}, _, from}) ->
@@ -645,6 +676,11 @@ defmodule DBConnection.Connection do
     :ok
   end
 
+  defp pool_update(state, %{pool: pool, tag: tag, mod: mod} = s) do
+    ref = ConnectionPool.update(pool, tag, mod, state)
+    {:noreply, %{s | client: {ref, :pool}, state: state}, :hibernate}
+  end
+
   defp normal_status(mod, pdict, state) do
     try do
       mod.format_status(:normal, [pdict, state])