view_manager.hpp

Go to the documentation of this file.

#pragma once

#include <map>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <thread>
#include <vector>

#include "../subgroup_info.hpp"
#include "../view.hpp"
#include "derecho_internal.hpp"
#include "locked_reference.hpp"
#include "multicast_group.hpp"
#include "restart_state.hpp"
#include <derecho/conf/conf.hpp>

#include <derecho/mutils-serialization/SerializationSupport.hpp>
#include <spdlog/spdlog.h>

namespace derecho {

template <typename T>
class Replicated;
template <typename T>
class ExternalCaller;

class ReplicatedObject;

namespace rpc {
class RPCManager;
}

template <typename T>
class LockedQueue {
private:
    using unique_lock_t = std::unique_lock<std::mutex>;
    std::mutex mutex;
    std::list<T> underlying_list;

public:
    struct LockedListAccess {
    private:
        unique_lock_t lock;

    public:
        std::list<T>& access;
        LockedListAccess(std::mutex& m, std::list<T>& a) : lock(m), access(a){};
    };
    LockedListAccess locked() {
        return LockedListAccess{mutex, underlying_list};
    }
};

enum class JoinResponseCode {
    OK,              
    TOTAL_RESTART,   
    ID_IN_USE,       
    LEADER_REDIRECT  
};

struct JoinResponse {
    JoinResponseCode code;
    node_id_t leader_id;
};

template <typename T>
using SharedLockedReference = LockedReference<std::shared_lock<std::shared_timed_mutex>, T>;

using view_upcall_t = std::function<void(const View&)>;

using vector_int64_2d = std::vector<std::vector<int64_t>>;

class ViewManager {
private:
    using pred_handle = sst::Predicates<DerechoSST>::pred_handle;

    using initialize_rpc_objects_t = std::function<void(node_id_t, const View&, const std::vector<std::vector<int64_t>>&)>;

    //Allow RPCManager and Replicated to access curr_view and view_mutex directly
    friend class rpc::RPCManager;
    template <typename T>
    friend class Replicated;
    template <typename T>
    friend class ExternalCaller;

    friend class PersistenceManager;

    friend class RestartLeaderState;

    std::shared_timed_mutex view_mutex;
    std::condition_variable_any view_change_cv;

    std::unique_ptr<View> curr_view;
    std::unique_ptr<View> next_view;

    LockedQueue<tcp::socket> pending_join_sockets;

    std::queue<std::unique_ptr<View>> old_views;
    std::mutex old_views_mutex;
    std::condition_variable old_views_cv;

    std::list<tcp::socket> proposed_join_sockets;
    std::vector<bool> last_suspected;

    tcp::connection_listener server_socket;
    std::atomic<bool> thread_shutdown;
    std::thread client_listener_thread;
    std::thread old_view_cleanup_thread;

    const bool disable_partitioning_safety;

    //Handles for all the predicates the GMS registered with the current view's SST.
    pred_handle suspected_changed_handle;
    pred_handle start_join_handle;
    pred_handle reject_join_handle;
    pred_handle change_commit_ready_handle;
    pred_handle leader_proposed_handle;
    pred_handle leader_committed_handle;

    std::vector<view_upcall_t> view_upcalls;
    const SubgroupInfo subgroup_info;
    std::vector<std::type_index> subgroup_type_order;

    std::shared_ptr<tcp::tcp_connections> tcp_sockets;

    using ReplicatedObjectReferenceMap = std::map<subgroup_id_t, std::reference_wrapper<ReplicatedObject>>;
    ReplicatedObjectReferenceMap& subgroup_objects;
    initialize_rpc_objects_t initialize_subgroup_objects;
    const bool any_persistent_objects;

    bool in_total_restart;

    std::unique_ptr<RestartLeaderState> restart_leader_state_machine;

    std::unique_ptr<RestartState> restart_state;

    bool active_leader;

    persistence_manager_callbacks_t persistence_manager_callbacks;

    std::vector<std::vector<int64_t>> prior_view_shard_leaders;

    std::atomic<bool> bSilent = false;

    bool has_pending_join() { return pending_join_sockets.locked().access.size() > 0; }

