warm_pool.hpp

1#pragma once23#include "resource.hpp"4#if defined(BRQ_ENABLE_POOL_DEPTH_STATS)5#include "depth_stats.hpp"6#endif7#include "intrusive_stack.hpp"8#include "intrusive_queue.hpp"9#include "TwoBufQueue/src/utils/wait.hpp"10#include "TwoBufQueue/src/utils/no_lock.hpp"11#include <chrono>12#include <cstdio>13#include <exception>14#include <limits>15#include <mutex>16#include <stdexcept>17#include <thread>18#include <utility>19#include <vector>2021namespace TwoBufQueue {2223enum class PoolGetStatus { Success, Limit, ResourceError };2425// The pool lock protects containers/statistics; Node locks protect resource operations using Loan synchronization.26// Lock a Loan before checking validity; callers using ordinary raw pointers remain responsible for lifetime.27// Creation, reset and Factory::dispose run outside the pool lock; factories must support concurrent calls.28template<PoolResource TResource, bool TIsConcurrent = true,29         class TFactory = DefaultResourceFactory<TResource>>30class WarmPool {31    using Lock = std::conditional_t<TIsConcurrent, SpinLock, NoLock>;32    struct Node;33    // Allocate alongside creation so destruction registration needs no allocation. Old records are independent of a reused Node::record.34    struct Record {35        Node* node;36        TResource* resource = nullptr;37        Record* next = nullptr;38    };39    struct Node {40        Lock lock_; // Stable address until pool destruction, allowing old Loans to lock safely.41        Record* record = nullptr;42        Node* next = nullptr;43        std::uint64_t generation = 0;44        bool in_use = false;45    };46    friend struct WarmPoolTestAccess;47    static_assert(requires(TFactory& factory, TResource* resource) {48        { factory.dispose(resource) } noexcept -> std::same_as<void>;49    }, "Factory::dispose(Resource*) must be noexcept and destroy the complete resource");5051public:52    static constexpr bool concurrent = TIsConcurrent;53    using ResourceType = TResource;54    using Loan = TwoBufQueue::Loan<TResource>;55    using ReleaseCallback = void (*)(void*) noexcept;5657    WarmPool(std::size_t min_count, std::size_t max_count, TFactory factory = {},58        std::size_t max_free_count = std::numeric_limits<std::size_t>::max())59        : min_count_(min_count), max_count_(max_count), max_free_count_(max_free_count),60          factory_(std::move(factory)) {61        if (!max_count || min_count > max_count || max_count == std::numeric_limits<std::size_t>::max())62            throw std::invalid_argument("WarmPool requires 0 <= min_count <= max_count < SIZE_MAX and max_count > 0");63        nodes_.reserve(max_count);64        try {65            if (!preallocate(min_count)) throw std::bad_alloc{};66        } catch (...) {67            destroy_all(); // Raw-pointer members do not clean themselves up; failed construction does not invoke this object's destructor.68            throw;69        }70    }71    WarmPool(const WarmPool&) = delete;72    WarmPool& operator=(const WarmPool&) = delete;73    ~WarmPool() noexcept {74        // Destroy the pool only after all calls/Loan-lock operations stop and every loan is returned.75        if (in_use_count_ || pending_count_) {76            std::fputs("destroying WarmPool while resources are still in use\n", stderr);77            std::terminate();78        }79        destroy_all(); // Destroy the final pending objects now rather than waiting for another registration.80    }8182    [[nodiscard]] Loan get() noexcept {83        PoolGetStatus ignored;84        return get(ignored);85    }86    // Record the reason at the failure site instead of inferring concurrent state from counters afterward.87    [[nodiscard]] Loan get(PoolGetStatus& status) noexcept {88        status = PoolGetStatus::ResourceError;89        auto* node = reserve_node(true, &status);90        if (!node) return {};91        // Wait for prior-generation node operations to leave, without holding the pool lock or allocating under the node lock.92        { std::lock_guard<Lock> guard(node->lock_); }93        if (!prepare(node)) return {};94        auto loan = make_loan(node, node->record->resource);95        if constexpr (requires(TResource& r) { { r.bind(loan) } noexcept -> std::same_as<void>; })96            node->record->resource->bind(loan);97        if (!node->record->resource->reset(loan.generation())) {98            discard(node);99            return {};100        }101        std::lock_guard<Lock> guard(lock_);102        node->in_use = true;103        ++in_use_count_;104        --pending_count_;105#if defined(BRQ_ENABLE_POOL_DEPTH_STATS)106        if (depth_stats_enabled_) depth_stats_.record(in_use_count_, resource_count_, free_count_);107#endif108        status = PoolGetStatus::Success;109        return loan;110    }111    template<class TRep, class TPeriod>112    [[nodiscard]] Loan wait_get(std::chrono::duration<TRep, TPeriod> timeout) noexcept {113        using Clock = std::chrono::steady_clock;114        const auto deadline = deadline_after<Clock>(Clock::now(), wait_duration<Clock>(timeout));115        const auto interval = wait_duration<Clock>(std::chrono::milliseconds{1});116        for (;;) {117            if (auto loan = get()) return loan;118            const auto now = Clock::now();119            if (now >= deadline) return {};120            const auto retry = deadline_after<Clock>(now, interval);121            std::this_thread::sleep_until(retry < deadline ? retry : deadline);122        }123    }124125    // O(1) snapshot check of stable nodes only; the resource pointer may already be stale.126    template<class TOtherValue> [[nodiscard]] bool is_valid(const TwoBufQueue::Loan<TOtherValue>& loan) const noexcept {127        if (!owns(loan)) return false;128        return is_valid_node(static_cast<const Node*>(loan.node_), loan.generation_);129    }130    [[nodiscard]] bool is_valid(const TResource& resource) const noexcept requires BoundResource<TResource> {131        return is_valid(resource.loan());132    }133    [[nodiscard]] bool is_valid(const TResource* resource) const noexcept requires BoundResource<TResource> {134        return resource && is_valid(*resource);135    }136    // Claim return at most once per generation. Stop using the resource afterward; ordinary accesses require caller serialization.137    // A thread holding an old Loan lock may unlock it; get/destruction waits or skips the node so generations never overlap.138    template<class TOtherValue> [[nodiscard]] bool recycle(const TwoBufQueue::Loan<TOtherValue>& loan) noexcept {139        return owns(loan) && recycle_node(static_cast<Node*>(loan.node_), loan.generation_);140    }141    [[nodiscard]] bool recycle(const TResource& resource) noexcept requires BoundResource<TResource> {142        const auto loan = resource.loan();143        return recycle(loan);144    }145    [[nodiscard]] bool recycle(const TResource* resource) noexcept requires BoundResource<TResource> {146        return resource && recycle(*resource);147    }148    // Call during initialization/quiescence; count is the pool's total resource count, excluding old objects pending destruction.149    [[nodiscard]] bool preallocate(std::size_t count) noexcept {150        if (count > max_count_) return false;151        while (resource_count() < count) {152            auto* node = reserve_node(false);153            if (!node || !prepare(node)) return false;154            std::lock_guard<Lock> guard(lock_);155            put_free_without_lock(node);156        }157        return true;158    }159    // Notify once after logical return; physical destruction may be deferred. Query only; do not reenter mutations.160    void set_release_callback(ReleaseCallback callback, void* context) noexcept {161        std::lock_guard<Lock> guard(lock_);162        release_callback_ = callback;163        release_context_ = context;164    }165    [[nodiscard]] const TFactory& factory() const noexcept { return factory_; }166#if defined(BRQ_ENABLE_POOL_DEPTH_STATS)167    // Begin only after warmup and input preparation. The initial snapshot168    // contributes to maxima, not to acquisition-weighted averages.169    void begin_depth_stats() noexcept {170        std::lock_guard<Lock> guard(lock_);171        depth_stats_.begin(in_use_count_, resource_count_);172        depth_stats_enabled_ = true;173    }174    [[nodiscard]] PoolDepthStats end_depth_stats() noexcept {175        std::lock_guard<Lock> guard(lock_);176        depth_stats_enabled_ = false;177        return depth_stats_;178    }179#endif180    [[nodiscard]] std::size_t min_count() const noexcept { return min_count_; }181    [[nodiscard]] std::size_t max_count() const noexcept { return max_count_; }182    [[nodiscard]] std::size_t resource_count() const noexcept {183        std::lock_guard<Lock> guard(lock_); return resource_count_;184    }185    [[nodiscard]] std::size_t free_resource_count() const noexcept {186        std::lock_guard<Lock> guard(lock_); return free_count_;187    }188    [[nodiscard]] std::size_t in_use_count() const noexcept {189        std::lock_guard<Lock> guard(lock_); return in_use_count_;190    }191    [[nodiscard]] std::size_t pending_count() const noexcept {192        std::lock_guard<Lock> guard(lock_); return pending_count_;193    }194    [[nodiscard]] std::size_t will_dispose_count() const noexcept {195        std::lock_guard<Lock> guard(lock_); return will_dispose_qu_.size();196    }197    [[nodiscard]] bool all_released() const noexcept {198        std::lock_guard<Lock> guard(lock_); return !in_use_count_ && !pending_count_;199    }200    template<class TOtherValue> [[nodiscard]] bool owns(const TwoBufQueue::Loan<TOtherValue>& loan) const noexcept {201        return loan.owner_ == this && loan.node_ != nullptr;202    }203    // The raw resource must still be alive; this overload cannot detect dangling pointers.204    [[nodiscard]] bool owns(const TResource& resource) const noexcept requires BoundResource<TResource> {205        return owns(resource.loan());206    }207    [[nodiscard]] bool owns(const TResource* resource) const noexcept requires BoundResource<TResource> {208        return resource && owns(*resource);209    }210211private:212    Loan make_loan(Node* node, TResource* resource) noexcept {213        SpinLock* gate = nullptr;214        if constexpr (TIsConcurrent) gate = &node->lock_;215        return Loan(this, node, resource, node->generation, gate, &operations_);216    }217    bool is_valid_node(const Node* node, std::uint64_t generation) const noexcept {218        std::lock_guard<Lock> guard(lock_);219        return node->in_use && node->generation == generation;220    }221    bool recycle_node(Node* node, std::uint64_t generation) noexcept {222        Record* retiring = nullptr;223        ReleaseCallback callback;224        void* context;225        {226            std::lock_guard<Lock> guard(lock_);227            if (!node->in_use || node->generation != generation) return false;228            node->in_use = false;229            --in_use_count_;230            ++pending_count_;231            if (free_count_ >= max_free_count_ && resource_count_ > min_count_) {232                retiring = std::exchange(node->record, nullptr);233                --resource_count_; // Reserve shrinking immediately so concurrent returns cannot go below min_count.234            } else {235                put_free_without_lock(node);236            }237            callback = release_callback_;238            context = release_context_;239        }240        if (retiring) {241            register_dispose(retiring);242            std::lock_guard<Lock> guard(lock_);243            dormant_.push(node); // The queue retains the old Record while the node may create a new resource.244            --pending_count_;245        }246        if (callback) callback(context);247        return true;248    }249    void put_free_without_lock(Node* node) noexcept {250        free_.push(node); ++free_count_; --pending_count_;251    }252    Node* reserve_node(bool borrow, PoolGetStatus* status = nullptr) noexcept {253        std::uint64_t generation = 0;254        {255            std::lock_guard<Lock> guard(lock_);256            if (borrow && next_generation_ == std::numeric_limits<std::uint64_t>::max()) return nullptr;257            auto* node = borrow ? free_.pop() : nullptr;258            if (node) --free_count_;259            else node = dormant_.pop();260            if (node) {261                ++pending_count_;262                if (borrow) node->generation = ++next_generation_;263                return node;264            }265            if (nodes_.size() + creating_count_ >= max_count_) {266                if (status) *status = PoolGetStatus::Limit;267                return nullptr;268            }269            ++creating_count_; ++pending_count_;270            if (borrow) generation = ++next_generation_;271        }272        Node* node = nullptr;273        try { node = new Node; } catch (...) {}274        std::lock_guard<Lock> guard(lock_);275        --creating_count_;276        if (!node) { --pending_count_; return nullptr; }277        node->generation = generation;278        nodes_.push_back(node); // reserve(max_count) ran during construction; no allocation here.279        return node;280    }281    bool prepare(Node* node) noexcept {282        if (node->record) return true;283        Record* record = nullptr;284        try {285            record = new Record{node};286            const auto seed = make_loan(node, nullptr);287            if constexpr (requires { factory_.create(TwoBufQueue::Loan<void>(seed)); })288                record->resource = factory_.create(TwoBufQueue::Loan<void>(seed));289            else record->resource = factory_.create();290        } catch (...) {}291        if (!record || !record->resource) {292            delete record;293            std::lock_guard<Lock> guard(lock_);294            dormant_.push(node); --pending_count_;295            return false;296        }297        std::lock_guard<Lock> guard(lock_);298        node->record = record; ++resource_count_;299        return true;300    }301    void discard(Node* node) noexcept {302        Record* record;303        {304            std::lock_guard<Lock> guard(lock_);305            record = std::exchange(node->record, nullptr);306            --resource_count_;307        }308        register_dispose(record);309        std::lock_guard<Lock> guard(lock_);310        dormant_.push(node); --pending_count_;311    }312    void register_dispose(Record* record) noexcept {313        Record* candidates[2]{};314        {315            std::lock_guard<Lock> guard(lock_);316            // Remove two distinct old records at once; do not recheck a busy, requeued record during the same pass.317            candidates[0] = will_dispose_qu_.pop();318            candidates[1] = will_dispose_qu_.pop();319        }320        for (auto* old : candidates) {321            if (!old) continue;322            if (old->node->lock_.try_lock()) {323                // The old generation is invalid and detached from its node; acquiring the stable node lock proves old operations have exited.324                // Unlock before destruction. Later old Loans fail validation and cannot touch the old object.325                old->node->lock_.unlock();326                factory_.dispose(old->resource);327                delete old;328            } else {329                std::lock_guard<Lock> guard(lock_);330                will_dispose_qu_.push(old);331            }332        }333        std::lock_guard<Lock> guard(lock_);334        will_dispose_qu_.push(record);335    }336    void destroy_all() noexcept {337        while (auto* record = will_dispose_qu_.pop()) {338            factory_.dispose(record->resource); delete record;339        }340        for (auto* node : nodes_) {341            if (node->record) {342                factory_.dispose(node->record->resource);343                delete node->record;344            }345            delete node;346        }347        nodes_.clear();348    }349    static bool validate_loan(const LoanOperations&, void* owner, void* node, std::uint64_t generation) noexcept {350        return static_cast<WarmPool*>(owner)->is_valid_node(static_cast<Node*>(node), generation);351    }352    static bool recycle_loan(const LoanOperations&, void* owner, void* node, std::uint64_t generation) noexcept {353        return static_cast<WarmPool*>(owner)->recycle_node(static_cast<Node*>(node), generation);354    }355    inline static constexpr LoanOperations operations_{&validate_loan, &recycle_loan};356    const std::size_t min_count_;357    const std::size_t max_count_;358    const std::size_t max_free_count_;359    TFactory factory_;360    mutable Lock lock_;361    std::vector<Node*> nodes_;362    SingleThreadStack<Node, &Node::next> free_;363    SingleThreadStack<Node, &Node::next> dormant_;364    SingleThreadQueue<Record, &Record::next> will_dispose_qu_;365    std::size_t free_count_ = 0, in_use_count_ = 0, resource_count_ = 0;366    std::size_t creating_count_ = 0, pending_count_ = 0;367    std::uint64_t next_generation_ = 0;368    ReleaseCallback release_callback_ = nullptr;369    void* release_context_ = nullptr;370#if defined(BRQ_ENABLE_POOL_DEPTH_STATS)371    PoolDepthStats depth_stats_{};372    bool depth_stats_enabled_ = false;373#endif374};375} // namespace TwoBufQueue