libpmemkv.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2017-2020, Intel Corporation */
 
#ifndef LIBPMEMKV_HPP
#define LIBPMEMKV_HPP
 
#include <functional>
#include <iostream>
#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);
 
typedef int comparator_function(string_view key1, string_view key2);
 
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 config {
public:
    config() noexcept;
    explicit config(pmemkv_config *cfg) noexcept;
 
    ~config();
 
    config(const config &other) = delete;
    config(config &&other) noexcept;
 
    config &operator=(const config &other) = delete;
    config &operator=(config &&other) 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;
 
    pmemkv_config *_config;
};
 
class db {
public:
    db() noexcept;
    ~db();
 
    db(const db &other) = delete;
    db(db &&other) noexcept;
 
    db &operator=(const db &other) = delete;
    db &operator=(db &&other) noexcept;
 
    status open(const std::string &engine_name) noexcept;
    status open(const std::string &engine_name, config &&cfg) 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);
 
    std::string errormsg();
 
private:
    pmemkv_db *_db;
};
 
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
{
    this->_config = nullptr;
}
 
inline config::config(config &&other) noexcept
{
    this->_config = other._config;
    other._config = nullptr;
}
 
inline config &config::operator=(config &&other) noexcept
{
    if (this == &other)
        return *this;
 
    if (this->_config)
        pmemkv_config_delete(this->_config);
 
    this->_config = other._config;
    other._config = nullptr;
 
    return *this;
}
 
inline config::config(pmemkv_config *cfg) noexcept
{
    this->_config = cfg;
}
 
inline config::~config()
{
    if (this->_config)
        pmemkv_config_delete(this->_config);
}
 
inline int config::init() noexcept
{
    if (this->_config == nullptr) {
        this->_config = pmemkv_config_new();
 
        if (this->_config == 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, 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, 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, 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, "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, 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, 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, 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, oid));
}
 
template <typename T>
inline status config::get_data(const std::string &key, T *&value,
                   std::size_t &count) const noexcept
{
    if (this->_config == nullptr)
        return status::NOT_FOUND;
 
    std::size_t size;
    auto s = static_cast<status>(pmemkv_config_get_data(
        this->_config, 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 == nullptr)
        return status::NOT_FOUND;
 
    auto s = static_cast<status>(
        pmemkv_config_get_object(this->_config, key.data(), (void **)&value));
 
    return s;
}
 
inline status config::get_uint64(const std::string &key, std::uint64_t &value) const
    noexcept
{
    if (this->_config == nullptr)
        return status::NOT_FOUND;
 
    return static_cast<status>(
        pmemkv_config_get_uint64(this->_config, key.data(), &value));
}
 
inline status config::get_int64(const std::string &key, std::int64_t &value) const
    noexcept
{
    if (this->_config == nullptr)
        return status::NOT_FOUND;
 
    return static_cast<status>(
        pmemkv_config_get_int64(this->_config, key.data(), &value));
}
 
inline status config::get_string(const std::string &key, std::string &value) const
    noexcept
{
    if (this->_config == nullptr)
        return status::NOT_FOUND;
 
    const char *data;
 
    auto s = static_cast<status>(
        pmemkv_config_get_string(this->_config, key.data(), &data));
 
    if (s != status::OK)
        return s;
 
    value = data;
 
    return status::OK;
}
 
inline pmemkv_config *config::release() noexcept
{
    auto c = this->_config;
    this->_config = nullptr;
    return c;
}
 
/*
 * 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
{
    this->_db = nullptr;
}
 
inline db::db(db &&other) noexcept
{
    this->_db = other._db;
    other._db = nullptr;
}
 
inline db &db::operator=(db &&other) noexcept
{
    if (this == &other)
        return *this;
 
    close();
 
    std::swap(this->_db, other._db);
 
    return *this;
}
 
inline status db::open(const std::string &engine_name) noexcept
{
    return static_cast<status>(
        pmemkv_open(engine_name.c_str(), nullptr, &(this->_db)));
}
 
inline status db::open(const std::string &engine_name, config &&cfg) noexcept
{
    return static_cast<status>(
        pmemkv_open(engine_name.c_str(), cfg.release(), &(this->_db)));
}
 
inline void db::close() noexcept
{
    if (this->_db != nullptr)
        pmemkv_close(this->_db);
 
    this->_db = nullptr;
}
 
inline db::~db()
{
    close();
}
 
inline status db::count_all(std::size_t &cnt) noexcept
{
    return static_cast<status>(pmemkv_count_all(this->_db, &cnt));
}
 
inline status db::count_above(string_view key, std::size_t &cnt) noexcept
{
    return static_cast<status>(
        pmemkv_count_above(this->_db, 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, 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, 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, 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, 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, callback, arg));
}
 
inline status db::get_all(std::function<get_kv_function> f) noexcept
{
    return static_cast<status>(pmemkv_get_all(this->_db, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, key.data(), key.size(),
                          value.data(), value.size()));
}
 
inline status db::remove(string_view key) noexcept
{
    return static_cast<status>(pmemkv_remove(this->_db, key.data(), key.size()));
}
 
inline status db::defrag(double start_percent, double amount_percent)
 
{
    return static_cast<status>(
        pmemkv_defrag(this->_db, start_percent, amount_percent));
}
 
inline std::string db::errormsg()
{
    return std::string(pmemkv_errormsg());
}
 
static inline std::string errormsg()
{
    return std::string(pmemkv_errormsg());
}
 
} /* namespace kv */
} /* namespace pmem */
 
#endif /* LIBPMEMKV_HPP */