Pooling#

Opening a connection costs several round trips and a new server process, so a service that answers many requests should keep connections open and lend them out. ConnectionPool does that. It wraps psycopg’s pool and adds what a graph service needs on top: the graph selected on every connection, one budget for waiting and running, a way to retire every connection at once, and the safe handling of interrupted statements.

Class
`ConnectionPool`	keeps between `min_size` and `max_size` connections open and lends them out
`NullConnectionPool`	keeps none: opens a connection per borrow and closes it after
`AsyncConnectionPool`, `AsyncNullConnectionPool`	the same, for awaiting code

Creating and opening a pool#

import agensgraph

pool = agensgraph.ConnectionPool(
    "host=localhost dbname=docs user=agens",
    graph="moviekb",
    min_size=2,
    max_size=8,
    kwargs={"autocommit": True},
)
pool.open(wait=True)

Creating a pool connects nothing. open() starts it. It first makes one connection of its own and closes it again, so a server that cannot be reached, or that the driver cannot read, fails there, at startup, with the real reason, rather than being retried in the background for minutes while requests queue. wait=True then waits until min_size connections exist.

graph= is selected on each connection when it is opened, and committed, so a rollback during a borrow cannot undo it. kwargs are passed to connect() for every connection.

Prefer autocommit for a pool, as above, and open a conn.transaction() where a borrow needs one. A pool that is not in autocommit commits each borrow when its block ends and rolls back on an exception, which is also correct, but then every borrow is already a transaction, and a transaction() block inside one is only a savepoint.

Borrowing a connection#

>>> with pool.connection() as conn:
...     conn.execute_query("MATCH (p:person) RETURN count(p)").records
[(28,)]

The connection goes back to the pool when the block ends. For a single statement, pool.execute_query() borrows, runs and returns in one call:

>>> pool.execute_query("MATCH (m:movie) WHERE m.year > %s RETURN count(m)", (2010,)).records
[(5,)]

Do not keep the handle past the block. The connection may already be lent to someone else, and work done on it then would land in their transaction. While it sits idle in the pool, using it is refused:

>>> with pool.connection() as conn:
...     kept = conn
>>> kept.execute_query("RETURN 1")
Traceback (most recent call last):
  ...
agensgraph.errors.ReleasedConnection: this connection has gone back to the pool, and may already have been lent to somebody else. Take one with `with pool.connection() as conn:` for each piece of work, and do not keep the handle past the block

Once the pool has lent it to someone else, the driver can no longer tell the two holders apart, so the refusal is a safeguard and not a guarantee.

A budget for the whole borrow#

Pass a Deadline to bound the wait for a connection and the statements run on it together:

>>> from agensgraph import Deadline
>>> with pool.connection(deadline=Deadline(5.0)) as conn:
...     conn.execute("SHOW statement_timeout").fetchone()
('4499ms',)

Time spent waiting for a connection is time the statements no longer have. The rest of the budget reaches the server as statement_timeout, set once when the connection is handed out, a little inside what is left so the server gives up before you do, and put back before the next borrower gets the connection. Timeouts and cancellation explains the details.

timeout= on connection() is a budget too: it bounds the wait and sets the statement limit the same way.

Hooks#

Three functions run at fixed points in a connection’s life:

Argument	Runs	For
`configure=`	once, when a connection is opened, after the graph is selected	anything a connection keeps: `register_vectors()`, `SET application_name`, adapters
`setup=`	each time a connection is lent	per-borrow settings
`reset=`	each time a connection comes back	cleaning up after a borrower

vector_pool = agensgraph.ConnectionPool(
    "host=localhost dbname=docs user=agens",
    graph="moviekb",
    kwargs={"autocommit": True},
    configure=lambda conn: conn.register_vectors(),
)

In the blocking pools these are plain functions; in the awaiting pools they are coroutine functions and are awaited. If reset= raises, the connection is closed instead of reused.

The graph is selected once per connection, not on each borrow. A borrower that switches graphs with conn.graph() leaves the connection on that graph for the next borrower; if borrowers switch, select the graph again in setup=.

