libpmemkv.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2017-2021, Intel Corporation */
 
#ifndef LIBPMEMKV_HPP
#define LIBPMEMKV_HPP
 
#include <cassert>
#include <functional>
#include <iostream>
#include <libpmemobj++/slice.hpp>
#include <libpmemobj++/string_view.hpp>
#include <memory>
#include <stdexcept>
#include <string>
#include <utility>
 
#include "libpmemkv.h"
#include <libpmemobj/pool_base.h>
 
namespace pmem
{
namespace kv
{
using string_view = obj::string_view;
 
typedef int get_kv_function(string_view key, string_view value);
typedef void get_v_function(string_view value);
 
using get_kv_callback = pmemkv_get_kv_callback;
using get_v_callback = pmemkv_get_v_callback;
 
enum class status {
    OK = PMEMKV_STATUS_OK,               
    UNKNOWN_ERROR = PMEMKV_STATUS_UNKNOWN_ERROR, 
    NOT_FOUND = PMEMKV_STATUS_NOT_FOUND, 
    NOT_SUPPORTED = PMEMKV_STATUS_NOT_SUPPORTED, 
    INVALID_ARGUMENT = PMEMKV_STATUS_INVALID_ARGUMENT, 
    CONFIG_PARSING_ERROR =
        PMEMKV_STATUS_CONFIG_PARSING_ERROR, 
    CONFIG_TYPE_ERROR =
        PMEMKV_STATUS_CONFIG_TYPE_ERROR, 
    STOPPED_BY_CB = PMEMKV_STATUS_STOPPED_BY_CB, 
    OUT_OF_MEMORY =
        PMEMKV_STATUS_OUT_OF_MEMORY, 
    WRONG_ENGINE_NAME =
        PMEMKV_STATUS_WRONG_ENGINE_NAME, 
    TRANSACTION_SCOPE_ERROR =
        PMEMKV_STATUS_TRANSACTION_SCOPE_ERROR, 
    DEFRAG_ERROR = PMEMKV_STATUS_DEFRAG_ERROR, 
    COMPARATOR_MISMATCH =
        PMEMKV_STATUS_COMPARATOR_MISMATCH, 
};
 
inline std::ostream &operator<<(std::ostream &os, const status &s)
{
    static const std::string statuses[] = {"OK",
                           "UNKNOWN_ERROR",
                           "NOT_FOUND",
                           "NOT_SUPPORTED",
                           "INVALID_ARGUMENT",
                           "CONFIG_PARSING_ERROR",
                           "CONFIG_TYPE_ERROR",
                           "STOPPED_BY_CB",
                           "OUT_OF_MEMORY",
                           "WRONG_ENGINE_NAME",
                           "TRANSACTION_SCOPE_ERROR",
                           "DEFRAG_ERROR",
                           "COMPARATOR_MISMATCH"};
 
    int status_no = static_cast<int>(s);
    os << statuses[status_no] << " (" << status_no << ")";
 
    return os;
}
 
class bad_result_access : public std::runtime_error {
public:
    bad_result_access(const char *what_arg) : std::runtime_error(what_arg)
    {
    }
 
    const char *what() const noexcept final
    {
        return std::runtime_error::what();
    }
 
private:
};
 
template <typename T>
class result {
    union {
        T value;
    };
 
    status s;
 
public:
    result(const T &val) noexcept(noexcept(T(std::declval<T>())));
    result(const status &err) noexcept;
    result(const result &other) noexcept(noexcept(T(std::declval<T>())));
    result(result &&other) noexcept(noexcept(T(std::declval<T>())));
    result(T &&val) noexcept(noexcept(T(std::declval<T>())));
 
    ~result();
 
    result &
    operator=(const result &other) noexcept(noexcept(std::declval<T>().~T()) &&
                        noexcept(T(std::declval<T>())));
    result &operator=(result &&other) noexcept(noexcept(std::declval<T>().~T()) &&
                           noexcept(T(std::declval<T>())));
 
    bool is_ok() const noexcept;
 
    const T &get_value() const &;
    T &get_value() &;
 
    T &&get_value() &&;
 
