concurrent_skip_list_impl.hpp

// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2020-2021, Intel Corporation */

#ifndef PMEMOBJ_CONCURRENT_SKIP_LIST_IMPL_HPP
#define PMEMOBJ_CONCURRENT_SKIP_LIST_IMPL_HPP

#include <algorithm>
#include <array>
#include <atomic>
#include <cstdlib>
#include <limits>
#include <mutex> /* for std::unique_lock */
#include <random>
#include <type_traits>

#include <libpmemobj++/detail/common.hpp>
#include <libpmemobj++/detail/enumerable_thread_specific.hpp>
#include <libpmemobj++/detail/life.hpp>
#include <libpmemobj++/detail/pair.hpp>
#include <libpmemobj++/detail/template_helpers.hpp>
#include <libpmemobj++/mutex.hpp>
#include <libpmemobj++/persistent_ptr.hpp>
#include <libpmemobj++/pool.hpp>
#include <libpmemobj++/transaction.hpp>

#include <libpmemobj++/experimental/atomic_self_relative_ptr.hpp>
#include <libpmemobj++/experimental/self_relative_ptr.hpp>

/* Windows has a max and a min macros which collides with min() and max()
 * methods of default_random_generator */
#if defined(_WIN32)
#if defined(max)
#undef max
#endif
#if defined(min)
#undef min
#endif
#endif

namespace pmem
{
namespace detail
{

#ifndef NDEBUG
inline void
try_insert_node_finish_marker()
{
}
#endif

template <typename MyAlloc, typename OtherAlloc>
inline void
allocator_copy_assignment(MyAlloc &my_allocator, OtherAlloc &other_allocator,
              std::true_type)
{
    my_allocator = other_allocator;
}

template <typename MyAlloc, typename OtherAlloc>
inline void
allocator_copy_assignment(MyAlloc &, OtherAlloc &, std::false_type)
{ /* NO COPY */
}

template <typename MyAlloc, typename OtherAlloc>
inline void
allocator_move_assignment(MyAlloc &my_allocator, OtherAlloc &other_allocator,
              std::true_type)
{
    my_allocator = std::move(other_allocator);
}

template <typename MyAlloc, typename OtherAlloc>
inline void
allocator_move_assignment(MyAlloc &, OtherAlloc &, std::false_type)
{ /* NO MOVE */
}

template <typename MyAlloc, typename OtherAlloc>
inline void
allocator_swap(MyAlloc &my_allocator, OtherAlloc &other_allocator,
           std::true_type)
{
    std::swap(my_allocator, other_allocator);
}

template <typename MyAlloc, typename OtherAlloc>
inline void
allocator_swap(MyAlloc &, OtherAlloc &, std::false_type)
{ /* NO SWAP */
}

template <typename Value, typename Mutex = pmem::obj::mutex,
      typename LockType = std::unique_lock<Mutex>>
class skip_list_node {
public:
    using value_type = Value;
    using size_type = std::size_t;
    using reference = value_type &;
    using const_reference = const value_type &;
    using pointer = value_type *;
    using const_pointer = const value_type *;
    using node_pointer =
        obj::experimental::self_relative_ptr<skip_list_node>;
    using atomic_node_pointer = std::atomic<node_pointer>;
    using mutex_type = Mutex;
    using lock_type = LockType;

    skip_list_node(size_type levels) : height_(levels)
    {
        for (size_type lev = 0; lev < height_; ++lev)
            detail::create<atomic_node_pointer>(&get_next(lev),
                                nullptr);

        assert(height() == levels);
#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
        /*
         * Valgrind does not understand atomic semantic and reports
         * false-postives in drd and helgrind tools.
         */
        for (size_type lev = 0; lev < height_; ++lev) {
            VALGRIND_HG_DISABLE_CHECKING(&get_next(lev),
                             sizeof(get_next(lev)));
        }
#endif
    }

    skip_list_node(size_type levels, const node_pointer *new_nexts)
        : height_(levels)
    {
        for (size_type lev = 0; lev < height_; ++lev)
            detail::create<atomic_node_pointer>(&get_next(lev),
                                new_nexts[lev]);

        assert(height() == levels);
#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
        /*
         * Valgrind does not understand atomic semantic and reports
         * false-postives in drd and helgrind tools.
         */
        for (size_type lev = 0; lev < height_; ++lev) {
            VALGRIND_HG_DISABLE_CHECKING(&get_next(lev),
                             sizeof(get_next(lev)));
        }
#endif
    }

    ~skip_list_node()
    {
        for (size_type lev = 0; lev < height_; ++lev)
            detail::destroy<atomic_node_pointer>(get_next(lev));
    }

    skip_list_node(const skip_list_node &) = delete;

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

    pointer
    get() noexcept
    {
        return &val;
    }

    const_pointer
    get() const noexcept
    {
        return &val;
    }

    reference
    value()
    {
        return *get();
    }

    node_pointer
    next(size_type level) const
    {
        assert(level < height());
        return get_next(level).load(std::memory_order_acquire);
    }

    void
    set_next_tx(size_type level, node_pointer next)
    {
        assert(level < height());
        assert(pmemobj_tx_stage() == TX_STAGE_WORK);
        auto &node = get_next(level);
        obj::flat_transaction::snapshot<atomic_node_pointer>(&node);
        node.store(next, std::memory_order_release);
    }

    void
    set_next(obj::pool_base pop, size_type level, node_pointer next)
    {
        assert(level < height());
        auto &node = get_next(level);
        node.store(next, std::memory_order_release);
        pop.persist(&node, sizeof(node));
    }

    void
    set_nexts(const node_pointer *new_nexts, size_type h)
    {
        assert(h == height());
        auto *nexts = get_nexts();

        for (size_type i = 0; i < h; i++) {
            nexts[i].store(new_nexts[i], std::memory_order_relaxed);
        }
    }

    void
    set_nexts(obj::pool_base pop, const node_pointer *new_nexts,
          size_type h)
    {
        set_nexts(new_nexts, h);

        auto *nexts = get_nexts();
        pop.persist(nexts, sizeof(nexts[0]) * h);
    }

    size_type
    height() const
    {
        return height_;
    }

    lock_type
    acquire()
    {
        return lock_type(mutex);
    }

private:
    atomic_node_pointer *
    get_nexts()
    {
        return reinterpret_cast<atomic_node_pointer *>(this + 1);
    }

    atomic_node_pointer &
    get_next(size_type level)
    {
        auto *arr = get_nexts();
        return arr[level];
    }

