enumerable_thread_specific.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2019-2020, Intel Corporation */
 
#ifndef LIBPMEMOBJ_CPP_ENUMERABLE_THREAD_SPECIFIC_HPP
#define LIBPMEMOBJ_CPP_ENUMERABLE_THREAD_SPECIFIC_HPP
 
#include <libpmemobj++/container/segment_vector.hpp>
#include <libpmemobj++/detail/common.hpp>
#include <libpmemobj++/mutex.hpp>
#include <libpmemobj++/shared_mutex.hpp>
 
#include <cassert>
#include <deque>
#include <mutex>
#include <numeric>
#include <thread>
#include <unordered_map>
 
namespace pmem
{
namespace detail
{
 
struct id_manager {
    id_manager();
 
    id_manager(const id_manager &) = delete;
    id_manager &operator=(const id_manager &) = delete;
 
    size_t get();
    void release(size_t id);
 
private:
    static constexpr size_t initial_queue_capacity = 1024;
 
    std::mutex mutex;
    std::size_t queue_capacity;
    std::deque<size_t> queue;
};
 
struct thread_id_type {
    thread_id_type();
 
    thread_id_type(const thread_id_type &) = delete;
    thread_id_type &operator=(const thread_id_type &) = delete;
 
    ~thread_id_type();
 
    size_t get();
 
private:
    size_t id;
 
    static id_manager &get_id_manager();
};
 
template <typename T, typename Mutex = obj::shared_mutex,
      typename Storage =
          obj::segment_vector<T, obj::exponential_size_array_policy<>>>
class enumerable_thread_specific {
    using storage_type = Storage;
    using mutex_type = Mutex;
 
public:
    /* traits */
    using value_type = T;
    using size_type = typename storage_type::size_type;
    using difference_type = typename storage_type::difference_type;
    using reference = value_type &;
    using const_reference = const value_type &;
    using pointer = value_type *;
    using const_pointer = const value_type *;
    using iterator = typename storage_type::iterator;
    using const_iterator = typename storage_type::const_iterator;
 
    /* initialization */
    template <typename Handler>
    void initialize(Handler handler = [](reference) {});
 
    /* ctors & dtor */
    enumerable_thread_specific();
    ~enumerable_thread_specific() = default;
 
    /* access */
    reference local();
 
    /* size */
    bool empty() const;
    void clear();
    size_type size() const;
 
    /* iterators */
    iterator begin();
    iterator end();
    const_iterator begin() const;
    const_iterator end() const;
 
private:
    /* private helper methods */
    obj::pool_base get_pool() const noexcept;
    void set_cached_size(size_t s);
    size_t get_cached_size();
 
    mutex_type _mutex;
    storage_type _storage;
 