    status get_status() const noexcept;
};
 
template <typename T>
result<T>::result(const T &val) noexcept(noexcept(T(std::declval<T>())))
    : value(val), s(status::OK)
{
}
 
template <typename T>
result<T>::result(const status &status) noexcept : s(status)
{
    assert(s != status::OK);
}
 
template <typename T>
result<T>::result(const result &other) noexcept(noexcept(T(std::declval<T>())))
    : s(other.s)
{
    if (s == status::OK)
        new (&value) T(other.value);
}
 
template <typename T>
result<T>::result(result &&other) noexcept(noexcept(T(std::declval<T>()))) : s(other.s)
{
    if (s == status::OK)
        new (&value) T(std::move(other.value));
 
    other.s = status::UNKNOWN_ERROR;
}
 
template <typename T>
result<T>::~result()
{
    if (s == status::OK)
        value.~T();
}
 
template <typename T>
result<T> &
result<T>::operator=(const result &other) noexcept(noexcept(std::declval<T>().~T()) &&
                           noexcept(T(std::declval<T>())))
{
    if (s == status::OK && other.is_ok())
        value = other.value;
    else if (other.is_ok())
        new (&value) T(other.value);
    else if (s == status::OK)
        value.~T();
 
    s = other.s;
 
    return *this;
}
 
template <typename T>
result<T> &
result<T>::operator=(result &&other) noexcept(noexcept(std::declval<T>().~T()) &&
                          noexcept(T(std::declval<T>())))
{
    if (s == status::OK && other.is_ok())
        value = std::move(other.value);
    else if (other.is_ok())
        new (&value) T(std::move(other.value));
    else if (s == status::OK)
        value.~T();
 
    s = other.s;
    other.s = status::UNKNOWN_ERROR;
 
    return *this;
}
 
template <typename T>
result<T>::result(T &&val) noexcept(noexcept(T(std::declval<T>())))
    : value(std::move(val)), s(status::OK)
{
}
 
template <typename T>
bool result<T>::is_ok() const noexcept
{
    return s == status::OK;
}
 
template <typename T>
const T &result<T>::get_value() const &
{
    if (s == status::OK)
        return value;
    else
        throw bad_result_access("bad_result_access: value doesn't exist");
}
 
template <typename T>
T &result<T>::get_value() &
{
    if (s == status::OK)
        return value;
    else
        throw bad_result_access("bad_result_access: value doesn't exist");
}
 
template <typename T>
T &&result<T>::get_value() &&
{
    if (s == status::OK) {
        s = status::UNKNOWN_ERROR;
        return std::move(value);
    } else
        throw bad_result_access("bad_result_access: value doesn't exist");
}
 
template <typename T>
status result<T>::get_status() const noexcept
{
    return s;
}
 
template <typename T>
bool operator==(const result<T> &lhs, const status &rhs)
{
    return lhs.get_status() == rhs;
}
 
template <typename T>
bool operator==(const status &lhs, const result<T> &rhs)
{
    return lhs == rhs.get_status();
}
 
template <typename T>
bool operator!=(const result<T> &lhs, const status &rhs)
{
    return lhs.get_status() != rhs;
}
 
template <typename T>
bool operator!=(const status &lhs, const result<T> &rhs)
{
    return lhs != rhs.get_status();
}
 
class config {
public:
    config() noexcept;
    explicit config(pmemkv_config *cfg) noexcept;
 
    template <typename T>
    status put_data(const std::string &key, const T *value,
            const std::size_t number = 1) noexcept;
 
    template <typename T>
    status put_object(const std::string &key, T *value,
              void (*deleter)(void *)) noexcept;
    template <typename T, typename D>
    status put_object(const std::string &key, std::unique_ptr<T, D> object) noexcept;
    status put_uint64(const std::string &key, std::uint64_t value) noexcept;
    status put_int64(const std::string &key, std::int64_t value) noexcept;
    status put_string(const std::string &key, const std::string &value) noexcept;
 
    status put_size(std::uint64_t size) noexcept;
    status put_path(const std::string &path) noexcept;
    status put_force_create(bool value) noexcept;
    status put_oid(PMEMoid *oid) noexcept;
    template <typename Comparator>
    status put_comparator(Comparator &&comparator);
 
    template <typename T>
    status get_data(const std::string &key, T *&value, std::size_t &number) const
        noexcept;
    template <typename T>
    status get_object(const std::string &key, T *&value) const noexcept;
 
    status get_uint64(const std::string &key, std::uint64_t &value) const noexcept;
    status get_int64(const std::string &key, std::int64_t &value) const noexcept;
    status get_string(const std::string &key, std::string &value) const noexcept;
 
    pmemkv_config *release() noexcept;
 
private:
    int init() noexcept;
 
    std::unique_ptr<pmemkv_config, decltype(&pmemkv_config_delete)> config_;
};
 
class tx {
public:
    tx(pmemkv_tx *tx_) noexcept;
 
    status put(string_view key, string_view value) noexcept;
    status remove(string_view key) noexcept;
    status commit() noexcept;
    void abort() noexcept;
 
private:
    std::unique_ptr<pmemkv_tx, decltype(&pmemkv_tx_end)> tx_;
};
 
class db {
    template <bool IsConst>
    class iterator;
 
public:
    using read_iterator = iterator<true>;
    using write_iterator = iterator<false>;
 
    db() noexcept;
 
    status open(const std::string &engine_name, config &&cfg = config{}) noexcept;
 
    void close() noexcept;
 