    const atomic_node_pointer &
    get_next(size_type level) const
    {
        auto *arr =
            reinterpret_cast<const atomic_node_pointer *>(this + 1);
        return arr[level];
    }

    mutex_type mutex;
    union {
        value_type val;
    };
    size_type height_;
};

template <typename NodeType, bool is_const>
class skip_list_iterator {
    using node_type = NodeType;
    using node_ptr = typename std::conditional<is_const, const node_type *,
                           node_type *>::type;
    friend class skip_list_iterator<node_type, true>;

public:
    using value_type = typename node_type::value_type;
    using iterator_category = std::forward_iterator_tag;
    using difference_type = std::ptrdiff_t;
    using reference =
        typename std::conditional<is_const,
                      typename node_type::const_reference,
                      typename node_type::reference>::type;
    using pointer = typename std::conditional<is_const, const value_type *,
                          value_type *>::type;

    skip_list_iterator() : node(nullptr)
    {
    }

    skip_list_iterator(const skip_list_iterator &other) : node(other.node)
    {
    }

    template <typename U = void,
          typename = typename std::enable_if<is_const, U>::type>
    skip_list_iterator(const skip_list_iterator<node_type, false> &other)
        : node(other.node)
    {
    }

    reference operator*() const
    {
        return *(node->get());
    }

    pointer operator->() const
    {
        return node->get();
    }

    skip_list_iterator &
    operator++()
    {
        assert(node != nullptr);
        node = node->next(0).get();
        return *this;
    }

    skip_list_iterator
    operator++(int)
    {
        skip_list_iterator tmp = *this;
        ++*this;
        return tmp;
    }

    skip_list_iterator &
    operator=(const skip_list_iterator &other)
    {
        node = other.node;
        return *this;
    }

private:
    explicit skip_list_iterator(node_type *n) : node(n)
    {
    }

    template <typename T = void,
          typename = typename std::enable_if<is_const, T>::type>
    explicit skip_list_iterator(const node_type *n) : node(n)
    {
    }

    node_ptr node;

    template <typename Traits>
    friend class concurrent_skip_list;

    template <typename T, bool M, bool U>
    friend bool operator==(const skip_list_iterator<T, M> &lhs,
                   const skip_list_iterator<T, U> &rhs);

    template <typename T, bool M, bool U>
    friend bool operator!=(const skip_list_iterator<T, M> &lhs,
                   const skip_list_iterator<T, U> &rhs);
};

template <typename T, bool M, bool U>
bool
operator==(const skip_list_iterator<T, M> &lhs,
       const skip_list_iterator<T, U> &rhs)
{
    return lhs.node == rhs.node;
}

template <typename T, bool M, bool U>
bool
operator!=(const skip_list_iterator<T, M> &lhs,
       const skip_list_iterator<T, U> &rhs)
{
    return lhs.node != rhs.node;
}

struct default_random_generator {
    using gen_type = std::mt19937_64;
    using result_type = typename gen_type::result_type;

    size_t
    operator()()
    {
        static thread_local gen_type engine(
            static_cast<size_t>(time(0)));

        return engine();
    }

    static constexpr result_type
    min()
    {
        return gen_type::min();
    }

    static constexpr result_type
    max()
    {
        return gen_type::max();
    }
};

template <typename RndGenerator, size_t MAX_LEVEL>
class geometric_level_generator {
public:
    using rnd_generator_type = RndGenerator;

    static constexpr size_t max_level = MAX_LEVEL;

    size_t
    operator()()
    {
        /* rnd_generator_type should be thread-safe random number
         * generator. */
        static rnd_generator_type gen;
        /* std::geometric_distribution is not thread-safe. We mark it as
         * a thread_local to avoid data races. */
        static thread_local std::geometric_distribution<size_t> d;

        return (d(gen) % MAX_LEVEL) + 1;
    }
};

template <typename Traits>
class concurrent_skip_list {
protected:
    using traits_type = Traits;
    using key_type = typename traits_type::key_type;
    using mapped_type = typename traits_type::mapped_type;
    using value_type = typename traits_type::value_type;
    using size_type = std::size_t;
    using difference_type = std::ptrdiff_t;
    using key_compare = typename traits_type::compare_type;
    using allocator_type = typename traits_type::allocator_type;
    using allocator_traits_type = std::allocator_traits<allocator_type>;

    using reference = value_type &;
    using const_reference = const value_type &;
    using pointer = typename allocator_traits_type::pointer;
    using const_pointer = typename allocator_traits_type::const_pointer;

    using list_node_type = skip_list_node<value_type>;

    using iterator = skip_list_iterator<list_node_type, false>;
    using const_iterator = skip_list_iterator<list_node_type, true>;

    static constexpr size_type MAX_LEVEL = traits_type::max_level;

    using random_level_generator_type = geometric_level_generator<
        typename traits_type::random_generator_type, MAX_LEVEL>;
    using node_allocator_type = typename std::allocator_traits<
        allocator_type>::template rebind_alloc<uint8_t>;
    using node_allocator_traits = typename std::allocator_traits<
        allocator_type>::template rebind_traits<uint8_t>;
    using node_ptr = list_node_type *;
    using const_node_ptr = const list_node_type *;
    using persistent_node_ptr =
        obj::experimental::self_relative_ptr<list_node_type>;

    using prev_array_type = std::array<node_ptr, MAX_LEVEL>;
    using next_array_type = std::array<persistent_node_ptr, MAX_LEVEL>;
    using node_lock_type = typename list_node_type::lock_type;
    using lock_array = std::array<node_lock_type, MAX_LEVEL>;

public:
    static constexpr bool allow_multimapping =
        traits_type::allow_multimapping;

    concurrent_skip_list()
    {
        check_tx_stage_work();
        init();
    }

    explicit concurrent_skip_list(
        const key_compare &comp,
        const allocator_type &alloc = allocator_type())
        : _node_allocator(alloc), _compare(comp)
    {
        check_tx_stage_work();
        init();
    }

    template <class InputIt>
    concurrent_skip_list(InputIt first, InputIt last,
                 const key_compare &comp = key_compare(),
                 const allocator_type &alloc = allocator_type())
        : _node_allocator(alloc), _compare(comp)
    {
        check_tx_stage_work();
        init();
        while (first != last)
            internal_unsafe_emplace(*first++);
    }