    obj::p<std::atomic<size_t>> _storage_size;
};
 
inline id_manager::id_manager()
    : queue_capacity(initial_queue_capacity), queue(initial_queue_capacity, 0)
{
    /* Convert 0, 0, 0, ..., 0 into 0, 1, 2, ..., N */
    std::iota(queue.begin(), queue.end(), 0);
}
 
inline size_t
id_manager::get()
{
    std::unique_lock<std::mutex> lock(mutex);
 
    if (queue.empty())
        queue.push_front(queue_capacity++);
 
    auto front = queue.front();
    queue.pop_front();
 
    return front;
}
 
inline void
id_manager::release(size_t id)
{
    std::unique_lock<std::mutex> lock(mutex);
 
    queue.push_front(id);
}
 
inline id_manager &
thread_id_type::get_id_manager()
{
    static id_manager manager;
    return manager;
}
 
inline thread_id_type::thread_id_type()
{
    auto &manager = get_id_manager();
 
    /*
     * Drd had a bug related to static object initialization and reported
     * conflicting load on std::mutex access inside id_manager.
     */
#if LIBPMEMOBJ_CPP_VG_DRD_ENABLED
    ANNOTATE_BENIGN_RACE_SIZED(
        &manager, sizeof(std::mutex),
        "https://bugs.kde.org/show_bug.cgi?id=416286");
#endif
 
    id = manager.get();
}
 
inline thread_id_type::~thread_id_type()
{
    get_id_manager().release(id);
}
 
inline size_t
thread_id_type::get()
{
    return id;
}
 
template <typename T, typename Mutex, typename Storage>
enumerable_thread_specific<T, Mutex, Storage>::enumerable_thread_specific()
{
    _storage_size.get_rw() = 0;
}
 
template <typename T, typename Mutex, typename Storage>
void
enumerable_thread_specific<T, Mutex, Storage>::set_cached_size(size_t s)
{
    auto pop = get_pool();
 
    /* Helgrind does not understand std::atomic */
#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
    VALGRIND_HG_DISABLE_CHECKING(&_storage_size, sizeof(_storage_size));
#endif
 
#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED || LIBPMEMOBJ_CPP_VG_DRD_ENABLED
    ANNOTATE_HAPPENS_BEFORE(&_storage_size);
#endif
 
    _storage_size.get_rw().store(s);
    pop.persist(_storage_size);
}
 
template <typename T, typename Mutex, typename Storage>
size_t
enumerable_thread_specific<T, Mutex, Storage>::get_cached_size()
{
    auto s = _storage_size.get_ro().load();
 
#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED || LIBPMEMOBJ_CPP_VG_DRD_ENABLED
    ANNOTATE_HAPPENS_AFTER(&_storage_size);
#endif
 
    return s;
}
 
template <typename T, typename Mutex, typename Storage>
template <typename Handler>
void
enumerable_thread_specific<T, Mutex, Storage>::initialize(Handler handler)
{
    for (reference e : *this) {
        handler(e);
    }
    clear();
}
 
template <typename T, typename Mutex, typename Storage>
typename enumerable_thread_specific<T, Mutex, Storage>::reference
enumerable_thread_specific<T, Mutex, Storage>::local()
{
    assert(pmemobj_tx_stage() != TX_STAGE_WORK);
 
    static thread_local thread_id_type tid;
    auto index = tid.get();
 
    auto cached_size = get_cached_size();
 
    if (index >= cached_size) {
        std::unique_lock<mutex_type> lock(_mutex);
 
        /* Size of the storage could have changed before we obtained the
         * lock. That's why we read size once again. */
        auto size = _storage.size();
 
        if (index >= size) {
            _storage.resize(index + 1);
            set_cached_size(index + 1);
        } else if (size != cached_size) {
            set_cached_size(size);
        }
    }
 
    /*
     * Because _storage can only grow (unless clear() was called which
     * should not happen simultaneously with this operation), index must be
     * less than _storage.size().
     */
    return _storage[index];
}
 
template <typename T, typename Mutex, typename Storage>
void
enumerable_thread_specific<T, Mutex, Storage>::clear()
{
    auto pop = get_pool();
 
    obj::flat_transaction::run(pop, [&] {
        _storage_size.get_rw() = 0;
        _storage.clear();
    });
}
 
template <typename T, typename Mutex, typename Storage>
typename enumerable_thread_specific<T, Mutex, Storage>::size_type
enumerable_thread_specific<T, Mutex, Storage>::size() const
{
    return _storage.size();
}
 
template <typename T, typename Mutex, typename Storage>
bool
enumerable_thread_specific<T, Mutex, Storage>::empty() const
{
    return _storage.size() == 0;
}
 
template <typename T, typename Mutex, typename Storage>
typename enumerable_thread_specific<T, Mutex, Storage>::iterator
enumerable_thread_specific<T, Mutex, Storage>::begin()
{
    return _storage.begin();
}
 
template <typename T, typename Mutex, typename Storage>
typename enumerable_thread_specific<T, Mutex, Storage>::iterator
enumerable_thread_specific<T, Mutex, Storage>::end()
{
    return _storage.end();
}
 
template <typename T, typename Mutex, typename Storage>
typename enumerable_thread_specific<T, Mutex, Storage>::const_iterator
enumerable_thread_specific<T, Mutex, Storage>::begin() const
{
    return _storage.begin();
}
 
template <typename T, typename Mutex, typename Storage>
typename enumerable_thread_specific<T, Mutex, Storage>::const_iterator
enumerable_thread_specific<T, Mutex, Storage>::end() const
{
    return _storage.end();
}
 
template <typename T, typename Mutex, typename Storage>
obj::pool_base
enumerable_thread_specific<T, Mutex, Storage>::get_pool() const noexcept
{
    auto pop = pmemobj_pool_by_ptr(this);
    assert(pop != nullptr);
    return obj::pool_base(pop);
}
 
} /* namespace detail */
} /* namespace pmem */
 
#endif