From 01134b6677a76d4eee11056d19371ecb8f565c57 Mon Sep 17 00:00:00 2001 From: KinutaDev Date: Mon, 17 Aug 2026 13:53:27 +0900 Subject: [PATCH 1/5] feat: add DBConnection.SingleConnection for AtomVM - Add a one-connection pool selected via :pool - Add tests for checkout, FIFO queue, metrics, and disconnect_all --- lib/db_connection.ex | 2 + lib/db_connection/single_connection.ex | 238 +++++++++++++++++ test/db_connection/single_connection_test.exs | 250 ++++++++++++++++++ test/test_helper.exs | 4 + 4 files changed, 494 insertions(+) create mode 100644 lib/db_connection/single_connection.ex create mode 100644 test/db_connection/single_connection_test.exs diff --git a/lib/db_connection.ex b/lib/db_connection.ex index 58718943..629958bf 100644 --- a/lib/db_connection.ex +++ b/lib/db_connection.ex @@ -76,6 +76,8 @@ defmodule DBConnection do DBConnection connections support using different pools via the `:pool` option passed to `start_link/2`. The default pool is `DBConnection.ConnectionPool`. Another supported pool that is commonly used for tests is `DBConnection.Ownership`. + `DBConnection.SingleConnection` is a one-connection pool without CoDel + queueing or `DynamicSupervisor`, intended for constrained runtimes. For now, using *custom* pools is not supported since the API for pools is not public. diff --git a/lib/db_connection/single_connection.ex b/lib/db_connection/single_connection.ex new file mode 100644 index 00000000..204c6025 --- /dev/null +++ b/lib/db_connection/single_connection.ex @@ -0,0 +1,238 @@ +defmodule DBConnection.SingleConnection do + @moduledoc """ + A single-connection pool for constrained runtimes. + + Unlike `DBConnection.ConnectionPool`, this pool keeps exactly one connection, + does not use CoDel queueing, idle polling, or `DynamicSupervisor`. + + Pass this module as the `:pool` option to `DBConnection.start_link/2`. + The default pool remains `DBConnection.ConnectionPool`. `:pool_size` is + ignored except that values below 1 raise, same as the default pool. + """ + + use GenServer + alias DBConnection.Holder + alias DBConnection.Util + + @behaviour DBConnection.Pool + + @doc false + def start_link({mod, opts}) do + GenServer.start_link(__MODULE__, {mod, opts}, start_opts(opts)) + end + + @doc false + @impl DBConnection.Pool + def checkout(pool, callers, opts) do + Holder.checkout(pool, callers, opts) + end + + @doc false + @impl DBConnection.Pool + def disconnect_all(pool, interval, _opts) do + GenServer.call(pool, {:disconnect_all, interval}, :infinity) + end + + @doc false + @impl DBConnection.Pool + def get_connection_metrics(pool) do + GenServer.call(pool, :get_connection_metrics, :infinity) + end + + @impl GenServer + def init({mod, opts}) do + Process.flag(:trap_exit, true) + DBConnection.register_as_pool(mod) + + size = Keyword.get(opts, :pool_size, 1) + + if size < 1 do + raise ArgumentError, "pool size must be greater or equal to 1, got #{size}" + end + + tag = make_ref() + + state = %{ + mod: mod, + opts: opts, + tag: tag, + holder: nil, + waiters: :queue.new(), + ts: {nil, max_lifetime(opts)}, + conn: nil, + restarts: [], + max_restarts: Keyword.get(opts, :max_restarts, 3), + max_seconds: Keyword.get(opts, :max_seconds, 5) + } + + {:ok, pid} = start_connection(state) + {:ok, %{state | conn: pid}} + end + + @impl GenServer + def handle_call(:get_connection_metrics, _from, state) do + metrics = %{ + source: {:pool, self()}, + ready_conn_count: if(state.holder, do: 1, else: 0), + checkout_queue_length: :queue.len(state.waiters) + } + + {:reply, [metrics], state} + end + + def handle_call({:disconnect_all, interval}, _from, state) do + {_, max_lifetime} = state.ts + ts = {{System.monotonic_time(), interval}, max_lifetime} + {:reply, :ok, %{state | ts: ts}} + end + + @impl GenServer + def handle_info({:db_connection, from, {:checkout, _callers, _now, queue?}}, state) do + {:noreply, checkout_request(from, queue?, state)} + end + + def handle_info({:"ETS-TRANSFER", holder, _pid, {msg, tag, extra}}, %{tag: tag} = state) do + case msg do + :checkin -> + owner = self() + + case :ets.info(holder, :owner) do + ^owner -> + {interval, max_lifetime} = state.ts + + if Holder.maybe_disconnect(holder, interval, max_lifetime) do + {:noreply, %{state | holder: nil}} + else + handle_checkin(holder, extra, state) + end + + :undefined -> + {:noreply, state} + end + + :disconnect -> + Holder.handle_disconnect(holder, extra) + {:noreply, %{state | holder: nil}} + + :stop -> + Holder.handle_stop(holder, extra) + {:noreply, %{state | holder: nil}} + end + end + + def handle_info({:"ETS-TRANSFER", holder, pid, tag}, %{tag: tag} = state) do + message = "client #{Util.inspect_pid(pid)} exited" + err = DBConnection.ConnectionError.exception(message: message, severity: :info) + Holder.handle_disconnect(holder, err) + {:noreply, %{state | holder: nil}} + end + + def handle_info({:timeout, deadline, {tag, holder, pid, len}}, %{tag: tag} = state) do + if Holder.handle_deadline(holder, deadline) do + message = + "client #{Util.inspect_pid(pid)} timed out because " <> + "it queued and checked out the connection for longer than #{len}ms" + + exc = + case Process.info(pid, :current_stacktrace) do + {:current_stacktrace, stacktrace} -> + message <> + "\n\n#{Util.inspect_pid(pid)} was at location:\n\n" <> + Exception.format_stacktrace(stacktrace) + + _ -> + message + end + |> DBConnection.ConnectionError.exception() + + Holder.handle_disconnect(holder, exc) + end + + {:noreply, state} + end + + def handle_info({:EXIT, pid, reason}, %{conn: pid} = state) do + restart_connection(reason, state) + end + + def handle_info({:EXIT, _pid, reason}, state) do + {:stop, reason, state} + end + + @impl GenServer + def terminate(_reason, %{conn: pid}) when is_pid(pid) do + Process.exit(pid, :shutdown) + :ok + end + + def terminate(_reason, _state), do: :ok + + defp checkout_request(from, queue?, %{holder: {holder, checked_in_at}} = state) do + if Holder.handle_checkout(holder, from, state.tag, checked_in_at) do + %{state | holder: nil} + else + checkout_request(from, queue?, %{state | holder: nil}) + end + end + + defp checkout_request(from, true, state) do + %{state | waiters: :queue.in(from, state.waiters)} + end + + defp checkout_request(from, false, state) do + message = "connection not available and queuing is disabled" + Holder.reply_error(from, DBConnection.ConnectionError.exception(message)) + state + end + + defp handle_checkin(holder, now_in_native, state) do + case :queue.out(state.waiters) do + {:empty, waiters} -> + {:noreply, %{state | holder: {holder, now_in_native}, waiters: waiters}} + + {{:value, from}, waiters} -> + state = %{state | waiters: waiters} + + if Holder.handle_checkout(holder, from, state.tag, now_in_native) do + {:noreply, %{state | holder: nil}} + else + handle_checkin(holder, now_in_native, state) + end + end + end + + defp restart_connection(reason, state) do + now = System.monotonic_time(:second) + cutoff = now - state.max_seconds + restarts = Enum.filter(state.restarts, &(&1 >= cutoff)) + + if length(restarts) >= state.max_restarts do + {:stop, reason, state} + else + {:ok, pid} = start_connection(state) + {:noreply, %{state | conn: pid, holder: nil, restarts: [now | restarts]}} + end + end + + defp start_connection(%{mod: mod, opts: opts, tag: tag}) do + DBConnection.Connection.start_link(mod, Keyword.put(opts, :pool_index, 1), self(), tag) + end + + defp max_lifetime(opts) do + case Keyword.fetch(opts, :max_lifetime) do + {:ok, %Range{first: first, last: last, step: 1}} when first >= 0 and last >= first -> + {System.convert_time_unit(first, :millisecond, :native), last - first} + + {:ok, invalid} -> + raise ArgumentError, + "invalid value for :max_lifetime, expected a non-negative step-1 range, got: #{inspect(invalid)}" + + :error -> + nil + end + end + + defp start_opts(opts) do + Keyword.take(opts, [:name, :spawn_opt]) + end +end diff --git a/test/db_connection/single_connection_test.exs b/test/db_connection/single_connection_test.exs new file mode 100644 index 00000000..35c2a1b0 --- /dev/null +++ b/test/db_connection/single_connection_test.exs @@ -0,0 +1,250 @@ +defmodule DBConnection.SingleConnectionTest do + use ExUnit.Case, async: true + + alias TestSingleConnection, as: P + alias TestAgent, as: A + alias TestQuery, as: Q + alias TestResult, as: R + + test "start_link workflow with unregistered name" do + stack = [{:ok, :state}] + {:ok, agent} = A.start_link(stack) + + opts = [agent: agent, pool_size: 1] + {:ok, conn} = P.start_link(opts) + + {:links, links} = Process.info(self(), :links) + assert conn in links + + _ = :sys.get_state(conn) + assert [{:connect, [connect_opts]}] = A.record(agent) + assert Keyword.get(connect_opts, :pool_index) == 1 + assert Keyword.get(connect_opts, :agent) == agent + end + + test "start_link workflow with registered name", %{test: name} do + stack = [{:ok, :state}] + {:ok, agent} = A.start_link(stack) + + opts = [agent: agent, name: name, pool_size: 1] + {:ok, conn} = P.start_link(opts) + + assert Process.info(conn, :registered_name) == {:registered_name, name} + + _ = :sys.get_state(conn) + assert [{:connect, [connect_opts]}] = A.record(agent) + assert Keyword.get(connect_opts, :pool_index) == 1 + end + + test "connection_module/1 returns the connection module" do + {:ok, agent} = A.start_link([{:ok, :state}, {:idle, :state}, {:idle, :state}]) + {:ok, pool} = P.start_link(agent: agent) + + assert {:ok, TestConnection} = DBConnection.connection_module(pool) + + P.run(pool, fn conn -> + assert {:ok, TestConnection} = DBConnection.connection_module(conn) + end) + end + + test "checkout and checkin via run" do + stack = [{:ok, :state}] ++ List.duplicate({:idle, :state}, 8) + {:ok, agent} = A.start_link(stack) + {:ok, pool} = P.start_link(agent: agent) + + assert P.run(pool, fn _conn -> :result end) == :result + assert P.run(pool, fn _conn -> :again end) == :again + end + + test "execute after checkout" do + stack = [ + {:ok, :state}, + {:ok, %Q{}, %R{}, :state}, + {:ok, %Q{}, %R{}, :state} + ] + + {:ok, agent} = A.start_link(stack) + {:ok, pool} = P.start_link(agent: agent) + + assert P.execute(pool, %Q{}, [:param1]) == {:ok, %Q{}, %R{}} + assert P.execute(pool, %Q{}, [:param2]) == {:ok, %Q{}, %R{}} + + assert [ + connect: [_], + handle_execute: [%Q{}, [:param1], _, :state], + handle_execute: [%Q{}, [:param2], _, :state] + ] = A.record(agent) + end + + test "queue: false raises when the connection is busy" do + stack = [{:ok, :state}, {:idle, :state}, {:idle, :state}] + {:ok, agent} = A.start_link(stack) + {:ok, pool} = P.start_link(agent: agent) + + P.run(pool, fn _ -> + {queue_time, _} = + :timer.tc(fn -> + opts = [queue: false] + + assert_raise DBConnection.ConnectionError, + ~r"connection not available and queuing is disabled", + fn -> P.run(pool, fn _ -> flunk("got connection") end, opts) end + end) + + assert queue_time <= 1_000_000, "request was queued" + end) + end + + test "queues waiters and serves them in FIFO order" do + stack = [{:ok, :state}] ++ List.duplicate({:idle, :state}, 20) + {:ok, agent} = A.start_link(stack) + {:ok, pool} = P.start_link(agent: agent) + + parent = self() + + holder = + spawn_link(fn -> + Process.put(:agent, agent) + + P.run(pool, fn _ -> + send(parent, {:held, self()}) + + receive do + {:release, ^parent} -> :ok + end + end) + end) + + assert_receive {:held, ^holder} + + first = + Task.async(fn -> + Process.put(:agent, agent) + P.run(pool, fn _ -> send(parent, {:ran, 1}) end) + :ok + end) + + TestHelpers.poll(fn -> + assert [%{checkout_queue_length: 1, ready_conn_count: 0}] = + P.get_connection_metrics(pool) + end) + + second = + Task.async(fn -> + Process.put(:agent, agent) + P.run(pool, fn _ -> send(parent, {:ran, 2}) end) + :ok + end) + + TestHelpers.poll(fn -> + assert [%{checkout_queue_length: 2, ready_conn_count: 0}] = + P.get_connection_metrics(pool) + end) + + send(holder, {:release, self()}) + assert_receive {:ran, 1} + assert_receive {:ran, 2} + + assert :ok = Task.await(first) + assert :ok = Task.await(second) + end + + test "reports connection metrics" do + stack = [ + {:ok, :state}, + fn _, _, _, _ -> + receive do + :continue -> {:ok, %Q{}, %R{}, :state} + end + end + ] + + {:ok, agent} = A.start_link(stack) + {:ok, pool} = P.start_link(agent: agent) + + TestHelpers.poll(fn -> + assert [%{source: {:pool, ^pool}, checkout_queue_length: 0, ready_conn_count: 1}] = + P.get_connection_metrics(pool) + end) + + query = + spawn_link(fn -> + Process.put(:agent, agent) + assert P.execute(pool, %Q{}, [:client]) + end) + + TestHelpers.poll(fn -> + assert [%{source: {:pool, ^pool}, checkout_queue_length: 0, ready_conn_count: 0}] = + P.get_connection_metrics(pool) + end) + + send(query, :continue) + + TestHelpers.poll(fn -> + assert [%{source: {:pool, ^pool}, checkout_queue_length: 0, ready_conn_count: 1}] = + P.get_connection_metrics(pool) + end) + end + + test "disconnect_all disconnects on checkin" do + stack = [ + {:ok, :state}, + {:ok, %Q{}, %R{}, :new_state1}, + {:ok, %Q{}, %R{}, :new_state2}, + :ok, + {:ok, :final_state}, + {:ok, %Q{}, %R{}, :final_state1}, + {:ok, %Q{}, %R{}, :final_state2} + ] + + {:ok, agent} = A.start_link(stack) + {:ok, pool} = P.start_link(agent: agent) + + assert P.execute(pool, %Q{}, [:param]) == {:ok, %Q{}, %R{}} + P.disconnect_all(pool, 0) + assert P.execute(pool, %Q{}, [:param]) == {:ok, %Q{}, %R{}} + assert P.execute(pool, %Q{}, [:param]) == {:ok, %Q{}, %R{}} + assert P.execute(pool, %Q{}, [:param]) == {:ok, %Q{}, %R{}} + + err = %DBConnection.ConnectionError{message: "disconnect_all requested", severity: :debug} + + assert [ + connect: [_], + handle_execute: [_, _, _, :state], + handle_execute: [_, _, _, :new_state1], + disconnect: [^err, :new_state2], + connect: [_], + handle_execute: [_, _, _, :final_state], + handle_execute: [_, _, _, :final_state1] + ] = A.record(agent) + end + + test "reconnects when the client exits" do + stack = [ + {:ok, :state}, + {:idle, :state}, + :ok, + fn opts -> + send(opts[:parent], :reconnected) + {:ok, :state} + end, + {:idle, :state}, + {:idle, :state} + ] + + {:ok, agent} = A.start_link(stack) + {:ok, pool} = P.start_link(agent: agent, parent: self()) + + _ = + spawn(fn -> + _ = Process.put(:agent, agent) + + P.run(pool, fn _ -> + Process.exit(self(), :shutdown) + end) + end) + + assert_receive :reconnected + assert P.run(pool, fn _ -> :result end) == :result + end +end diff --git a/test/test_helper.exs b/test/test_helper.exs index 989768b2..1c6ee2ee 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -5,3 +5,7 @@ Code.require_file("test_support.exs", __DIR__) defmodule TestPool do use TestConnection, pool: DBConnection.ConnectionPool, pool_size: 1 end + +defmodule TestSingleConnection do + use TestConnection, pool: DBConnection.SingleConnection, pool_size: 1 +end From 3f860a3b7c3040e8e2e6ee17e44a8701fe9d0973 Mon Sep 17 00:00:00 2001 From: KinutaDev Date: Mon, 17 Aug 2026 17:06:59 +0900 Subject: [PATCH 2/5] feat: adapt DBConnection.Holder for AtomVM ETS - Change to use set tables instead of ordered_set - Change to send ETS-TRANSFER when give_away is missing - Change to look up owner without ets.info as a fallback --- lib/db_connection/connection_pool.ex | 2 +- lib/db_connection/holder.ex | 78 ++++++++++++++++++++++---- lib/db_connection/single_connection.ex | 2 +- 3 files changed, 68 insertions(+), 14 deletions(-) diff --git a/lib/db_connection/connection_pool.ex b/lib/db_connection/connection_pool.ex index e760a46e..989a9e13 100644 --- a/lib/db_connection/connection_pool.ex +++ b/lib/db_connection/connection_pool.ex @@ -161,7 +161,7 @@ defmodule DBConnection.ConnectionPool do :checkin -> owner = self() - case :ets.info(holder, :owner) do + case Holder.owner(holder) do ^owner -> {interval, max_lifetime} = ts diff --git a/lib/db_connection/holder.ex b/lib/db_connection/holder.ex index 0d69a51d..8a31a769 100644 --- a/lib/db_connection/holder.ex +++ b/lib/db_connection/holder.ex @@ -15,7 +15,8 @@ defmodule DBConnection.Holder do :lock, :connected_at, deadline: nil, - status: :ok + status: :ok, + owner: nil ]) Record.defrecord(:pool_ref, [:pool, :reference, :deadline, :holder, :lock]) @@ -28,13 +29,21 @@ defmodule DBConnection.Holder do @spec new(pid, reference, module, term) :: t @spec new(pid, reference, module, term, integer) :: t def new(pool, ref, mod, state, connected_at \\ System.monotonic_time()) do - # Insert before setting heir so that pool can't receive empty table - holder = :ets.new(__MODULE__, [:public, :ordered_set, decentralized_counters: true]) + # Insert before setting heir so that pool can't receive empty table. + # AtomVM has no ordered_set; this table holds a single :conn row so :set is enough. + holder = :ets.new(__MODULE__, [:public, :set, decentralized_counters: true]) + + conn = + conn( + connection: self(), + module: mod, + state: state, + connected_at: connected_at, + owner: self() + ) - conn = conn(connection: self(), module: mod, state: state, connected_at: connected_at) true = :ets.insert_new(holder, conn) - - :ets.setopts(holder, {:heir, pool, ref}) + maybe_set_heir(holder, pool, ref) holder end @@ -44,7 +53,7 @@ defmodule DBConnection.Holder do holder = new(pool, ref, mod, state, connected_at) try do - :ets.give_away(holder, pool, {:checkin, ref, System.monotonic_time()}) + transfer(holder, pool, {:checkin, ref, System.monotonic_time()}) {:ok, holder} rescue ArgumentError -> :error @@ -207,12 +216,25 @@ defmodule DBConnection.Holder do :ok end + @spec owner(t) :: pid | :undefined + def owner(holder) do + if function_exported?(:ets, :info, 2) do + :ets.info(holder, :owner) + else + try do + :ets.lookup_element(holder, :conn, conn(:owner) + 1) + rescue + ArgumentError -> :undefined + end + end + end + @spec handle_checkout(t, {pid, reference}, reference, checkin_time) :: boolean def handle_checkout(holder, {pid, mref}, ref, checkin_time) do - :ets.give_away(holder, pid, {mref, ref, checkin_time}) + transfer(holder, pid, {mref, ref, checkin_time}) rescue ArgumentError -> - if Process.alive?(pid) or :ets.info(holder, :owner) != self() do + if Process.alive?(pid) or owner(holder) != self() do raise ArgumentError, no_holder(holder, pid) else false @@ -362,7 +384,7 @@ defmodule DBConnection.Holder do defp no_holder(holder, maybe_pid) do reason = - case :ets.info(holder, :owner) do + case owner(holder) do :undefined -> "does not exist" ^maybe_pid -> "is being given to its current owner" owner when owner != self() -> "does not belong to the giving process" @@ -382,7 +404,7 @@ defmodule DBConnection.Holder do #{call_reason}. SELF: #{Util.inspect_pid(self())} - ETS INFO: #{inspect(:ets.info(holder))} + ETS INFO: #{inspect(ets_info(holder))} Please report at https://github.com/elixir-ecto/db_connection/issues" """ @@ -431,7 +453,7 @@ defmodule DBConnection.Holder do try do :ets.update_element(holder, :conn, [{conn(:deadline) + 1, nil} | ops]) - :ets.give_away(holder, pool, {tag, ref, info}) + transfer(holder, pool, {tag, ref, info}) rescue ArgumentError -> :ok else @@ -485,4 +507,36 @@ defmodule DBConnection.Holder do hash = :erlang.phash2(pid, interval_ms) System.convert_time_unit(hash, :millisecond, :native) end + + defp maybe_set_heir(holder, pool, ref) do + if function_exported?(:ets, :setopts, 2) do + :ets.setopts(holder, {:heir, pool, ref}) + else + :ok + end + end + + defp transfer(table, pid, gift_data) do + if function_exported?(:ets, :give_away, 3) do + :ets.give_away(table, pid, gift_data) + else + _ = put_owner(table, pid) + send(pid, {:"ETS-TRANSFER", table, self(), gift_data}) + true + end + end + + defp put_owner(table, pid) do + :ets.update_element(table, :conn, {conn(:owner) + 1, pid}) + rescue + ArgumentError -> false + end + + defp ets_info(holder) do + if function_exported?(:ets, :info, 1) do + :ets.info(holder) + else + %{owner: owner(holder)} + end + end end diff --git a/lib/db_connection/single_connection.ex b/lib/db_connection/single_connection.ex index 204c6025..d541a4f0 100644 --- a/lib/db_connection/single_connection.ex +++ b/lib/db_connection/single_connection.ex @@ -96,7 +96,7 @@ defmodule DBConnection.SingleConnection do :checkin -> owner = self() - case :ets.info(holder, :owner) do + case Holder.owner(holder) do ^owner -> {interval, max_lifetime} = state.ts From ab4a894f1c69f384e03b45d31b815e85fec155fb Mon Sep 17 00:00:00 2001 From: KinutaDev Date: Mon, 17 Aug 2026 18:49:40 +0900 Subject: [PATCH 3/5] feat: run DBConnection.Connection as a GenServer - Change to use GenServer instead of gen_statem handle_event_function - Change connect and disconnect to handle_continue - Change backoff from a gen_statem named timeout to Process.send_after - Change pool check-in to return without hibernate --- lib/db_connection/connection.ex | 178 +++++++++++++++----------------- 1 file changed, 83 insertions(+), 95 deletions(-) diff --git a/lib/db_connection/connection.ex b/lib/db_connection/connection.ex index 5b6abf7a..55fb28a3 100644 --- a/lib/db_connection/connection.ex +++ b/lib/db_connection/connection.ex @@ -1,7 +1,7 @@ defmodule DBConnection.Connection do @moduledoc false - @behaviour :gen_statem + use GenServer require Logger alias DBConnection.Backoff @@ -14,7 +14,7 @@ defmodule DBConnection.Connection do def start_link(mod, opts, pool, tag) do start_opts = Keyword.take(opts, [:debug, :spawn_opt]) sensitive_options = %DBConnection.SensitiveData{data: opts} - :gen_statem.start_link(__MODULE__, {mod, sensitive_options, pool, tag}, start_opts) + GenServer.start_link(__MODULE__, {mod, sensitive_options, pool, tag}, start_opts) end @doc false @@ -27,27 +27,20 @@ defmodule DBConnection.Connection do @doc false def disconnect({pid, ref}, err, state) do - :gen_statem.cast(pid, {:disconnect, ref, err, state}) + GenServer.cast(pid, {:disconnect, ref, err, state}) end @doc false def stop({pid, ref}, err, state) do - :gen_statem.cast(pid, {:stop, ref, err, state}) + GenServer.cast(pid, {:stop, ref, err, state}) end @doc false def ping({pid, ref}, state) do - :gen_statem.cast(pid, {:ping, ref, state}) + GenServer.cast(pid, {:ping, ref, state}) end - ## gen_statem API - - @doc false - @impl :gen_statem - def callback_mode, do: :handle_event_function - - @doc false - @impl :gen_statem + @impl GenServer def init({mod, %DBConnection.SensitiveData{data: opts}, pool, tag}) do pool_index = Keyword.get(opts, :pool_index) label = if pool_index, do: "db_conn_#{pool_index}", else: "db_conn" @@ -61,6 +54,7 @@ defmodule DBConnection.Connection do pool: pool, tag: tag, timer: nil, + backoff_ref: nil, connected_at: nil, backoff: Backoff.new(opts), connection_listeners: Keyword.get(opts, :connection_listeners, []), @@ -68,13 +62,11 @@ defmodule DBConnection.Connection do after_connect_timeout: Keyword.get(opts, :after_connect_timeout, @timeout) } - {:ok, :no_state, s, {:next_event, :internal, {:connect, :init}}} + {:ok, s, {:continue, {:connect, :init}}} end - @impl :gen_statem - def handle_event(type, info, state, s) - - def handle_event(:internal, {:connect, _info}, :no_state, s) do + @impl GenServer + def handle_continue({:connect, _info}, s) do %{mod: mod, opts: opts, backoff: backoff, after_connect: after_connect} = s try do @@ -87,16 +79,19 @@ defmodule DBConnection.Connection do {:ok, state} when after_connect != nil -> ref = make_ref() connected_at = System.monotonic_time() - :gen_statem.cast(self(), {:after_connect, ref}) - {:keep_state, %{s | state: state, client: {ref, :connect}, connected_at: connected_at}} + GenServer.cast(self(), {:after_connect, ref}) + s = cancel_backoff(s) + + {:noreply, %{s | state: state, client: {ref, :connect}, connected_at: connected_at}} {:ok, state} -> backoff = backoff && Backoff.reset(backoff) ref = make_ref() connected_at = System.monotonic_time() - :gen_statem.cast(self(), {:connected, ref}) + GenServer.cast(self(), {:connected, ref}) + s = cancel_backoff(s) - {:keep_state, + {:noreply, %{ s | state: state, @@ -135,12 +130,11 @@ defmodule DBConnection.Connection do crash_reason: {err, []} ) - {timeout, backoff} = Backoff.backoff(backoff) - {:keep_state, %{s | backoff: backoff}, {{:timeout, :backoff}, timeout, nil}} + {:noreply, schedule_backoff(s)} end end - def handle_event(:internal, {:disconnect, {log, err}}, :no_state, %{mod: mod} = s) do + def handle_continue({:disconnect, {log, err}}, %{mod: mod} = s) do if log == :log do severity = case err do @@ -174,33 +168,29 @@ defmodule DBConnection.Connection do {:stop, {:shutdown, err}, s} {_, :after_connect} -> - {timeout, backoff} = Backoff.backoff(backoff) - {:keep_state, %{s | backoff: backoff}, {{:timeout, :backoff}, timeout, nil}} + {:noreply, schedule_backoff(s)} _ -> - {:keep_state, s, {:next_event, :internal, {:connect, :disconnect}}} + {:noreply, s, {:continue, {:connect, :disconnect}}} end end - def handle_event({:timeout, :backoff}, _content, :no_state, s) do - {:keep_state, s, {:next_event, :internal, {:connect, :backoff}}} - end - - def handle_event(:cast, {:ping, ref, state}, :no_state, %{client: {ref, :pool}, mod: mod} = s) do + @impl GenServer + def handle_cast({:ping, ref, state}, %{client: {ref, :pool}, mod: mod} = s) do case apply(mod, :ping, [state]) do {:ok, state} -> pool_update(state, s) {:disconnect, err, state} -> - {:keep_state, %{s | state: state}, {:next_event, :internal, {:disconnect, {:log, err}}}} + {:noreply, %{s | state: state}, {:continue, {:disconnect, {:log, err}}}} end end - def handle_event(:cast, {:disconnect, ref, err, state}, :no_state, %{client: {ref, _}} = s) do - {:keep_state, %{s | state: state}, {:next_event, :internal, {:disconnect, {:log, err}}}} + def handle_cast({:disconnect, ref, err, state}, %{client: {ref, _}} = s) do + {:noreply, %{s | state: state}, {:continue, {:disconnect, {:log, err}}}} end - def handle_event(:cast, {:stop, ref, err, state}, :no_state, %{client: {ref, _}} = s) do + def handle_cast({:stop, ref, err, state}, %{client: {ref, _}} = s) do {_, stack} = :erlang.process_info(self(), :current_stacktrace) case err do @@ -228,11 +218,11 @@ defmodule DBConnection.Connection do {:stop, {err, stack}, %{s | state: state}} end - def handle_event(:cast, {tag, _, _, _}, :no_state, s) when tag in [:disconnect, :stop] do - handle_timeout(s) + def handle_cast({tag, _, _, _}, s) when tag in [:disconnect, :stop] do + {:noreply, s} end - def handle_event(:cast, {:after_connect, ref}, :no_state, %{client: {ref, :connect}} = s) do + def handle_cast({:after_connect, ref}, %{client: {ref, :connect}} = s) do %{ mod: mod, state: state, @@ -249,18 +239,18 @@ defmodule DBConnection.Connection do {pid, ref} = DBConnection.Task.run_child(mod, state, after_connect, opts) timer = start_timer(pid, timeout) s = %{s | client: {ref, :after_connect}, timer: timer, state: state} - {:keep_state, s} + {:noreply, s} {:disconnect, err, state} -> - {:keep_state, %{s | state: state}, {:next_event, :internal, {:disconnect, {:log, err}}}} + {:noreply, %{s | state: state}, {:continue, {:disconnect, {:log, err}}}} end end - def handle_event(:cast, {:after_connect, _}, :no_state, _s) do - :keep_state_and_data + def handle_cast({:after_connect, _}, s) do + {:noreply, s} end - def handle_event(:cast, {:connected, ref}, :no_state, %{client: {ref, :connect}} = s) do + def handle_cast({:connected, ref}, %{client: {ref, :connect}} = s) do %{mod: mod, state: state} = s notify_connection_listeners(:connected, s) @@ -270,45 +260,39 @@ defmodule DBConnection.Connection do pool_update(state, s) {:disconnect, err, state} -> - {:keep_state, %{s | state: state}, {:next_event, :internal, {:disconnect, {:log, err}}}} + {:noreply, %{s | state: state}, {:continue, {:disconnect, {:log, err}}}} end end - def handle_event(:cast, {:connected, _}, :no_state, _s) do - :keep_state_and_data + def handle_cast({:connected, _}, s) do + {:noreply, s} end - def handle_event( - :info, - {:DOWN, ref, _, pid, reason}, - :no_state, - %{client: {ref, :after_connect}} = s - ) do + @impl GenServer + def handle_info(:backoff, s) do + {:noreply, %{s | backoff_ref: nil}, {:continue, {:connect, :backoff}}} + end + + def handle_info({:DOWN, ref, _, pid, reason}, %{client: {ref, :after_connect}} = s) do message = "client #{Util.inspect_pid(pid)} exited: " <> Exception.format_exit(reason) err = DBConnection.ConnectionError.exception(message) - {:keep_state, %{s | client: {nil, :after_connect}}, - {:next_event, :internal, {:disconnect, {down_log(reason), err}}}} + {:noreply, %{s | client: {nil, :after_connect}}, + {:continue, {:disconnect, {down_log(reason), err}}}} end - def handle_event(:info, {:DOWN, mon, _, pid, reason}, :no_state, %{client: {ref, mon}} = s) do + def handle_info({:DOWN, mon, _, pid, reason}, %{client: {ref, mon}} = s) do message = "client #{Util.inspect_pid(pid)} exited: " <> Exception.format_exit(reason) err = DBConnection.ConnectionError.exception(message) - {:keep_state, %{s | client: {ref, nil}}, - {:next_event, :internal, {:disconnect, {down_log(reason), err}}}} + {:noreply, %{s | client: {ref, nil}}, {:continue, {:disconnect, {down_log(reason), err}}}} end - def handle_event( - :info, - {:timeout, timer, {__MODULE__, pid, timeout}}, - :no_state, - %{timer: timer} = s - ) + def handle_info({:timeout, timer, {__MODULE__, pid, timeout}}, %{timer: timer} = s) when is_reference(timer) do message = "client #{Util.inspect_pid(pid)} timed out because it checked out " <> @@ -326,13 +310,11 @@ defmodule DBConnection.Connection do end |> DBConnection.ConnectionError.exception() - {:keep_state, %{s | timer: nil}, {:next_event, :internal, {:disconnect, {:log, exc}}}} + {:noreply, %{s | timer: nil}, {:continue, {:disconnect, {:log, exc}}}} end - def handle_event( - :info, + def handle_info( {:"ETS-TRANSFER", holder, _pid, {msg, ref, extra}}, - :no_state, %{client: {ref, :after_connect}, timer: timer} = s ) do {_, state} = Holder.delete(holder) @@ -341,31 +323,27 @@ defmodule DBConnection.Connection do case msg do :checkin -> handle_checkin(state, s) - :disconnect -> handle_event(:cast, {:disconnect, ref, extra, state}, :no_state, s) - :stop -> handle_event(:cast, {:stop, ref, extra, state}, :no_state, s) + :disconnect -> handle_cast({:disconnect, ref, extra, state}, s) + :stop -> handle_cast({:stop, ref, extra, state}, s) end end - # We discard EXIT messages which may arrive if the process is trapping exits - def handle_event(:info, {:EXIT, _, _}, :no_state, s) do - handle_timeout(s) + def handle_info({:EXIT, _, _}, s) do + {:noreply, s} end - def handle_event(:info, msg, :no_state, %{mod: mod} = s) do + def handle_info(msg, %{mod: mod} = s) do Logger.info(fn -> [inspect(mod), ?\s, ?(, Util.inspect_pid(self()), ") missed message: " | inspect(msg)] end) - handle_timeout(s) + {:noreply, s} end - @doc false - @impl :gen_statem - # If client is :closed then the connection was previously disconnected - # and cleanup is not required. - def terminate(_, _, %{client: :closed}), do: :ok + @impl GenServer + def terminate(_reason, %{client: :closed}), do: :ok - def terminate(reason, _, s) do + def terminate(reason, s) do %{mod: mod, state: state} = s msg = "connection exited: " <> Exception.format_exit(reason) @@ -374,16 +352,16 @@ defmodule DBConnection.Connection do |> mod.disconnect(state) end - @doc false - @impl :gen_statem - def format_status(info, [_, :no_state, %{client: :closed, mod: mod}]) do - case info do - :normal -> [{:data, [{~c"Module", mod}]}] - :terminate -> mod - end + @impl GenServer + def format_status(:normal, [_pdict, %{client: :closed, mod: mod}]) do + [{:data, [{~c"Module", mod}]}] + end + + def format_status(:terminate, [_pdict, %{client: :closed, mod: mod}]) do + mod end - def format_status(info, [pdict, :no_state, %{mod: mod, state: state}]) do + def format_status(info, [pdict, %{mod: mod, state: state}]) do case function_exported?(mod, :format_status, 2) do true when info == :normal -> normal_status(mod, pdict, state) @@ -399,8 +377,6 @@ defmodule DBConnection.Connection do end end - ## Helpers - defp maybe_sanitize_exception(e, stack, opts) do if Keyword.get(opts, :show_sensitive_data_on_connection_error, false) do {e, stack} @@ -436,8 +412,6 @@ defmodule DBConnection.Connection do defp down_log({:shutdown, _}), do: :nolog defp down_log(_), do: :log - defp handle_timeout(s), do: {:keep_state, s} - defp demonitor({_, mon}) when is_reference(mon) do Process.demonitor(mon, [:flush]) end @@ -474,6 +448,20 @@ defmodule DBConnection.Connection do end end + defp schedule_backoff(%{backoff: backoff} = s) do + {timeout, backoff} = Backoff.backoff(backoff) + s = cancel_backoff(s) + ref = Process.send_after(self(), :backoff, timeout) + %{s | backoff: backoff, backoff_ref: ref} + end + + defp cancel_backoff(%{backoff_ref: ref} = s) when is_reference(ref) do + _ = Process.cancel_timer(ref) + %{s | backoff_ref: nil} + end + + defp cancel_backoff(s), do: s + defp handle_checkin(state, s) do %{backoff: backoff, client: client} = s backoff = backoff && Backoff.reset(backoff) @@ -484,7 +472,7 @@ defmodule DBConnection.Connection do defp pool_update(state, %{pool: pool, tag: tag, mod: mod, connected_at: connected_at} = s) do case Holder.update(pool, tag, mod, state, connected_at) do {:ok, ref} -> - {:keep_state, %{s | client: {ref, :pool}, state: state}, :hibernate} + {:noreply, %{s | client: {ref, :pool}, state: state}} :error -> {:stop, {:shutdown, :no_more_pool}, s} From 2dcae802a11982be5953ab129a1340082b17ab5d Mon Sep 17 00:00:00 2001 From: KinutaDev Date: Mon, 17 Aug 2026 18:56:17 +0900 Subject: [PATCH 4/5] feat: replace DynamicSupervisor and Task.Supervisor for AtomVM - Change to add DBConnection.DynamicSupervisor, which starts children later via Supervisor - Change to add DBConnection.TaskSupervisor, which starts after_connect processes via Supervisor - Change App, Watcher, and Ownership to start children through those modules --- lib/db_connection/app.ex | 15 ++++------ lib/db_connection/dynamic_supervisor.ex | 34 +++++++++++++++++++++ lib/db_connection/ownership/manager.ex | 2 +- lib/db_connection/task.ex | 2 +- lib/db_connection/task_supervisor.ex | 39 +++++++++++++++++++++++++ lib/db_connection/watcher.ex | 2 +- test/db_connection/watcher_test.exs | 4 +-- 7 files changed, 83 insertions(+), 15 deletions(-) create mode 100644 lib/db_connection/dynamic_supervisor.ex create mode 100644 lib/db_connection/task_supervisor.ex diff --git a/lib/db_connection/app.ex b/lib/db_connection/app.ex index 5bf64664..3927609c 100644 --- a/lib/db_connection/app.ex +++ b/lib/db_connection/app.ex @@ -5,19 +5,14 @@ defmodule DBConnection.App do @impl true def start(_type, _args) do children = [ - {Task.Supervisor, name: DBConnection.Task}, - dynamic_supervisor(DBConnection.Ownership.Supervisor), - dynamic_supervisor(DBConnection.ConnectionPool.Supervisor), + {DBConnection.TaskSupervisor, name: DBConnection.Task}, + {DBConnection.DynamicSupervisor, + name: DBConnection.Ownership.Supervisor, strategy: :one_for_one}, + {DBConnection.DynamicSupervisor, + name: DBConnection.ConnectionPool.Supervisor, strategy: :one_for_one}, DBConnection.Watcher ] Supervisor.start_link(children, strategy: :one_for_all, name: __MODULE__) end - - defp dynamic_supervisor(name) do - Supervisor.child_spec( - {DynamicSupervisor, name: name, strategy: :one_for_one}, - id: name - ) - end end diff --git a/lib/db_connection/dynamic_supervisor.ex b/lib/db_connection/dynamic_supervisor.ex new file mode 100644 index 00000000..a173ecb8 --- /dev/null +++ b/lib/db_connection/dynamic_supervisor.ex @@ -0,0 +1,34 @@ +defmodule DBConnection.DynamicSupervisor do + @moduledoc false + + use Supervisor + + def child_spec(opts) do + name = Keyword.fetch!(opts, :name) + + %{ + id: name, + start: {__MODULE__, :start_link, [opts]}, + type: :supervisor + } + end + + def start_link(opts) do + name = Keyword.fetch!(opts, :name) + Supervisor.start_link(__MODULE__, opts, name: name) + end + + def start_child(supervisor, child_spec) do + spec = Supervisor.child_spec(child_spec, id: make_ref()) + Supervisor.start_child(supervisor, spec) + end + + def which_children(supervisor) do + Supervisor.which_children(supervisor) + end + + @impl true + def init(opts) do + Supervisor.init([], strategy: Keyword.get(opts, :strategy, :one_for_one)) + end +end diff --git a/lib/db_connection/ownership/manager.ex b/lib/db_connection/ownership/manager.ex index a6f31031..cd5f07af 100644 --- a/lib/db_connection/ownership/manager.ex +++ b/lib/db_connection/ownership/manager.ex @@ -258,7 +258,7 @@ defmodule DBConnection.Ownership.Manager do %{pool: pool, checkouts: checkouts, owners: owners, ets: ets, log: log, mode: mode} = state {:ok, proxy} = - DynamicSupervisor.start_child( + DBConnection.DynamicSupervisor.start_child( DBConnection.Ownership.Supervisor, {DBConnection.Ownership.Proxy, {caller, pool, opts}} ) diff --git a/lib/db_connection/task.ex b/lib/db_connection/task.ex index 6d6671f7..acd2e8b7 100644 --- a/lib/db_connection/task.ex +++ b/lib/db_connection/task.ex @@ -6,7 +6,7 @@ defmodule DBConnection.Task do def run_child(mod, state, fun, opts) do arg = [fun, self(), %DBConnection.SensitiveData{data: opts}] - {:ok, pid} = Task.Supervisor.start_child(@name, __MODULE__, :init, arg) + {:ok, pid} = DBConnection.TaskSupervisor.start_child(@name, __MODULE__, :init, arg) ref = Process.monitor(pid) _ = DBConnection.Holder.update(pid, ref, mod, state) {pid, ref} diff --git a/lib/db_connection/task_supervisor.ex b/lib/db_connection/task_supervisor.ex new file mode 100644 index 00000000..ad74618f --- /dev/null +++ b/lib/db_connection/task_supervisor.ex @@ -0,0 +1,39 @@ +defmodule DBConnection.TaskSupervisor do + @moduledoc false + + use Supervisor + + def child_spec(opts) do + name = Keyword.fetch!(opts, :name) + + %{ + id: name, + start: {__MODULE__, :start_link, [opts]}, + type: :supervisor + } + end + + def start_link(opts) do + name = Keyword.fetch!(opts, :name) + Supervisor.start_link(__MODULE__, :ok, name: name) + end + + def start_child(supervisor, module, function_name, args) do + spec = %{ + id: make_ref(), + start: {__MODULE__, :start_task, [module, function_name, args]}, + restart: :temporary + } + + Supervisor.start_child(supervisor, spec) + end + + def start_task(module, function_name, args) do + {:ok, :proc_lib.spawn_link(module, function_name, args)} + end + + @impl true + def init(:ok) do + Supervisor.init([], strategy: :one_for_one) + end +end diff --git a/lib/db_connection/watcher.ex b/lib/db_connection/watcher.ex index 547add0a..eb9761b4 100644 --- a/lib/db_connection/watcher.ex +++ b/lib/db_connection/watcher.ex @@ -24,7 +24,7 @@ defmodule DBConnection.Watcher do @impl true def handle_call({:watch, supervisor, args}, {caller_pid, _ref}, {caller_refs, started_refs}) do - case DynamicSupervisor.start_child(supervisor, args) do + case DBConnection.DynamicSupervisor.start_child(supervisor, args) do {:ok, started_pid} -> Process.link(caller_pid) caller_ref = Process.monitor(caller_pid) diff --git a/test/db_connection/watcher_test.exs b/test/db_connection/watcher_test.exs index f31a718b..be4be399 100644 --- a/test/db_connection/watcher_test.exs +++ b/test/db_connection/watcher_test.exs @@ -29,7 +29,7 @@ defmodule DBConnection.WatcherTest do # Find the pool sup for pool1 by matching the owner pid in child spec ids. pool_sup_ref = DBConnection.ConnectionPool.Supervisor - |> DynamicSupervisor.which_children() + |> DBConnection.DynamicSupervisor.which_children() |> Enum.find_value(fn {_, sup_pid, _, _} -> try do has_pool1? = @@ -54,7 +54,7 @@ defmodule DBConnection.WatcherTest do C.start_link(agent: agent2, pool_size: 1, idle_interval: 100_000) end) - # If the DynamicSupervisor or the watcher is blocked, this hangs + # If the supervisor or the watcher is blocked, this hangs assert {:ok, _pool2} = Task.await(task, 1_000) # Ensure the pool supervisor actually terminates From 553fd57b23a5fc644626c9b8fb2bfd877aade654 Mon Sep 17 00:00:00 2001 From: BlackTanuTanuNinjas Date: Tue, 15 Sep 2026 18:47:27 +0900 Subject: [PATCH 5/5] feat: fold SingleConnection into ConnectionPool and harden Holder - Change to absorb single-connection pool behavior into ConnectionPool - Change to delete DBConnection.SingleConnection and ConnectionPool.Pool - Change to soft-give-away holders with heir reclaim on client down - Change to use relative timers and millisecond monotonic units - Change to log connection stop via :erlang.display instead of error_logger --- lib/db_connection.ex | 2 - lib/db_connection/connection.ex | 8 +- lib/db_connection/connection_pool.ex | 385 +++++------------- lib/db_connection/connection_pool/pool.ex | 48 --- lib/db_connection/holder.ex | 141 +++++-- lib/db_connection/single_connection.ex | 238 ----------- ...tion_test.exs => connection_pool_test.exs} | 4 +- test/test_helper.exs | 4 - 8 files changed, 221 insertions(+), 609 deletions(-) delete mode 100644 lib/db_connection/connection_pool/pool.ex delete mode 100644 lib/db_connection/single_connection.ex rename test/db_connection/{single_connection_test.exs => connection_pool_test.exs} (98%) diff --git a/lib/db_connection.ex b/lib/db_connection.ex index 629958bf..58718943 100644 --- a/lib/db_connection.ex +++ b/lib/db_connection.ex @@ -76,8 +76,6 @@ defmodule DBConnection do DBConnection connections support using different pools via the `:pool` option passed to `start_link/2`. The default pool is `DBConnection.ConnectionPool`. Another supported pool that is commonly used for tests is `DBConnection.Ownership`. - `DBConnection.SingleConnection` is a one-connection pool without CoDel - queueing or `DynamicSupervisor`, intended for constrained runtimes. For now, using *custom* pools is not supported since the API for pools is not public. diff --git a/lib/db_connection/connection.ex b/lib/db_connection/connection.ex index 55fb28a3..555fe578 100644 --- a/lib/db_connection/connection.ex +++ b/lib/db_connection/connection.ex @@ -207,12 +207,7 @@ defmodule DBConnection.Connection do _other -> "** #{inspect(err)}" end - format = - ~c"** State machine ~p terminating~n" ++ - ~c"** Reason for termination ==~n" ++ - ~c"~s~n" - - :error_logger.format(format, [self(), reason]) + :erlang.display({:db_connection_terminating, self(), reason}) end {:stop, {err, stack}, %{s | state: state}} @@ -472,6 +467,7 @@ defmodule DBConnection.Connection do defp pool_update(state, %{pool: pool, tag: tag, mod: mod, connected_at: connected_at} = s) do case Holder.update(pool, tag, mod, state, connected_at) do {:ok, ref} -> + # Omit :hibernate — AtomVM gen_server does not support it. {:noreply, %{s | client: {ref, :pool}, state: state}} :error -> diff --git a/lib/db_connection/connection_pool.ex b/lib/db_connection/connection_pool.ex index 989a9e13..8ee12d4c 100644 --- a/lib/db_connection/connection_pool.ex +++ b/lib/db_connection/connection_pool.ex @@ -1,8 +1,9 @@ defmodule DBConnection.ConnectionPool do @moduledoc """ - The default connection pool. + The default connection pool for AtomVM. - The queueing algorithm is based on [CoDel](https://queue.acm.org/appendices/codel.html). + Keeps exactly one connection. Does not use CoDel queueing, idle polling, + or a nested connection supervisor. You're not supposed to call any functions on this pool directly, but only pass this as the value of the `:pool` option in functions such as `DBConnection.start_link/2`. @@ -14,12 +15,6 @@ defmodule DBConnection.ConnectionPool do @behaviour DBConnection.Pool - @queue_target 50 - @queue_interval 2000 - @idle_interval 1000 - @time_unit 1000 - @watcher_ref {__MODULE__, :watcher_ref} - @doc false def start_link({mod, opts}) do GenServer.start_link(__MODULE__, {mod, opts}, start_opts(opts)) @@ -43,150 +38,105 @@ defmodule DBConnection.ConnectionPool do GenServer.call(pool, :get_connection_metrics, :infinity) end - ## GenServer api - @impl GenServer def init({mod, opts}) do Process.flag(:trap_exit, true) DBConnection.register_as_pool(mod) - queue = :ets.new(__MODULE__.Queue, [:protected, :ordered_set, decentralized_counters: true]) - - max_lifetime = - case Keyword.fetch(opts, :max_lifetime) do - {:ok, %Range{first: first, last: last, step: 1}} when first >= 0 and last >= first -> - {System.convert_time_unit(first, :millisecond, :native), last - first} - - {:ok, invalid} -> - raise ArgumentError, - "invalid value for :max_lifetime, expected a non-negative step-1 range, got: #{inspect(invalid)}" - - :error -> - nil - end - - ts = {nil, max_lifetime} - {:ok, watcher_ref} = DBConnection.ConnectionPool.Pool.start_supervised(queue, mod, opts) - Process.put(@watcher_ref, watcher_ref) - target = Keyword.get(opts, :queue_target, @queue_target) - interval = Keyword.get(opts, :queue_interval, @queue_interval) - idle_interval = Keyword.get(opts, :idle_interval, @idle_interval) - idle_limit = Keyword.get_lazy(opts, :idle_limit, fn -> Keyword.get(opts, :pool_size, 1) end) - now_in_native = System.monotonic_time() - now_in_ms = System.convert_time_unit(now_in_native, :native, @time_unit) - - codel = %{ - target: target, - interval: interval, - delay: 0, - slow: false, - next: now_in_ms, - poll: nil, - idle_interval: idle_interval, - idle_limit: idle_limit, - idle: nil + size = Keyword.get(opts, :pool_size, 1) + + if size < 1 do + raise ArgumentError, "pool size must be greater or equal to 1, got #{size}" + end + + tag = make_ref() + + state = %{ + mod: mod, + opts: opts, + tag: tag, + holder: nil, + out: nil, + waiters: :queue.new(), + ts: {nil, max_lifetime(opts)}, + conn: nil, + restarts: [], + max_restarts: Keyword.get(opts, :max_restarts, 3), + max_seconds: Keyword.get(opts, :max_seconds, 5) } - codel = start_idle(now_in_native, start_poll(now_in_ms, now_in_ms, codel)) - {:ok, {:busy, queue, codel, ts}} + {:ok, pid} = start_connection(state) + {:ok, %{state | conn: pid}} end @impl GenServer - def handle_call(:get_connection_metrics, _from, {status, queue, _, _} = state) do - {ready_conn_count, checkout_queue_length} = - case status do - :busy -> - {0, :ets.select_count(queue, [{{{:_, :_, :_}}, [], [true]}])} - - :ready -> - {:ets.select_count(queue, [{{{:_, :_}}, [], [true]}]), 0} - end - + def handle_call(:get_connection_metrics, _from, state) do metrics = %{ source: {:pool, self()}, - ready_conn_count: ready_conn_count, - checkout_queue_length: checkout_queue_length + ready_conn_count: if(state.holder, do: 1, else: 0), + checkout_queue_length: :queue.len(state.waiters) } {:reply, [metrics], state} end - def handle_call({:disconnect_all, interval}, _from, {type, queue, codel, ts}) do - {_, max_lifetime} = ts + def handle_call({:disconnect_all, interval}, _from, state) do + {_, max_lifetime} = state.ts ts = {{System.monotonic_time(), interval}, max_lifetime} - {:reply, :ok, {type, queue, codel, ts}} + {:reply, :ok, %{state | ts: ts}} end @impl GenServer - def handle_info( - {:db_connection, from, {:checkout, _caller, now, queue?}}, - {:busy, queue, _, _} = busy - ) do - case queue? do - true -> - :ets.insert(queue, {{now, System.unique_integer(), from}}) - {:noreply, busy} - - false -> - message = "connection not available and queuing is disabled" - err = DBConnection.ConnectionError.exception(message) - Holder.reply_error(from, err) - {:noreply, busy} - end + def handle_info({:db_connection, from, {:checkout, _callers, _now, queue?}}, state) do + {:noreply, checkout_request(from, queue?, state)} end - def handle_info( - {:db_connection, from, {:checkout, _caller, _now, _queue?}} = checkout, - {:ready, queue, _codel, _ts} = ready - ) do - case :ets.first(queue) do - {queued_in_native, holder} = key -> - Holder.handle_checkout(holder, from, queue, queued_in_native) and :ets.delete(queue, key) - {:noreply, ready} - - :"$end_of_table" -> - handle_info(checkout, put_elem(ready, 0, :busy)) - end - end - - def handle_info({:"ETS-TRANSFER", holder, pid, queue}, {_, queue, _, _} = data) do - message = "client #{Util.inspect_pid(pid)} exited" - err = DBConnection.ConnectionError.exception(message: message, severity: :info) - Holder.handle_disconnect(holder, err) - {:noreply, data} - end - - def handle_info({:"ETS-TRANSFER", holder, _, {msg, queue, extra}}, {_, queue, _, ts} = data) do + def handle_info({:"ETS-TRANSFER", holder, _pid, {msg, tag, extra}}, %{tag: tag} = state) do case msg do :checkin -> owner = self() + _ = Holder.unwatch_client(holder) case Holder.owner(holder) do ^owner -> - {interval, max_lifetime} = ts + {interval, max_lifetime} = state.ts if Holder.maybe_disconnect(holder, interval, max_lifetime) do - {:noreply, data} + {:noreply, %{state | holder: nil, out: nil}} else - handle_checkin(holder, extra, data) + handle_checkin(holder, extra, %{state | out: nil}) end :undefined -> - {:noreply, data} + {:noreply, %{state | out: nil}} end :disconnect -> + _ = Holder.unwatch_client(holder) Holder.handle_disconnect(holder, extra) - {:noreply, data} + {:noreply, %{state | holder: nil, out: nil}} :stop -> + _ = Holder.unwatch_client(holder) Holder.handle_stop(holder, extra) - {:noreply, data} + {:noreply, %{state | holder: nil, out: nil}} end end - def handle_info({:timeout, deadline, {queue, holder, pid, len}}, {_, queue, _, _} = data) do - # Check that timeout refers to current holder (and not previous) + def handle_info({:"ETS-TRANSFER", holder, pid, tag}, %{tag: tag} = state) do + _ = Holder.unwatch_client(holder) + message = "client #{Util.inspect_pid(pid)} exited" + err = DBConnection.ConnectionError.exception(message: message, severity: :info) + Holder.handle_disconnect(holder, err) + {:noreply, %{state | holder: nil, out: nil}} + end + + def handle_info({:DOWN, mon, :process, pid, _reason}, %{out: {holder, mon}} = state) do + _ = Holder.reclaim_on_client_down(holder, pid) + {:noreply, %{state | out: nil}} + end + + def handle_info({:timeout, deadline, {tag, holder, pid, len}}, %{tag: tag} = state) do if Holder.handle_deadline(holder, deadline) do message = "client #{Util.inspect_pid(pid)} timed out because " <> @@ -207,29 +157,11 @@ defmodule DBConnection.ConnectionPool do Holder.handle_disconnect(holder, exc) end - {:noreply, data} + {:noreply, state} end - def handle_info({:timeout, poll, {time, last_sent}}, {_, _, %{poll: poll}, _} = data) do - {status, queue, codel, ts} = data - - # If no queue progress since last poll check queue - case :ets.first(queue) do - {sent, _, _} when sent <= last_sent and status == :busy -> - delay = time - sent - timeout(delay, time, queue, start_poll(time, sent, codel), ts) - - {sent, _, _} -> - {:noreply, {status, queue, start_poll(time, sent, codel), ts}} - - _ -> - {:noreply, {status, queue, start_poll(time, time, codel), ts}} - end - end - - def handle_info({:timeout, idle, past_in_native}, {_, _, %{idle: idle}, _} = data) do - {status, queue, %{idle_limit: limit} = codel, ts} = data - drop_idle(past_in_native, limit, status, queue, codel, ts) + def handle_info({:EXIT, pid, reason}, %{conn: pid} = state) do + restart_connection(reason, state) end def handle_info({:EXIT, _pid, reason}, state) do @@ -237,182 +169,83 @@ defmodule DBConnection.ConnectionPool do end @impl GenServer - def terminate(_reason, _state) do - if watcher_ref = Process.get(@watcher_ref) do - DBConnection.ConnectionPool.Pool.stop_supervised(watcher_ref) - end - + def terminate(_reason, %{conn: pid}) when is_pid(pid) do + Process.exit(pid, :shutdown) :ok end - defp drop_idle(past_in_native, limit, status, queue, codel, ts) do - with true <- status == :ready and limit > 0, - {queued_in_native, holder} = key when queued_in_native <= past_in_native <- - :ets.first(queue) do - :ets.delete(queue, key) - {interval, max_lifetime} = ts - Holder.maybe_disconnect(holder, interval, max_lifetime) or Holder.handle_ping(holder) - drop_idle(past_in_native, limit - 1, status, queue, codel, ts) - else - _ -> - {:noreply, {status, queue, start_idle(System.monotonic_time(), codel), ts}} - end - end + def terminate(_reason, _state), do: :ok - defp timeout(delay, time, queue, codel, ts) do - case codel do - %{delay: min_delay, next: next, target: target, interval: interval} - when time >= next and min_delay > target -> - codel = %{codel | slow: true, delay: delay, next: time + interval} - drop_slow(time, target * 2, queue) - {:noreply, {:busy, queue, codel, ts}} - - %{next: next, interval: interval} when time >= next -> - codel = %{codel | slow: false, delay: delay, next: time + interval} - {:noreply, {:busy, queue, codel, ts}} - - _ -> - {:noreply, {:busy, queue, codel, ts}} + defp checkout_request(from, queue?, %{holder: {holder, checked_in_at}} = state) do + if Holder.handle_checkout(holder, from, state.tag, checked_in_at) do + {client, _} = from + mon = Holder.watch_client(holder, client) + %{state | holder: nil, out: {holder, mon}} + else + checkout_request(from, queue?, %{state | holder: nil}) end end - defp drop_slow(time, timeout, queue) do - min_sent = time - timeout - match = {{:"$1", :_, :"$2"}} - guards = [{:<, :"$1", min_sent}] - select_slow = [{match, guards, [{{:"$1", :"$2"}}]}] - - for {sent, from} <- :ets.select(queue, select_slow) do - drop(time - sent, from) - end - - :ets.select_delete(queue, [{match, guards, [true]}]) + defp checkout_request(from, true, state) do + %{state | waiters: :queue.in(from, state.waiters)} end - defp handle_checkin(holder, now_in_native, {:ready, queue, _, _} = data) do - :ets.insert(queue, {{now_in_native, holder}}) - {:noreply, data} + defp checkout_request(from, false, state) do + message = "connection not available and queuing is disabled" + Holder.reply_error(from, DBConnection.ConnectionError.exception(message)) + state end - defp handle_checkin(holder, now_in_native, {:busy, queue, codel, ts}) do - now_in_ms = System.convert_time_unit(now_in_native, :native, @time_unit) - - case dequeue(now_in_ms, holder, queue, codel, ts) do - {:busy, _, _, _} = busy -> - {:noreply, busy} + defp handle_checkin(holder, now_in_native, state) do + case :queue.out(state.waiters) do + {:empty, waiters} -> + {:noreply, %{state | holder: {holder, now_in_native}, waiters: waiters}} - {:ready, _, _, _} = ready -> - :ets.insert(queue, {{now_in_native, holder}}) - {:noreply, ready} - end - end + {{:value, from}, waiters} -> + state = %{state | waiters: waiters} - defp dequeue(time, holder, queue, codel, ts) do - case codel do - %{next: next, delay: delay, target: target} when time >= next -> - dequeue_first(time, delay > target, holder, queue, codel, ts) - - %{slow: false} -> - dequeue_fast(time, holder, queue, codel, ts) - - %{slow: true, target: target} -> - dequeue_slow(time, target * 2, holder, queue, codel, ts) - end - end - - defp dequeue_first(time, slow?, holder, queue, codel, ts) do - %{interval: interval} = codel - next = time + interval - - case :ets.first(queue) do - {sent, _, from} = key -> - :ets.delete(queue, key) - delay = time - sent - codel = %{codel | next: next, delay: delay, slow: slow?} - go(delay, from, time, holder, queue, codel, ts) - - :"$end_of_table" -> - codel = %{codel | next: next, delay: 0, slow: slow?} - {:ready, queue, codel, ts} + if Holder.handle_checkout(holder, from, state.tag, now_in_native) do + {client, _} = from + mon = Holder.watch_client(holder, client) + {:noreply, %{state | holder: nil, out: {holder, mon}}} + else + handle_checkin(holder, now_in_native, state) + end end end - defp dequeue_fast(time, holder, queue, codel, ts) do - case :ets.first(queue) do - {sent, _, from} = key -> - :ets.delete(queue, key) - go(time - sent, from, time, holder, queue, codel, ts) + defp restart_connection(reason, state) do + now = System.monotonic_time(:second) + cutoff = now - state.max_seconds + restarts = Enum.filter(state.restarts, &(&1 >= cutoff)) - :"$end_of_table" -> - {:ready, queue, %{codel | delay: 0}, ts} + if length(restarts) >= state.max_restarts do + {:stop, reason, state} + else + {:ok, pid} = start_connection(state) + {:noreply, %{state | conn: pid, holder: nil, out: nil, restarts: [now | restarts]}} end end - defp dequeue_slow(time, timeout, holder, queue, codel, ts) do - case :ets.first(queue) do - {sent, _, from} = key when time - sent > timeout -> - :ets.delete(queue, key) - drop(time - sent, from) - dequeue_slow(time, timeout, holder, queue, codel, ts) - - {sent, _, from} = key -> - :ets.delete(queue, key) - go(time - sent, from, time, holder, queue, codel, ts) - - :"$end_of_table" -> - {:ready, queue, %{codel | delay: 0}, ts} - end + defp start_connection(%{mod: mod, opts: opts, tag: tag}) do + DBConnection.Connection.start_link(mod, Keyword.put(opts, :pool_index, 1), self(), tag) end - defp go(delay, from, time, holder, queue, %{delay: min} = codel, ts) do - case Holder.handle_checkout(holder, from, queue, 0) do - true when delay < min -> - {:busy, queue, %{codel | delay: delay}, ts} + defp max_lifetime(opts) do + case Keyword.fetch(opts, :max_lifetime) do + {:ok, %Range{first: first, last: last, step: 1}} when first >= 0 and last >= first -> + {System.convert_time_unit(first, :millisecond, :native), last - first} - true -> - {:busy, queue, codel, ts} + {:ok, invalid} -> + raise ArgumentError, + "invalid value for :max_lifetime, expected a non-negative step-1 range, got: #{inspect(invalid)}" - false -> - dequeue(time, holder, queue, codel, ts) + :error -> + nil end end - defp drop(delay, from) do - message = """ - [#{ancestor()}] connection not available and request was dropped from queue after #{delay}ms. \ - This means requests are coming in and your connection pool cannot serve them fast enough. \ - You can address this by: - - 1. Ensuring your database is available and that you can connect to it - 2. Tracking down slow queries and making sure they are running fast enough - 3. Increasing the pool_size (although this increases resource consumption) - 4. Allowing requests to wait longer by increasing :queue_target and :queue_interval - - See DBConnection.start_link/2 for more information - """ - - err = DBConnection.ConnectionError.exception(message, :queue_timeout) - - Holder.reply_error(from, err) - end - - defp ancestor do - Process.get(:"$ancestors", []) |> Enum.find(&is_atom/1) - end - defp start_opts(opts) do Keyword.take(opts, [:name, :spawn_opt]) end - - defp start_poll(now, last_sent, %{interval: interval} = codel) do - timeout = now + interval - poll = :erlang.start_timer(timeout, self(), {timeout, last_sent}, abs: true) - %{codel | poll: poll} - end - - defp start_idle(now_in_native, %{idle_interval: interval} = codel) do - timeout = System.convert_time_unit(now_in_native, :native, :millisecond) + interval - idle = :erlang.start_timer(timeout, self(), now_in_native, abs: true) - %{codel | idle: idle} - end end diff --git a/lib/db_connection/connection_pool/pool.ex b/lib/db_connection/connection_pool/pool.ex deleted file mode 100644 index 8da2a0d3..00000000 --- a/lib/db_connection/connection_pool/pool.ex +++ /dev/null @@ -1,48 +0,0 @@ -defmodule DBConnection.ConnectionPool.Pool do - @moduledoc false - use Supervisor, restart: :temporary - - def start_supervised(tag, mod, opts) do - DBConnection.Watcher.watch( - DBConnection.ConnectionPool.Supervisor, - {DBConnection.ConnectionPool.Pool, {self(), tag, mod, opts}} - ) - end - - def stop_supervised(ref) do - case DBConnection.Watcher.unwatch(ref) do - pid when is_pid(pid) -> wait_until_down(pid) - nil -> :ok - end - end - - def start_link(arg) do - Supervisor.start_link(__MODULE__, arg) - end - - @impl true - def init({owner, tag, mod, opts}) do - size = Keyword.get(opts, :pool_size, 1) - - if size < 1, do: raise(ArgumentError, "pool size must be greater or equal to 1, got #{size}") - - children = for id <- 1..size, do: conn(owner, tag, id, mod, opts) - sup_opts = [strategy: :one_for_one] ++ Keyword.take(opts, [:max_restarts, :max_seconds]) - Supervisor.init(children, sup_opts) - end - - ## Helpers - - defp conn(owner, tag, id, mod, opts) do - child_opts = [id: {mod, owner, id}] ++ Keyword.take(opts, [:shutdown]) - DBConnection.Connection.child_spec(mod, [pool_index: id] ++ opts, owner, tag, child_opts) - end - - defp wait_until_down(pid) do - ref = Process.monitor(pid) - - receive do - {:DOWN, ^ref, _, _, _} -> :ok - end - end -end diff --git a/lib/db_connection/holder.ex b/lib/db_connection/holder.ex index 8a31a769..6351f450 100644 --- a/lib/db_connection/holder.ex +++ b/lib/db_connection/holder.ex @@ -6,7 +6,9 @@ defmodule DBConnection.Holder do @queue true @timeout 15000 - @time_unit 1000 + # AtomVM System.monotonic_time/1 only accepts atom units (:millisecond etc.), + # not integer "parts per second" like 1000. + @time_unit :millisecond Record.defrecord(:conn, [ :connection, @@ -16,7 +18,10 @@ defmodule DBConnection.Holder do :connected_at, deadline: nil, status: :ok, - owner: nil + owner: nil, + heir_pid: nil, + heir_data: nil, + client_mon: nil ]) Record.defrecord(:pool_ref, [:pool, :reference, :deadline, :holder, :lock]) @@ -39,7 +44,9 @@ defmodule DBConnection.Holder do module: mod, state: state, connected_at: connected_at, - owner: self() + owner: self(), + heir_pid: pool, + heir_data: ref ) true = :ets.insert_new(holder, conn) @@ -218,15 +225,63 @@ defmodule DBConnection.Holder do @spec owner(t) :: pid | :undefined def owner(holder) do - if function_exported?(:ets, :info, 2) do - :ets.info(holder, :owner) - else - try do - :ets.lookup_element(holder, :conn, conn(:owner) + 1) - rescue - ArgumentError -> :undefined - end + soft_owner(holder) + end + + @doc false + @spec watch_client(t, pid) :: reference | nil + def watch_client(holder, client_pid) when is_pid(client_pid) do + mon = Process.monitor(client_pid) + true = :ets.update_element(holder, :conn, {conn(:client_mon) + 1, mon}) + mon + rescue + ArgumentError -> nil + end + + @doc false + @spec unwatch_client(t) :: :ok + def unwatch_client(holder) do + case soft_client_mon(holder) do + mon when is_reference(mon) -> + Process.demonitor(mon, [:flush]) + _ = :ets.update_element(holder, :conn, {conn(:client_mon) + 1, nil}) + :ok + + _ -> + :ok end + rescue + ArgumentError -> :ok + end + + @doc false + @spec reclaim_on_client_down(t, pid) :: boolean + def reclaim_on_client_down(holder, client_pid) when is_pid(client_pid) do + case :ets.lookup(holder, :conn) do + [ + conn( + owner: ^client_pid, + heir_pid: heir_pid, + heir_data: heir_data, + client_mon: mon + ) + ] + when is_pid(heir_pid) -> + if is_reference(mon), do: Process.demonitor(mon, [:flush]) + _ = :ets.update_element(holder, :conn, {conn(:client_mon) + 1, nil}) + + if put_owner(holder, heir_pid) do + send(heir_pid, {:"ETS-TRANSFER", holder, client_pid, heir_data}) + true + else + false + end + + _ -> + false + end + rescue + ArgumentError -> false end @spec handle_checkout(t, {pid, reference}, reference, checkin_time) :: boolean @@ -486,9 +541,9 @@ defmodule DBConnection.Holder do end defp start_deadline(timeout, pid, ref, holder, start) do - deadline = - :erlang.start_timer(timeout, pid, {ref, holder, self(), timeout - start}, abs: true) - + now = System.monotonic_time(@time_unit) + delay = max(timeout - now, 0) + deadline = :erlang.start_timer(delay, pid, {ref, holder, self(), timeout - start}) {deadline, [{conn(:deadline) + 1, deadline}]} end @@ -497,7 +552,8 @@ defmodule DBConnection.Holder do end defp cancel_deadline(deadline) do - :erlang.cancel_timer(deadline, async: true, info: false) + _ = :erlang.cancel_timer(deadline) + :ok end defp hash_holder(_holder, 0), do: 0 @@ -508,21 +564,32 @@ defmodule DBConnection.Holder do System.convert_time_unit(hash, :millisecond, :native) end - defp maybe_set_heir(holder, pool, ref) do - if function_exported?(:ets, :setopts, 2) do - :ets.setopts(holder, {:heir, pool, ref}) - else - :ok - end - end + defp maybe_set_heir(_holder, _pool, _ref), do: :ok defp transfer(table, pid, gift_data) do - if function_exported?(:ets, :give_away, 3) do - :ets.give_away(table, pid, gift_data) - else - _ = put_owner(table, pid) - send(pid, {:"ETS-TRANSFER", table, self(), gift_data}) - true + soft_give_away(table, pid, gift_data) + end + + # Approximate :ets.give_away/3 failure conditions so checkout/checkin + # rescue paths behave like OTP when give_away is missing. + defp soft_give_away(table, pid, gift_data) do + cond do + not is_pid(pid) or not Process.alive?(pid) -> + raise ArgumentError + + pid == self() -> + raise ArgumentError + + soft_owner(table) != self() -> + raise ArgumentError + + true -> + if put_owner(table, pid) do + send(pid, {:"ETS-TRANSFER", table, self(), gift_data}) + true + else + raise ArgumentError + end end end @@ -532,11 +599,19 @@ defmodule DBConnection.Holder do ArgumentError -> false end + defp soft_owner(holder) do + :ets.lookup_element(holder, :conn, conn(:owner) + 1) + rescue + ArgumentError -> :undefined + end + + defp soft_client_mon(holder) do + :ets.lookup_element(holder, :conn, conn(:client_mon) + 1) + rescue + ArgumentError -> nil + end + defp ets_info(holder) do - if function_exported?(:ets, :info, 1) do - :ets.info(holder) - else - %{owner: owner(holder)} - end + %{owner: owner(holder)} end end diff --git a/lib/db_connection/single_connection.ex b/lib/db_connection/single_connection.ex deleted file mode 100644 index d541a4f0..00000000 --- a/lib/db_connection/single_connection.ex +++ /dev/null @@ -1,238 +0,0 @@ -defmodule DBConnection.SingleConnection do - @moduledoc """ - A single-connection pool for constrained runtimes. - - Unlike `DBConnection.ConnectionPool`, this pool keeps exactly one connection, - does not use CoDel queueing, idle polling, or `DynamicSupervisor`. - - Pass this module as the `:pool` option to `DBConnection.start_link/2`. - The default pool remains `DBConnection.ConnectionPool`. `:pool_size` is - ignored except that values below 1 raise, same as the default pool. - """ - - use GenServer - alias DBConnection.Holder - alias DBConnection.Util - - @behaviour DBConnection.Pool - - @doc false - def start_link({mod, opts}) do - GenServer.start_link(__MODULE__, {mod, opts}, start_opts(opts)) - end - - @doc false - @impl DBConnection.Pool - def checkout(pool, callers, opts) do - Holder.checkout(pool, callers, opts) - end - - @doc false - @impl DBConnection.Pool - def disconnect_all(pool, interval, _opts) do - GenServer.call(pool, {:disconnect_all, interval}, :infinity) - end - - @doc false - @impl DBConnection.Pool - def get_connection_metrics(pool) do - GenServer.call(pool, :get_connection_metrics, :infinity) - end - - @impl GenServer - def init({mod, opts}) do - Process.flag(:trap_exit, true) - DBConnection.register_as_pool(mod) - - size = Keyword.get(opts, :pool_size, 1) - - if size < 1 do - raise ArgumentError, "pool size must be greater or equal to 1, got #{size}" - end - - tag = make_ref() - - state = %{ - mod: mod, - opts: opts, - tag: tag, - holder: nil, - waiters: :queue.new(), - ts: {nil, max_lifetime(opts)}, - conn: nil, - restarts: [], - max_restarts: Keyword.get(opts, :max_restarts, 3), - max_seconds: Keyword.get(opts, :max_seconds, 5) - } - - {:ok, pid} = start_connection(state) - {:ok, %{state | conn: pid}} - end - - @impl GenServer - def handle_call(:get_connection_metrics, _from, state) do - metrics = %{ - source: {:pool, self()}, - ready_conn_count: if(state.holder, do: 1, else: 0), - checkout_queue_length: :queue.len(state.waiters) - } - - {:reply, [metrics], state} - end - - def handle_call({:disconnect_all, interval}, _from, state) do - {_, max_lifetime} = state.ts - ts = {{System.monotonic_time(), interval}, max_lifetime} - {:reply, :ok, %{state | ts: ts}} - end - - @impl GenServer - def handle_info({:db_connection, from, {:checkout, _callers, _now, queue?}}, state) do - {:noreply, checkout_request(from, queue?, state)} - end - - def handle_info({:"ETS-TRANSFER", holder, _pid, {msg, tag, extra}}, %{tag: tag} = state) do - case msg do - :checkin -> - owner = self() - - case Holder.owner(holder) do - ^owner -> - {interval, max_lifetime} = state.ts - - if Holder.maybe_disconnect(holder, interval, max_lifetime) do - {:noreply, %{state | holder: nil}} - else - handle_checkin(holder, extra, state) - end - - :undefined -> - {:noreply, state} - end - - :disconnect -> - Holder.handle_disconnect(holder, extra) - {:noreply, %{state | holder: nil}} - - :stop -> - Holder.handle_stop(holder, extra) - {:noreply, %{state | holder: nil}} - end - end - - def handle_info({:"ETS-TRANSFER", holder, pid, tag}, %{tag: tag} = state) do - message = "client #{Util.inspect_pid(pid)} exited" - err = DBConnection.ConnectionError.exception(message: message, severity: :info) - Holder.handle_disconnect(holder, err) - {:noreply, %{state | holder: nil}} - end - - def handle_info({:timeout, deadline, {tag, holder, pid, len}}, %{tag: tag} = state) do - if Holder.handle_deadline(holder, deadline) do - message = - "client #{Util.inspect_pid(pid)} timed out because " <> - "it queued and checked out the connection for longer than #{len}ms" - - exc = - case Process.info(pid, :current_stacktrace) do - {:current_stacktrace, stacktrace} -> - message <> - "\n\n#{Util.inspect_pid(pid)} was at location:\n\n" <> - Exception.format_stacktrace(stacktrace) - - _ -> - message - end - |> DBConnection.ConnectionError.exception() - - Holder.handle_disconnect(holder, exc) - end - - {:noreply, state} - end - - def handle_info({:EXIT, pid, reason}, %{conn: pid} = state) do - restart_connection(reason, state) - end - - def handle_info({:EXIT, _pid, reason}, state) do - {:stop, reason, state} - end - - @impl GenServer - def terminate(_reason, %{conn: pid}) when is_pid(pid) do - Process.exit(pid, :shutdown) - :ok - end - - def terminate(_reason, _state), do: :ok - - defp checkout_request(from, queue?, %{holder: {holder, checked_in_at}} = state) do - if Holder.handle_checkout(holder, from, state.tag, checked_in_at) do - %{state | holder: nil} - else - checkout_request(from, queue?, %{state | holder: nil}) - end - end - - defp checkout_request(from, true, state) do - %{state | waiters: :queue.in(from, state.waiters)} - end - - defp checkout_request(from, false, state) do - message = "connection not available and queuing is disabled" - Holder.reply_error(from, DBConnection.ConnectionError.exception(message)) - state - end - - defp handle_checkin(holder, now_in_native, state) do - case :queue.out(state.waiters) do - {:empty, waiters} -> - {:noreply, %{state | holder: {holder, now_in_native}, waiters: waiters}} - - {{:value, from}, waiters} -> - state = %{state | waiters: waiters} - - if Holder.handle_checkout(holder, from, state.tag, now_in_native) do - {:noreply, %{state | holder: nil}} - else - handle_checkin(holder, now_in_native, state) - end - end - end - - defp restart_connection(reason, state) do - now = System.monotonic_time(:second) - cutoff = now - state.max_seconds - restarts = Enum.filter(state.restarts, &(&1 >= cutoff)) - - if length(restarts) >= state.max_restarts do - {:stop, reason, state} - else - {:ok, pid} = start_connection(state) - {:noreply, %{state | conn: pid, holder: nil, restarts: [now | restarts]}} - end - end - - defp start_connection(%{mod: mod, opts: opts, tag: tag}) do - DBConnection.Connection.start_link(mod, Keyword.put(opts, :pool_index, 1), self(), tag) - end - - defp max_lifetime(opts) do - case Keyword.fetch(opts, :max_lifetime) do - {:ok, %Range{first: first, last: last, step: 1}} when first >= 0 and last >= first -> - {System.convert_time_unit(first, :millisecond, :native), last - first} - - {:ok, invalid} -> - raise ArgumentError, - "invalid value for :max_lifetime, expected a non-negative step-1 range, got: #{inspect(invalid)}" - - :error -> - nil - end - end - - defp start_opts(opts) do - Keyword.take(opts, [:name, :spawn_opt]) - end -end diff --git a/test/db_connection/single_connection_test.exs b/test/db_connection/connection_pool_test.exs similarity index 98% rename from test/db_connection/single_connection_test.exs rename to test/db_connection/connection_pool_test.exs index 35c2a1b0..8af32de0 100644 --- a/test/db_connection/single_connection_test.exs +++ b/test/db_connection/connection_pool_test.exs @@ -1,7 +1,7 @@ -defmodule DBConnection.SingleConnectionTest do +defmodule DBConnection.ConnectionPoolTest do use ExUnit.Case, async: true - alias TestSingleConnection, as: P + alias TestPool, as: P alias TestAgent, as: A alias TestQuery, as: Q alias TestResult, as: R diff --git a/test/test_helper.exs b/test/test_helper.exs index 1c6ee2ee..989768b2 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -5,7 +5,3 @@ Code.require_file("test_support.exs", __DIR__) defmodule TestPool do use TestConnection, pool: DBConnection.ConnectionPool, pool_size: 1 end - -defmodule TestSingleConnection do - use TestConnection, pool: DBConnection.SingleConnection, pool_size: 1 -end