minato_pool (minato v0.18.6)
View SourceA pool of connections, handed out one caller at a time.
The pool holds connections and lends them; it does not run queries. That follows
from the caller owning the socket: a pool that ran the query would be back to
one process serialising every caller, which is the arrangement the whole client
is written to avoid. checkout/2 hands over the socket as well as the
connection term, checkin/2 hands it back, and minato wraps both so that
the ordinary case is one call.
Waiting
A checkout with no connection free waits in a FIFO queue until one is checked in or the timeout runs out. First in, first served, and a caller that gives up is taken out of the queue rather than handed a connection nobody is waiting for.
The queue is deliberately plain. A CoDel queue, which drops from the head when the minimum delay over a window stays above target, is the thing to build here when there is a load to measure it against; a bounded FIFO with a timeout is what that will replace, and it is honest in the meantime.
Connections come and go
Connecting happens in the background and is retried with exponential backoff, so
a pool started before its database is a pool that fills up when the database
arrives rather than a supervisor that gives up. checkout/2 waits for that
exactly as it waits for a busy connection.
A borrower that dies takes the socket with it, because the socket belongs to the process that holds it: the pool notices through its monitor and starts a replacement.
What a connection is like when it comes back
Checked in with a transaction open, it is rolled back before it is lent again. A caller that returns a connection mid-transaction has already made a mistake, and the alternative is the next borrower silently joining a transaction it did not open. A connection that cannot be rolled back is closed and replaced, because there is nothing else to be sure of about it.
Summary
Types
How the pool is registered and asked for.
What to open, and how many.
What the pool has been doing since it started.
Functions
Give a connection back.
checkout/2 with the default timeout of five seconds.
Borrow a connection, waiting for one if they are all out.
Give a connection back to be closed rather than lent again.
Hand a borrowed connection to another process, which then owes it back.
Start a pool under a supervisor.
What the pool holds and what it has done.
Stop the pool and close every connection it holds.
Types
-type cast() :: {checkin, pid(), minato_conn:conn()} | {discard, pid(), minato_conn:conn()} | {gave_up, pid()}.
-type info() :: {connected, minato_conn:conn()} | {connect_failed, term()} | retry | sweep | {'DOWN', reference(), process, pid(), term()}.
-type name() :: atom().
How the pool is registered and asked for.
-type opts() :: #{size => pos_integer(), min_size => non_neg_integer(), max_age => timeout(), max_idle => timeout(), disconnected => fail | wait, connection := minato_conn:opts()}.
What to open, and how many.
#{size => pos_integer(),
connection => minato_conn:opts()}connection is passed to minato_conn:connect/1 unchanged. size is how many
connections the pool keeps. How long to wait for a free one is the caller's to
say, in checkout/2.
min_size is how many connections the pool opens before anybody asks, and it
defaults to one. The rest are opened when a caller finds none free, so a node
with twenty pools does not want a hundred connections the moment it boots, while
a pool that cannot connect at all still says so at start up rather than at the
first query. 0 opens nothing until asked.
disconnected decides what a checkout gets when the pool holds no connections
and the last attempt to open one failed. fail is the default and answers
{error, disconnected} at once, because a caller that cannot be served should
find out in microseconds rather than after the whole checkout timeout - during
an outage the alternative is every caller in the system holding a process for
five seconds to learn the same thing. wait queues anyway, for work that would
rather be slow than fail.
max_idle closes a connection that has sat unused for that many milliseconds,
down to min_size. A pool that grew to meet a busy hour otherwise holds that
hour's connections until the process dies, which is a cost the server pays and
nobody sees. It is one minute by default, and infinity keeps whatever the pool
has grown to.
max_age retires a connection when it is checked in older than that many
milliseconds, and opens a replacement. It is infinity by default, because a
connection to PostgreSQL itself is good indefinitely and churning them would be
work nobody asked for. Set it when something sits in the middle - a proxy, a
load balancer, a NAT - since those have idle and lifetime limits of their own,
and a connection cut by one of them is discovered by a query failing.
-type stats() :: #{size := non_neg_integer(), idle := non_neg_integer(), borrowed := non_neg_integer(), waiting := non_neg_integer(), checkouts := non_neg_integer(), waits := non_neg_integer(), timeouts := non_neg_integer(), disconnects := non_neg_integer(), retired := non_neg_integer()}.
What the pool has been doing since it started.
disconnects is connections lost: a connect that failed, a borrower that died,
a connection given back in a state it could not be reset from. retired is
connections closed on purpose because they reached max_age. The two are
separate because one of them is a problem and the other is the pool working.
-type waiter() :: {gen_server:from(), pid(), reference()}.
Functions
-spec checkin(name(), minato_conn:conn()) -> ok.
Give a connection back.
The socket goes back to the pool first, because the pool cannot read a socket it does not own. A connection with a transaction open is rolled back before it is lent again, and one that cannot be is closed and replaced.
-spec checkout(name()) -> {ok, minato_conn:conn()} | {error, term()}.
checkout/2 with the default timeout of five seconds.
-spec checkout(name(), timeout()) -> {ok, minato_conn:conn()} | {error, term()}.
Borrow a connection, waiting for one if they are all out.
The socket is handed over with it, so the calling process can read and write it.
Give it back with checkin/2; a caller that dies instead is noticed, and the
connection is replaced rather than leaked.
-spec discard(name(), minato_conn:conn()) -> ok.
Give a connection back to be closed rather than lent again.
For a borrower that knows the connection is no longer trustworthy: a read that timed out, an exception part way through a result set, anything that leaves the stream at an offset nobody knows. The pool closes it and opens another.
-spec give_away(name(), minato_conn:conn(), pid()) -> ok | {error, term()}.
Hand a borrowed connection to another process, which then owes it back.
Two things move and both have to: the socket, which belongs to whoever holds it, and the pool's idea of who to watch. The pool stops watching the caller and starts watching the new owner, so the connection is replaced if that process dies rather than the one that borrowed it.
The connection term itself is the caller's to send on. Sending it without calling this leaves the receiver holding a term whose socket belongs to somebody else, and the pool waiting on a process that has moved on.
For a worker that borrows a connection and hands it to the process that will use
it, and for LISTEN on a connection taken out of the pool.
-spec handle_call(request(), gen_server:from(), #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}) -> {reply, term(), #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}} | {noreply, #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}}.
-spec handle_cast(cast(), #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}) -> {noreply, #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}}.
-spec handle_info(info(), #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}) -> {noreply, #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}}.
-spec init({name(), opts()}) -> {ok, #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}}.
-spec start_link(name(), opts()) -> gen_server:start_ret().
Start a pool under a supervisor.
What the pool holds and what it has done.
-spec stop(name()) -> ok.
Stop the pool and close every connection it holds.
-spec terminate(term(), #state{name :: name(), opts :: minato_conn:opts(), size :: pos_integer(), min_size :: non_neg_integer(), max_age :: timeout(), disconnected :: fail | wait, max_idle :: timeout(), failures :: non_neg_integer(), measured :: boolean(), idle :: [{minato_conn:conn(), integer()}], borrowed :: #{pid() => #borrowed{monitor :: reference(), count :: pos_integer()}}, waiting :: queue:queue(waiter()), connecting :: non_neg_integer(), backoff :: pos_integer(), counters :: counters:counters_ref()}) -> ok.