11-module (roadrunner_httparena_db ).
22
3- -export ([ start_pool / 0 , query / 2 ] ).
3+ -behaviour ( ecpool_worker ).
44
5- % % Connections live in a persistent_term tuple, addressed lock-free per
6- % % request (no central checkout broker, so nothing serializes the pool).
7- -define (CONNS_KEY , {? MODULE , conns }).
5+ -export ([start_pool /0 , connect /1 , query /2 ]).
6+
7+ % % `connect/1` is the ecpool_worker callback, invoked by ecpool via the
8+ % % `Mod:connect/1` MFA when it opens (or reopens) a pooled connection, not
9+ % % by a static call, so xref cannot see the reference.
10+ -ignore_xref ([{connect , 1 }]).
11+
12+ -define (POOL , httparena_pg ).
813
914% % Every SQL the adapter runs, keyed by an atom name. Each statement is
10- % % parsed once per connection at connect time (named after its key), so the
11- % % hot path runs `epgsql:prepared_query/3` against a cached `#statement{}`
12- % % record instead of re-parsing on every request.
15+ % % parsed once per pooled connection at connect time (named after its key),
16+ % % so the hot path runs `epgsql:prepared_query/3` against a cached
17+ % % `#statement{}` record instead of re-parsing on every request.
1318statements () ->
1419 #{
1520 read => ~ """
@@ -51,32 +56,42 @@ statements() ->
5156 fortunes => ~ " SELECT id, message FROM fortune"
5257 }.
5358
59+ % % Start an ecpool pool of `pool_size()` connections. ecpool routes each
60+ % % request to a worker via `gproc_pool` (an ETS lookup, no central checkout
61+ % % broker), so concurrent callers spread across connections instead of
62+ % % serializing through one coordinator. `auto_reconnect` makes a worker
63+ % % reopen (and re-prepare) its connection if the backend drops it.
5464-spec start_pool () -> ok | disabled .
5565start_pool () ->
5666 case os :getenv (" DATABASE_URL" of
5767 false ->
5868 disabled ;
5969 Url ->
6070 ConnMap = parse_url (Url ),
61- Size = pool_size (),
62- Conns = list_to_tuple ([connect (ConnMap ) || _ <- lists :seq (1 , Size )]),
63- persistent_term :put (? CONNS_KEY , Conns ),
71+ Opts = [
72+ {pool_size , pool_size ()},
73+ {pool_type , random },
74+ {auto_reconnect , 2 },
75+ {conn_map , ConnMap }
76+ ],
77+ {ok , _ } = ecpool :start_sup_pool (? POOL , ? MODULE , Opts ),
6478 ok
6579 end .
6680
67- % % Open a connection and parse every statement on it (the prepared
68- % % statement must exist on each backend connection). The resulting
69- % % `#statement{}` metadata is connection-independent, so we cache one per
70- % % name in `persistent_term`; concurrent connects racing to put the same
71- % % value is harmless.
72- -spec connect (epgsql :connect_opts ()) -> epgsql :connection ().
73- connect (ConnMap ) ->
81+ % % ecpool_worker callback: open a connection and parse every statement on it
82+ % % (the prepared statement must exist on each backend connection). The
83+ % % resulting `#statement{}` metadata is connection-independent, so we cache
84+ % % one per name in `persistent_term`; connections racing to put the same
85+ % % value (including after a reconnect) is harmless.
86+ -spec connect ([term ()]) -> {ok , epgsql :connection ()}.
87+ connect (Opts ) ->
88+ {conn_map , ConnMap } = lists :keyfind (conn_map , 1 , Opts ),
7489 {ok , Conn } = epgsql :connect (ConnMap ),
7590 ok = maps :foreach (
7691 fun (Name , Sql ) -> prepare (Conn , Name , Sql ) end ,
7792 statements ()
7893 ),
79- Conn .
94+ { ok , Conn } .
8095
8196prepare (Conn , Name , Sql ) ->
8297 {ok , Stmt } = epgsql :parse (Conn , atom_to_list (Name ), Sql , []),
@@ -86,17 +101,18 @@ prepare(Conn, Name, Sql) ->
86101 _ -> ok
87102 end .
88103
89- % % Pick a connection at random and run the prepared statement on it
90- % % directly. epgsql serializes commands per connection (its gen_server
91- % % mailbox) , so concurrent callers queue on the chosen connection instead
92- % % of funneling through one checkout broker — no broker bottleneck and no
93- % % `no_members` refusals under high concurrency .
104+ % % Run the prepared statement on a pool-picked connection. ecpool hands the
105+ % % callback the connection chosen for this request; epgsql serializes
106+ % % commands per connection through its own mailbox, so callers queue on the
107+ % % chosen connection, never on a shared broker. A worker mid-reconnect
108+ % % yields `{error, disconnected}`, which the handlers map to a 5xx .
94109-spec query (atom (), [term ()]) ->
95110 {ok , list (), list ()} | {ok , non_neg_integer ()} | {error , term ()}.
96111query (Name , Params ) ->
97- Conns = persistent_term :get (? CONNS_KEY ),
98- Conn = element (rand :uniform (tuple_size (Conns )), Conns ),
99- epgsql :prepared_query (Conn , persistent_term :get ({? MODULE , stmt , Name }), Params ).
112+ Stmt = persistent_term :get ({? MODULE , stmt , Name }),
113+ ecpool :with_client (? POOL , fun (Conn ) ->
114+ epgsql :prepared_query (Conn , Stmt , Params )
115+ end ).
100116
101117parse_url (Url ) ->
102118 Parsed = uri_string :parse (Url ),
0 commit comments