    concurrent_skip_list(const concurrent_skip_list &other)
        : _node_allocator(node_allocator_traits::
                      select_on_container_copy_construction(
                          other._node_allocator)),
          _compare(other._compare),
          _rnd_generator(other._rnd_generator)
    {
        check_tx_stage_work();
        init();
        internal_copy(other);
        assert(_size == other._size);
    }

    concurrent_skip_list(const concurrent_skip_list &other,
                 const allocator_type &alloc)
        : _node_allocator(alloc),
          _compare(other._compare),
          _rnd_generator(other._rnd_generator)
    {
        check_tx_stage_work();
        init();
        internal_copy(other);
        assert(_size == other._size);
    }

    concurrent_skip_list(concurrent_skip_list &&other)
        : _node_allocator(std::move(other._node_allocator)),
          _compare(other._compare),
          _rnd_generator(other._rnd_generator)
    {
        check_tx_stage_work();
        init();
        internal_move(std::move(other));
    }

    concurrent_skip_list(concurrent_skip_list &&other,
                 const allocator_type &alloc)
        : _node_allocator(alloc),
          _compare(other._compare),
          _rnd_generator(other._rnd_generator)
    {
        check_tx_stage_work();
        init();
        if (alloc == other.get_allocator()) {
            internal_move(std::move(other));
        } else {
            init();
            internal_copy(std::make_move_iterator(other.begin()),
                      std::make_move_iterator(other.end()));
        }
    }

    void
    runtime_initialize()
    {
        tls_restore();

        assert(this->size() ==
               size_type(std::distance(this->begin(), this->end())));
    }

    void
    free_data()
    {
        if (dummy_head == nullptr)
            return;

        auto pop = get_pool_base();
        obj::flat_transaction::run(pop, [&] {
            clear();
            delete_dummy_head();
        });
    }

    ~concurrent_skip_list()
    {
        try {
            free_data();
        } catch (...) {
            std::terminate();
        }
    }

    concurrent_skip_list &
    operator=(const concurrent_skip_list &other)
    {
        if (this == &other)
            return *this;

        obj::pool_base pop = get_pool_base();
        obj::flat_transaction::run(pop, [&] {
            using pocca_t = typename node_allocator_traits::
                propagate_on_container_copy_assignment;
            clear();
            allocator_copy_assignment(_node_allocator,
                          other._node_allocator,
                          pocca_t());
            _compare = other._compare;
            _rnd_generator = other._rnd_generator;

            internal_copy(other);
        });
        return *this;
    }

    concurrent_skip_list &
    operator=(concurrent_skip_list &&other)
    {
        if (this == &other)
            return *this;

        obj::pool_base pop = get_pool_base();
        obj::flat_transaction::run(pop, [&] {
            using pocma_t = typename node_allocator_traits::
                propagate_on_container_move_assignment;
            clear();
            if (pocma_t::value ||
                _node_allocator == other._node_allocator) {
                delete_dummy_head();
                allocator_move_assignment(_node_allocator,
                              other._node_allocator,
                              pocma_t());
                _compare = other._compare;
                _rnd_generator = other._rnd_generator;
                internal_move(std::move(other));
            } else {
                internal_copy(
                    std::make_move_iterator(other.begin()),
                    std::make_move_iterator(other.end()));
            }
        });
        return *this;
    }

    concurrent_skip_list &
    operator=(std::initializer_list<value_type> il)
    {
        obj::pool_base pop = get_pool_base();
        obj::flat_transaction::run(pop, [&] {
            clear();
            for (auto it = il.begin(); it != il.end(); ++it)
                internal_unsafe_emplace(*it);
        });
        return *this;
    }

    std::pair<iterator, bool>
    insert(const value_type &value)
    {
        return internal_insert(value.first, value);
    }

    template <typename P,
          typename std::enable_if<
              std::is_constructible<value_type, P &&>::value>::type>
    std::pair<iterator, bool>
    insert(P &&value)
    {
        return emplace(std::forward<P>(value));
    }

    std::pair<iterator, bool>
    insert(value_type &&value)
    {
        return internal_insert(value.first, std::move(value));
    }

    iterator
    insert(const_iterator hint, const_reference value)
    {
        /* Ignore hint */
        return insert(value).first;
    }

    template <typename P,
          typename std::enable_if<
              std::is_constructible<value_type, P &&>::value>::type>
    iterator
    insert(const_iterator hint, P &&value)
    {
        return emplace_hint(hint, std::forward<P>(value));
    }

    template <typename InputIterator>
    void
    insert(InputIterator first, InputIterator last)
    {
        for (InputIterator it = first; it != last; ++it)
            insert(*it);
    }

    void
    insert(std::initializer_list<value_type> ilist)
    {
        insert(ilist.begin(), ilist.end());
    }

    template <typename... Args>
    std::pair<iterator, bool>
    emplace(Args &&... args)
    {
        return internal_emplace(std::forward<Args>(args)...);
    }

    template <typename... Args>
    iterator
    emplace_hint(const_iterator hint, Args &&... args)
    {
        /* Ignore hint */
        return emplace(std::forward<Args>(args)...).first;
    }

    template <typename... Args>
    std::pair<iterator, bool>
    try_emplace(const key_type &k, Args &&... args)
    {
        return internal_try_emplace(k, std::forward<Args>(args)...);
    }

    template <typename... Args>
    std::pair<iterator, bool>
    try_emplace(key_type &&k, Args &&... args)
    {
        return internal_try_emplace(std::move(k),
                        std::forward<Args>(args)...);
    }

    template <typename K, typename... Args>
    typename std::enable_if<
        has_is_transparent<key_compare>::value &&
            std::is_constructible<key_type, K &&>::value,
        std::pair<iterator, bool>>::type
    try_emplace(K &&k, Args &&... args)
    {
        return internal_try_emplace(std::forward<K>(k),
                        std::forward<Args>(args)...);
    }

    iterator
    unsafe_erase(iterator pos)
    {
        check_outside_tx();
        auto &size_diff = tls_data.local().size_diff;
        return internal_erase(pos, size_diff);
    }

    iterator
    unsafe_erase(const_iterator pos)
    {
        return unsafe_erase(get_iterator(pos));
    }

    iterator
    unsafe_erase(const_iterator first, const_iterator last)
    {
        check_outside_tx();
        obj::pool_base pop = get_pool_base();
        auto &size_diff = tls_data.local().size_diff;

        obj::flat_transaction::run(pop, [&] {
            while (first != last) {
                first = internal_erase(first, size_diff);
            }
        });

        return get_iterator(first);
    }