    /* ---------------------------- View-management triggers ---------------------------- */
    void new_suspicion(DerechoSST& gmsSST);
    void leader_start_join(DerechoSST& gmsSST);
    void redirect_join_attempt(DerechoSST& gmsSST);
    void new_leader_takeover(DerechoSST& gmsSST);
    void leader_commit_change(DerechoSST& gmsSST);
    void acknowledge_proposed_change(DerechoSST& gmsSST);
    void start_meta_wedge(DerechoSST& gmsSST);
    void terminate_epoch(DerechoSST& gmsSST);
    void echo_ragged_trim(std::shared_ptr<std::map<subgroup_id_t, uint32_t>> follower_subgroups_and_shards,
                          DerechoSST& gmsSST);
    void deliver_ragged_trim(DerechoSST& gmsSST);
    void finish_view_change(DerechoSST& gmsSST);

    /* ---------------------------------------------------------------------------------- */
    /* ------------------- Helper methods for view-management triggers ------------------ */

    bool receive_join(DerechoSST& gmsSST, tcp::socket& client_socket);
    void update_tcp_connections(const View& new_view);

    void send_objects_to_new_members(const View& new_view, const vector_int64_2d& old_shard_leaders);

    void send_subgroup_object(subgroup_id_t subgroup_id, node_id_t new_node_id);

    void send_view(const View& new_view, tcp::socket& client_socket);

    void log_ragged_trim(const int shard_leader_rank,
                         const subgroup_id_t subgroup_num,
                         const uint32_t num_received_offset,
                         const uint num_shard_senders);
    void deliver_in_order(const int shard_leader_rank,
                          const subgroup_id_t subgroup_num, const uint32_t num_received_offset,
                          const std::vector<node_id_t>& shard_members, uint num_shard_senders);
    void leader_ragged_edge_cleanup(const subgroup_id_t subgroup_num,
                                    const uint32_t num_received_offset,
                                    const std::vector<node_id_t>& shard_members,
                                    uint num_shard_senders);
    void follower_ragged_edge_cleanup(const subgroup_id_t subgroup_num,
                                      uint shard_leader_rank,
                                      const uint32_t num_received_offset,
                                      uint num_shard_senders);

    /* -- Static helper methods that implement chunks of view-management functionality -- */
    static bool suspected_not_equal(const DerechoSST& gmsSST, const std::vector<bool>& old);
    static void copy_suspected(const DerechoSST& gmsSST, std::vector<bool>& old);
    static bool changes_contains(const DerechoSST& gmsSST, const node_id_t q);
    static int min_acked(const DerechoSST& gmsSST, const std::vector<char>& failed);
    static bool previous_leaders_suspected(const DerechoSST& gmsSST, const View& curr_view);

    static bool copy_prior_leader_proposals(DerechoSST& gmsSST);

    static std::unique_ptr<View> make_next_view(const std::unique_ptr<View>& curr_view,
                                                const DerechoSST& gmsSST);

    /* ---------------------------------------------------------------------------------- */

    /* ------------------------ Setup/constructor helpers ------------------------------- */

    void create_threads();
    void register_predicates();
    void load_ragged_trim();
    void await_first_view(const node_id_t my_id);

    void receive_view_and_leaders(const node_id_t my_id, tcp::socket& leader_connection);

    void truncate_persistent_logs(const ragged_trim_map_t& logged_ragged_trim);

    void initialize_rdmc_sst();
    void receive_initial_view(node_id_t my_id, tcp::socket& leader_connection);

    void setup_initial_tcp_connections(const View& initial_view, node_id_t my_id);

    void reinit_tcp_connections(const View& initial_view, node_id_t my_id);
    void construct_multicast_group(CallbackSet callbacks,
                                   const std::map<subgroup_id_t, SubgroupSettings>& subgroup_settings,
                                   const uint32_t num_received_size,
                                   const uint32_t slot_size);

    void transition_multicast_group(const std::map<subgroup_id_t, SubgroupSettings>& new_subgroup_settings,
                                    const uint32_t new_num_received_size,
                                    const uint32_t new_slot_size);
    static void make_subgroup_maps(const SubgroupInfo& subgroup_info,
                                   const std::unique_ptr<View>& prev_view,
                                   View& curr_view);

    std::pair<uint32_t, uint32_t> derive_subgroup_settings(View& curr_view,
                                                           std::map<subgroup_id_t, SubgroupSettings>& subgroup_settings);

    static uint32_t compute_num_received_size(const View& view);

