restart_state.hpp

Go to the documentation of this file.

#pragma once

#include <cstdint>
#include <map>
#include <memory>
#include <sstream>
#include <string>
#include <vector>

#include "../subgroup_info.hpp"
#include "../view.hpp"
#include "derecho_internal.hpp"

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

namespace derecho {
struct RaggedTrim : public mutils::ByteRepresentable {
    subgroup_id_t subgroup_id;
    uint32_t shard_num;
    int vid;
    int32_t leader_id;  //Signed instead of unsigned so it can have the special value -1
    std::vector<int32_t> max_received_by_sender;
    RaggedTrim(subgroup_id_t subgroup_id, uint32_t shard_num, int vid,
               int32_t leader_id, std::vector<int32_t> max_received_by_sender)
            : subgroup_id(subgroup_id), shard_num(shard_num), vid(vid), leader_id(leader_id), max_received_by_sender(max_received_by_sender) {}
    DEFAULT_SERIALIZATION_SUPPORT(RaggedTrim, subgroup_id, shard_num, vid, leader_id, max_received_by_sender);
};

enum class CommitMessage {
    PREPARE,  
    COMMIT,   
    ABORT,    
    ACK       
};

inline std::string ragged_trim_filename(subgroup_id_t subgroup_num, uint32_t shard_num) {
    std::ostringstream string_builder;
    string_builder << "RaggedTrim_" << subgroup_num << "_" << shard_num;
    return string_builder.str();
}

using ragged_trim_map_t = std::map<subgroup_id_t, std::map<uint32_t, std::unique_ptr<RaggedTrim>>>;

struct RestartState {
    ragged_trim_map_t logged_ragged_trim;
    std::vector<std::vector<int64_t>> restart_shard_leaders;
    void load_ragged_trim(const View& curr_view);
    static persistent::version_t ragged_trim_to_latest_version(const int32_t view_id,
                                                               const std::vector<int32_t>& max_received_by_sender);
};

class RestartLeaderState {
private:
    std::unique_ptr<View> curr_view;
    RestartState& restart_state;
    const SubgroupInfo& subgroup_info;

    std::unique_ptr<View> restart_view;
    std::map<node_id_t, tcp::socket> waiting_join_sockets;
    std::map<node_id_t, std::tuple<ip_addr_t, uint16_t, uint16_t, uint16_t, uint16_t>> rejoined_node_ips_and_ports;
    std::set<node_id_t> members_sent_restart_view;
    std::set<node_id_t> rejoined_node_ids;
    std::set<node_id_t> last_known_view_members;
    std::vector<std::vector<persistent::version_t>> longest_log_versions;
    std::vector<std::vector<int64_t>> nodes_with_longest_log;
    const node_id_t my_id;

    void receive_joiner_logs(const node_id_t& joiner_id, tcp::socket& client_socket);

    bool compute_restart_view();

    std::unique_ptr<View> update_curr_and_next_restart_view();

public:
    static const int RESTART_LEADER_TIMEOUT = 2000;
    RestartLeaderState(std::unique_ptr<View> _curr_view, RestartState& restart_state,
                       const SubgroupInfo& subgroup_info,
                       const node_id_t my_id);
    void await_quorum(tcp::connection_listener& server_socket);

    bool has_restart_quorum();

    bool resend_view_until_quorum_lost();

    int64_t send_restart_view();

    void send_abort();

    int64_t send_prepare();

    void send_commit();
    const View& get_curr_view() const { return *curr_view; }
    const View& get_restart_view() const { return *restart_view; }
    std::unique_ptr<View> take_restart_view() { return std::move(restart_view); }

    void print_longest_logs() const;

    static std::unique_ptr<View> make_next_view(const std::unique_ptr<View>& curr_view,
                                                const std::vector<node_id_t>& joiner_ids,
                                                const std::vector<std::tuple<ip_addr_t, uint16_t, uint16_t, uint16_t, uint16_t>>& joiner_ips_and_ports);
    static bool contains_at_least_one_member_per_subgroup(std::set<node_id_t> rejoined_node_ids,
                                                          const View& last_view);
};

} /* namespace derecho */