replicated.hpp

Go to the documentation of this file.

#pragma once

#include <functional>
#include <memory>
#include <tuple>
#include <type_traits>

#include "derecho_exception.hpp"
#include "detail/derecho_internal.hpp"
#include "detail/remote_invocable.hpp"
#include "detail/replicated_interface.hpp"
#include "detail/rpc_manager.hpp"
#include "detail/rpc_utils.hpp"

#include <derecho/persistent/Persistent.hpp>
#include <derecho/tcp/tcp.hpp>

namespace derecho {

class _Group;
class GroupReference;

class PersistsFields {};

template <typename T>
using has_persistent_fields = std::is_base_of<PersistsFields, T>;

struct RawObject {
    static auto register_functions() { return std::tuple<>(); };
};

inline std::unique_ptr<RawObject> raw_object_factory(persistent::PersistentRegistry*) {
    return std::make_unique<RawObject>();
}

template <typename T>
class Replicated : public ReplicatedObject, public persistent::ITemporalQueryFrontierProvider {
private:
    std::unique_ptr<persistent::PersistentRegistry> persistent_registry_ptr;
#if defined(_PERFORMANCE_DEBUG) || !defined(NDEBUG)
public:
#endif

    std::unique_ptr<std::unique_ptr<T>> user_object_ptr;
#if defined(_PERFORMANCE_DEBUG) || !defined(NDEBUG)
private:
#endif

    const node_id_t node_id;
    const subgroup_id_t subgroup_id;
    const uint32_t subgroup_index;
    const uint32_t shard_num;
    rpc::RPCManager& group_rpc_manager;
    std::unique_ptr<rpc::RemoteInvocableOf<T>> wrapped_this;
    _Group* group;
    persistent::version_t next_version = INVALID_VERSION;
    uint64_t next_timestamp_us = 0;

public:
    Replicated(subgroup_type_id_t type_id, node_id_t nid, subgroup_id_t subgroup_id, uint32_t subgroup_index,
               uint32_t shard_num, rpc::RPCManager& group_rpc_manager,
               Factory<T> client_object_factory, _Group* group);

    Replicated(subgroup_type_id_t type_id, node_id_t nid, subgroup_id_t subgroup_id, uint32_t subgroup_index,
               uint32_t shard_num, rpc::RPCManager& group_rpc_manager, _Group* group);

    Replicated(Replicated&& rhs);
    Replicated(const Replicated&) = delete;
    virtual ~Replicated();

    virtual bool is_persistent() const {
        return has_persistent_fields<T>::value;
    }

    bool is_valid() const {
        return *user_object_ptr && true;
    }

    uint32_t get_shard_num() const {
        return shard_num;
    }

    template <rpc::FunctionTag tag, typename... Args>
    auto p2p_send(node_id_t dest_node, Args&&... args);

    template <rpc::FunctionTag tag, typename... Args>
    auto ordered_send(Args&&... args);

    void send(unsigned long long int payload_size, const std::function<void(char* buf)>& msg_generator);

    std::size_t object_size() const;

    void send_object(tcp::socket& receiver_socket) const;

    void send_object_raw(tcp::socket& receiver_socket) const;

    std::size_t receive_object(char* buffer);

    const uint64_t compute_global_stability_frontier();

    inline const HLC getFrontier() {
        // transform from ns to us:
        HLC hlc(this->compute_global_stability_frontier() / 1e3, 0);
        return hlc;
    }

    const persistent::version_t get_minimum_latest_persisted_version() noexcept(false);

    virtual void make_version(const persistent::version_t& ver, const HLC& hlc) noexcept(false) {
        persistent_registry_ptr->makeVersion(ver, hlc);
    };

    virtual void persist(const persistent::version_t version) noexcept(false);

    virtual void trim(const persistent::version_t& earliest_version) noexcept(false) {
        persistent_registry_ptr->trim(earliest_version);
    };

    virtual void truncate(const persistent::version_t& latest_version) {
        persistent_registry_ptr->truncate(latest_version);
    }

    virtual void post_next_version(const persistent::version_t& version, const uint64_t & ts_us) {
        next_version = version;
        next_timestamp_us = ts_us;
    }

    virtual std::tuple<persistent::version_t,uint64_t> get_next_version() {
        return std::tie(next_version,next_timestamp_us);
    }

    virtual void register_persistent_member(const char* object_name,
                                            const persistent::VersionFunc& vf,
                                            const persistent::PersistFunc& pf,
                                            const persistent::TrimFunc& tf,
                                            const persistent::LatestPersistedGetterFunc& gf,
                                            persistent::TruncateFunc tcf) noexcept(false) {
        this->persistent_registry_ptr->registerPersist(object_name, vf, pf, tf, gf, tcf);
    }
};

template <typename T>
class ExternalCaller {
private:
    const node_id_t node_id;
    subgroup_id_t subgroup_id;
    rpc::RPCManager& group_rpc_manager;
    std::unique_ptr<rpc::RemoteInvokerFor<T>> wrapped_this;

public:
    ExternalCaller(uint32_t type_id, node_id_t nid, subgroup_id_t subgroup_id, rpc::RPCManager& group_rpc_manager);

    ExternalCaller(ExternalCaller&&) = default;
    ExternalCaller(const ExternalCaller&) = delete;

    template <rpc::FunctionTag tag, typename... Args>
    auto p2p_send(node_id_t dest_node, Args&&... args);

    bool is_valid() const { return true; }
};

template <typename T>
class ShardIterator {
private:
    ExternalCaller<T>& EC;
    const std::vector<node_id_t> shard_reps;

public:
    ShardIterator(ExternalCaller<T>& EC, std::vector<node_id_t> shard_reps)
            : EC(EC),
              shard_reps(shard_reps) {}
    template <rpc::FunctionTag tag, typename... Args>
    auto p2p_send(Args&&... args);
};
}  // namespace derecho

#include "detail/replicated_impl.hpp"