Persistent.hpp

Go to the documentation of this file.

#pragma once
#ifndef PERSISTENT_HPP
#define PERSISTENT_HPP

#include "detail/FilePersistLog.hpp"
#include "HLC.hpp"
#include "PersistException.hpp"
#include "detail/PersistLog.hpp"
#include "PersistNoLog.hpp"
#include "PersistentTypenames.hpp"
#include <derecho/mutils-serialization/SerializationSupport.hpp>
#include <functional>
#include <inttypes.h>
#include <iostream>
#include <map>
#include <memory>
#include <pthread.h>
#include <string>
#include <sys/types.h>
#include <time.h>
#include <typeindex>

#include <derecho/utils/logger.hpp>

#if defined(_PERFORMANCE_DEBUG) || defined(DERECHO_DEBUG) || !defined(NDEBUG)
  #include <derecho/utils/time.h>
#endif  //_PERFORMANCE_DEBUG

namespace persistent {

// #define DEFINE_PERSIST_VAR(_t,_n) DEFINE_PERSIST_VAR(_t,_n,ST_FILE)
#define DEFINE_PERSIST_VAR(_t, _n, _s) \
    Persistent<_t, _s> _n(#_n)
#define DECLARE_PERSIST_VAR(_t, _n, _s) \
    extern DEFINE_PERSIST_VAR(_t, _n, _s)

class ITemporalQueryFrontierProvider {
public:
    virtual const HLC getFrontier() = 0;
};

template <typename int_type>
version_t combine_int32s(const int_type high_bits, const int_type low_bits);

template <typename int_type>
std::pair<int_type, int_type> unpack_version(const version_t packed_int);

class PersistentRegistry : public mutils::RemoteDeserializationContext {
public:
    /* Constructor */
    PersistentRegistry(
        ITemporalQueryFrontierProvider* tqfp,
        const std::type_index& subgroup_type,
        uint32_t subgroup_index, uint32_t shard_num);

    /* Destructor */
    virtual ~PersistentRegistry();

    void makeVersion(const int64_t& ver, const HLC& mhlc) noexcept(false);

    const int64_t persist() noexcept(false);

    void trim(const int64_t& earliest_version) noexcept(false);

    const int64_t getMinimumLatestPersistedVersion() noexcept(false);

    static void setEarliestVersionToSerialize(const int64_t& ver) noexcept(true);

    static void resetEarliestVersionToSerialize() noexcept(true);

    static int64_t getEarliestVersionToSerialize() noexcept(true);

    void truncate(const int64_t& last_version);

    void registerPersist(const char* obj_name,
                         const VersionFunc& vf,
                         const PersistFunc& pf,
                         const TrimFunc& tf,
                         const LatestPersistedGetterFunc& lpgf,
                         const TruncateFunc& tcf) noexcept(false);

    void unregisterPersist(const char* obj_name) noexcept(false);

    inline const HLC getFrontier() {
        if(_temporal_query_frontier_provider != nullptr) {
#ifndef NDEBUG
            const HLC r = _temporal_query_frontier_provider->getFrontier();
            dbg_default_warn("temporal_query_frontier=HLC({},{})", r.m_rtc_us, r.m_logic);
            return r;
#else
            return _temporal_query_frontier_provider->getFrontier();
#endif  //NDEBUG
        } else {
            struct timespec t;
            clock_gettime(CLOCK_REALTIME, &t);
            return HLC((uint64_t)(t.tv_sec * 1e6 + t.tv_nsec / 1e3), (uint64_t)0);
        }
    }

    void updateTemporalFrontierProvider(ITemporalQueryFrontierProvider* tqfp);

    PersistentRegistry(PersistentRegistry&&) = default;

    PersistentRegistry(const PersistentRegistry&) = delete;

    const char* get_subgroup_prefix();

    static std::string generate_prefix(const std::type_index& subgroup_type, uint32_t subgroup_index, uint32_t shard_num) noexcept(true);

    static bool match_prefix(const std::string str, const std::type_index& subgroup_type, uint32_t subgroup_index, uint32_t shard_num) noexcept(true);

protected:
    const std::string _subgroup_prefix;  

    ITemporalQueryFrontierProvider* _temporal_query_frontier_provider;

    std::map<std::size_t, std::tuple<VersionFunc, PersistFunc, TrimFunc, LatestPersistedGetterFunc, TruncateFunc>> _registry;

    template <int funcIdx, typename... Args>
    void callFunc(Args... args);
    
    template <int funcIdx, typename ReturnType, typename... Args>
    ReturnType callFuncMin(Args... args);

    static thread_local int64_t earliest_version_to_serialize;
};

// If the type T in persistent<T> is a big object and the operations are small
// updates, for example, an object store or a file system, then T would better
// implement the IDeltaSupport interface. This interface allows persistent<T>
// only store the delta in the log, avoiding huge duplicated data wasting
// storage space as well as I/O bandwidth.
//
// The idea is that T is responsible of keeping track of the updates in the form
// of a byte array - the DELTA, as long as the update should be persisted. Each
// time Persistent<T> trying to make a version, it collects the DELTA and write
// it to the log. On reloading data from persistent storage, the DELTAs in the
// log entries are applied in order. TODO: use checkpointing to accelerate it!
//
// There are three method included in this interface:
// - 'finalizeCurrentDelta'     This method is called when Persistent<T> trying to
//   make a version. Once done, the delta needs to be cleared.
// - 'applyDelta' This method is called on object construction from the disk
// - 'create' This static method is used to create an empty object from deserialization
//   manager.
using DeltaFinalizer = std::function<void(char const* const, std::size_t)>;

template <typename DeltaObjectType>
class IDeltaObjectFactory {
public:
    static std::unique_ptr<DeltaObjectType> create(mutils::DeserializationManager* dm) {
        return DeltaObjectType::create(dm);
    }
};

template <typename ObjectType>
class IDeltaSupport : public IDeltaObjectFactory<ObjectType> {
public:
    virtual void finalizeCurrentDelta(const DeltaFinalizer&) = 0;
    virtual void applyDelta(char const* const) = 0;
};


// _NameMaker is a tool makeing the name for the log corresponding to a
// given Persistent<ObjectType> object.
template <typename ObjectType, StorageType storageType>
class _NameMaker {
public:
    // Constructor
    _NameMaker() noexcept(false);

    // Destructor
    virtual ~_NameMaker() noexcept(true);

    // guess a name
    std::unique_ptr<std::string> make(const char* prefix) noexcept(false);

private:
    int m_iCounter;
    const char* m_sObjectTypeName;
    pthread_spinlock_t m_oLck;
};


// Persistent represents a variable backed up by persistent storage. The
// backend is PersistLog class. PersistLog handles only raw bytes and this
// class is repsonsible for converting it back and forth between raw bytes
// and ObjectType. But, the serialization/deserialization functionality is
// actually defined by ObjectType and provided by Persistent users.
// - ObjectType: user-defined type of the variable it is required to support
//   serialization and deserialization as follows:
//   // serialize
//   void * ObjectType::serialize(const ObjectType & obj, uint64_t *psize)
//   - obj: obj is the reference to the object to be serialized
//   - psize: psize is a uint64_t pointer to receive the size of the serialized
//     data.
//   - Return value is a pointer to a new malloced buffer with the serialized
//     //TODO: this may not be efficient for large object...open to be changed.
//   // deserialize
//   ObjectType * ObjectType::deserialize(const void *pdata)
//   - pdata: a buffer of the serialized data
//   - Return value is a pointer to a new created ObjectType deserialized from
//     'pdata' buffer.
// - StorageType: storage type is defined in PersistLog. The value could be
//   ST_FILE/ST_MEM/ST_3DXP ... I will start with ST_FILE and extend it to
//   other persistent Storage.
// TODO:comments
//TODO: Persistent<T> has to be serializable, extending from mutils::ByteRepresentable
template <typename ObjectType,
          StorageType storageType = ST_FILE>
class Persistent : public mutils::ByteRepresentable {
protected:

    inline void initialize_log(const char* object_name) noexcept(false);

    inline void initialize_object_from_log(const std::function<std::unique_ptr<ObjectType>(void)>& object_factory,
                                           mutils::DeserializationManager* dm);

    inline void register_callbacks() noexcept(false);

    inline void unregister_callbacks() noexcept(false);

public:
    Persistent(
        const std::function<std::unique_ptr<ObjectType>(void)>& object_factory,
        const char* object_name = nullptr,
        PersistentRegistry* persistent_registry = nullptr,
        mutils::DeserializationManager dm = {{}}) noexcept(false);

    Persistent(Persistent&& other) noexcept(false);

    Persistent(
        const char* object_name,
        std::unique_ptr<ObjectType>& wrapped_obj_ptr,
        const char* log_tail = nullptr,
        PersistentRegistry* persistent_registry = nullptr,
        mutils::DeserializationManager dm = {{}}) noexcept(false);

    Persistent(const Persistent&) = delete;

    virtual ~Persistent() noexcept(true);

    ObjectType& operator*();

    ObjectType* operator->();

    const ObjectType& getConstRef() const;

    const std::string& getObjectName();

    template <typename Func>
    auto get(
        const Func& fun,
        mutils::DeserializationManager* dm = nullptr) noexcept(false);

    std::unique_ptr<ObjectType> get(mutils::DeserializationManager* dm = nullptr);

    template <typename Func>
    auto getByIndex(
        int64_t idx,
        const Func& fun,
        mutils::DeserializationManager* dm = nullptr) noexcept(false);

    std::unique_ptr<ObjectType> getByIndex(
            int64_t idx,
            mutils::DeserializationManager* dm = nullptr) noexcept(false);

    template <typename Func>
    auto get(
        const int64_t& ver,
        const Func& fun,
        mutils::DeserializationManager* dm = nullptr) noexcept(false);

    std::unique_ptr<ObjectType> get(
        const int64_t& ver,
        mutils::DeserializationManager* dm = nullptr) noexcept(false);

    template <typename TKey>
    void trim(const TKey& k) noexcept(false);

    void truncate(const int64_t& ver);

    template <typename Func>
    auto get(
        const HLC& hlc,
        const Func& fun,
        mutils::DeserializationManager* dm = nullptr) noexcept(false);

    std::unique_ptr<ObjectType> get(
            const HLC& hlc,
            mutils::DeserializationManager* dm = nullptr) noexcept(false);

    std::unique_ptr<ObjectType> operator[](const int64_t ver) noexcept(false) {
        return this->get(ver);
    }

    std::unique_ptr<ObjectType> operator[](const HLC& hlc) noexcept(false) {
        return this->get(hlc);
    }

    virtual int64_t getNumOfVersions() noexcept(false);

    virtual int64_t getEarliestIndex() noexcept(false);

    virtual int64_t getEarliestVersion() noexcept(false);

    virtual int64_t getLatestIndex() noexcept(false);

    virtual int64_t getLatestVersion() noexcept(false);

    virtual const int64_t getLastPersisted() noexcept(false);

    virtual void set(ObjectType& v, const version_t& ver, const HLC& mhlc) noexcept(false);

    virtual void set(ObjectType& v, const version_t& ver) noexcept(false);

    virtual void version(const version_t& ver) noexcept(false);

    virtual const int64_t persist() noexcept(false);

public:
    // wrapped objected
    std::unique_ptr<ObjectType> m_pWrappedObject;

protected:
    // PersistLog
    std::unique_ptr<PersistLog> m_pLog;
    // Persistence Registry
    PersistentRegistry* m_pRegistry;
    // get the static name maker.
    static _NameMaker<ObjectType,storageType>& getNameMaker(const std::string& prefix = std::string(""));

    //serialization supports
public:
    // Serialization and Deserialization of Persistent<T>
    // Serialization of the persistent<T> is packed in the following order
    // 1) the log name
    // 2) current state of the object
    // 3) number of log entries
    // 4) the log entries from the earliest to the latest
    // TODO.
    //Note: this rely on PersistentRegistry::earliest_version_to_serialize
    std::size_t to_bytes(char* ret) const;
    std::size_t bytes_size() const;
    void post_object(const std::function<void(char const* const, std::size_t)>& f) const;
    // NOTE: we do not set up the registry here. This will only happen in the
    // construction of Replicated<T>
    static std::unique_ptr<Persistent> from_bytes(mutils::DeserializationManager* dsm, char const* v);
    // derived from ByteRepresentable
    virtual void ensure_registered(mutils::DeserializationManager&) {}
    // apply the serialized log tail to existing log
    // @dsm - deserialization manager
    // @v - bytes representation of the log tail)
    void applyLogTail(mutils::DeserializationManager* dsm, char const* v);

#if defined(_PERFORMANCE_DEBUG) || !defined(NDEBUG)
    uint64_t ns_in_persist = 0ul;
    uint64_t ns_in_set = 0ul;
    uint64_t cnt_in_persist = 0ul;
    uint64_t cnt_in_set = 0ul;
    virtual void print_performance_stat();
#endif  //_PERFORMANCE_DEBUG
};

template <typename ObjectType>
class Volatile : public Persistent<ObjectType, ST_MEM> {
public:
    Volatile(
            const std::function<std::unique_ptr<ObjectType>(void)>& object_factory,
            const char* object_name = nullptr,
            PersistentRegistry* persistent_registry = nullptr,
            mutils::DeserializationManager dm = {{}}) noexcept(false)
            : Persistent<ObjectType, ST_MEM>(object_factory, object_name, persistent_registry) {}

    Volatile(Volatile&& other) noexcept(false)
            : Persistent<ObjectType, ST_MEM>(other) {}

    Volatile(
            const std::function<std::unique_ptr<ObjectType>(void)>& object_factory,
            const char* object_name,
            std::unique_ptr<ObjectType>& wrapped_obj_ptr,
            std::unique_ptr<PersistLog>& log_ptr,
            PersistentRegistry* persistent_registry = nullptr,
            mutils::DeserializationManager dm = {{}}) noexcept(false)
            : Persistent<ObjectType, ST_MEM>(object_factory, object_name, wrapped_obj_ptr, log_ptr, persistent_registry) {}

    Volatile(const Volatile&) = delete;

    // destructor:
    virtual ~Volatile() noexcept(true){
            // do nothing
    };
};

/* Utilities for manage a single "ByteRepresentable" persistent object. */
template <typename ObjectType, StorageType storageType = ST_FILE>
void saveObject(ObjectType& obj, const char* object_name = nullptr) noexcept(false);

template <typename ObjectType, StorageType storageType = ST_FILE>
std::unique_ptr<ObjectType> loadObject(const char* object_name = nullptr) noexcept(false);

template <StorageType storageType = ST_FILE>
const typename std::enable_if<(storageType == ST_FILE || storageType == ST_MEM), version_t>::type getMinimumLatestPersistedVersion(const std::type_index& subgroup_type, uint32_t subgroup_index, uint32_t shard_num);

}  // namespace persistent

#include "detail/Persistent_impl.hpp"

#endif  //PERSISTENT_HPP