<commit_before>#ifndef LIBTEN_SHARED_POOL_HH
#define LIBTEN_SHARED_POOL_HH

#include "task/rendez.hh"
#include "ten/logging.hh"

#include <boost/call_traits.hpp>
#include <set>
#include <deque>

namespace ten {

namespace detail {
    template <typename T> class scoped_resource;
}

//! thread and task safe pool of shared resources
//
//! useful for connection pools and other types of shared resources
template <typename ResourceT, typename ScopeT = detail::scoped_resource<ResourceT> >
class shared_pool {
public:
    // use this scoped_resource type to RAII resources from the pool
    typedef ScopeT scoped_resource;
    typedef std::deque<std::shared_ptr<ResourceT> > queue_type;
    typedef std::set<std::shared_ptr<ResourceT> > set_type;
    typedef std::function<std::shared_ptr<ResourceT> ()> alloc_func;
protected:
    template <typename TT> friend class detail::scoped_resource;

    qutex _mut;
    rendez _not_empty;
    queue_type _q;
    set_type _set;
    std::string _name;
    alloc_func _new_resource;
    ssize_t _max;

public:
    shared_pool(const std::string &name_,
            const alloc_func &alloc_,
            ssize_t max_ = -1)
        : _name(name_),
        _new_resource(alloc_),
        _max(max_)
    {}

    shared_pool(const shared_pool &) = delete;
    shared_pool &operator =(const shared_pool &) = delete;

    size_t size() {
        std::unique_lock<qutex> lk(_mut);
        return _set.size();
    }

    void clear() {
        std::unique_lock<qutex> lk(_mut);
        _q.clear();
        _set.clear();
    }

    const std::string &name() const { return _name; }

protected:
    bool is_not_empty() const { return !_q.empty(); }

    std::shared_ptr<ResourceT> acquire() {
        std::unique_lock<qutex> lk(_mut);
        return create_or_acquire_nolock(lk);
    }

    // internal, does not lock mutex
    std::shared_ptr<ResourceT> create_or_acquire_nolock(std::unique_lock<qutex> &lk) {
        std::shared_ptr<ResourceT> c;
        if (_q.empty()) {
            // need to create a new resource
            if (_max < 0 || _set.size() < (size_t)_max) {
                c = add_new_resource(lk);
            } else {
                // can't create anymore we're at max, try waiting
                _not_empty.sleep(lk, [&] {
                    return _q.empty();
                });
                DCHECK(!_q.empty());
                c = _q.front();
                _q.pop_front();
            }
        } else {
            // pop resource from front of queue
            c = _q.front();
            _q.pop_front();
        }
        CHECK(c) << "acquire shared resource failed in pool: " << _name;
        return c;
    }

    void release(std::shared_ptr<ResourceT> &c) {
        std::unique_lock<qutex> lk(_mut);
        // don't add resource to queue if it was removed from _set
        if (_set.count(c)) {
            _q.push_front(c);
            _not_empty.wakeup();
        }
    }

    void destroy(std::shared_ptr<ResourceT> &c) {
        std::unique_lock<qutex> lk(_mut);
        // remove bad resource
        DVLOG(4) << "shared_pool(" << _name
            << ") destroy in set? " << _set.count(c)
            << " : " << c << " rc: " << c.use_count();
        size_t count = _set.erase(c);
        if (count) {
            LOG(WARNING) << "destroying shared resource from pool " << _name;
        }
        typename queue_type::iterator i = std::find(_q.begin(), _q.end(), c);
        if (i!=_q.end()) { _q.erase(i); }

        c.reset();

        // replace the destroyed resource
        std::shared_ptr<ResourceT> cc = add_new_resource(lk);
        _q.push_front(cc);
        cc.reset();
        _not_empty.wakeup();
    }

