class ConnectionPool:
"""A bounded, thread-safe pool of ready Connection objects (POOL-01..04).
``params`` is the same keyword dict :func:`saprfclib.connect` accepts (D-10).
``min_size`` connections are opened eagerly at construction (warm-up); the
pool then grows lazily up to ``max_size`` under demand. Acquire/release are
safe from any number of threads concurrently.
"""
def __init__(
self,
params: dict[str, Any],
min_size: int = 1,
max_size: int = 10,
) -> None:
if min_size < 0:
raise ValueError("min_size must be >= 0")
if max_size < 1:
raise ValueError("max_size must be >= 1")
if min_size > max_size:
raise ValueError("min_size must be <= max_size")
self._params: dict[str, Any] = dict(params)
self._max_size = max_size
# One descriptor cache for the whole pool. A FunctionDesc describes the
# system, not the socket it arrived over, so a per-connection cache made
# a pool of N connections pay N round-trips to learn each interface. The
# key hint covers the case where the system reports no sys_id: these
# connections are opened from identical parameters, so they reach the
# same system by construction and may share a bucket.
self._metadata_cache = MetadataCache()
self._metadata_cache_key = f"\x00pool-{uuid.uuid4().hex}"
self._cv = threading.Condition()
self._idle: deque[Any] = deque()
self._in_use: set[Any] = set()
self._created = 0
self._closed = False
self.metrics = PoolMetrics()
# Warm-up (POOL-01): pre-open min_size connections before any acquire.
for _ in range(min_size):
conn = self._open()
self._idle.append(conn)
self._created += 1
# ------------------------------------------------------------------ #
# Connection lifecycle seams (all real network/Connection touch points)
# ------------------------------------------------------------------ #
def _open(self) -> Any:
"""Open a fresh Connection via the saprfclib.connect() factory (D-10).
Dispatches through the ``saprfclib.connection`` module attribute (not a
bound name) so the factory seam stays patchable for offline tests.
"""
return _connection.connect(
**self._params,
metadata_cache=self._metadata_cache,
metadata_cache_key=self._metadata_cache_key,
)
def _ping_ok(self, conn: Any) -> bool:
"""True iff ``conn.ping()`` reports liveness. Any False/exception is dead.
Called OUTSIDE the Condition lock (Pitfall 2). Treats both a falsy ping
result and any raised exception as a dead connection.
"""
if _is_retired(conn):
_logger.debug("pool: discarding a connection whose session is BROKEN")
return False
try:
return bool(conn.ping())
except Exception as exc: # noqa: BLE001 — any failure means "do not lend it"
# Discarding the connection is right; doing it without a word is not.
# A pool that quietly bins every connection it checks looks to the caller
# like a slow pool rather than a broken one.
_logger.debug(
"pool: discarding a connection that failed its health check (%s: %s)",
type(exc).__name__,
exc,
)
return False
def _safe_close(self, conn: Any) -> None:
"""Close a connection, swallowing any error (close is best-effort)."""
try:
conn.close()
except Exception as exc: # noqa: BLE001 — close is best-effort
_logger.debug("pool: error while closing a discarded connection: %s", exc)
# ------------------------------------------------------------------ #
# Public acquire/release surface
# ------------------------------------------------------------------ #
@contextmanager
def acquire(self, timeout: float = 30.0) -> Iterator[Any]:
"""Lend a single-owner Connection for the duration of the ``with`` block.
Blocks up to ``timeout`` seconds for a connection to become available,
ping-checking it before lending (D-12); auto-releases on block exit,
including on exception (D-11, POOL-02). Raises :class:`PoolTimeoutError`
if the deadline elapses while the pool is exhausted.
"""
conn = self._checkout(timeout)
try:
yield conn
finally:
self.release(conn)
def _checkout(self, timeout: float) -> Any:
"""Check out a healthy connection or raise PoolTimeoutError on deadline.
Predicate loop under the Condition (D-13):
1. Reuse an idle connection, ping-checked OUTSIDE the lock (Pitfall 2):
pop under lock + provisionally count out, release lock, ping, then
re-acquire to lend (healthy) or discard+replace (dead, POOL-03).
2. Otherwise lazily grow toward max_size (POOL-01).
3. Otherwise wait on the condition until a release or the deadline.
"""
started = time.monotonic()
deadline = started + timeout
discarded = 0
with self._cv:
while True:
if self._closed:
raise RuntimeError("pool is closed")
# 1) Try to reuse an idle connection, ping-checked outside the lock.
if self._idle:
candidate = self._idle.popleft()
# Provisionally mark it out so no other thread can grab it and
# so _created accounting stays consistent across the lock gap.
self._in_use.add(candidate)
self._cv.release()
try:
healthy = self._ping_ok(candidate)
finally:
self._cv.acquire()
if healthy:
self.metrics.hits += 1
self.metrics.acquires += 1
self.metrics._record_wait(time.monotonic() - started)
return candidate
# Dead connection: discard + replace (POOL-03).
self._in_use.discard(candidate)
self._created -= 1
discarded += 1
self.metrics.discards += 1
self._cv.release()
try:
self._safe_close(candidate)
finally:
self._cv.acquire()
# A slot freed up; let a waiter know and re-loop.
self._cv.notify()
continue
# 2) Lazily grow toward max_size.
if self._created < self._max_size:
self._created += 1
self.metrics.creates += 1
self._cv.release()
try:
fresh = self._open()
except Exception:
# Open failed: undo the reservation and surface the error.
self._cv.acquire()
self._created -= 1
# The reservation is undone, so the create never happened.
# Leaving it counted would show a pool steadily opening
# connections that do not exist.
self.metrics.creates -= 1
self._cv.notify()
raise
else:
self._cv.acquire()
self._in_use.add(fresh)
self.metrics.acquires += 1
self.metrics._record_wait(time.monotonic() - started)
return fresh
# 3) Exhausted: wait for a release or the deadline.
remaining = deadline - time.monotonic()
if remaining <= 0:
self.metrics.timeouts += 1
self.metrics._record_wait(time.monotonic() - started)
raise PoolTimeoutError(
waited=timeout,
discarded=discarded,
active=len(self._in_use),
idle=len(self._idle),
max_size=self._max_size,
)
self._cv.wait(remaining)
def release(self, conn: Any) -> None:
"""Return a lent connection to the idle set and wake one waiter (POOL-04)."""
with self._cv:
self._in_use.discard(conn)
if self._closed:
# Pool shut down while this connection was out: close it instead of
# returning it to a defunct idle set.
self._created -= 1
self._cv.release()
try:
self._safe_close(conn)
finally:
self._cv.acquire()
self._cv.notify()
return
self._idle.append(conn)
self._cv.notify()
# ------------------------------------------------------------------ #
# Live gauges #
# ------------------------------------------------------------------ #
@property
def in_use(self) -> int:
"""Connections currently lent out."""
return len(self._in_use)
@property
def idle(self) -> int:
"""Connections sitting in the pool, ready to lend."""
return len(self._idle)
@property
def size(self) -> int:
"""Connections the pool currently owns, lent or idle."""
return self._created
@property
def max_size(self) -> int:
"""The ceiling the pool will not grow past."""
return self._max_size
def stats(self) -> dict[str, float | int]:
"""Counters and live gauges together, flat, for an exporter.
The gauges are read without the lock. They are a snapshot either way --
by the time a caller acts on them another thread may have acquired or
released -- so taking the lock would buy a consistency that cannot
survive the return statement, at the cost of contending with the acquire
path this is meant to observe.
"""
return {
**self.metrics.as_dict(),
"in_use": self.in_use,
"idle": self.idle,
"size": self.size,
"max_size": self.max_size,
}
def close(self) -> None:
"""Close every pooled connection (idle + in-use) and mark the pool closed.
Idle connections are closed immediately. In-use connections are closed on
their next ``release()``. Idempotent.
"""
with self._cv:
if self._closed:
return
self._closed = True
idle_conns = list(self._idle)
self._idle.clear()
self._created -= len(idle_conns)
self._cv.notify_all()
# Close outside the lock (close may do I/O).
for conn in idle_conns:
self._safe_close(conn)