After the server restarts#

A server restart breaks every connection a pool holds, and each would fail in turn, once for each caller who happened to borrow it. invalidate() retires all of them in one call:

>>> pool.generation
0
>>> pool.invalidate()
1

It closes nothing and waits for nothing. Each connection belongs to the generation it was opened in; a borrow never receives one from an older generation, and a connection in use when you called invalidate() stays usable until its holder is done, then is closed instead of returned. The pool replaces them as it needs to.

drain() does a different job: it closes every connection and opens the same number again, without changing the generation. Use it when the connections are not wrong but you want new ones built differently, such as after changing what configure= does, or after rotating the password behind a connection string given as a function. (A conninfo given as a function is called for every new connection.)

A statement interrupted mid-flight#

A connection whose statement was interrupted while the driver was sending it and waiting for the answer, by a cancelled task, a timeout around the call, or KeyboardInterrupt, is closed when it comes back rather than lent again. Its state can no longer be described: the statement may still be running, and an unread result may be waiting on the socket. (A connection that comes back in the middle of any other operation, such as a stream or a copy, is not in step either, and psycopg’s pool closes it in turn.) Replacing it costs one connection on an event that is rare. A statement that merely failed, with an error from the server or a value the driver refused, keeps its connection, which is idle and in step.

Cancellation describes how to stop a statement so that its connection can be kept.

Watching a pool#

>>> stats = pool.get_stats()
>>> sorted(stats)
['connections_errors', 'connections_interrupted', 'connections_lost', 'connections_ms', 'connections_num', 'connections_retired', 'generation', 'pool_available', 'pool_max', 'pool_min', 'pool_size', 'requests_errors', 'requests_num', 'requests_queued', 'requests_wait_ms', 'requests_waiting', 'returns_bad', 'usage_ms']

psycopg’s own counters, plus generation, connections_retired (closed for being of an old generation) and connections_interrupted. Every counter is always present, at zero when nothing has moved it, so an exporter can read them on a timer without checking. agensgraph.pool.COUNTERS names the ten that accumulate, and pop_stats() returns the same with those reset, for reporting an interval.

A pool that is too small looks like a slow database. Read requests_wait_ms, the time borrowers spent waiting, against requests_queued, how many had to wait. Measured with six slow borrowers against a pool of two: almost nine seconds of waiting across five queued requests, while a trivial query took nearly three seconds just to get a connection.

Without a pool: `NullConnectionPool`#

For a process that handles one request and exits, or one that may be frozen between requests, keeping connections is waste. NullConnectionPool opens one per borrow and closes it after. Everything else is the same, so moving between the two is a change of class:

per_request = agensgraph.NullConnectionPool(
    "host=localhost dbname=docs user=agens", graph="moviekb"
)

A borrow and one statement then cost about eight times what they cost from a pool that keeps its connections, which is what connecting costs. Do not choose it for a process that lives long enough to reuse a connection.

Awaiting code#

pool = agensgraph.AsyncConnectionPool("host=localhost dbname=docs user=agens", graph="moviekb",
                                      kwargs={"autocommit": True})
await pool.open(wait=True)

async with pool.connection() as conn:
    result = await conn.execute_query("MATCH (p:person) RETURN count(p)")

await pool.close()

open(), wait(), close(), check(), drain(), resize() and execute_query() are awaited; invalidate(), get_stats() and pop_stats() are not.

Sizing#

min_size defaults to 4 and max_size to 16. A min_size above 16 needs a max_size too, or the pool is refused. The other psycopg settings pass straight through: timeout (the longest a borrow waits by default, 30 seconds), max_waiting, max_lifetime, max_idle, reconnect_timeout, num_workers and name. check_connections=True, the default, checks each connection as it is lent, so a connection the network has broken is replaced rather than handed to a caller; it costs an empty statement per borrow.

Closing#

Close the pool when the service stops:

pool.close()

A pool is also a context manager, with agensgraph.ConnectionPool(...) as pool:, which opens it without waiting, so call pool.wait() inside it to fail at startup, and closes it at the end.