    template <PORT_TYPE port_index>
    static std::map<node_id_t, std::pair<ip_addr_t, uint16_t>>
    make_member_ips_and_ports_map(const View& view) {
        std::map<node_id_t, std::pair<ip_addr_t, uint16_t>> member_ips_and_ports_map;
        size_t num_members = view.members.size();
        for(uint i = 0; i < num_members; ++i) {
            if(!view.failed[i]) {
                member_ips_and_ports_map[view.members[i]] = std::pair<ip_addr_t, uint16_t>{
                        std::get<0>(view.member_ips_and_ports[i]),
                        std::get<port_index>(view.member_ips_and_ports[i])};
            }
        }
        return member_ips_and_ports_map;
    }
    static vector_int64_2d old_shard_leaders_by_new_ids(const View& curr_view, const View& next_view);

    template <typename ValueType>
    static std::unique_ptr<std::vector<std::vector<ValueType>>> receive_vector2d(tcp::socket& socket) {
        std::size_t buffer_size;
        socket.read(buffer_size);
        if(buffer_size == 0) {
            return std::make_unique<std::vector<std::vector<ValueType>>>();
        }
        char buffer[buffer_size];
        socket.read(buffer, buffer_size);
        return mutils::from_bytes<std::vector<std::vector<ValueType>>>(nullptr, buffer);
    }

public:
    ViewManager(const SubgroupInfo& subgroup_info,
                const std::vector<std::type_index>& subgroup_type_order,
                const bool any_persistent_objects,
                const std::shared_ptr<tcp::tcp_connections>& group_tcp_sockets,
                ReplicatedObjectReferenceMap& object_reference_map,
                const persistence_manager_callbacks_t& _persistence_manager_callbacks,
                std::vector<view_upcall_t> _view_upcalls = {});

    ViewManager(tcp::socket& leader_connection,
                const SubgroupInfo& subgroup_info,
                const std::vector<std::type_index>& subgroup_type_order,
                const bool any_persistent_objects,
                const std::shared_ptr<tcp::tcp_connections>& group_tcp_sockets,
                ReplicatedObjectReferenceMap& object_reference_map,
                const persistence_manager_callbacks_t& _persistence_manager_callbacks,
                std::vector<view_upcall_t> _view_upcalls = {});

    ~ViewManager();

    void await_rejoining_nodes(const node_id_t my_id);

    bool check_view_committed(tcp::socket& leader_connection);

    bool leader_prepare_initial_view(bool& leader_has_quorum);

    void leader_commit_initial_view();

    void truncate_logs();

    void send_logs();

    void initialize_multicast_groups(CallbackSet callbacks);

    void finish_setup();

    void start();

    const vector_int64_2d& get_old_shard_leaders() const { return prior_view_shard_leaders; }

    void leave();
    std::vector<node_id_t> get_members();
    int32_t get_my_rank();
    std::vector<std::vector<node_id_t>> get_subgroup_members(subgroup_type_id_t subgroup_type, uint32_t subgroup_index);
    std::size_t get_number_of_shards_in_subgroup(subgroup_type_id_t subgroup_type, uint32_t subgroup_index);
    int32_t get_my_shard(subgroup_type_id_t subgroup_type, uint32_t subgroup_index);
    void send(subgroup_id_t subgroup_num, long long unsigned int payload_size,
              const std::function<void(char* buf)>& msg_generator, bool cooked_send = false);

    const uint64_t compute_global_stability_frontier(subgroup_id_t subgroup_num);

    SharedLockedReference<View> get_current_view();

    SharedLockedReference<const View> get_current_view_const();

    void add_view_upcall(const view_upcall_t& upcall);

    void report_failure(const node_id_t who);
    void barrier_sync();

    void register_initialize_objects_upcall(initialize_rpc_objects_t upcall) {
        initialize_subgroup_objects = std::move(upcall);
    }

    void silence();

    void debug_print_status() const;

    // UGLY - IMPROVE LATER
    std::map<subgroup_id_t, uint64_t> max_payload_sizes;
    std::map<subgroup_id_t, uint64_t> get_max_payload_sizes();
    // max of max_payload_sizes
    uint64_t view_max_rpc_reply_payload_size = 0;
    uint32_t view_max_rpc_window_size = 0;
};

} /* namespace derecho */