171 using Clock = std::chrono::steady_clock;
175 static constexpr bool TracksTime = Config.maxIdleTimeMs > 0 || Config.maxLifetimeMs > 0;
192 std::unique_ptr<DataMapper> mapper;
193 Clock::time_point createdAt {};
194 Clock::time_point idleSince {};
195 std::uint32_t generation {};
198 bool unusable =
false;
218 _entry { std::move(entry) },
230 _entry { std::move(other._entry) },
231 _pool { other._pool }
234 PooledDataMapper& operator=(PooledDataMapper
const&) =
delete;
235 PooledDataMapper& operator=(PooledDataMapper&&) =
delete;
236 ~PooledDataMapper() noexcept
245 return _entry.mapper.get();
253 return *_entry.mapper;
257 void ReturnToPool() noexcept
259 _pool.Return(std::move(_entry));
260 _entry.mapper =
nullptr;
282 static void ResetForReuse(Entry& entry)
noexcept
284 auto& connection = entry.mapper->Connection();
285 connection.DisableAsync();
286 if (!connection.RollbackOpenTransaction())
287 entry.unusable =
true;
292 [[nodiscard]] Clock::time_point NowIfTracking() const noexcept
294 if constexpr (!TracksTime)
306 [[nodiscard]] Entry MakeEntry()
const
308 auto const now = NowIfTracking();
313 auto mapper = std::make_unique<DataMapper>();
317 if constexpr (Config.preparedStatementCacheCapacity != 0)
318 mapper->Connection().SetPreparedStatementCacheCapacity(Config.preparedStatementCacheCapacity);
321 detail::AdoptRelationLoadSource(
323 detail::PinToDefaultConnectionString(_relationLoadSource, mapper->Connection().ConnectionString(), generation));
324 return Entry { .mapper = std::move(mapper), .createdAt = now, .idleSince = now, .generation = generation };
336 [[nodiscard]]
bool IsUsable(Entry
const& entry, Clock::time_point now)
const noexcept
338 if (MustRetire(entry, now))
340 if constexpr (Config.maxIdleTimeMs > 0)
342 if (now - entry.idleSince >= Config.MaxIdleTime())
347 if (!entry.mapper->Connection().IsAlive())
362 [[nodiscard]] Entry TakeIdleLocked(std::vector<Entry>& retired)
364 auto const now = NowIfTracking();
365 while (!_idleDataMappers.empty())
367 auto entry = std::move(_idleDataMappers.back());
368 _idleDataMappers.pop_back();
369 if (IsUsable(entry, now))
371 retired.push_back(std::move(entry));
378 [[nodiscard]]
static bool IsCurrentGeneration(Entry
const& entry)
noexcept
393 [[nodiscard]]
static bool MustRetire(Entry
const& entry, [[maybe_unused]] Clock::time_point now)
noexcept
395 if (entry.unusable || !IsCurrentGeneration(entry))
397 if constexpr (Config.maxLifetimeMs > 0)
398 return now - entry.createdAt >= Config.MaxLifetime();
404 void Return(Entry entry)
noexcept
407 ResetForReuse(entry);
408 auto const now = NowIfTracking();
409 if (MustRetire(entry, now))
413 LIGHTWEIGHT_STATS_POOL_RELEASE(
true);
416 entry.idleSince = now;
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);
426 void Return(Entry entry)
noexcept
429 ResetForReuse(entry);
431 std::shared_ptr<WaiterNode> toResume;
433 std::scoped_lock
const lock(_mutex);
434 toResume = ReturnLocked(std::move(entry), retired);
438 toResume->resume->Resume(toResume->handle);
447 PooledDataMapper TakeHandedOff(WaiterNode& node, std::unique_lock<std::mutex>& lock)
450 if (node.entry.mapper)
451 return PooledDataMapper(*
this, std::move(node.entry));
453 return PooledDataMapper(*
this, ConnectForTransferredSlot());
463 [[nodiscard]] Entry AcquireReadyLocked(std::vector<Entry>& retired)
466 if (
auto entry = TakeIdleLocked(retired); entry.mapper)
472 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 },
true,
false);
473 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
476 if (_checkedOut < Config.maxSize)
480 auto fresh = MakeEntry();
482 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 },
false,
false);
483 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
500 std::shared_ptr<WaiterNode> ReturnLocked(Entry entry, Entry& retired)
noexcept
503 if (entry.unusable || !IsCurrentGeneration(entry))
509 retired = std::move(entry);
510 LIGHTWEIGHT_STATS_POOL_RELEASE(
true);
511 return ReleaseSlotLocked();
513 if (
auto node = NextParkedWaiterLocked())
524 LIGHTWEIGHT_STATS_POOL_RELEASE(
false);
525 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
526 return FulfilLocked(std::move(node), std::move(entry));
531 auto const now = NowIfTracking();
532 if (MustRetire(entry, now))
534 retired = std::move(entry);
537 LIGHTWEIGHT_STATS_POOL_RELEASE(
true);
538 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
541 entry.idleSince = now;
543 _idleDataMappers.push_back(std::move(entry));
544 LIGHTWEIGHT_STATS_POOL_RELEASE(
false);
545 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
556 std::shared_ptr<WaiterNode> ReleaseSlotLocked() noexcept
559 if (
auto node = NextParkedWaiterLocked())
560 return FulfilLocked(std::move(node), Entry {});
562 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
570 [[nodiscard]] std::shared_ptr<WaiterNode> NextParkedWaiterLocked() noexcept
572 while (!_waiters.empty())
574 auto node = std::move(_waiters.front());
575 _waiters.pop_front();
578 if (node->state == WaiterNode::State::Parked)
591 [[nodiscard]]
static std::shared_ptr<WaiterNode> FulfilLocked(std::shared_ptr<WaiterNode> node, Entry entry)
noexcept
593 node->state = WaiterNode::State::Fulfilled;
594 node->entry = std::move(entry);
595 if (node->kind == WaiterNode::Kind::Async)
597 node->cv.notify_one();
608 [[nodiscard]] Entry ConnectForTransferredSlot()
617 std::shared_ptr<WaiterNode> toResume;
619 std::scoped_lock
const lock(_mutex);
620 toResume = ReleaseSlotLocked();
623 toResume->resume->Resume(toResume->handle);
629 void Return(Entry entry)
noexcept
632 ResetForReuse(entry);
633 auto const now = NowIfTracking();
634 if (MustRetire(entry, now))
638 LIGHTWEIGHT_STATS_POOL_RELEASE(
true);
641 entry.idleSince = now;
642 std::scoped_lock lock(_mutex);
643 if (_idleDataMappers.size() < Config.maxSize)
646 _idleDataMappers.push_back(std::move(entry));
647 LIGHTWEIGHT_STATS_POOL_RELEASE(
false);
653 LIGHTWEIGHT_STATS_POOL_RELEASE(
true);
655 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
663 _idleDataMappers.reserve(Config.initialSize);
664 for ([[maybe_unused]]
auto const _: std::views::iota(0U, Config.initialSize))
665 _idleDataMappers.push_back(MakeEntry());
674 _relationLoadSource->Detach();
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");
689 Pool& operator=(
Pool const&) =
delete;
702 std::vector<Entry> retired;
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)
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; });
714 LIGHTWEIGHT_STATS_POOL_ACQUIRE(
715 std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - acquireStartedAt),
718 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
719 return TakeHandedOff(*node, lock);
730 [[nodiscard]] std::expected<PooledDataMapper, PoolError>
Acquire(std::chrono::milliseconds timeout)
733 std::vector<Entry> retired;
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)
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; }))
746 node->state = WaiterNode::State::Abandoned;
747 std::erase(_waiters, node);
752 LIGHTWEIGHT_STATS_POOL_ACQUIRE(
753 std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - acquireStartedAt),
756 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
757 return TakeHandedOff(*node, lock);
766 std::vector<Entry> retired;
767 std::scoped_lock lock(_mutex);
768 auto entry = TakeIdleLocked(retired);
772 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 },
false,
false);
773 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
777 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 },
true,
false);
778 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(_idleDataMappers.size(), _checkedOut);
790 [[nodiscard]] std::expected<PooledDataMapper, PoolError>
Acquire([[maybe_unused]] std::chrono::milliseconds timeout)
809 return AcquireAsyncImpl(&dbWorkers, &resume);
828 _asyncDbWorkers = &dbWorkers;
829 _asyncResume = &resume;
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."
847 return AcquireAsyncImpl(_asyncDbWorkers, _asyncResume);
857 void SetClock(std::function<Clock::time_point()> clock)
noexcept
859 _clock = std::move(clock);
874 return _relationLoadSource;
877#if defined(BUILD_TESTS)
878 [[nodiscard]]
size_t IdleCount() noexcept
880 std::scoped_lock lock(_mutex);
881 return _idleDataMappers.size();
886 [[nodiscard]]
size_t WaiterCount() noexcept
888 std::scoped_lock lock(_mutex);
889 return _waiters.size();
899 class PoolRelationLoadSource final:
900 public detail::RelationLoadSource,
901 public std::enable_shared_from_this<PoolRelationLoadSource>
904 explicit PoolRelationLoadSource(
Pool& pool)
noexcept:
909 [[nodiscard]] RelationResult<std::shared_ptr<DataMapper>> Borrow()
override
915 catch (std::exception
const& error)
924 void Detach() noexcept
926 auto const lock = std::unique_lock { _mutex };
932 [[nodiscard]] std::shared_ptr<DataMapper> Lend()
936 auto const lock = std::shared_lock { _mutex };
938 return OwnConnection();
945 if (
auto pooled = _pool->Acquire(std::chrono::milliseconds { 0 }); pooled)
946 return Share(std::move(*pooled));
947 return OwnConnection();
950 return Share(_pool->Acquire());
954 [[nodiscard]]
static std::shared_ptr<DataMapper> Share(PooledDataMapper pooled)
956 auto holder = std::make_shared<PooledDataMapper>(std::move(pooled));
957 return { holder, &holder->Get() };
962 [[nodiscard]] std::shared_ptr<DataMapper> OwnConnection()
965 auto mapper = std::make_shared<DataMapper>();
966 detail::AdoptRelationLoadSource(*mapper,
967 detail::PinToDefaultConnectionString(this->shared_from_this(),
968 mapper->Connection().ConnectionString(),
973 std::shared_mutex _mutex;
987 enum class Kind : std::uint8_t
994 enum class State : std::uint8_t
1002 State state = State::Parked;
1006 std::coroutine_handle<> handle {};
1007 Async::IResumeScheduler* resume =
nullptr;
1011 std::condition_variable cv {};
1013 explicit WaiterNode(Kind nodeKind)
noexcept:
1024 struct AsyncAcquireAwaitable
1027 Async::IResumeScheduler& resume;
1029 std::shared_ptr<WaiterNode> node {};
1030 std::vector<Entry> retired {};
1033 std::chrono::steady_clock::time_point parkedAt {};
1035 AsyncAcquireAwaitable(
Pool& poolRef, Async::IResumeScheduler& resumeRef)
noexcept:
1037 resume { resumeRef }
1041 AsyncAcquireAwaitable(AsyncAcquireAwaitable
const&) =
delete;
1042 AsyncAcquireAwaitable& operator=(AsyncAcquireAwaitable
const&) =
delete;
1043 AsyncAcquireAwaitable(AsyncAcquireAwaitable&&) =
delete;
1044 AsyncAcquireAwaitable& operator=(AsyncAcquireAwaitable&&) =
delete;
1057 ~AsyncAcquireAwaitable()
1062 std::shared_ptr<WaiterNode> toResume;
1064 std::scoped_lock
const lock(pool._mutex);
1065 switch (node->state)
1067 case WaiterNode::State::Parked:
1070 node->state = WaiterNode::State::Abandoned;
1071 std::erase(pool._waiters, node);
1073 case WaiterNode::State::Fulfilled:
1076 node->state = WaiterNode::State::Abandoned;
1079 if (node->entry.mapper)
1080 toResume = pool.ReturnLocked(std::move(node->entry), reclaimed);
1082 toResume = pool.ReleaseSlotLocked();
1085 case WaiterNode::State::Abandoned:
1090 toResume->resume->Resume(toResume->handle);
1093 [[nodiscard]]
bool await_ready() const noexcept
1098 bool await_suspend(std::coroutine_handle<> handle)
1100 std::scoped_lock
const lock(pool._mutex);
1103 acquired = pool.TakeIdleLocked(retired);
1104 if (acquired.mapper)
1109 LIGHTWEIGHT_STATS_POOL_ACQUIRE(std::chrono::microseconds { 0 },
true,
false);
1110 LIGHTWEIGHT_STATS_POOL_OCCUPANCY(pool._idleDataMappers.size(), pool._checkedOut);
1118 if (pool._checkedOut >= Config.maxSize)
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();
1132 auto fresh = pool.MakeEntry();
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);
1141 Entry await_resume() noexcept
1149 LIGHTWEIGHT_STATS_POOL_ACQUIRE(
1150 std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - parkedAt),
1153 return std::move(node->entry);
1155 return std::move(acquired);
1159 Async::Task<PooledDataMapper> AcquireAsyncImpl(Async::IExecutor* dbWorkers, Async::IResumeScheduler* resume)
1161 auto entry =
co_await AsyncAcquireAwaitable { *
this, *resume };
1166 entry = ConnectForTransferredSlot();
1171 auto pooled = PooledDataMapper(*
this, std::move(entry));
1172 pooled->Connection().EnableAsync(*dbWorkers, *resume);
1173 co_return std::move(pooled);
1177 std::vector<Entry> _idleDataMappers;
1178 size_t _checkedOut {};
1180 std::shared_ptr<PoolRelationLoadSource> _relationLoadSource = std::make_shared<PoolRelationLoadSource>(*
this);
1187 std::function<Clock::time_point()> _clock {};
1190 Async::IExecutor* _asyncDbWorkers =
nullptr;
1191 Async::IResumeScheduler* _asyncResume =
nullptr;
1194 std::deque<std::shared_ptr<WaiterNode>> _waiters;