Lightweight 0.20261105.1
Loading...
Searching...
No Matches
Pool.hpp
1// SPDX-License-Identifier: Apache-2.0
2#pragma once
3
4#include "../Async/Executor.hpp"
5#include "../Async/Task.hpp"
6#include "../SqlConnectInfo.hpp"
7#include "../SqlLogger.hpp"
8#include "../SqlStatistics.hpp"
9#include "DataMapper.hpp"
10
11#include <cassert>
12#include <chrono>
13#include <condition_variable>
14#include <coroutine>
15#include <cstdint>
16#include <deque>
17#include <expected>
18#include <format>
19#include <functional>
20#include <memory>
21#include <mutex>
22#include <ranges>
23#include <shared_mutex>
24#include <stdexcept>
25#include <vector>
26
27/// @defgroup ConnectionPool Connection Pooling
28/// @brief A thread-safe pool of @c DataMapper instances, configured at compile time.
29///
30/// The growth strategy, initial size and maximum size are supplied as a @c PoolConfig
31/// template parameter, so the policy is fixed at the type level rather than at runtime.
32
33namespace Lightweight
34{
35
36/// @ingroup ConnectionPool
37/// Enum to define growth strategies of the pool
38///
39enum class GrowthStrategy : uint8_t
40{
41 /// Pre-create initialSize objects. Allow the total count
42 /// to grow up to maxSize. Once maxSize objects exist, callers BLOCK
43 /// until one is returned.
45
46 /// Pre-create initialSize objects. The pool stores up to maxSize
47 /// objects. If none are idle, a fresh object is ALWAYS created (no
48 /// waiting). On return: kept if the idle set is below maxSize, otherwise
49 /// destroyed.
51
52 /// Pre-create initialSize objects. Grow without limit. Every returned
53 /// object is always kept in the pool.
55};
56
57/// @ingroup ConnectionPool
58/// Whether the pool checks that a connection is still live before handing it to a caller.
59enum class ValidateOnBorrow : uint8_t
60{
61 /// Hand the connection out without checking it.
62 No,
63
64 /// Check SqlConnection::IsAlive() first and discard a connection reported dead, transparently
65 /// serving the caller from the next idle connection or a freshly created one.
66 ///
67 /// The check reads the driver-local @c SQL_ATTR_CONNECTION_DEAD attribute, so it costs no round
68 /// trip to the server. By the same token it is only as good as the driver's own bookkeeping:
69 /// several drivers mark a connection dead only after an operation has already failed, so a
70 /// connection whose peer vanished silently (a firewall or NAT dropping the flow without sending
71 /// FIN or RST) can still pass this check. Pair it with @ref PoolConfig::maxIdleTimeMs to retire
72 /// such connections before they are ever handed out.
73 Yes,
74};
75
76/// @ingroup ConnectionPool
77/// Reason an @ref Pool::Acquire call with a timeout failed to produce a data mapper.
78enum class PoolError : uint8_t
79{
80 /// The timeout elapsed before a data mapper became available.
81 Timeout,
82};
83
84/// @ingroup ConnectionPool
85/// Structure to hold the configuration of the pool, including the initial size, maximum size and growth strategy.
86/// Structure is used as a template parameter for the Pool class to configure its behavior at compile time.
87///
88/// @note The lifetime bounds are expressed as plain millisecond counts rather than
89/// @c std::chrono::milliseconds because this structure is used as a non-type template
90/// parameter: @c std::chrono::duration keeps its representation private and is therefore not a
91/// structural type. Use @ref PoolConfig::MaxIdleTime and @ref PoolConfig::MaxLifetime to read
92/// them back as durations.
94{
95 /// Initial number of data mappers to pre-create and store in the pool, must be less than or equal to maxSize
96 size_t initialSize {};
97 /// Maximum number of data mappers that can exist in the pool, must be greater than or equal to initialSize
98 /// this is used for the Bounded* strategies to determine when to block or when to stop accepting returned data mappers,
99 /// for the UnboundedGrow strategy this is ignored
100 size_t maxSize {};
101 /// Strategy to determine how the pool should grow when there are no idle data mappers available, default is BoundedWait
102 /// which blocks until a data mapper is returned to the pool
104
105 /// Whether a connection is checked for liveness before it is handed to a caller, enabled by default.
106 ///
107 /// @see ValidateOnBorrow
109
110 /// Maximum time in milliseconds a connection may sit idle in the pool before it is retired
111 /// instead of handed out again; 0 (the default) disables the bound.
112 ///
113 /// Set this below any idle timeout imposed by the network path (a firewall or NAT dropping idle
114 /// flows) or by the server, so a connection is never idle long enough to be reaped behind the
115 /// pool's back. This removes the failure mode that @ref ValidateOnBorrow can only detect, and
116 /// then only when the driver has noticed.
117 ///
118 /// @note Retirement is lazy: it happens when the pool is next used, so a pool that goes
119 /// completely idle keeps its connections until the next @ref Pool::Acquire. Correctness is
120 /// unaffected — an expired connection is discarded rather than handed out — but the pool
121 /// does not shrink on its own.
122 std::chrono::milliseconds::rep maxIdleTimeMs {};
123
124 /// Maximum total age in milliseconds of a connection, counted from when it was created, after
125 /// which it is retired rather than reused; 0 (the default) disables the bound.
126 ///
127 /// Unlike @ref validateOnBorrow, this retires connections that are perfectly alive but no longer
128 /// appropriate: after a failover or a rolling restart a pooled connection stays bound to the old
129 /// node, and nothing else in the pool will ever move it. Setting this shorter than any
130 /// connection-age ceiling imposed by the database or the infrastructure also means connections
131 /// are retired while idle, which is free, rather than being cut mid-query by something else.
132 ///
133 /// @note Retirement is lazy, as described for @ref maxIdleTimeMs.
134 std::chrono::milliseconds::rep maxLifetimeMs {};
135
136 /// Prepared-statement cache capacity given to the connection of every data mapper this pool creates,
137 /// i.e. how many already-prepared ODBC statement handles that connection keeps for reuse. Zero (the
138 /// default) leaves the cache disabled, exactly as an unpooled connection.
139 ///
140 /// The bound is per connection, not per pool: a pool may hold up to `maxSize` connections, each with
141 /// its own cache of this size, so the live prepared handles a fully warmed pool holds on the server
142 /// are `maxSize * preparedStatementCacheCapacity`. Size it against the backend's per-session limit
143 /// on prepared statements, not against the number of distinct queries alone.
144 ///
145 /// @see SqlConnection::SetPreparedStatementCacheCapacity for what enabling the cache implies.
146 size_t preparedStatementCacheCapacity { PreparedStatementCacheCapacityDefault };
147
148 /// @return @ref maxIdleTimeMs as a duration.
149 [[nodiscard]] constexpr std::chrono::milliseconds MaxIdleTime() const noexcept
150 {
151 return std::chrono::milliseconds { maxIdleTimeMs };
152 }
153
154 /// @return @ref maxLifetimeMs as a duration.
155 [[nodiscard]] constexpr std::chrono::milliseconds MaxLifetime() const noexcept
156 {
157 return std::chrono::milliseconds { maxLifetimeMs };
158 }
159};
160
161/// @ingroup ConnectionPool
162/// A thread-safe pool of DataMapper instances with the policy configured by the PoolConfig template parameter.
163/// The pool allows acquiring and returning DataMapper instances, and manages the lifecycle of these instances according to
164/// the specified growth strategy.
165template <PoolConfig Config>
166class Pool
167{
168 private:
169 /// Clock the idle-time and lifetime bounds are measured against. Monotonic, so the bounds are
170 /// immune to wall-clock adjustments.
171 using Clock = std::chrono::steady_clock;
172
173 /// True when this configuration enables at least one time-based bound, and the pool therefore
174 /// has to timestamp its connections. When false, no clock is ever read.
175 static constexpr bool TracksTime = Config.maxIdleTimeMs > 0 || Config.maxLifetimeMs > 0;
176
177 /// A pooled DataMapper together with the timestamps the health bounds are evaluated against.
178 ///
179 /// @c createdAt travels with the mapper across checkout and return, so @ref PoolConfig::maxLifetimeMs
180 /// measures the connection's total age rather than the time since it was last idled.
181 /// @c idleSince is refreshed each time the mapper is stored in the idle set.
182 ///
183 /// @c generation is the @ref SqlConnection::DefaultConnectionStringGeneration the connection was
184 /// made under. Once the application replaces the default connection string, entries of an older
185 /// generation point at a database it has switched away from, and are retired rather than reused.
186 ///
187 /// An entry whose @c mapper is null is the empty result of @ref TakeIdleLocked, not a pooled entry.
188 /// Handed to a waiter, it transfers a checked-out slot without a connection (see
189 /// @ref ReleaseSlotLocked): the waiter then connects for itself.
190 struct Entry
191 {
192 std::unique_ptr<DataMapper> mapper;
193 Clock::time_point createdAt {};
194 Clock::time_point idleSince {};
195 std::uint32_t generation {};
196 /// Set on return when the connection could not be brought back to a clean state; such an
197 /// entry is retired rather than idled or handed to a waiter.
198 bool unusable = false;
199 };
200
201 public:
202 /// @ingroup ConnectionPool
203 /// A wrapper around a DataMapper that returns it to the pool when destroyed
204 /// can be created only from the Pool and is move-only to ensure it is always
205 /// returned to the pool when it goes out of scope
206 ///
207 /// On return, a transaction still open on the connection is rolled back, so the next caller
208 /// starts outside any transaction. That covers transactions ODBC knows about - a
209 /// @ref SqlTransaction, or autocommit switched off by hand. A transaction opened by executing a
210 /// raw @c BEGIN statement is invisible to ODBC: ending it is the caller's responsibility, and
211 /// @ref SqlTransaction is the way to have it ended on every exit path.
213 {
214 private:
215 friend class Pool;
216
217 explicit PooledDataMapper(Pool& pool, Entry entry) noexcept:
218 _entry { std::move(entry) },
219 _pool { pool }
220 {
221 }
222
223 public:
224 PooledDataMapper() = delete;
225 PooledDataMapper(PooledDataMapper const&) = delete;
226
227 /// Move constructor for the pooled data mapper, the only public
228 /// constructor, allows moving the pooled data mapper but not copying it
230 _entry { std::move(other._entry) },
231 _pool { other._pool }
232 {
233 }
234 PooledDataMapper& operator=(PooledDataMapper const&) = delete;
235 PooledDataMapper& operator=(PooledDataMapper&&) = delete;
236 ~PooledDataMapper() noexcept
237 {
238 if (_entry.mapper)
239 ReturnToPool();
240 }
241
242 /// Access the underlying data mapper via pointer semantics
243 DataMapper* operator->() const noexcept
244 {
245 return _entry.mapper.get();
246 }
247
248 /// Access the underlying data mapper via reference semantics
249 /// This is useful for passing the pooled data mapper to functions
250 /// that expect a DataMapper reference
251 [[nodiscard]] DataMapper& Get() const noexcept
252 {
253 return *_entry.mapper;
254 }
255
256 private:
257 void ReturnToPool() noexcept
258 {
259 _pool.Return(std::move(_entry));
260 _entry.mapper = nullptr;
261 }
262
263 Entry _entry;
264 Pool& _pool;
265 };
266
267 private:
268 struct WaiterNode; // defined below; referenced by ReturnLocked's signature.
269
270 /// Brings a returned mapper's connection back to the state a fresh one is in, before it is idled
271 /// or handed off. Shared by every @c Return overload.
272 ///
273 /// - Detaches the async backend, so a recycled connection never carries references to executors
274 /// that may since have been destroyed (the next @c AcquireAsync re-enables it fresh).
275 /// - Rolls back a transaction left open on it, so the next, unrelated caller does not silently run
276 /// inside it (#583). A connection where that fails is marked @c unusable and retired.
277 ///
278 /// @warning The caller must not return a mapper that still has an async operation in flight on it:
279 /// dropping the backend destroys the strand/executors an outstanding offloaded step references and
280 /// races the worker still touching the ODBC handle. Await every async op before returning.
281 /// @param entry The returned entry.
282 static void ResetForReuse(Entry& entry) noexcept
283 {
284 auto& connection = entry.mapper->Connection();
285 connection.DisableAsync();
286 if (!connection.RollbackOpenTransaction())
287 entry.unusable = true;
288 }
289
290 /// @return The current time, or a default-constructed time point when this configuration enables
291 /// no time-based bound and therefore never inspects timestamps.
292 [[nodiscard]] Clock::time_point NowIfTracking() const noexcept
293 {
294 if constexpr (!TracksTime)
295 return {};
296 else
297 {
298 if (_clock)
299 return _clock();
300 return Clock::now();
301 }
302 }
303
304 /// Creates a fresh entry, stamped with the current time.
305 /// @return The new entry; its mapper is never null.
306 [[nodiscard]] Entry MakeEntry() const
307 {
308 auto const now = NowIfTracking();
309 // Read before the mapper reads the default connection string: the setter publishes the string
310 // first and the generation second, so this order can only stamp an entry older than it is
311 // (retired early, harmlessly), never newer (reused against the wrong database).
312 auto const generation = SqlConnection::DefaultConnectionStringGeneration();
313 auto mapper = std::make_unique<DataMapper>();
314 // The one place a pooled connection comes into existence, so a pool-wide connection setting is
315 // applied here rather than at each call site. Compile-time gated, so a pool left at the default
316 // capacity emits exactly the code it did before the setting existed.
317 if constexpr (Config.preparedStatementCacheCapacity != 0)
318 mapper->Connection().SetPreparedStatementCacheCapacity(Config.preparedStatementCacheCapacity);
319 // Records read through this mapper load their relations from this pool, not from the mapper
320 // itself, which goes back to the pool (and to other callers) long before a record dies.
321 detail::AdoptRelationLoadSource(
322 *mapper,
323 detail::PinToDefaultConnectionString(_relationLoadSource, mapper->Connection().ConnectionString(), generation));
324 return Entry { .mapper = std::move(mapper), .createdAt = now, .idleSince = now, .generation = generation };
325 }
326
327 /// Decides whether an idle connection may still be handed to a caller.
328 ///
329 /// Applied only to connections coming out of the idle set. A connection handed straight from
330 /// @ref Return to a parked waiter deliberately bypasses this: see @ref ReturnLocked.
331 ///
332 /// @param entry The idle entry under consideration.
333 /// @param now The current time, as returned by @ref NowIfTracking.
334 /// @return true when the entry is within both configured bounds and, if validation is enabled,
335 /// its connection is still reported alive.
336 [[nodiscard]] bool IsUsable(Entry const& entry, Clock::time_point now) const noexcept
337 {
338 if (MustRetire(entry, now))
339 return false;
340 if constexpr (Config.maxIdleTimeMs > 0)
341 {
342 if (now - entry.idleSince >= Config.MaxIdleTime())
343 return false;
344 }
345 if constexpr (Config.validateOnBorrow == ValidateOnBorrow::Yes)
346 {
347 if (!entry.mapper->Connection().IsAlive())
348 return false;
349 }
350 return true;
351 }
352
353 /// Pops the most recently idled usable connection, retiring any expired or dead entries it passes.
354 ///
355 /// Retired entries are moved into @p retired rather than destroyed here, so the ODBC disconnect
356 /// they trigger happens after the caller has released @c _mutex instead of blocking every other
357 /// thread for the duration of a network teardown.
358 ///
359 /// @pre @c _mutex is held by the caller.
360 /// @param retired Collects the retired entries; must outlive the caller's lock.
361 /// @return A usable entry, or an entry with a null mapper when the idle set holds none.
362 [[nodiscard]] Entry TakeIdleLocked(std::vector<Entry>& retired)
363 {
364 auto const now = NowIfTracking();
365 while (!_idleDataMappers.empty())
366 {
367 auto entry = std::move(_idleDataMappers.back());
368 _idleDataMappers.pop_back();
369 if (IsUsable(entry, now))
370 return entry;
371 retired.push_back(std::move(entry));
372 }
373 return {};
374 }
375
376 /// @param entry The entry under consideration.
377 /// @return true when @p entry was made from the current default connection string.
378 [[nodiscard]] static bool IsCurrentGeneration(Entry const& entry) noexcept
379 {
380 return entry.generation == SqlConnection::DefaultConnectionStringGeneration();
381 }
382
383 /// The retirement rules that hold whether a connection is being borrowed or returned. Checking
384 /// them on return as well as on borrow releases the connection as soon as it is no longer
385 /// wanted, rather than holding it until the next acquire. The idle bound and the liveness check
386 /// apply to borrowing only (see @ref IsUsable): a returned entry has been idle for zero time.
387 ///
388 /// @param entry The entry under consideration.
389 /// @param now The current time, as returned by @ref NowIfTracking.
390 /// @return true when the connection has outlived @ref PoolConfig::maxLifetimeMs, was made from
391 /// a default connection string the application has since replaced, or could not be reset
392 /// on return (see @ref ResetForReuse).
393 [[nodiscard]] static bool MustRetire(Entry const& entry, [[maybe_unused]] Clock::time_point now) noexcept
394 {
395 if (entry.unusable || !IsCurrentGeneration(entry))
396 return true;
397 if constexpr (Config.maxLifetimeMs > 0)
398 return now - entry.createdAt >= Config.MaxLifetime();
399 else
400 return false;
401 }
402
403 /// always return the data mapper to the pool for this strategy
404 void Return(Entry entry) noexcept
405 requires(Config.growthStrategy == GrowthStrategy::UnboundedGrow)
406 {
407 ResetForReuse(entry);
408 auto const now = NowIfTracking();
409 if (MustRetire(entry, now))
410 {
411 // Retired rather than idled: destroyed here, so it counts as a discarded release. The
412 // idle set is untouched, so the occupancy gauge needs no update (and _mutex is not held).
413 LIGHTWEIGHT_STATS_POOL_RELEASE(true);
414 return; // retired here, outside the lock
415 }
416 entry.idleSince = now;
417 SqlLogger::GetLogger().OnConnectionIdle(entry.mapper->Connection());
418 std::scoped_lock lock(_mutex);
419 _idleDataMappers.push_back(std::move(entry));
420 LIGHTWEIGHT_STATS_POOL_RELEASE(false);
421 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
422 }
423
424 /// for bounded wait strategy, return the data mapper to the pool: hand it to the next FIFO waiter
425 /// (sync or async) or idle it.
426 void Return(Entry entry) noexcept
427 requires(Config.growthStrategy == GrowthStrategy::BoundedWait)
428 {
429 ResetForReuse(entry);
430 Entry retired; // declared before the lock so its disconnect runs after the lock is released
431 std::shared_ptr<WaiterNode> toResume;
432 {
433 std::scoped_lock const lock(_mutex);
434 toResume = ReturnLocked(std::move(entry), retired);
435 }
436 // Resume outside the lock to avoid re-entrancy (the resumed coroutine may call back into the pool).
437 if (toResume)
438 toResume->resume->Resume(toResume->handle);
439 }
440
441 /// Takes what @ref ReturnLocked or @ref ReleaseSlotLocked handed a fulfilled sync waiter: a
442 /// connection, or a bare slot to connect for (outside @c _mutex, see @ref ConnectForTransferredSlot).
443 ///
444 /// @param node The fulfilled waiter node.
445 /// @param lock The caller's lock on @c _mutex; released before a connect.
446 /// @return The pooled data mapper.
447 PooledDataMapper TakeHandedOff(WaiterNode& node, std::unique_lock<std::mutex>& lock)
448 requires(Config.growthStrategy == GrowthStrategy::BoundedWait)
449 {
450 if (node.entry.mapper)
451 return PooledDataMapper(*this, std::move(node.entry));
452 lock.unlock();
453 return PooledDataMapper(*this, ConnectForTransferredSlot());
454 }
455
456 /// Produces a data mapper for a caller without waiting: reuses a usable idle one, otherwise
457 /// creates a fresh one while the pool is below capacity.
458 ///
459 /// @pre @c _mutex is held by the caller.
460 /// @param retired Collects entries retired while scanning the idle set; must outlive the lock.
461 /// @return A ready entry, or an entry with a null mapper when the pool is at capacity and the
462 /// caller must park.
463 [[nodiscard]] Entry AcquireReadyLocked(std::vector<Entry>& retired)
464 requires(Config.growthStrategy == GrowthStrategy::BoundedWait)
465 {
466 if (auto entry = TakeIdleLocked(retired); entry.mapper)
467 {
468 ++_checkedOut;
469 SqlLogger::GetLogger().OnConnectionReuse(entry.mapper->Connection());
470 // Only ever reached before a caller parks, so the wait is zero by construction; the
471 // parking paths record their own sample once they are handed a connection.
472 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 }, true, false);
473 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
474 return entry;
475 }
476 if (_checkedOut < Config.maxSize)
477 {
478 // below capacity: create a fresh data mapper. Claim the slot only once the connection
479 // actually stands up, so a failing connect does not leak capacity.
480 auto fresh = MakeEntry();
481 ++_checkedOut;
482 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 }, false, false);
483 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
484 return fresh;
485 }
486 // At capacity: the caller parks and records its own acquire sample.
487 return {};
488 }
489
490 /// Hands @p entry to the next FIFO waiter (transferring the checked-out count) or idles it. Serving
491 /// @c _waiters in arrival order keeps sync @ref Acquire and async @ref AcquireAsync waiters fair.
492 ///
493 /// @pre @c _mutex is held by the caller.
494 /// @param entry The entry to return; its mapper's async backend must already be disabled.
495 /// @param retired Receives the entry when it is retired for having outlived
496 /// @ref PoolConfig::maxLifetimeMs; must outlive the caller's lock.
497 /// @return The async waiter node handed the mapper, to be resumed by the caller after releasing
498 /// @c _mutex; @c nullptr if a sync waiter was woken in place, or the entry was idled or
499 /// retired.
500 std::shared_ptr<WaiterNode> ReturnLocked(Entry entry, Entry& retired) noexcept
501 requires(Config.growthStrategy == GrowthStrategy::BoundedWait)
502 {
503 if (entry.unusable || !IsCurrentGeneration(entry))
504 {
505 // Made from a default connection string the application has replaced since, or left in a
506 // transaction that could not be rolled back: it must reach neither the idle set nor a
507 // waiter, who would silently talk to the old database or run inside that transaction. The
508 // slot it held is passed on bare instead, and the waiter connects for itself.
509 retired = std::move(entry); // destroyed by the caller, after _mutex is released
510 LIGHTWEIGHT_STATS_POOL_RELEASE(true);
511 return ReleaseSlotLocked();
512 }
513 if (auto node = NextParkedWaiterLocked())
514 {
515 // A direct hand-off deliberately skips the health bounds and the liveness check. A waiter
516 // is blocked on a predicate only a hand-off satisfies, so retiring the connection here
517 // would strand it, and manufacturing a replacement means a DataMapper construction that
518 // may throw inside this noexcept path. The connection was in active use moments ago, and
519 // it is still checked the next time it comes out of the idle set.
520 //
521 // Handed directly to a waiter, never idled: a reuse, not an idle transition.
522 SqlLogger::GetLogger().OnConnectionReuse(entry.mapper->Connection());
523 // The hand-off is both a release by the returner and a (reusing) acquire by the waiter.
524 LIGHTWEIGHT_STATS_POOL_RELEASE(false);
525 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
526 return FulfilLocked(std::move(node), std::move(entry));
527 }
528 // No waiter: the connection goes idle, so the lifetime bound applies. Releasing the slot
529 // matters either way — a retired connection frees capacity just as an idled one does.
530 --_checkedOut;
531 auto const now = NowIfTracking();
532 if (MustRetire(entry, now))
533 {
534 retired = std::move(entry); // destroyed by the caller, after _mutex is released
535 // Retired rather than idled: the connection is destroyed, so it counts as a discard.
536 // Without this an idle count reconstructed from these events alone would drift.
537 LIGHTWEIGHT_STATS_POOL_RELEASE(true);
538 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
539 return nullptr;
540 }
541 entry.idleSince = now;
542 SqlLogger::GetLogger().OnConnectionIdle(entry.mapper->Connection());
543 _idleDataMappers.push_back(std::move(entry));
544 LIGHTWEIGHT_STATS_POOL_RELEASE(false);
545 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
546 return nullptr;
547 }
548
549 /// Releases a checked-out slot that no connection comes back with: hands it to the next FIFO
550 /// waiter as an entry with a null mapper (the waiter then connects for itself, see
551 /// @ref ConnectForTransferredSlot), or frees it when nobody waits.
552 ///
553 /// @pre @c _mutex is held by the caller.
554 /// @return The async waiter node handed the slot, to be resumed by the caller after releasing
555 /// @c _mutex; @c nullptr if a sync waiter was woken in place, or the slot was freed.
556 std::shared_ptr<WaiterNode> ReleaseSlotLocked() noexcept
557 requires(Config.growthStrategy == GrowthStrategy::BoundedWait)
558 {
559 if (auto node = NextParkedWaiterLocked())
560 return FulfilLocked(std::move(node), Entry {}); // the slot alone: _checkedOut stays (transferred)
561 --_checkedOut;
562 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
563 return nullptr;
564 }
565
566 /// Takes the longest-waiting parked acquirer off @c _waiters.
567 ///
568 /// @pre @c _mutex is held by the caller.
569 /// @return The waiter, or @c nullptr when none is parked.
570 [[nodiscard]] std::shared_ptr<WaiterNode> NextParkedWaiterLocked() noexcept
571 {
572 while (!_waiters.empty())
573 {
574 auto node = std::move(_waiters.front());
575 _waiters.pop_front();
576 // _waiters only ever holds parked nodes (an async awaitable de-registers itself on
577 // abandonment, and so does a timed-out Acquire(timeout)), but guard defensively.
578 if (node->state == WaiterNode::State::Parked)
579 return node;
580 }
581 return nullptr;
582 }
583
584 /// Hands @p entry - a connection, or a bare slot - to @p node, transferring the checked-out slot.
585 ///
586 /// @pre @c _mutex is held by the caller.
587 /// @param node The parked waiter to serve.
588 /// @param entry What it receives.
589 /// @return @p node when it is an async waiter, to be resumed by the caller after releasing
590 /// @c _mutex; @c nullptr for a sync waiter, which is woken here.
591 [[nodiscard]] static std::shared_ptr<WaiterNode> FulfilLocked(std::shared_ptr<WaiterNode> node, Entry entry) noexcept
592 {
593 node->state = WaiterNode::State::Fulfilled;
594 node->entry = std::move(entry);
595 if (node->kind == WaiterNode::Kind::Async)
596 return node; // resumed by the caller outside the lock
597 node->cv.notify_one(); // wake the blocked Acquire(); it consumes node->entry
598 return nullptr;
599 }
600
601 /// Connects for a waiter that was handed a bare slot by @ref ReleaseSlotLocked. The slot is
602 /// already counted in @c _checkedOut; if the connect fails, it is released again (to the next
603 /// waiter, or freed) so the pool's capacity does not shrink.
604 ///
605 /// @pre @c _mutex is not held by the caller.
606 /// @return The fresh entry; its mapper is never null.
607 /// @throws Whatever @ref MakeEntry throws, after releasing the slot.
608 [[nodiscard]] Entry ConnectForTransferredSlot()
609 requires(Config.growthStrategy == GrowthStrategy::BoundedWait)
610 {
611 try
612 {
613 return MakeEntry();
614 }
615 catch (...)
616 {
617 std::shared_ptr<WaiterNode> toResume;
618 {
619 std::scoped_lock const lock(_mutex);
620 toResume = ReleaseSlotLocked();
621 }
622 if (toResume)
623 toResume->resume->Resume(toResume->handle);
624 throw;
625 }
626 }
627
628 /// for bounded overflow strategy, only return to pool if we have capacity, otherwise just destroy the data mapper
629 void Return(Entry entry) noexcept
630 requires(Config.growthStrategy == GrowthStrategy::BoundedOverflow)
631 {
632 ResetForReuse(entry);
633 auto const now = NowIfTracking();
634 if (MustRetire(entry, now))
635 {
636 // Retired rather than idled: destroyed here, so it counts as a discarded release. The
637 // idle set is untouched, so the occupancy gauge needs no update (and _mutex is not held).
638 LIGHTWEIGHT_STATS_POOL_RELEASE(true);
639 return; // retired here, outside the lock
640 }
641 entry.idleSince = now;
642 std::scoped_lock lock(_mutex);
643 if (_idleDataMappers.size() < Config.maxSize)
644 {
645 SqlLogger::GetLogger().OnConnectionIdle(entry.mapper->Connection());
646 _idleDataMappers.push_back(std::move(entry));
647 LIGHTWEIGHT_STATS_POOL_RELEASE(false);
648 }
649 else
650 {
651 // Over capacity: the mapper is destroyed here rather than idled. Without this counter an
652 // idle-count reconstructed from idle/reuse events alone would silently drift.
653 LIGHTWEIGHT_STATS_POOL_RELEASE(true);
654 }
655 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
656 }
657
658 public:
659 /// Default constructor that pre-creates the initial number of data mappers and stores them in the pool
660 /// No other constructors are provided, as the pool is configured at compile time via the template parameter
661 explicit Pool()
662 {
663 _idleDataMappers.reserve(Config.initialSize);
664 for ([[maybe_unused]] auto const _: std::views::iota(0U, Config.initialSize))
665 _idleDataMappers.push_back(MakeEntry());
666 }
667
668 /// Destructor. The pool manages the lifecycle of the idle data mappers; be aware that any
669 /// acquired data mappers not returned to the pool are destroyed when the pool is destroyed,
670 /// which may leak resources if not handled properly.
671 ~Pool() noexcept
672 {
673 // Records read through this pool may outlive it; their relation loads fall back from now on.
674 _relationLoadSource->Detach();
675 // A parked AcquireAsync coroutine (or a thread blocked in Acquire) holds a reference back to this
676 // pool, so destroying the pool out from under it is undefined: drive every AcquireAsync task to
677 // completion and let every blocked Acquire() return first. The assert catches this in debug; the
678 // warning surfaces it in release (where the later access would be a use-after-free).
679 if (!_waiters.empty())
681 "Pool destroyed while acquirers are still waiting on it (coroutines parked in AcquireAsync "
682 "and/or threads blocked in Acquire); the pool must outlive every acquirer (drive each "
683 "AcquireAsync task to completion or destroy it first, and never destroy the pool while a "
684 "thread is blocked in Acquire). This is undefined behavior.");
685 assert(_waiters.empty() && "Pool destroyed while acquirers are still waiting on it");
686 }
687
688 Pool(Pool const&) = delete;
689 Pool& operator=(Pool const&) = delete;
690 Pool(Pool&&) = delete;
691 Pool& operator=(Pool&&) = delete;
692
693 /// Function to acquire a data mapper from the pool, the behavior of this function depends on the growth strategy
694 /// this is a specific implementation for the BoundedWait strategy, which blocks until a data mapper is available if the
695 /// pool is at maximum capacity
696 ///
697 /// Prefer the @ref Acquire(std::chrono::milliseconds) overload in production code: this one waits
698 /// indefinitely, so an exhausted pool parks the calling thread with no diagnostic.
700 requires(Config.growthStrategy == GrowthStrategy::BoundedWait)
701 {
702 std::vector<Entry> retired; // declared before the lock: disconnects run after it is released
703 [[maybe_unused]] auto const acquireStartedAt = std::chrono::steady_clock::now();
704 std::unique_lock lock(_mutex);
705 if (auto entry = AcquireReadyLocked(retired); entry.mapper)
706 return PooledDataMapper(*this, std::move(entry));
707
708 // Pool exhausted: park as a FIFO waiter (fair with AcquireAsync waiters) and block until a
709 // mapper is handed to this node. The hand-off transfers a checked-out slot, so no ++_checkedOut.
710 auto node = std::make_shared<WaiterNode>(WaiterNode::Kind::Sync);
711 _waiters.push_back(node);
712 node->cv.wait(lock, [&node] { return node->state == WaiterNode::State::Fulfilled; });
713 // Only this branch actually blocked, so only this branch contributes a wait-latency sample.
714 LIGHTWEIGHT_STATS_POOL_ACQUIRE(
715 std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - acquireStartedAt),
716 true,
717 true);
718 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
719 return TakeHandedOff(*node, lock);
720 }
721
722 /// Acquires a data mapper, giving up if none becomes available within @p timeout.
723 ///
724 /// Bounds how long an exhausted BoundedWait pool may park the calling thread, so a stuck or
725 /// overloaded pool surfaces as an error the caller can act on rather than as an indefinite hang.
726 ///
727 /// @param timeout How long to wait for a data mapper to be returned. A non-positive value makes
728 /// this a pure try-acquire.
729 /// @return The acquired data mapper, or @ref PoolError::Timeout if @p timeout elapsed first.
730 [[nodiscard]] std::expected<PooledDataMapper, PoolError> Acquire(std::chrono::milliseconds timeout)
731 requires(Config.growthStrategy == GrowthStrategy::BoundedWait)
732 {
733 std::vector<Entry> retired; // declared before the lock: disconnects run after it is released
734 [[maybe_unused]] auto const acquireStartedAt = std::chrono::steady_clock::now();
735 std::unique_lock lock(_mutex);
736 if (auto entry = AcquireReadyLocked(retired); entry.mapper)
737 return PooledDataMapper(*this, std::move(entry));
738
739 auto node = std::make_shared<WaiterNode>(WaiterNode::Kind::Sync);
740 _waiters.push_back(node);
741 if (!node->cv.wait_for(lock, timeout, [&node] { return node->state == WaiterNode::State::Fulfilled; }))
742 {
743 // wait_for evaluates the predicate under the lock and reports its final value, so a false
744 // result proves this node is still Parked and no hand-off can be in flight. De-registering
745 // it here is therefore race-free: a later Return will never see it.
746 node->state = WaiterNode::State::Abandoned;
747 std::erase(_waiters, node);
748 // A timed-out acquire yields no connection, so it contributes no acquire sample.
749 return std::unexpected { PoolError::Timeout };
750 }
751 // Reached only after actually blocking, so this contributes a wait-latency sample.
752 LIGHTWEIGHT_STATS_POOL_ACQUIRE(
753 std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - acquireStartedAt),
754 true,
755 true);
756 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
757 return TakeHandedOff(*node, lock);
758 }
759
760 /// Function to acquire a data mapper from the pool, the behavior of this function depends on the growth strategy
761 /// this is a specific implementation for the strategies that do not block, which always creates a new data mapper if
762 /// the pool is empty, regardless of the maximum capacity
764 requires(Config.growthStrategy != GrowthStrategy::BoundedWait)
765 {
766 std::vector<Entry> retired; // declared before the lock: disconnects run after it is released
767 std::scoped_lock lock(_mutex);
768 auto entry = TakeIdleLocked(retired);
769 if (!entry.mapper)
770 {
771 // no usable idle data mapper: create a new one and return it
772 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 }, false, false);
773 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
774 return PooledDataMapper(*this, MakeEntry());
775 }
776 SqlLogger::GetLogger().OnConnectionReuse(entry.mapper->Connection());
777 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 }, true, false);
778 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
779 return PooledDataMapper(*this, std::move(entry));
780 }
781
782 /// Acquires a data mapper, for the strategies that never wait.
783 ///
784 /// Provided so call sites can be written without knowing the strategy. These strategies create a
785 /// fresh data mapper whenever no idle one is available, so the timeout can never elapse and the
786 /// result always holds a value.
787 ///
788 /// @param timeout Ignored; see above.
789 /// @return The acquired data mapper.
790 [[nodiscard]] std::expected<PooledDataMapper, PoolError> Acquire([[maybe_unused]] std::chrono::milliseconds timeout)
791 requires(Config.growthStrategy != GrowthStrategy::BoundedWait)
792 {
793 return Acquire();
794 }
795
796 /// Asynchronously acquires a DataMapper from the pool without blocking the calling thread.
797 ///
798 /// If the pool is exhausted (BoundedWait at capacity), the awaiting coroutine is suspended and
799 /// resumed — via @p resume — when a mapper is returned, rather than parking a thread. The acquired
800 /// mapper's connection is wired for async via SqlConnection::EnableAsync(@p dbWorkers, @p resume),
801 /// so the caller can immediately co_await its async methods.
802 ///
803 /// @param dbWorkers The worker pool used to run the acquired mapper's blocking ODBC calls.
804 /// @param resume The scheduler used to resume coroutines (typically the app run loop).
805 /// @return A Task yielding a pooled DataMapper.
807 {
808 // Forward to a coroutine taking pointers (coroutines must not take reference parameters).
809 return AcquireAsyncImpl(&dbWorkers, &resume);
810 }
811
812 /// Configures the executors that the no-argument @ref AcquireAsync() overload wires acquired
813 /// mappers for, so async consumers of this pool no longer repeat the executors at every call.
814 ///
815 /// This is opt-in and scoped to the pool: only pools configured this way hand out async-enabled
816 /// mappers via the no-arg overload; synchronous @ref Acquire and connections outside the pool are
817 /// unaffected. Unlike a process-global default, the executors' lifetime is tied to this pool,
818 /// which already must outlive every acquirer.
819 ///
820 /// @warning @p dbWorkers and @p resume must outlive this pool's async use (the same contract the
821 /// explicit-argument @ref AcquireAsync overload already implies). Only references are
822 /// retained. Intended to be called once during setup, before any concurrent
823 /// @ref AcquireAsync(); it is not synchronized against in-flight acquirers.
824 /// @param dbWorkers The worker pool used to run acquired mappers' blocking ODBC calls.
825 /// @param resume The scheduler used to resume coroutines (typically the app run loop).
827 {
828 _asyncDbWorkers = &dbWorkers;
829 _asyncResume = &resume;
830 }
831
832 /// Asynchronously acquires a DataMapper using the executors previously set via
833 /// @ref SetAsyncExecutors, without blocking the calling thread.
834 ///
835 /// Equivalent to the explicit-argument @ref AcquireAsync overload with the pool's stored
836 /// executors; pass the executors to that overload to override them for a single call.
837 ///
838 /// @return A Task yielding a pooled DataMapper.
839 /// @throws std::logic_error if @ref SetAsyncExecutors has not been called on this pool.
841 {
842 if (!_asyncDbWorkers || !_asyncResume)
843 throw std::logic_error {
844 "Pool::AcquireAsync(): no async executors configured; call Pool::SetAsyncExecutors(...) first "
845 "or use the explicit AcquireAsync(dbWorkers, resume) overload."
846 };
847 return AcquireAsyncImpl(_asyncDbWorkers, _asyncResume);
848 }
849
850 /// Overrides the clock the idle-time and lifetime bounds are measured against, so eviction can be
851 /// driven deterministically (from a test, or from a simulated clock) instead of by sleeping.
852 ///
853 /// @warning Like @ref SetAsyncExecutors, this is setup, not a runtime knob: it is not
854 /// synchronized against in-flight acquirers, so call it before any concurrent use of the
855 /// pool. Advancing whatever time @p clock reports is the caller's business afterwards.
856 /// @param clock Source of the current time; pass @c {} to restore the real clock.
857 void SetClock(std::function<Clock::time_point()> clock) noexcept
858 {
859 _clock = std::move(clock);
860 }
861
862 /// The source through which records read from this pool's mappers load their relations.
863 ///
864 /// Each lazy relation load borrows a mapper from this pool for its own duration, so it counts
865 /// against the pool's capacity, is health-checked like any other checkout, and uses the pool's
866 /// per-connection settings. A load never waits for a connection, though: when a
867 /// @ref GrowthStrategy::BoundedWait pool is at capacity - typically because the caller still holds
868 /// the mapper the record was read through - the load runs on a one-off connection instead.
869 ///
870 /// @return The source; never null. Outlives the pool as long as records hold it, falling back
871 /// to a fresh connection with the default connection string once the pool is destroyed.
872 [[nodiscard]] std::shared_ptr<detail::RelationLoadSource> LoadSourceForRelations() const noexcept
873 {
874 return _relationLoadSource;
875 }
876
877#if defined(BUILD_TESTS)
878 [[nodiscard]] size_t IdleCount() noexcept
879 {
880 std::scoped_lock lock(_mutex);
881 return _idleDataMappers.size();
882 }
883
884 /// @return the number of parked acquirers (blocked Acquire() threads + suspended AcquireAsync
885 /// coroutines); lets tests observe parking/fairness deterministically.
886 [[nodiscard]] size_t WaiterCount() noexcept
887 {
888 std::scoped_lock lock(_mutex);
889 return _waiters.size();
890 }
891#endif
892
893 private:
894 /// Lends this pool's mappers to relation auto-loaders, one load at a time.
895 ///
896 /// @ref MakeEntry attaches it - pinned to the entry's connection string - to every mapper this pool
897 /// creates; the loaders that mapper then installs on the records it reads call @c Borrow(). See
898 /// @c detail::RelationLoadSource for the full call path.
899 class PoolRelationLoadSource final:
900 public detail::RelationLoadSource,
901 public std::enable_shared_from_this<PoolRelationLoadSource>
902 {
903 public:
904 explicit PoolRelationLoadSource(Pool& pool) noexcept:
905 _pool { &pool }
906 {
907 }
908
909 [[nodiscard]] RelationResult<std::shared_ptr<DataMapper>> Borrow() override
910 {
911 try
912 {
913 return Lend();
914 }
915 catch (std::exception const& error)
916 {
917 // Connecting failed (no idle connection to reuse): reported, not thrown.
918 SqlLogger::GetLogger().OnWarning(std::format("Connecting for a relation load failed: {}", error.what()));
919 return std::unexpected { RelationError::QueryFailed };
920 }
921 }
922
923 /// Called by the pool's destructor; later loads connect on their own.
924 void Detach() noexcept
925 {
926 auto const lock = std::unique_lock { _mutex };
927 _pool = nullptr;
928 }
929
930 private:
931 /// @return A mapper for one load; throws whatever connecting throws.
932 [[nodiscard]] std::shared_ptr<DataMapper> Lend()
933 {
934 // Shared: concurrent loads borrow in parallel; only Detach() excludes them, so the pool
935 // cannot be destroyed while one of them is still inside Acquire().
936 auto const lock = std::shared_lock { _mutex };
937 if (!_pool)
938 return OwnConnection();
939 if constexpr (Config.growthStrategy == GrowthStrategy::BoundedWait)
940 {
941 // Never wait: the caller commonly still holds the pooled mapper the record came from,
942 // and a load nested in Each() holds one more, so waiting on a small pool for a slot the
943 // caller itself occupies would deadlock. At capacity, a one-off connection serves this
944 // load and is closed afterwards.
945 if (auto pooled = _pool->Acquire(std::chrono::milliseconds { 0 }); pooled)
946 return Share(std::move(*pooled));
947 return OwnConnection();
948 }
949 else
950 return Share(_pool->Acquire());
951 }
952
953 /// @return @p pooled as a shared mapper that goes back to the pool with its last copy.
954 [[nodiscard]] static std::shared_ptr<DataMapper> Share(PooledDataMapper pooled)
955 {
956 auto holder = std::make_shared<PooledDataMapper>(std::move(pooled));
957 return { holder, &holder->Get() };
958 }
959
960 /// @return A connection of its own, for a load the pool cannot serve without waiting (or once
961 /// the pool is gone). Records read through it still load their relations from here.
962 [[nodiscard]] std::shared_ptr<DataMapper> OwnConnection()
963 {
964 auto const generation = SqlConnection::DefaultConnectionStringGeneration();
965 auto mapper = std::make_shared<DataMapper>();
966 detail::AdoptRelationLoadSource(*mapper,
967 detail::PinToDefaultConnectionString(this->shared_from_this(),
968 mapper->Connection().ConnectionString(),
969 generation));
970 return mapper;
971 }
972
973 std::shared_mutex _mutex;
974 Pool* _pool;
975 };
976
977 /// A parked acquirer awaiting a DataMapper — a suspended @ref AcquireAsync coroutine (@c Kind::Async)
978 /// or a blocked synchronous @ref Acquire thread (@c Kind::Sync). Both share one FIFO queue
979 /// (@c _waiters), served in arrival order so neither kind starves the other.
980 ///
981 /// Heap-allocated and shared with the pool. Holding the handed-off mapper and @c state in this node
982 /// (not via a pointer into a coroutine frame) lets @ref Return and the awaitable's destructor
983 /// coordinate purely through node state under @c _mutex, never touching a possibly-destroyed frame.
984 struct WaiterNode
985 {
986 /// Whether this waiter is a suspended coroutine or a blocked synchronous Acquire() thread.
987 enum class Kind : std::uint8_t
988 {
989 Sync, ///< A blocked @ref Acquire thread; woken via @c cv.
990 Async, ///< A suspended @ref AcquireAsync coroutine; resumed via @c resume / @c handle.
991 };
992
993 /// Liveness of the waiter, transitioned only under @c Pool::_mutex.
994 enum class State : std::uint8_t
995 {
996 Parked, ///< Registered in @c _waiters, awaiting a mapper.
997 Fulfilled, ///< Return handed it a mapper (in @c mapper) and woke/scheduled it.
998 Abandoned, ///< The awaiting async task was destroyed (or its entry consumed); inert.
999 };
1000
1001 Kind kind;
1002 State state = State::Parked;
1003 Entry entry {}; ///< Filled by Return on hand-off; lives outside any frame.
1004
1005 // Async waiter only:
1006 std::coroutine_handle<> handle {};
1007 Async::IResumeScheduler* resume = nullptr;
1008
1009 // Sync waiter only: the blocked Acquire() waits on this CV (under Pool::_mutex). One waiter per
1010 // CV, so Return's notify_one wakes exactly the served thread.
1011 std::condition_variable cv {};
1012
1013 explicit WaiterNode(Kind nodeKind) noexcept:
1014 kind { nodeKind }
1015 {
1016 }
1017 };
1018
1019 /// Awaitable that acquires a DataMapper, suspending only when the pool is at capacity.
1020 ///
1021 /// Non-copyable/non-movable: constructed in place in the co_await expression. On suspension it
1022 /// registers a shared @ref WaiterNode (@c Kind::Async) in pool._waiters; the node carries the
1023 /// handed-off mapper and liveness state so Return() and this destructor coordinate safely.
1024 struct AsyncAcquireAwaitable
1025 {
1026 Pool& pool;
1027 Async::IResumeScheduler& resume;
1028 Entry acquired {}; ///< Entry obtained without suspending (idle/fresh).
1029 std::shared_ptr<WaiterNode> node {}; ///< Set only while parked; shared with the pool.
1030 std::vector<Entry> retired {}; ///< Entries retired while scanning the idle set.
1031 /// When this acquisition parked, used to attribute its wait latency. Only read when @c node
1032 /// is set, so it needs no value on the non-parking paths.
1033 std::chrono::steady_clock::time_point parkedAt {};
1034
1035 AsyncAcquireAwaitable(Pool& poolRef, Async::IResumeScheduler& resumeRef) noexcept:
1036 pool { poolRef },
1037 resume { resumeRef }
1038 {
1039 }
1040
1041 AsyncAcquireAwaitable(AsyncAcquireAwaitable const&) = delete;
1042 AsyncAcquireAwaitable& operator=(AsyncAcquireAwaitable const&) = delete;
1043 AsyncAcquireAwaitable(AsyncAcquireAwaitable&&) = delete;
1044 AsyncAcquireAwaitable& operator=(AsyncAcquireAwaitable&&) = delete;
1045
1046 /// Cleans up if the awaiting coroutine is destroyed before it consumes its mapper.
1047 ///
1048 /// Under pool._mutex: if still parked, de-registers the node so a later Return() never hands
1049 /// off to a dead frame. If Return() already handed off a mapper (Fulfilled) that await_resume
1050 /// never consumed, reclaims it into the pool so the BoundedWait checked-out count is not
1051 /// leaked (possibly handing it straight to the next waiter, resumed after the lock is released).
1052 ///
1053 /// @warning A task that has already been handed a mapper must still be driven to completion:
1054 /// the resumption Return() scheduled cannot be cancelled, so a coroutine frame with a pending
1055 /// resumption must not be freed (do not destroy such a task concurrently with, or right after,
1056 /// the hand-off). Likewise the pool must outlive every task acquired from it.
1057 ~AsyncAcquireAwaitable()
1058 {
1059 if (!node)
1060 return;
1061 Entry reclaimed; // declared before the lock so its disconnect runs after the lock is released
1062 std::shared_ptr<WaiterNode> toResume;
1063 {
1064 std::scoped_lock const lock(pool._mutex);
1065 switch (node->state)
1066 {
1067 case WaiterNode::State::Parked:
1068 // Never fulfilled: remove ourselves so Return() won't hand off to a dead frame.
1069 // Parking never incremented _checkedOut, so there is nothing to release.
1070 node->state = WaiterNode::State::Abandoned;
1071 std::erase(pool._waiters, node);
1072 break;
1073 case WaiterNode::State::Fulfilled:
1074 // Handed a mapper but the task is dropped before consuming it: reclaim it,
1075 // releasing this acquisition's checked-out count so the pool does not leak.
1076 node->state = WaiterNode::State::Abandoned;
1077 if constexpr (Config.growthStrategy == GrowthStrategy::BoundedWait)
1078 {
1079 if (node->entry.mapper)
1080 toResume = pool.ReturnLocked(std::move(node->entry), reclaimed);
1081 else
1082 toResume = pool.ReleaseSlotLocked(); // handed a bare slot
1083 }
1084 break;
1085 case WaiterNode::State::Abandoned:
1086 break;
1087 }
1088 }
1089 if (toResume)
1090 toResume->resume->Resume(toResume->handle);
1091 }
1092
1093 [[nodiscard]] bool await_ready() const noexcept
1094 {
1095 return false;
1096 }
1097
1098 bool await_suspend(std::coroutine_handle<> handle)
1099 {
1100 std::scoped_lock const lock(pool._mutex);
1101 // Retired entries land in `retired`, a member of this awaitable, so the disconnects they
1102 // trigger happen when the awaitable dies rather than under pool._mutex.
1103 acquired = pool.TakeIdleLocked(retired);
1104 if (acquired.mapper)
1105 {
1106 if constexpr (Config.growthStrategy == GrowthStrategy::BoundedWait)
1107 ++pool._checkedOut;
1108 SqlLogger::GetLogger().OnConnectionReuse(acquired.mapper->Connection());
1109 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 }, true, false);
1110 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(pool._idleDataMappers.size(), pool._checkedOut);
1111 return false; // do not suspend — resume immediately
1112 }
1113 // Only BoundedWait bounds the pool and parks coroutines on exhaustion. The non-blocking
1114 // strategies (BoundedOverflow — the default — and UnboundedGrow) always create a fresh
1115 // mapper here, matching the synchronous Acquire() overloads, which also never suspend.
1116 if constexpr (Config.growthStrategy == GrowthStrategy::BoundedWait)
1117 {
1118 if (pool._checkedOut >= Config.maxSize)
1119 {
1120 node = std::make_shared<WaiterNode>(WaiterNode::Kind::Async);
1121 node->handle = handle;
1122 node->resume = &resume;
1123 pool._waiters.push_back(node);
1124 parkedAt = std::chrono::steady_clock::now();
1125 return true; // suspend until a mapper is returned
1126 }
1127 }
1128 // Below capacity: create a fresh data mapper. As in AcquireReadyLocked, claim the slot
1129 // only once the connection actually stands up — MakeEntry() connects and may throw, and a
1130 // slot claimed before that would never be released, permanently shrinking a BoundedWait
1131 // pool's capacity.
1132 auto fresh = pool.MakeEntry();
1133 if constexpr (Config.growthStrategy == GrowthStrategy::BoundedWait)
1134 ++pool._checkedOut;
1135 acquired = std::move(fresh);
1136 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 }, false, false);
1137 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(pool._idleDataMappers.size(), pool._checkedOut);
1138 return false;
1139 }
1140
1141 Entry await_resume() noexcept
1142 {
1143 // If we suspended, Return() placed the entry in the shared node; take it here (on the
1144 // resuming thread, with no concurrent access per the destruction contract). That leaves
1145 // node->entry empty, so the destructor treats the node as already consumed.
1146 if (node)
1147 {
1148 // This acquisition actually parked the coroutine, so it contributes a wait sample.
1149 LIGHTWEIGHT_STATS_POOL_ACQUIRE(
1150 std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - parkedAt),
1151 true,
1152 true);
1153 return std::move(node->entry);
1154 }
1155 return std::move(acquired);
1156 }
1157 };
1158
1159 Async::Task<PooledDataMapper> AcquireAsyncImpl(Async::IExecutor* dbWorkers, Async::IResumeScheduler* resume)
1160 {
1161 auto entry = co_await AsyncAcquireAwaitable { *this, *resume };
1162 if constexpr (Config.growthStrategy == GrowthStrategy::BoundedWait)
1163 {
1164 // Handed a bare slot: the connection returned for this waiter was stale (see ReturnLocked).
1165 if (!entry.mapper)
1166 entry = ConnectForTransferredSlot();
1167 }
1168 // Wrap in the RAII PooledDataMapper BEFORE the throwing EnableAsync call: if EnableAsync
1169 // throws (e.g. bad_alloc), ~PooledDataMapper returns the mapper to the pool, decrementing
1170 // _checkedOut and avoiding a permanent BoundedWait capacity leak.
1171 auto pooled = PooledDataMapper(*this, std::move(entry));
1172 pooled->Connection().EnableAsync(*dbWorkers, *resume);
1173 co_return std::move(pooled);
1174 }
1175
1176 std::mutex _mutex;
1177 std::vector<Entry> _idleDataMappers;
1178 size_t _checkedOut {};
1179 /// Shared with every record read through this pool; see @ref LoadSourceForRelations().
1180 std::shared_ptr<PoolRelationLoadSource> _relationLoadSource = std::make_shared<PoolRelationLoadSource>(*this);
1181 /// Injected clock; the real @c Clock::now is used when unset. @see SetClock
1182 ///
1183 /// Deliberately not compiled out in non-test builds: this is a data member of a class template
1184 /// instantiated both inside the library (@ref GlobalDataMapperPool) and in consumer translation
1185 /// units, so making its presence depend on a translation-unit-local macro would give the same
1186 /// specialization two different layouts and two different definitions.
1187 std::function<Clock::time_point()> _clock {};
1188 /// Executors used by the no-argument @ref AcquireAsync() overload; set via @ref SetAsyncExecutors.
1189 /// Null until configured. Only references are held; they must outlive the pool's async use.
1190 Async::IExecutor* _asyncDbWorkers = nullptr;
1191 Async::IResumeScheduler* _asyncResume = nullptr;
1192 /// FIFO of parked acquirers (sync @ref Acquire threads and async @ref AcquireAsync coroutines) in
1193 /// arrival order. Each sync waiter owns its CV inside its @ref WaiterNode, so no shared CV is needed.
1194 std::deque<std::shared_ptr<WaiterNode>> _waiters;
1195};
1196
1197// Default pool configuration, configurable via CMake options:
1198// LIGHTWEIGHT_POOL_INITIAL_SIZE (default: 4)
1199// LIGHTWEIGHT_POOL_MAX_SIZE (default: 16)
1200// LIGHTWEIGHT_POOL_GROWTH_STRATEGY (default: BoundedOverflow)
1201// Accepted values: BoundedWait, BoundedOverflow, UnboundedGrow
1202// LIGHTWEIGHT_POOL_VALIDATE_ON_BORROW (default: Yes)
1203// Accepted values: Yes, No
1204// LIGHTWEIGHT_POOL_MAX_IDLE_TIME_MS (default: 0, meaning no bound)
1205// LIGHTWEIGHT_POOL_MAX_LIFETIME_MS (default: 0, meaning no bound)
1206// LIGHTWEIGHT_POOL_PREPARED_STATEMENT_CACHE_CAPACITY (default: 0, i.e. disabled)
1207
1208#if !defined(LIGHTWEIGHT_POOL_INITIAL_SIZE)
1209 #define LIGHTWEIGHT_POOL_INITIAL_SIZE 4
1210#endif
1211
1212#if !defined(LIGHTWEIGHT_POOL_MAX_SIZE)
1213 #define LIGHTWEIGHT_POOL_MAX_SIZE 16
1214#endif
1215
1216#if !defined(LIGHTWEIGHT_POOL_GROWTH_STRATEGY)
1217 #define LIGHTWEIGHT_POOL_GROWTH_STRATEGY BoundedOverflow
1218#endif
1219
1220#if !defined(LIGHTWEIGHT_POOL_VALIDATE_ON_BORROW)
1221 #define LIGHTWEIGHT_POOL_VALIDATE_ON_BORROW Yes
1222#endif
1223
1224// The lifetime bounds default to 0 (disabled): a default recycle window would silently change the
1225// behaviour of every existing deployment, and the right value depends on the infrastructure the
1226// connections traverse. See PoolConfig for how to choose one.
1227#if !defined(LIGHTWEIGHT_POOL_MAX_IDLE_TIME_MS)
1228 #define LIGHTWEIGHT_POOL_MAX_IDLE_TIME_MS 0
1229#endif
1230
1231#if !defined(LIGHTWEIGHT_POOL_MAX_LIFETIME_MS)
1232 #define LIGHTWEIGHT_POOL_MAX_LIFETIME_MS 0
1233#endif
1234
1235#if !defined(LIGHTWEIGHT_POOL_PREPARED_STATEMENT_CACHE_CAPACITY)
1236 #define LIGHTWEIGHT_POOL_PREPARED_STATEMENT_CACHE_CAPACITY 0
1237#endif
1238
1239inline constexpr PoolConfig DefaultPoolConfig {
1240 .initialSize = LIGHTWEIGHT_POOL_INITIAL_SIZE,
1241 .maxSize = LIGHTWEIGHT_POOL_MAX_SIZE,
1242 .growthStrategy = GrowthStrategy::LIGHTWEIGHT_POOL_GROWTH_STRATEGY,
1243 .validateOnBorrow = ValidateOnBorrow::LIGHTWEIGHT_POOL_VALIDATE_ON_BORROW,
1244 .maxIdleTimeMs = LIGHTWEIGHT_POOL_MAX_IDLE_TIME_MS,
1245 .maxLifetimeMs = LIGHTWEIGHT_POOL_MAX_LIFETIME_MS,
1246 .preparedStatementCacheCapacity = LIGHTWEIGHT_POOL_PREPARED_STATEMENT_CACHE_CAPACITY,
1247};
1248
1249using DataMapperPool = Pool<DefaultPoolConfig>;
1250
1251/// Returns the process-wide global DataMapper pool.
1252///
1253/// The pool is configured at compile time via the LIGHTWEIGHT_POOL_* defines.
1254/// Because the singleton lives inside the Lightweight library, it is shared
1255/// correctly across shared-library boundaries.
1256LIGHTWEIGHT_API DataMapperPool& GlobalDataMapperPool();
1257
1258} // namespace Lightweight
Main API for mapping records to and from the database using high level C++ syntax.
DataMapper * operator->() const noexcept
Access the underlying data mapper via pointer semantics.
Definition Pool.hpp:243
PooledDataMapper(PooledDataMapper &&other) noexcept
Definition Pool.hpp:229
DataMapper & Get() const noexcept
Definition Pool.hpp:251
~Pool() noexcept
Definition Pool.hpp:671
std::expected< PooledDataMapper, PoolError > Acquire(std::chrono::milliseconds timeout)
Definition Pool.hpp:730
std::expected< PooledDataMapper, PoolError > Acquire(std::chrono::milliseconds timeout)
Definition Pool.hpp:790
std::shared_ptr< detail::RelationLoadSource > LoadSourceForRelations() const noexcept
Definition Pool.hpp:872
PooledDataMapper Acquire()
Definition Pool.hpp:763
Async::Task< PooledDataMapper > AcquireAsync(Async::IExecutor &dbWorkers, Async::IResumeScheduler &resume)
Definition Pool.hpp:806
void SetClock(std::function< Clock::time_point()> clock) noexcept
Definition Pool.hpp:857
PooledDataMapper Acquire()
Definition Pool.hpp:699
void SetAsyncExecutors(Async::IExecutor &dbWorkers, Async::IResumeScheduler &resume) noexcept
Definition Pool.hpp:826
Async::Task< PooledDataMapper > AcquireAsync()
Definition Pool.hpp:840
static LIGHTWEIGHT_API std::uint32_t DefaultConnectionStringGeneration() noexcept
static LIGHTWEIGHT_API SqlLogger & GetLogger()
Retrieves the currently configured logger.
virtual void OnConnectionReuse(SqlConnection const &connection)=0
Invoked when a connection is reused.
virtual void OnConnectionIdle(SqlConnection const &connection)=0
Invoked when a connection is idle.
virtual void OnWarning(std::string_view const &message)=0
Invoked on a warning.
ValidateOnBorrow
Definition Pool.hpp:60
@ Timeout
The timeout elapsed before a data mapper became available.
constexpr std::chrono::milliseconds MaxIdleTime() const noexcept
Definition Pool.hpp:149
constexpr std::chrono::milliseconds MaxLifetime() const noexcept
Definition Pool.hpp:155
ValidateOnBorrow validateOnBorrow
Definition Pool.hpp:108
GrowthStrategy growthStrategy
Definition Pool.hpp:103
size_t initialSize
Initial number of data mappers to pre-create and store in the pool, must be less than or equal to max...
Definition Pool.hpp:96
size_t preparedStatementCacheCapacity
Definition Pool.hpp:146
std::chrono::milliseconds::rep maxLifetimeMs
Definition Pool.hpp:134
std::chrono::milliseconds::rep maxIdleTimeMs
Definition Pool.hpp:122