    size_type
    unsafe_erase(const key_type &key)
    {
        std::pair<iterator, iterator> range = equal_range(key);
        size_type sz = static_cast<size_type>(
            std::distance(range.first, range.second));
        unsafe_erase(range.first, range.second);
        return sz;
    }

    template <
        typename K,
        typename = typename std::enable_if<
            has_is_transparent<key_compare>::value &&
                !std::is_convertible<K, iterator>::value &&
                !std::is_convertible<K, const_iterator>::value,
            K>::type>
    size_type
    unsafe_erase(const K &key)
    {
        std::pair<iterator, iterator> range = equal_range(key);
        size_type sz = static_cast<size_type>(
            std::distance(range.first, range.second));
        unsafe_erase(range.first, range.second);
        return sz;
    }

    iterator
    lower_bound(const key_type &key)
    {
        return internal_get_bound(key, _compare);
    }

    const_iterator
    lower_bound(const key_type &key) const
    {
        return internal_get_bound(key, _compare);
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    iterator
    lower_bound(const K &x)
    {
        return internal_get_bound(x, _compare);
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    const_iterator
    lower_bound(const K &x) const
    {
        return internal_get_bound(x, _compare);
    }

    iterator
    find_higher_eq(const key_type &key)
    {
        return internal_get_bound(key, _compare);
    }

    const_iterator
    find_higher_eq(const key_type &key) const
    {
        return internal_get_bound(key, _compare);
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    iterator
    find_higher_eq(const K &x)
    {
        return internal_get_bound(x, _compare);
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    const_iterator
    find_higher_eq(const K &x) const
    {
        return internal_get_bound(x, _compare);
    }

    iterator
    upper_bound(const key_type &key)
    {
        return internal_get_bound(key, not_greater_compare(_compare));
    }

    const_iterator
    upper_bound(const key_type &key) const
    {
        return internal_get_bound(key, not_greater_compare(_compare));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    iterator
    upper_bound(const K &x)
    {
        return internal_get_bound(x, not_greater_compare(_compare));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    const_iterator
    upper_bound(const K &x) const
    {
        return internal_get_bound(x, not_greater_compare(_compare));
    }

    iterator
    find_higher(const key_type &key)
    {
        return internal_get_bound(key, not_greater_compare(_compare));
    }

    const_iterator
    find_higher(const key_type &key) const
    {
        return internal_get_bound(key, not_greater_compare(_compare));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    iterator
    find_higher(const K &x)
    {
        return internal_get_bound(x, not_greater_compare(_compare));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    const_iterator
    find_higher(const K &x) const
    {
        return internal_get_bound(x, not_greater_compare(_compare));
    }

    iterator
    find_lower(const key_type &key)
    {
        auto it = internal_get_biggest_less_than(key, _compare);
        return iterator(
            const_cast<typename iterator::node_ptr>(it.node));
    }

    const_iterator
    find_lower(const key_type &key) const
    {
        return internal_get_biggest_less_than(key, _compare);
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    iterator
    find_lower(const K &key)
    {
        auto it = internal_get_biggest_less_than(key, _compare);
        return iterator(
            const_cast<typename iterator::node_ptr>(it.node));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    const_iterator
    find_lower(const K &key) const
    {
        return internal_get_biggest_less_than(key, _compare);
    }

    iterator
    find_lower_eq(const key_type &key)
    {
        auto it = internal_get_biggest_less_than(
            key, not_greater_compare(_compare));
        return iterator(
            const_cast<typename iterator::node_ptr>(it.node));
    }

    const_iterator
    find_lower_eq(const key_type &key) const
    {
        return internal_get_biggest_less_than(
            key, not_greater_compare(_compare));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    iterator
    find_lower_eq(const K &key)
    {
        auto it = internal_get_biggest_less_than(
            key, not_greater_compare(_compare));
        return iterator(
            const_cast<typename iterator::node_ptr>(it.node));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    const_iterator
    find_lower_eq(const K &key) const
    {
        return internal_get_biggest_less_than(
            key, not_greater_compare(_compare));
    }

    iterator
    find(const key_type &key)
    {
        return internal_find(key);
    }

    const_iterator
    find(const key_type &key) const
    {
        return internal_find(key);
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    iterator
    find(const K &x)
    {
        return internal_find(x);
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    const_iterator
    find(const K &x) const
    {
        return internal_find(x);
    }

    size_type
    count(const key_type &key) const
    {
        return internal_count(key);
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    size_type
    count(const K &key) const
    {
        return internal_count(key);
    }

    bool
    contains(const key_type &key) const
    {
        return find(key) != end();
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    bool
    contains(const K &x) const
    {
        return find(x) != end();
    }

    void
    clear()
    {
        assert(dummy_head->height() > 0);
        obj::pool_base pop = get_pool_base();

        persistent_node_ptr current = dummy_head->next(0);

        obj::flat_transaction::run(pop, [&] {
            while (current) {
                assert(current->height() > 0);
                persistent_node_ptr next = current->next(0);
                delete_node(current);
                current = next;
            }

            node_ptr head = dummy_head.get();
            for (size_type i = 0; i < head->height(); ++i) {
                head->set_next_tx(i, nullptr);
            }

            on_init_size = 0;
            tls_data.clear();
            obj::flat_transaction::snapshot((size_t *)&_size);
            _size = 0;
        });
    }

    iterator
    begin()
    {
        return iterator(dummy_head.get()->next(0).get());
    }

    const_iterator
    begin() const
    {
        return const_iterator(dummy_head.get()->next(0).get());
    }

    const_iterator
    cbegin() const
    {
        return const_iterator(dummy_head.get()->next(0).get());
    }

    iterator
    end()
    {
        return iterator(nullptr);
    }

    const_iterator
    end() const
    {
        return const_iterator(nullptr);
    }

    const_iterator
    cend() const
    {
        return const_iterator(nullptr);
    }

    size_type
    size() const
    {
        return _size.load(std::memory_order_relaxed);
    }

    size_type
    max_size() const
    {
        return std::numeric_limits<difference_type>::max();
    }

    bool
    empty() const
    {
        return 0 == size();
    }

    allocator_type
    get_allocator() const
    {
        return allocator_type(_node_allocator);
    }

    void
    swap(concurrent_skip_list &other)
    {
        obj::pool_base pop = get_pool_base();
        obj::flat_transaction::run(pop, [&] {
            using pocs_t = typename node_allocator_traits::
                propagate_on_container_swap;
            allocator_swap(_node_allocator, other._node_allocator,
                       pocs_t());
            std::swap(_compare, other._compare);
            std::swap(_rnd_generator, other._rnd_generator);
            std::swap(dummy_head, other.dummy_head);
            on_init_size.swap(other.on_init_size);

            obj::flat_transaction::snapshot((size_t *)&_size);
            obj::flat_transaction::snapshot(
                (size_t *)&(other._size));
            _size = other._size.exchange(_size,
                             std::memory_order_relaxed);
        });
    }

    std::pair<iterator, iterator>
    equal_range(const key_type &key)
    {
        return std::pair<iterator, iterator>(lower_bound(key),
                             upper_bound(key));
    }

    std::pair<const_iterator, const_iterator>
    equal_range(const key_type &key) const
    {
        return std::pair<const_iterator, const_iterator>(
            lower_bound(key), upper_bound(key));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    std::pair<iterator, iterator>
    equal_range(const K &x)
    {
        return std::pair<iterator, iterator>(lower_bound(x),
                             upper_bound(x));
    }

    template <typename K,
          typename = typename std::enable_if<
              has_is_transparent<key_compare>::value, K>::type>
    std::pair<const_iterator, const_iterator>
    equal_range(const K &key) const
    {
        return std::pair<const_iterator, const_iterator>(
            lower_bound(key), upper_bound(key));
    }

    const key_compare &
    key_comp() const
    {
        return _compare;
    }

    key_compare &
    key_comp()
    {
        return _compare;
    }

private:
    /* Status flags stored in insert_stage field */
    enum insert_stage_type : uint8_t { not_started = 0, in_progress = 1 };
    /*
     * Structure of thread local data.
     * Size should be 64 bytes.
     */
    struct tls_entry_type {
        persistent_node_ptr ptr;
        obj::p<difference_type> size_diff;
        obj::p<insert_stage_type> insert_stage;

        char reserved[64 - sizeof(decltype(ptr)) -
                  sizeof(decltype(size_diff)) -
                  sizeof(decltype(insert_stage))];
    };
    static_assert(sizeof(tls_entry_type) == 64,
              "The size of tls_entry_type should be 64 bytes.");

    void
    check_tx_stage_work() const
    {
        if (pmemobj_tx_stage() != TX_STAGE_WORK)
            throw pmem::transaction_scope_error(
                "Function called out of transaction scope.");
    }

    /* Helper method which throws an exception when called in a tx */
    static inline void
    check_outside_tx()
    {
        if (pmemobj_tx_stage() != TX_STAGE_NONE)
            throw pmem::transaction_scope_error(
                "Function called inside transaction scope.");
    }

    void
    init()
    {
        if (pool_uuid == 0)
            throw pmem::pool_error("Invalid pool handle.");

        _size = 0;
        on_init_size = 0;
        create_dummy_head();
    }

    void
    internal_move(concurrent_skip_list &&other)
    {
        assert(this->empty());
        assert(pmemobj_tx_stage() == TX_STAGE_WORK);
        dummy_head = other.dummy_head;
        other.dummy_head = nullptr;
        other.create_dummy_head();

        _size.store(other._size.load(std::memory_order_relaxed),
                std::memory_order_relaxed);
        on_init_size = other.on_init_size;
    }

    static const_reference
    get_val(const_node_ptr n)
    {
        assert(n);
        return *(n->get());
    }

    static reference
    get_val(node_ptr n)
    {
        assert(n);
        return *(n->get());
    }

    static const key_type &
    get_key(const_node_ptr n)
    {
        assert(n);
        return traits_type::get_key(get_val(n));
    }

    template <typename K>
    iterator
    internal_find(const K &key)
    {
        iterator it = lower_bound(key);
        return (it == end() || _compare(key, traits_type::get_key(*it)))
            ? end()
            : it;
    }

    template <typename K>
    const_iterator
    internal_find(const K &key) const
    {
        const_iterator it = lower_bound(key);
        return (it == end() || _compare(key, traits_type::get_key(*it)))
            ? end()
            : it;
    }

    template <typename K>
    size_type
    internal_count(const K &key) const
    {
        if (allow_multimapping) {
            std::pair<const_iterator, const_iterator> range =
                equal_range(key);
            return static_cast<size_type>(
                std::distance(range.first, range.second));
        }
        return (find(key) == end()) ? size_type(0) : size_type(1);
    }

    template <typename K, typename pointer_type, typename comparator>
    persistent_node_ptr
    internal_find_position(size_type level, pointer_type &prev,
                   const K &key, const comparator &cmp) const
    {
        assert(level < prev->height());
        persistent_node_ptr next = prev->next(level);
        pointer_type curr = next.get();

        while (curr && cmp(get_key(curr), key)) {
            prev = curr;
            assert(level < prev->height());
            next = prev->next(level);
            curr = next.get();
        }

        return next;
    }

    template <typename K>
    void
    find_insert_pos(prev_array_type &prev_nodes,
            next_array_type &next_nodes, const K &key)
    {
        if (allow_multimapping) {
            fill_prev_next_arrays(prev_nodes, next_nodes, key,
                          not_greater_compare(_compare));
        } else {
            fill_prev_next_arrays(prev_nodes, next_nodes, key,
                          _compare);
        }
    }

    template <typename K, typename comparator>
    void
    fill_prev_next_arrays(prev_array_type &prev_nodes,
                  next_array_type &next_nodes, const K &key,
                  const comparator &cmp)
    {
        node_ptr prev = dummy_head.get();
        prev_nodes.fill(prev);
        next_nodes.fill(nullptr);

        for (size_type h = prev->height(); h > 0; --h) {
            persistent_node_ptr next =
                internal_find_position(h - 1, prev, key, cmp);
            prev_nodes[h - 1] = prev;
            next_nodes[h - 1] = next;
        }
    }

    template <typename K, typename... Args>
    std::pair<iterator, bool>
    internal_try_emplace(K &&key, Args &&... args)
    {
        return internal_insert(
            key, std::piecewise_construct,
            std::forward_as_tuple(std::forward<K>(key)),
            std::forward_as_tuple(std::forward<Args>(args)...));
    }

    template <typename... Args>
    std::pair<iterator, bool>
    internal_emplace(Args &&... args)
    {
        check_outside_tx();
        tls_entry_type &tls_entry = tls_data.local();
        obj::pool_base pop = get_pool_base();

        obj::flat_transaction::run(pop, [&] {
            assert(tls_entry.ptr == nullptr);
            tls_entry.ptr =
                create_node(std::forward<Args>(args)...);
            ++tls_entry.size_diff;
            tls_entry.insert_stage = not_started;
        });

        node_ptr n = tls_entry.ptr.get();
        size_type height = n->height();

        std::pair<iterator, bool> insert_result = internal_insert_node(
            get_key(n), height,
            [&](const next_array_type &next_nodes)
                -> persistent_node_ptr & {
                assert(tls_entry.insert_stage == not_started);
                assert(tls_entry.ptr != nullptr);

                n->set_nexts(pop, next_nodes.data(), height);

                tls_entry.insert_stage = in_progress;
                pop.persist(&(tls_entry.insert_stage),
                        sizeof(tls_entry.insert_stage));

                return tls_entry.ptr;
            });

        if (!insert_result.second) {
            assert(tls_entry.ptr != nullptr);
            assert(tls_entry.insert_stage == not_started);

            obj::flat_transaction::run(pop, [&] {
                --tls_entry.size_diff;
                delete_node(tls_entry.ptr);
                tls_entry.ptr = nullptr;
            });
        }

        assert(tls_entry.ptr == nullptr);
        return insert_result;
    }

    template <typename... Args>
    std::pair<iterator, bool>
    internal_unsafe_emplace(Args &&... args)
    {
        check_tx_stage_work();

        persistent_node_ptr new_node =
            create_node(std::forward<Args>(args)...);

        node_ptr n = new_node.get();
        size_type height = n->height();

        std::pair<iterator, bool> insert_result = internal_insert_node(
            get_key(n), height,
            [&](const next_array_type &next_nodes)
                -> persistent_node_ptr & {
                assert(new_node != nullptr);

                n->set_nexts(next_nodes.data(), height);

                return new_node;
            });

        if (insert_result.second) {
            ++on_init_size;
        } else {
            assert(new_node != nullptr);

            delete_node(new_node);
        }

        return insert_result;
    }

    template <typename K, typename... Args>
    std::pair<iterator, bool>
    internal_insert(const K &key, Args &&... args)
    {
        check_outside_tx();
        tls_entry_type &tls_entry = tls_data.local();
        assert(tls_entry.ptr == nullptr);

        size_type height = random_level();

        std::pair<iterator, bool> insert_result = internal_insert_node(
            key, height,
            [&](const next_array_type &next_nodes)
                -> persistent_node_ptr & {
                obj::pool_base pop = get_pool_base();

                obj::flat_transaction::manual tx(pop);
                tls_entry.ptr = create_node(
                    std::forward_as_tuple(
                        height, next_nodes.data()),
                    std::forward_as_tuple(
                        std::forward<Args>(args)...));

                ++(tls_entry.size_diff);
                tls_entry.insert_stage = in_progress;
                obj::flat_transaction::commit();

                assert(tls_entry.ptr != nullptr);
                return tls_entry.ptr;
            });

        assert(tls_entry.ptr == nullptr);

        return insert_result;
    }

    template <typename K, typename PrepareNode>
    std::pair<iterator, bool>
    internal_insert_node(const K &key, size_type height,
                 PrepareNode &&prepare_new_node)
    {
        prev_array_type prev_nodes;
        next_array_type next_nodes;
        node_ptr n = nullptr;

        do {
            find_insert_pos(prev_nodes, next_nodes, key);

            node_ptr next = next_nodes[0].get();
            if (next && !allow_multimapping &&
                !_compare(key, get_key(next))) {

                return std::pair<iterator, bool>(iterator(next),
                                 false);
            }

        } while ((n = try_insert_node(prev_nodes, next_nodes, height,
                          std::forward<PrepareNode>(
                              prepare_new_node))) ==
             nullptr);

        assert(n);
        return std::pair<iterator, bool>(iterator(n), true);
    }

    template <typename PrepareNode>
    node_ptr
    try_insert_node(prev_array_type &prev_nodes,
            const next_array_type &next_nodes, size_type height,
            PrepareNode &&prepare_new_node)
    {
        assert(dummy_head->height() >= height);

        lock_array locks;
        if (!try_lock_nodes(height, prev_nodes, next_nodes, locks)) {
            return nullptr;
        }

        node_lock_type new_node_lock;

        persistent_node_ptr &new_node = prepare_new_node(next_nodes);
        assert(new_node != nullptr);
        node_ptr n = new_node.get();

        /*
         * We need to hold lock to the new node until changes
         * are committed to persistent domain. Otherwise, the
         * new node would be visible to concurrent inserts
         * before it is persisted.
         */
        new_node_lock = n->acquire();

        obj::pool_base pop = get_pool_base();
        /*
         * In the loop below we are linking a new node to all layers of
         * the skip list. Transaction is not required because in case of
         * failure the node is reachable via a pointer from persistent
         * TLS. During recovery, we will complete the insert. It is also
         * OK if concurrent readers will see not a fully-linked node
         * because during recovery the insert procedure will be
         * completed.
         */
        for (size_type level = 0; level < height; ++level) {
            assert(prev_nodes[level]->height() > level);
            assert(prev_nodes[level]->next(level) ==
                   next_nodes[level]);
            assert(prev_nodes[level]->next(level) ==
                   n->next(level));
            prev_nodes[level]->set_next(pop, level, new_node);
        }

#ifndef NDEBUG
        try_insert_node_finish_marker();
#endif

        new_node = nullptr;
        /* We need to persist the node pointer. Otherwise, on a restart,
         * this pointer might be not null but the node can be already
         * deleted. */
        pop.persist(&new_node, sizeof(new_node));

        ++_size;
#if LIBPMEMOBJ_CPP_VG_PMEMCHECK_ENABLED
        VALGRIND_PMC_DO_FLUSH(&_size, sizeof(_size));
#endif

        assert(n);
        return n;
    }

    bool
    check_prev_array(const prev_array_type &prevs, size_type height)
    {
        for (size_type l = 1; l < height; ++l) {
            if (prevs[l] == dummy_head.get()) {
                continue;
            }

            assert(prevs[l - 1] != dummy_head.get());
            assert(!_compare(get_key(prevs[l - 1]),
                     get_key(prevs[l])));
        }

        return true;
    }

    bool
    try_lock_nodes(size_type height, prev_array_type &prevs,
               const next_array_type &nexts, lock_array &locks)
    {
        assert(check_prev_array(prevs, height));

        for (size_type l = 0; l < height; ++l) {
            if (l == 0 || prevs[l] != prevs[l - 1]) {
                locks[l] = prevs[l]->acquire();
            }

            persistent_node_ptr next = prevs[l]->next(l);
            if (next != nexts[l])
                /* Other thread inserted to this position and
                 * modified the pointer before we acquired the
                 * lock */
                return false;
        }

        return true;
    }

    template <typename K, typename comparator>
    const_iterator
    internal_get_bound(const K &key, const comparator &cmp) const
    {
        const_node_ptr prev = dummy_head.get();
        assert(prev->height() > 0);
        persistent_node_ptr next = nullptr;

        for (size_type h = prev->height(); h > 0; --h) {
            next = internal_find_position(h - 1, prev, key, cmp);
        }

        return const_iterator(next.get());
    }

    template <typename K, typename comparator>
    iterator
    internal_get_bound(const K &key, const comparator &cmp)
    {
        node_ptr prev = dummy_head.get();
        assert(prev->height() > 0);
        persistent_node_ptr next = nullptr;

        for (size_type h = prev->height(); h > 0; --h) {
            next = internal_find_position(h - 1, prev, key, cmp);
        }

        return iterator(next.get());
    }

    template <typename K, typename comparator>
    const_iterator
    internal_get_biggest_less_than(const K &key,
                       const comparator &cmp) const
    {
        const_node_ptr prev = dummy_head.get();
        assert(prev->height() > 0);

        for (size_type h = prev->height(); h > 0; --h) {
            internal_find_position(h - 1, prev, key, cmp);
        }

        if (prev == dummy_head.get())
            return end();

        return const_iterator(prev);
    }

    iterator
    internal_erase(const_iterator pos, obj::p<difference_type> &size_diff)
    {
        assert(pos != end());

        obj::pool_base pop = get_pool_base();

        std::pair<persistent_node_ptr, persistent_node_ptr>
            extract_result(nullptr, nullptr);

        obj::flat_transaction::run(pop, [&] {
            extract_result = internal_extract(pos);

            /* Make sure that node was extracted */
            assert(extract_result.first != nullptr);
            delete_node(extract_result.first);
            --size_diff;
            obj::flat_transaction::snapshot((size_type *)&_size);
            --_size;
        });

        return iterator(extract_result.second.get());
    }

    std::pair<persistent_node_ptr, persistent_node_ptr>
    internal_extract(const_iterator it)
    {
        assert(dummy_head->height() > 0);
        assert(it != end());
        assert(pmemobj_tx_stage() == TX_STAGE_WORK);

        const key_type &key = traits_type::get_key(*it);

        prev_array_type prev_nodes;
        next_array_type next_nodes;

        fill_prev_next_arrays(prev_nodes, next_nodes, key, _compare);

        node_ptr erase_node = next_nodes[0].get();
        assert(erase_node != nullptr);

        if (!_compare(key, get_key(erase_node))) {
            /* XXX: this assertion will fail in case of multimap
             * because we take the first node with the same key.
             * Need to extend algorithm for mutimap. */
            assert(erase_node == it.node);
            return internal_extract_node(prev_nodes, next_nodes,
                             erase_node);
        }

        return std::pair<persistent_node_ptr, persistent_node_ptr>(
            nullptr, nullptr);
    }

    std::pair<persistent_node_ptr, persistent_node_ptr>
    internal_extract_node(const prev_array_type &prev_nodes,
                  const next_array_type &next_nodes,
                  node_ptr erase_node)
    {
        assert(pmemobj_tx_stage() == TX_STAGE_WORK);
        assert(erase_node != nullptr);
        for (size_type level = 0; level < erase_node->height();
             ++level) {
            assert(prev_nodes[level]->height() > level);
            assert(next_nodes[level].get() == erase_node);
            prev_nodes[level]->set_next_tx(level,
                               erase_node->next(level));
        }

        return std::pair<persistent_node_ptr, persistent_node_ptr>(
            next_nodes[0], erase_node->next(0));
    }

    obj::pool_base
    get_pool_base() const
    {
        PMEMobjpool *pop = pmemobj_pool_by_ptr(this);
        return obj::pool_base(pop);
    }

    void
    internal_copy(const concurrent_skip_list &other)
    {
        internal_copy(other.begin(), other.end());
    }

    template <typename Iterator>
    void
    internal_copy(Iterator first, Iterator last)
    {
        assert(pmemobj_tx_stage() == TX_STAGE_WORK);

        prev_array_type prev_nodes;
        prev_nodes.fill(dummy_head.get());
        size_type sz = 0;

        for (; first != last; ++first, ++sz) {
            persistent_node_ptr new_node = create_node(*first);
            node_ptr n = new_node.get();
            for (size_type level = 0; level < n->height();
                 ++level) {
                prev_nodes[level]->set_next_tx(level, new_node);
                prev_nodes[level] = n;
            }
        }

        on_init_size = sz;
        /*
         * As internal_swap can only be called from one thread, and
         * there can be an outer transaction we must make sure that size
         * change is transactional
         */
        obj::flat_transaction::snapshot((size_type *)&_size);
        _size = sz;
        assert(std::is_sorted(
            this->begin(), this->end(),
            [&](const value_type &lhs, const value_type &rhs) {
                return lhs.first < rhs.first;
            }));
    }

    size_type
    random_level()
    {
        return _rnd_generator();
    }

    static size_type
    calc_node_size(size_type height)
    {
        return sizeof(list_node_type) +
            height * sizeof(typename list_node_type::node_pointer);
    }

    template <typename... Args>
    persistent_node_ptr
    create_node(Args &&... args)
    {
        size_type levels = random_level();

        return create_node(
            std::forward_as_tuple(levels),
            std::forward_as_tuple(std::forward<Args>(args)...));
    }

    template <typename... NodeArgs, typename... ValueArgs>
    persistent_node_ptr
    create_node(std::tuple<NodeArgs...> &&node_args,
            std::tuple<ValueArgs...> &&value_args)
    {
        assert(pmemobj_tx_stage() == TX_STAGE_WORK);

        persistent_node_ptr node = creates_dummy_node(
            std::forward<std::tuple<NodeArgs...>>(node_args),
            index_sequence_for<NodeArgs...>{});

        construct_value_type(
            node,
            std::forward<std::tuple<ValueArgs...>>(value_args),
            index_sequence_for<ValueArgs...>{});

        return node;
    }

    template <typename Tuple, size_t... I>
    void
    construct_value_type(persistent_node_ptr node, Tuple &&args,
                 index_sequence<I...>)
    {
        node_ptr new_node = node.get();

        node_allocator_traits::construct(
            _node_allocator, new_node->get(),
            std::get<I>(std::forward<Tuple>(args))...);
    }

    void
    create_dummy_head()
    {
        dummy_head = creates_dummy_node(MAX_LEVEL);
    }

    template <typename Tuple, size_t... I>
    persistent_node_ptr
    creates_dummy_node(Tuple &&args, index_sequence<I...>)
    {
        return creates_dummy_node(
            std::get<I>(std::forward<Tuple>(args))...);
    }

    template <typename... Args>
    persistent_node_ptr
    creates_dummy_node(size_type height, Args &&... args)
    {
        assert(pmemobj_tx_stage() == TX_STAGE_WORK);
        size_type sz = calc_node_size(height);

        persistent_node_ptr n =
            node_allocator_traits::allocate(_node_allocator, sz)
                .raw();

        assert(n != nullptr);

        node_allocator_traits::construct(_node_allocator, n.get(),
                         height,
                         std::forward<Args>(args)...);

        return n;
    }

    template <bool is_dummy = false>
    void
    delete_node(persistent_node_ptr &node)
    {
        assert(pmemobj_tx_stage() == TX_STAGE_WORK);
        node_ptr n = node.get();
        size_type sz = calc_node_size(n->height());

        /* Destroy value */
        if (!is_dummy)
            node_allocator_traits::destroy(_node_allocator,
                               n->get());
        /* Destroy node */
        node_allocator_traits::destroy(_node_allocator, n);
        /* Deallocate memory */
        deallocate_node(node, sz);
        node = nullptr;
    }

    void
    deallocate_node(persistent_node_ptr &node, size_type sz)
    {
        /*
         * Each node object has different size which depends on number
         * of layers the node is linked. Therefore, allocate/deallocate
         * just a raw byte array. persistent_ptr<uint8_t> is used as a
         * pointer to raw array of bytes.
         */
        obj::persistent_ptr<uint8_t> tmp =
            node.to_persistent_ptr().raw();
        node_allocator_traits::deallocate(_node_allocator, tmp, sz);
    }

    void
    delete_dummy_head()
    {
        assert(dummy_head != nullptr);
        delete_node<true>(dummy_head);
        assert(dummy_head == nullptr);
    }

    iterator
    get_iterator(const_iterator it)
    {
        return iterator(
            const_cast<typename iterator::node_ptr>(it.node));
    }

    void
    tls_restore()
    {
        int64_t last_run_size = 0;
        obj::pool_base pop = get_pool_base();

        for (auto &tls_entry : tls_data) {
            persistent_node_ptr &node = tls_entry.ptr;
            auto &size_diff = tls_entry.size_diff;
            if (node) {
                /*
                 * We are completing inserts which were in
                 * progress before the crash because readers
                 * might saw incompleted inserts before the
                 * crash. We set the in_progress flag inside
                 * try_insert_node function when we locked the
                 * predecessors for the new node, therefore,
                 * only single node with the same key might have
                 * the in_progress status.
                 */
                if (tls_entry.insert_stage == in_progress) {
                    complete_insert(tls_entry);
                } else {
                    obj::flat_transaction::run(pop, [&] {
                        --(tls_entry.size_diff);
                        delete_node(node);
                        node = nullptr;
                    });
                }
            }

            assert(node == nullptr);

            last_run_size += size_diff;
        }

        /* Make sure that on_init_size + last_run_size >= 0 */
        assert(last_run_size >= 0 ||
               on_init_size >
                   static_cast<size_type>(std::abs(last_run_size)));
        obj::flat_transaction::run(pop, [&] {
            tls_data.clear();
            on_init_size += static_cast<size_t>(last_run_size);
        });
        _size = on_init_size;
#if LIBPMEMOBJ_CPP_VG_PMEMCHECK_ENABLED
        VALGRIND_PMC_DO_FLUSH(&_size, sizeof(_size));
#endif
    }

    void
    complete_insert(tls_entry_type &tls_entry)
    {
        persistent_node_ptr &node = tls_entry.ptr;
        assert(node != nullptr);
        assert(tls_entry.insert_stage == in_progress);
        prev_array_type prev_nodes;
        next_array_type next_nodes;
        node_ptr n = node.get();
        const key_type &key = get_key(n);
        size_type height = n->height();

        fill_prev_next_arrays(prev_nodes, next_nodes, key, _compare);
        obj::pool_base pop = get_pool_base();

        /* Node was partially linked */
        for (size_type level = 0; level < height; ++level) {
            assert(prev_nodes[level]->height() > level);
            assert(prev_nodes[level]->next(level) ==
                   next_nodes[level]);

            if (prev_nodes[level]->next(level) != node) {
                /* Otherwise, node already linked on
                 * this layer */
                assert(n->next(level) == next_nodes[level]);
                prev_nodes[level]->set_next(pop, level, node);
            }
        }

        node = nullptr;
        pop.persist(&node, sizeof(node));
    }

    struct not_greater_compare {
        const key_compare &my_less_compare;

        not_greater_compare(const key_compare &less_compare)
            : my_less_compare(less_compare)
        {
        }

        template <typename K1, typename K2>
        bool
        operator()(const K1 &first, const K2 &second) const
        {
            return !my_less_compare(second, first);
        }
    };

    const uint64_t pool_uuid = pmemobj_oid(this).pool_uuid_lo;
    node_allocator_type _node_allocator;
    key_compare _compare;
    random_level_generator_type _rnd_generator;
    persistent_node_ptr dummy_head;

    enumerable_thread_specific<tls_entry_type> tls_data;

    std::atomic<size_type> _size;

    obj::p<size_type> on_init_size;
}; /* class concurrent_skip_list */

template <typename Key, typename Value, typename KeyCompare,
      typename RND_GENERATOR, typename Allocator, bool AllowMultimapping,
      size_t MAX_LEVEL>
class map_traits {
public:
    static constexpr size_t max_level = MAX_LEVEL;
    using random_generator_type = RND_GENERATOR;
    using key_type = Key;
    using mapped_type = Value;
    using compare_type = KeyCompare;
    using value_type = pair<const key_type, mapped_type>;
    using reference = value_type &;
    using const_reference = const value_type &;
    using allocator_type = Allocator;

    constexpr static bool allow_multimapping = AllowMultimapping;

    static const key_type &
    get_key(const_reference val)
    {
        return val.first;
    }
}; /* class map_traits */

} /* namespace detail */
} /* namespace pmem */

#endif /* PMEMOBJ_CONCURRENT_SKIP_LIST_IMPL_HPP */