    std::shared_ptr<ResourceT> add_new_resource(std::unique_lock<qutex> &lk) {
        std::shared_ptr<ResourceT> c;
        lk.unlock(); // unlock while newing resource
        try {
            c = _new_resource();
            CHECK(c) << "new_resource failed for pool: " << _name;
            DVLOG(4) << "inserting to shared_pool(" << _name << "): " << c;
        } catch (std::exception &e) {
            LOG(ERROR) << "exception creating new resource for pool: " <<_name << " " << e.what();
            throw;
        }
        lk.lock(); // re-lock before inserting to set
        _set.insert(c);
        return c;
    }

};

namespace detail {

// do not use this class directly, instead use your shared_pool<>::scoped_resource
template <typename T> class scoped_resource {
public:
    //typedef typename std::add_reference<shared_pool<T>>::type poolref;
    typedef typename boost::call_traits<shared_pool<T>>::reference poolref;
protected:
    poolref _pool;
    std::shared_ptr<T> _c;
    bool _success;
public:<commit_msg>more fixes to shared_pool logic<commit_after>#ifndef LIBTEN_SHARED_POOL_HH
#define LIBTEN_SHARED_POOL_HH

#include "task/rendez.hh"
#include "ten/logging.hh"

#include <boost/call_traits.hpp>
#include <set>
#include <deque>

namespace ten {

namespace detail {
    template <typename T> class scoped_resource;
}

//! thread and task safe pool of shared resources
//
//! useful for connection pools and other types of shared resources
template <typename ResourceT, typename ScopeT = detail::scoped_resource<ResourceT> >
class shared_pool {
public:
    // use this scoped_resource type to RAII resources from the pool
    typedef ScopeT scoped_resource;
    typedef std::deque<std::shared_ptr<ResourceT> > queue_type;
    typedef std::set<std::shared_ptr<ResourceT> > set_type;
    typedef std::function<std::shared_ptr<ResourceT> ()> alloc_func;
protected:
    template <typename TT> friend class detail::scoped_resource;

    qutex _mut;
    rendez _not_empty;
    queue_type _q;
    set_type _set;
    std::string _name;
    alloc_func _new_resource;
    ssize_t _max;

public:
    shared_pool(const std::string &name_,
            const alloc_func &alloc_,
            ssize_t max_ = -1)
        : _name(name_),
        _new_resource(alloc_),
        _max(max_)
    {}

    shared_pool(const shared_pool &) = delete;
    shared_pool &operator =(const shared_pool &) = delete;

    size_t size() {
        std::unique_lock<qutex> lk(_mut);
        return _set.size();
    }

    void clear() {
        std::unique_lock<qutex> lk(_mut);
        _q.clear();
        _set.clear();
    }

    const std::string &name() const { return _name; }

protected:
    bool is_not_empty() const { return !_q.empty(); }

    std::shared_ptr<ResourceT> acquire() {
        std::unique_lock<qutex> lk(_mut);
        return create_or_acquire_with_lock(lk);
    }

    // internal, does not lock mutex
    std::shared_ptr<ResourceT> create_or_acquire_with_lock(std::unique_lock<qutex> &lk) {
        while (_q.empty()) {
            if (_max < 0 || _set.size() < (size_t)_max) {
                // need to create a new resource
                return add_new_resource(lk);
                break;
            } else {
                // can't create anymore we're at max, try waiting
                _not_empty.sleep(lk, [&] {
                    return !_q.empty();
                });
            }
        }

        CHECK(!_q.empty());
        // pop resource from front of queue
        std::shared_ptr<ResourceT> c = _q.front();
        _q.pop_front();
        CHECK(c) << "acquire shared resource failed in pool: " << _name;
        return c;
    }

    void release(std::shared_ptr<ResourceT> &c) {
        std::unique_lock<qutex> lk(_mut);
        // don't add resource to queue if it was removed from _set
        if (_set.count(c)) {
            _q.push_front(c);
            _not_empty.wakeup();
        }
    }

    void destroy(std::shared_ptr<ResourceT> &c) {
        std::unique_lock<qutex> lk(_mut);
        // remove bad resource
        DVLOG(4) << "shared_pool(" << _name
            << ") destroy in set? " << _set.count(c)
            << " : " << c << " rc: " << c.use_count();
        size_t count = _set.erase(c);
        if (count) {
            LOG(WARNING) << "destroying shared resource from pool " << _name;
        }
        typename queue_type::iterator i = std::find(_q.begin(), _q.end(), c);
        if (i!=_q.end()) { _q.erase(i); }

        c.reset();

        // replace the destroyed resource
        std::shared_ptr<ResourceT> cc = add_new_resource(lk);
        _q.push_front(cc);
        cc.reset();
        _not_empty.wakeup();
    }

    std::shared_ptr<ResourceT> add_new_resource(std::unique_lock<qutex> &lk) {
        std::shared_ptr<ResourceT> c;
        lk.unlock(); // unlock while newing resource
        try {
            c = _new_resource();
            CHECK(c) << "new_resource failed for pool: " << _name;
            DVLOG(4) << "inserting to shared_pool(" << _name << "): " << c;
        } catch (std::exception &e) {
            LOG(ERROR) << "exception creating new resource for pool: " <<_name << " " << e.what();
            lk.lock();
            throw;
        }
        lk.lock(); // re-lock before inserting to set
        _set.insert(c);
        return c;
    }

};

namespace detail {

// do not use this class directly, instead use your shared_pool<>::scoped_resource
template <typename T> class scoped_resource {
public:
    //typedef typename std::add_reference<shared_pool<T>>::type poolref;
    typedef typename boost::call_traits<shared_pool<T>>::reference poolref;
protected:
    poolref _pool;
    std::shared_ptr<T> _c;
    bool _success;
public:
{'fraction_non_alphanumeric': 0.1290151595462737, 'fraction_numerical': 0.000848086504823492, 'mean_word_length': 2.366880799428979, 'pattern_counts': {'":': 0, '<': 165, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '743061', 'n_tokens_mistral': 3096, 'n_tokens_neox': 3000, 'n_words': 712}