    status count_all(std::size_t &cnt) noexcept;
    status count_above(string_view key, std::size_t &cnt) noexcept;
    status count_equal_above(string_view key, std::size_t &cnt) noexcept;
    status count_equal_below(string_view key, std::size_t &cnt) noexcept;
    status count_below(string_view key, std::size_t &cnt) noexcept;
    status count_between(string_view key1, string_view key2,
                 std::size_t &cnt) noexcept;
 
    status get_all(get_kv_callback *callback, void *arg) noexcept;
    status get_all(std::function<get_kv_function> f) noexcept;
 
    status get_above(string_view key, get_kv_callback *callback, void *arg) noexcept;
    status get_above(string_view key, std::function<get_kv_function> f) noexcept;
 
    status get_equal_above(string_view key, get_kv_callback *callback,
                   void *arg) noexcept;
    status get_equal_above(string_view key,
                   std::function<get_kv_function> f) noexcept;
 
    status get_equal_below(string_view key, get_kv_callback *callback,
                   void *arg) noexcept;
    status get_equal_below(string_view key,
                   std::function<get_kv_function> f) noexcept;
 
    status get_below(string_view key, get_kv_callback *callback, void *arg) noexcept;
    status get_below(string_view key, std::function<get_kv_function> f) noexcept;
 
    status get_between(string_view key1, string_view key2, get_kv_callback *callback,
               void *arg) noexcept;
    status get_between(string_view key1, string_view key2,
               std::function<get_kv_function> f) noexcept;
 
    status exists(string_view key) noexcept;
 
    status get(string_view key, get_v_callback *callback, void *arg) noexcept;
    status get(string_view key, std::function<get_v_function> f) noexcept;
    status get(string_view key, std::string *value) noexcept;
 
    status put(string_view key, string_view value) noexcept;
    status remove(string_view key) noexcept;
    status defrag(double start_percent = 0, double amount_percent = 100);
 
    result<tx> tx_begin() noexcept;
 
    result<read_iterator> new_read_iterator();
    result<write_iterator> new_write_iterator();
 
    std::string errormsg();
 
private:
    std::unique_ptr<pmemkv_db, decltype(&pmemkv_close)> db_;
};
 
template <bool IsConst>
class db::iterator {
    using iterator_type = typename std::conditional<IsConst, pmemkv_iterator,
                            pmemkv_write_iterator>::type;
 
    template <typename T>
    class OutputIterator;
 
public:
    iterator(iterator_type *it);
 
    status seek(string_view key) noexcept;
    status seek_lower(string_view key) noexcept;
    status seek_lower_eq(string_view key) noexcept;
    status seek_higher(string_view key) noexcept;
    status seek_higher_eq(string_view key) noexcept;
 
    status seek_to_first() noexcept;
    status seek_to_last() noexcept;
 
    status is_next() noexcept;
    status next() noexcept;
    status prev() noexcept;
 
    result<string_view> key() noexcept;
 
    result<string_view>
    read_range(size_t pos = 0,
           size_t n = std::numeric_limits<size_t>::max()) noexcept;
 
    template <bool IC = IsConst>
    typename std::enable_if<!IC, result<pmem::obj::slice<OutputIterator<char>>>>::type
    write_range(size_t pos = 0,
            size_t n = std::numeric_limits<size_t>::max()) noexcept;
 
    template <bool IC = IsConst>
    typename std::enable_if<!IC, status>::type commit() noexcept;
    template <bool IC = IsConst>
    typename std::enable_if<!IC>::type abort() noexcept;
 
private:
    std::unique_ptr<
        iterator_type,
        typename std::conditional<IsConst, decltype(&pmemkv_iterator_delete),
                      decltype(&pmemkv_write_iterator_delete)>::type>
        it_;
 
    pmemkv_iterator *get_raw_it();
};
 
template <bool IsConst>
template <typename T>
class db::iterator<IsConst>::OutputIterator {
    struct assign_only;
 
public:
    using reference = assign_only &;
    using pointer = void;
    using difference_type = std::ptrdiff_t;
    using value_type = void;
    using iterator_category = std::output_iterator_tag;
 
    OutputIterator(T *x);
 
    reference operator*();
 
    OutputIterator &operator++();
    OutputIterator operator++(int);
 
    OutputIterator &operator--();
    OutputIterator operator--(int);
 
    assign_only operator[](difference_type pos);
 
    difference_type operator-(const OutputIterator &other) const;
 
    bool operator!=(const OutputIterator &other) const;
 
private:
    struct assign_only {
        friend OutputIterator<T>;
 
        assign_only(T *x);
 
        assign_only &operator=(const T &x);
 
    private:
        T *c;
    };
 
    assign_only ao;
};
 
template <bool IsConst>
template <typename T>
db::iterator<IsConst>::OutputIterator<T>::OutputIterator(T *x) : ao(x)
{
}
 
template <bool IsConst>
template <typename T>
typename db::iterator<IsConst>::template OutputIterator<T>::reference
    db::iterator<IsConst>::OutputIterator<T>::operator*()
{
    return ao;
}
 
template <bool IsConst>
template <typename T>
typename db::iterator<IsConst>::template OutputIterator<T> &
db::iterator<IsConst>::OutputIterator<T>::operator++()
{
    ao.c += sizeof(T);
    return *this;
}
 
template <bool IsConst>
template <typename T>
typename db::iterator<IsConst>::template OutputIterator<T>
db::iterator<IsConst>::OutputIterator<T>::operator++(int)
{
    auto tmp = *this;
    ++(*this);
    return tmp;
}
 
template <bool IsConst>
template <typename T>
typename db::iterator<IsConst>::template OutputIterator<T> &
db::iterator<IsConst>::OutputIterator<T>::operator--()
{
    ao.c -= sizeof(T);
    return *this;
}
 
template <bool IsConst>
template <typename T>
typename db::iterator<IsConst>::template OutputIterator<T>
db::iterator<IsConst>::OutputIterator<T>::operator--(int)
{
    auto tmp = *this;
    --(*this);
    return tmp;
}
 
template <bool IsConst>
template <typename T>
typename db::iterator<IsConst>::template OutputIterator<T>::assign_only
    db::iterator<IsConst>::OutputIterator<T>::operator[](difference_type pos)
{
    return assign_only(ao.c + pos);
}
 
template <bool IsConst>
template <typename T>
typename db::iterator<IsConst>::template OutputIterator<T>::difference_type
db::iterator<IsConst>::OutputIterator<T>::operator-(const OutputIterator &other) const
{
    return this->ao.c - other.ao.c;
}
 
template <bool IsConst>
template <typename T>
bool db::iterator<IsConst>::OutputIterator<T>::operator!=(
    const OutputIterator &other) const
{
    return this->ao.c != other.ao.c;
}
 
template <bool IsConst>
template <typename T>
db::iterator<IsConst>::OutputIterator<T>::assign_only::assign_only(T *x) : c(x)
{
}
 
template <bool IsConst>
template <typename T>
typename db::iterator<IsConst>::template OutputIterator<T>::assign_only &
db::iterator<IsConst>::OutputIterator<T>::assign_only::operator=(const T &x)
{
    *c = x;
    return *this;
}
 
template <>
inline db::iterator<true>::iterator(iterator_type *it) : it_(it, &pmemkv_iterator_delete)
{
}
 
template <>
inline db::iterator<false>::iterator(iterator_type *it)
    : it_(it, &pmemkv_write_iterator_delete)
{
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::seek(string_view key) noexcept
{
    return static_cast<status>(
        pmemkv_iterator_seek(this->get_raw_it(), key.data(), key.size()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::seek_lower(string_view key) noexcept
{
    return static_cast<status>(
        pmemkv_iterator_seek_lower(this->get_raw_it(), key.data(), key.size()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::seek_lower_eq(string_view key) noexcept
{
    return static_cast<status>(pmemkv_iterator_seek_lower_eq(this->get_raw_it(),
                                 key.data(), key.size()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::seek_higher(string_view key) noexcept
{
    return static_cast<status>(
        pmemkv_iterator_seek_higher(this->get_raw_it(), key.data(), key.size()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::seek_higher_eq(string_view key) noexcept
{
    return static_cast<status>(pmemkv_iterator_seek_higher_eq(
        this->get_raw_it(), key.data(), key.size()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::seek_to_first() noexcept
{
    return static_cast<status>(pmemkv_iterator_seek_to_first(this->get_raw_it()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::seek_to_last() noexcept
{
    return static_cast<status>(pmemkv_iterator_seek_to_last(this->get_raw_it()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::is_next() noexcept
{
    return static_cast<status>(pmemkv_iterator_is_next(this->get_raw_it()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::next() noexcept
{
    return static_cast<status>(pmemkv_iterator_next(this->get_raw_it()));
}
 
template <bool IsConst>
inline status db::iterator<IsConst>::prev() noexcept
{
    return static_cast<status>(pmemkv_iterator_prev(this->get_raw_it()));
}
 
template <bool IsConst>
inline result<string_view> db::iterator<IsConst>::key() noexcept
{
    const char *c;
    size_t size;
    auto s = static_cast<status>(pmemkv_iterator_key(this->get_raw_it(), &c, &size));
 
    if (s == status::OK)
        return {string_view{c, size}};
    else
        return {s};
}
 
template <bool IsConst>
inline result<string_view> db::iterator<IsConst>::read_range(size_t pos,
                                 size_t n) noexcept
{
    const char *data;
    size_t size;
    auto s = static_cast<status>(
        pmemkv_iterator_read_range(this->get_raw_it(), pos, n, &data, &size));
 
    if (s == status::OK)
        return {string_view{data, size}};
    else
        return {s};
}
 
template <>
template <>
inline result<pmem::obj::slice<db::iterator<false>::OutputIterator<char>>>
db::iterator<false>::write_range(size_t pos, size_t n) noexcept
{
    char *data;
    size_t size;
    auto s = static_cast<status>(
        pmemkv_write_iterator_write_range(this->it_.get(), pos, n, &data, &size));
 
    if (s == status::OK) {
        try {
            return {{data, data + size}};
        } catch (std::out_of_range &e) {
            return {status::INVALID_ARGUMENT};
        }
    } else
        return {s};
}
 
template <>
template <>
inline status db::iterator<false>::commit() noexcept
{
    auto s = static_cast<status>(pmemkv_write_iterator_commit(this->it_.get()));
    return s;
}
 
template <>
template <>
inline void db::iterator<false>::abort() noexcept
{
    pmemkv_write_iterator_abort(this->it_.get());
}
 
template <>
inline pmemkv_iterator *db::iterator<true>::get_raw_it()
{
    return it_.get();
}
 
template <>
inline pmemkv_iterator *db::iterator<false>::get_raw_it()
{
    return it_.get()->iter;
}
 
namespace internal
{
 
/*
 * Abstracts unique_ptr - exposes only void *get() method and a destructor.
 * This class is needed for C callbacks which cannot be templated
 * (type of object and deleter must be abstracted away).
 */
struct unique_ptr_wrapper_base {
    virtual ~unique_ptr_wrapper_base()
    {
    }
 
    virtual void *get() = 0;
};
 
template <typename T, typename D>
struct unique_ptr_wrapper : public unique_ptr_wrapper_base {
    unique_ptr_wrapper(std::unique_ptr<T, D> ptr) : ptr(std::move(ptr))
    {
    }
 
    void *get() override
    {
        return ptr.get();
    }
 
    std::unique_ptr<T, D> ptr;
};
 
class comparator_base {
public:
    virtual ~comparator_base()
    {
    }
    virtual int compare(string_view key1, string_view key2) = 0;
};
 
template <typename Comparator>
struct comparator_wrapper : public comparator_base {
    comparator_wrapper(const Comparator &cmp) : cmp(cmp)
    {
    }
 
    comparator_wrapper(Comparator &&cmp) : cmp(std::move(cmp))
    {
    }
 
    int compare(string_view key1, string_view key2) override
    {
        return cmp.compare(key1, key2);
    }
 
    Comparator cmp;
};
 
struct comparator_config_entry : public unique_ptr_wrapper_base {
    comparator_config_entry(
        std::unique_ptr<comparator_base> ptr,
        std::unique_ptr<pmemkv_comparator, decltype(pmemkv_comparator_delete) *>
            c_cmp)
        : ptr(std::move(ptr)), c_cmp(std::move(c_cmp))
    {
    }
 
    void *get() override
    {
        return c_cmp.get();
    }
 
    std::unique_ptr<comparator_base> ptr;
    std::unique_ptr<pmemkv_comparator, decltype(pmemkv_comparator_delete) *> c_cmp;
};
 
/*
 * All functions which will be called by C code must be declared as extern "C"
 * to ensure they have C linkage. It is needed because it is possible that
 * C and C++ functions use different calling conventions.
 */
extern "C" {
static inline void call_up_destructor(void *object)
{
    auto *ptr = static_cast<unique_ptr_wrapper_base *>(object);
    delete ptr;
}
 
static inline void *call_up_get(void *object)
{
    auto *ptr = static_cast<unique_ptr_wrapper_base *>(object);
    return ptr->get();
}
 
static inline int call_comparator_function(const char *k1, size_t kb1, const char *k2,
                       size_t kb2, void *arg)
{
    auto *cmp = static_cast<comparator_base *>(arg);
    return cmp->compare(string_view(k1, kb1), string_view(k2, kb2));
}
} /* extern "C" */
} /* namespace internal */
 
inline config::config() noexcept : config_(nullptr, &pmemkv_config_delete)
{
}
 
inline config::config(pmemkv_config *cfg) noexcept : config_(cfg, &pmemkv_config_delete)
{
}
 
inline int config::init() noexcept
{
    if (this->config_.get() == nullptr) {
        this->config_ = {pmemkv_config_new(), &pmemkv_config_delete};
 
        if (this->config_.get() == nullptr)
            return 1;
    }
 
    return 0;
}
 
template <typename T>
inline status config::put_data(const std::string &key, const T *value,
                   const std::size_t count) noexcept
{
    if (init() != 0)
        return status::UNKNOWN_ERROR;
 
    return static_cast<status>(pmemkv_config_put_data(
        this->config_.get(), key.data(), (void *)value, count * sizeof(T)));
}
 
template <typename T>
inline status config::put_object(const std::string &key, T *value,
                 void (*deleter)(void *)) noexcept
{
    if (init() != 0)
        return status::UNKNOWN_ERROR;
 
    return static_cast<status>(pmemkv_config_put_object(
        this->config_.get(), key.data(), (void *)value, deleter));
}
 
template <typename T, typename D>
inline status config::put_object(const std::string &key,
                 std::unique_ptr<T, D> object) noexcept
{
    if (init() != 0)
        return status::UNKNOWN_ERROR;
 
    internal::unique_ptr_wrapper_base *wrapper;
 
    try {
        wrapper = new internal::unique_ptr_wrapper<T, D>(std::move(object));
    } catch (std::bad_alloc &e) {
        return status::OUT_OF_MEMORY;
    } catch (...) {
        return status::UNKNOWN_ERROR;
    }
 
    return static_cast<status>(pmemkv_config_put_object_cb(
        this->config_.get(), key.data(), (void *)wrapper, internal::call_up_get,
        internal::call_up_destructor));
}
 
template <typename Comparator>
inline status config::put_comparator(Comparator &&comparator)
{
    static_assert(
        std::is_same<decltype(std::declval<Comparator>().compare(
                     std::declval<string_view>(),
                     std::declval<string_view>())),
                 int>::value,
        "Comparator should implement `int compare(pmem::kv::string_view, pmem::kv::string_view)` method");
    static_assert(std::is_convertible<decltype(std::declval<Comparator>().name()),
                      std::string>::value,
              "Comparator should implement `std::string name()` method");
 
    std::unique_ptr<internal::comparator_base> wrapper;
 
    try {
        wrapper = std::unique_ptr<internal::comparator_base>(
            new internal::comparator_wrapper<Comparator>(
                std::forward<Comparator>(comparator)));
    } catch (std::bad_alloc &e) {
        return status::OUT_OF_MEMORY;
    } catch (...) {
        return status::UNKNOWN_ERROR;
    }
 
    auto cmp =
        std::unique_ptr<pmemkv_comparator, decltype(pmemkv_comparator_delete) *>(
            pmemkv_comparator_new(&internal::call_comparator_function,
                          std::string(comparator.name()).c_str(),
                          wrapper.get()),
            &pmemkv_comparator_delete);
    if (cmp == nullptr)
        return status::UNKNOWN_ERROR;
 
    internal::unique_ptr_wrapper_base *entry;
 
    try {
        entry = new internal::comparator_config_entry(std::move(wrapper),
                                  std::move(cmp));
    } catch (std::bad_alloc &e) {
        return status::OUT_OF_MEMORY;
    } catch (...) {
        return status::UNKNOWN_ERROR;
    }
 
    return static_cast<status>(pmemkv_config_put_object_cb(
        this->config_.get(), "comparator", (void *)entry, internal::call_up_get,
        internal::call_up_destructor));
}
 
inline status config::put_uint64(const std::string &key, std::uint64_t value) noexcept
{
    if (init() != 0)
        return status::UNKNOWN_ERROR;
 
    return static_cast<status>(
        pmemkv_config_put_uint64(this->config_.get(), key.data(), value));
}
 
inline status config::put_int64(const std::string &key, std::int64_t value) noexcept
{
    if (init() != 0)
        return status::UNKNOWN_ERROR;
 
    return static_cast<status>(
        pmemkv_config_put_int64(this->config_.get(), key.data(), value));
}
 
inline status config::put_string(const std::string &key,
                 const std::string &value) noexcept
{
    if (init() != 0)
        return status::UNKNOWN_ERROR;
 
    return static_cast<status>(
        pmemkv_config_put_string(this->config_.get(), key.data(), value.data()));
}
 
inline status config::put_size(std::uint64_t size) noexcept
{
    return put_uint64("size", size);
}
 
inline status config::put_path(const std::string &path) noexcept
{
    return put_string("path", path);
}
 
inline status config::put_force_create(bool value) noexcept
{
    return put_uint64("force_create", value ? 1 : 0);
}
 
inline status config::put_oid(PMEMoid *oid) noexcept
{
    if (init() != 0)
        return status::UNKNOWN_ERROR;
 
    return static_cast<status>(pmemkv_config_put_oid(this->config_.get(), oid));
}
 
template <typename T>
inline status config::get_data(const std::string &key, T *&value,
                   std::size_t &count) const noexcept
{
    if (this->config_.get() == nullptr)
        return status::NOT_FOUND;
 
    std::size_t size;
    auto s = static_cast<status>(pmemkv_config_get_data(
        this->config_.get(), key.data(), (const void **)&value, &size));
 
    if (s != status::OK)
        return s;
 
    count = size / sizeof(T);
 
    return status::OK;
}
 
template <typename T>
inline status config::get_object(const std::string &key, T *&value) const noexcept
{
    if (this->config_.get() == nullptr)
        return status::NOT_FOUND;
 
    auto s = static_cast<status>(pmemkv_config_get_object(
        this->config_.get(), key.data(), (void **)&value));
 
    return s;
}
 
inline status config::get_uint64(const std::string &key, std::uint64_t &value) const
    noexcept
{
    if (this->config_.get() == nullptr)
        return status::NOT_FOUND;
 
    return static_cast<status>(
        pmemkv_config_get_uint64(this->config_.get(), key.data(), &value));
}
 
inline status config::get_int64(const std::string &key, std::int64_t &value) const
    noexcept
{
    if (this->config_.get() == nullptr)
        return status::NOT_FOUND;
 
    return static_cast<status>(
        pmemkv_config_get_int64(this->config_.get(), key.data(), &value));
}
 
inline status config::get_string(const std::string &key, std::string &value) const
    noexcept
{
    if (this->config_.get() == nullptr)
        return status::NOT_FOUND;
 
    const char *data;
 
    auto s = static_cast<status>(
        pmemkv_config_get_string(this->config_.get(), key.data(), &data));
 
    if (s != status::OK)
        return s;
 
    value = data;
 
    return status::OK;
}
 
inline pmemkv_config *config::release() noexcept
{
    return this->config_.release();
}
 
inline tx::tx(pmemkv_tx *tx_) noexcept : tx_(tx_, &pmemkv_tx_end)
{
}
 
inline status tx::remove(string_view key) noexcept
{
    return static_cast<status>(pmemkv_tx_remove(tx_.get(), key.data(), key.size()));
}
 
inline status tx::put(string_view key, string_view value) noexcept
{
    return static_cast<status>(pmemkv_tx_put(tx_.get(), key.data(), key.size(),
                         value.data(), value.size()));
}
 
inline status tx::commit() noexcept
{
    return static_cast<status>(pmemkv_tx_commit(tx_.get()));
}
 
inline void tx::abort() noexcept
{
    pmemkv_tx_abort(tx_.get());
}
 
/*
 * All functions which will be called by C code must be declared as extern "C"
 * to ensure they have C linkage. It is needed because it is possible that
 * C and C++ functions use different calling conventions.
 */
extern "C" {
static inline int call_get_kv_function(const char *key, size_t keybytes,
                       const char *value, size_t valuebytes, void *arg)
{
    return (*reinterpret_cast<std::function<get_kv_function> *>(arg))(
        string_view(key, keybytes), string_view(value, valuebytes));
}
 
static inline void call_get_v_function(const char *value, size_t valuebytes, void *arg)
{
    (*reinterpret_cast<std::function<get_v_function> *>(arg))(
        string_view(value, valuebytes));
}
 
static inline void call_get_copy(const char *v, size_t vb, void *arg)
{
    auto c = reinterpret_cast<std::string *>(arg);
    c->assign(v, vb);
}
}
 
inline db::db() noexcept : db_(nullptr, &pmemkv_close)
{
}
 
inline status db::open(const std::string &engine_name, config &&cfg) noexcept
{
    pmemkv_db *db;
    auto s =
        static_cast<status>(pmemkv_open(engine_name.c_str(), cfg.release(), &db));
    if (s == pmem::kv::status::OK)
        this->db_.reset(db);
    return s;
}
 
inline void db::close() noexcept
{
    this->db_.reset(nullptr);
}
 
inline status db::count_all(std::size_t &cnt) noexcept
{
    return static_cast<status>(pmemkv_count_all(this->db_.get(), &cnt));
}
 
inline status db::count_above(string_view key, std::size_t &cnt) noexcept
{
    return static_cast<status>(
        pmemkv_count_above(this->db_.get(), key.data(), key.size(), &cnt));
}
 
inline status db::count_equal_above(string_view key, std::size_t &cnt) noexcept
{
    return static_cast<status>(
        pmemkv_count_equal_above(this->db_.get(), key.data(), key.size(), &cnt));
}
 
inline status db::count_equal_below(string_view key, std::size_t &cnt) noexcept
{
    return static_cast<status>(
        pmemkv_count_equal_below(this->db_.get(), key.data(), key.size(), &cnt));
}
 
inline status db::count_below(string_view key, std::size_t &cnt) noexcept
{
    return static_cast<status>(
        pmemkv_count_below(this->db_.get(), key.data(), key.size(), &cnt));
}
 
inline status db::count_between(string_view key1, string_view key2,
                std::size_t &cnt) noexcept
{
    return static_cast<status>(pmemkv_count_between(this->db_.get(), key1.data(),
                            key1.size(), key2.data(),
                            key2.size(), &cnt));
}
 
inline status db::get_all(get_kv_callback *callback, void *arg) noexcept
{
    return static_cast<status>(pmemkv_get_all(this->db_.get(), callback, arg));
}
 
inline status db::get_all(std::function<get_kv_function> f) noexcept
{
    return static_cast<status>(
        pmemkv_get_all(this->db_.get(), call_get_kv_function, &f));
}
 
inline status db::get_above(string_view key, get_kv_callback *callback,
                void *arg) noexcept
{
    return static_cast<status>(
        pmemkv_get_above(this->db_.get(), key.data(), key.size(), callback, arg));
}
 
inline status db::get_above(string_view key, std::function<get_kv_function> f) noexcept
{
    return static_cast<status>(pmemkv_get_above(
        this->db_.get(), key.data(), key.size(), call_get_kv_function, &f));
}
 
inline status db::get_equal_above(string_view key, get_kv_callback *callback,
                  void *arg) noexcept
{
    return static_cast<status>(pmemkv_get_equal_above(this->db_.get(), key.data(),
                              key.size(), callback, arg));
}
 
inline status db::get_equal_above(string_view key,
                  std::function<get_kv_function> f) noexcept
{
    return static_cast<status>(pmemkv_get_equal_above(
        this->db_.get(), key.data(), key.size(), call_get_kv_function, &f));
}
 
inline status db::get_equal_below(string_view key, get_kv_callback *callback,
                  void *arg) noexcept
{
    return static_cast<status>(pmemkv_get_equal_below(this->db_.get(), key.data(),
                              key.size(), callback, arg));
}
 
inline status db::get_equal_below(string_view key,
                  std::function<get_kv_function> f) noexcept
{
    return static_cast<status>(pmemkv_get_equal_below(
        this->db_.get(), key.data(), key.size(), call_get_kv_function, &f));
}
 
inline status db::get_below(string_view key, get_kv_callback *callback,
                void *arg) noexcept
{
    return static_cast<status>(
        pmemkv_get_below(this->db_.get(), key.data(), key.size(), callback, arg));
}
 
inline status db::get_below(string_view key, std::function<get_kv_function> f) noexcept
{
    return static_cast<status>(pmemkv_get_below(
        this->db_.get(), key.data(), key.size(), call_get_kv_function, &f));
}
 
inline status db::get_between(string_view key1, string_view key2,
                  get_kv_callback *callback, void *arg) noexcept
{
    return static_cast<status>(pmemkv_get_between(this->db_.get(), key1.data(),
                              key1.size(), key2.data(),
                              key2.size(), callback, arg));
}
inline status db::get_between(string_view key1, string_view key2,
                  std::function<get_kv_function> f) noexcept
{
    return static_cast<status>(
        pmemkv_get_between(this->db_.get(), key1.data(), key1.size(), key2.data(),
                   key2.size(), call_get_kv_function, &f));
}
 
inline status db::exists(string_view key) noexcept
{
    return static_cast<status>(
        pmemkv_exists(this->db_.get(), key.data(), key.size()));
}
 
inline status db::get(string_view key, get_v_callback *callback, void *arg) noexcept
{
    return static_cast<status>(
        pmemkv_get(this->db_.get(), key.data(), key.size(), callback, arg));
}
 
inline status db::get(string_view key, std::function<get_v_function> f) noexcept
{
    return static_cast<status>(pmemkv_get(this->db_.get(), key.data(), key.size(),
                          call_get_v_function, &f));
}
 
inline status db::get(string_view key, std::string *value) noexcept
{
    return static_cast<status>(pmemkv_get(this->db_.get(), key.data(), key.size(),
                          call_get_copy, value));
}
 
inline status db::put(string_view key, string_view value) noexcept
{
    return static_cast<status>(pmemkv_put(this->db_.get(), key.data(), key.size(),
                          value.data(), value.size()));
}
 
inline status db::remove(string_view key) noexcept
{
    return static_cast<status>(
        pmemkv_remove(this->db_.get(), key.data(), key.size()));
}
 
inline status db::defrag(double start_percent, double amount_percent)
{
    return static_cast<status>(
        pmemkv_defrag(this->db_.get(), start_percent, amount_percent));
}
 
inline result<db::write_iterator> db::new_write_iterator()
{
    pmemkv_write_iterator *tmp;
    auto ret = static_cast<status>(pmemkv_write_iterator_new(db_.get(), &tmp));
    if (static_cast<status>(ret) == status::OK)
        return {db::iterator<false>{tmp}};
    else
        return {ret};
}
 
inline result<db::read_iterator> db::new_read_iterator()
{
    pmemkv_iterator *tmp;
    auto ret = static_cast<status>(pmemkv_iterator_new(db_.get(), &tmp));
    if (ret == status::OK)
        return {db::iterator<true>{tmp}};
    else
        return {ret};
}
 
inline std::string db::errormsg()
{
    return std::string(pmemkv_errormsg());
}
 
static inline std::string errormsg()
{
    return std::string(pmemkv_errormsg());
}
 
inline result<tx> db::tx_begin() noexcept
{
    pmemkv_tx *tx_;
    auto s = static_cast<status>(pmemkv_tx_begin(db_.get(), &tx_));
 
    if (s == status::OK)
        return result<tx>(tx(tx_));
    else
        return result<tx>(s);
}
 
} /* namespace kv */
} /* namespace pmem */
 
#endif /* LIBPMEMKV_HPP */