multicast_group.hpp

Go to the documentation of this file.

#pragma once

#include <assert.h>
#include <condition_variable>
#include <functional>
#include <list>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <ostream>
#include <queue>
#include <set>
#include <tuple>
#include <vector>

#include "../derecho_modes.hpp"
#include "../subgroup_info.hpp"
#include "connection_manager.hpp"
#include "derecho_internal.hpp"
#include "derecho_sst.hpp"
#include <derecho/conf/conf.hpp>
#include <derecho/mutils-serialization/SerializationMacros.hpp>
#include <derecho/mutils-serialization/SerializationSupport.hpp>
#include <derecho/rdmc/rdmc.hpp>
#include <derecho/sst/multicast.hpp>
#include <derecho/sst/sst.hpp>
#include <spdlog/spdlog.h>

namespace derecho {

struct CallbackSet {
    message_callback_t global_stability_callback;
    persistence_callback_t local_persistence_callback = nullptr;
    persistence_callback_t global_persistence_callback = nullptr;
};

struct __attribute__((__packed__)) header {
    uint32_t header_size;
    int32_t index;
    uint64_t timestamp;
    bool cooked_send;
};

struct DerechoParams : public mutils::ByteRepresentable {
    uint64_t max_msg_size;
    uint64_t max_reply_msg_size;
    uint64_t sst_max_msg_size;
    uint64_t block_size;
    unsigned int window_size;
    unsigned int heartbeat_ms;
    rdmc::send_algorithm rdmc_send_algorithm;
    uint32_t rpc_port;

    static uint64_t compute_max_msg_size(
            const uint64_t max_payload_size,
            const uint64_t block_size,
            bool using_rdmc) {
        auto max_msg_size = max_payload_size + sizeof(header);
        if(using_rdmc) {
            if(max_msg_size % block_size != 0) {
                max_msg_size = (max_msg_size / block_size + 1) * block_size;
            }
        }
        return max_msg_size;
    }

    static rdmc::send_algorithm send_algorithm_from_string(const std::string& rdmc_send_algorithm_string) {
        if(rdmc_send_algorithm_string == "binomial_send") {
            return rdmc::send_algorithm::BINOMIAL_SEND;
        } else if(rdmc_send_algorithm_string == "chain_send") {
            return rdmc::send_algorithm::CHAIN_SEND;
        } else if(rdmc_send_algorithm_string == "sequential_send") {
            return rdmc::send_algorithm::SEQUENTIAL_SEND;
        } else if(rdmc_send_algorithm_string == "tree_send") {
            return rdmc::send_algorithm::TREE_SEND;
        } else {
            throw "wrong value for RDMC send algorithm: " + rdmc_send_algorithm_string + ". Check your config file.";
        }
    }

    DerechoParams(uint64_t max_payload_size,
                  uint64_t max_reply_payload_size,
                  uint64_t max_smc_payload_size,
                  uint64_t block_size,
                  unsigned int window_size,
                  unsigned int heartbeat_ms,
                  rdmc::send_algorithm rdmc_send_algorithm,
                  uint32_t rpc_port)
            : max_reply_msg_size(max_reply_payload_size + sizeof(header)),
              sst_max_msg_size(max_smc_payload_size + sizeof(header)),
              block_size(block_size),
              window_size(window_size),
              heartbeat_ms(heartbeat_ms),
              rdmc_send_algorithm(rdmc_send_algorithm),
              rpc_port(rpc_port) {
        //if this is initialized above, DerechoParams turns abstract. idk why.
        max_msg_size = compute_max_msg_size(max_payload_size, block_size,
                                            max_payload_size > max_smc_payload_size);
    }

    DerechoParams() {}

    static DerechoParams from_profile(const std::string& profile) {
        // Use the profile string to search the configuration file for the appropriate
        // settings. If they do not exist, then we should utilize the defaults
        std::string prefix = "SUBGROUP/" + profile + "/";
        for(auto& field : Conf::subgroupProfileFields) {
            if(!hasCustomizedConfKey(prefix + field)) {
                std::cout << "profile " << profile
                          << " not found in SUBGROUP section of derecho conf. "
                             "Look at derecho-sample.cfg for more information."
                          << std::endl;
                throw profile + " derecho subgroup profile not found";
            }
        }

        uint64_t max_payload_size = getConfUInt64(prefix + Conf::subgroupProfileFields[0]);
    uint64_t max_reply_payload_size = getConfUInt64(prefix + Conf::subgroupProfileFields[1]);
        uint64_t max_smc_payload_size = getConfUInt64(prefix + Conf::subgroupProfileFields[2]);
        uint64_t block_size = getConfUInt64(prefix + Conf::subgroupProfileFields[3]);
        uint32_t window_size = getConfUInt32(prefix + Conf::subgroupProfileFields[4]);
        uint32_t timeout_ms = getConfUInt32(CONF_DERECHO_HEARTBEAT_MS);
        const std::string& algorithm = getConfString(prefix + Conf::subgroupProfileFields[5]);
        uint32_t rpc_port = getConfUInt32(CONF_DERECHO_RPC_PORT);

        return DerechoParams{
                max_payload_size,
        max_reply_payload_size,
                max_smc_payload_size,
                block_size,
                window_size,
                timeout_ms,
                DerechoParams::send_algorithm_from_string(algorithm),
                rpc_port,
        };
    }

    DEFAULT_SERIALIZATION_SUPPORT(DerechoParams, max_msg_size, max_reply_msg_size,
                                  sst_max_msg_size, block_size, window_size,
                                  heartbeat_ms, rdmc_send_algorithm, rpc_port);
};

struct MessageBuffer {
    std::unique_ptr<char[]> buffer;
    std::shared_ptr<rdma::memory_region> mr;

    MessageBuffer() {}
    MessageBuffer(size_t size) {
        if(size != 0) {
            buffer = std::unique_ptr<char[]>(new char[size]);
            mr = std::make_shared<rdma::memory_region>(buffer.get(), size);
        }
    }
    MessageBuffer(const MessageBuffer&) = delete;
    MessageBuffer(MessageBuffer&&) = default;
    MessageBuffer& operator=(const MessageBuffer&) = delete;
    MessageBuffer& operator=(MessageBuffer&&) = default;
};

struct RDMCMessage {
    uint32_t sender_id;
    //long long int index;
    message_id_t index;
    long long unsigned int size;
    MessageBuffer message_buffer;
};

struct SSTMessage {
    uint32_t sender_id;
    int32_t index;
    long long unsigned int size;
    volatile char* buf;
};

struct SubgroupSettings {
    uint32_t shard_num;
    uint32_t shard_rank;
    std::vector<node_id_t> members;
    std::vector<int> senders;
    int sender_rank;
    uint32_t num_received_offset;
    uint32_t slot_offset;
    Mode mode;
    DerechoParams profile;
};

class MulticastGroup {
    friend class ViewManager;

private:
    std::vector<node_id_t> members;
    std::map<node_id_t, uint32_t> node_id_to_sst_index;
    const unsigned int num_members;
    const int member_index;
    const CallbackSet callbacks;
    uint32_t total_num_subgroups;
    const std::map<subgroup_id_t, SubgroupSettings> subgroup_settings_map;
    std::vector<std::list<int32_t>> received_intervals;
    std::map<subgroup_id_t, uint32_t> subgroup_to_rdmc_group;
    rpc_handler_t rpc_callback;

    uint16_t rdmc_group_num_offset;
    bool rdmc_sst_groups_created = false;
    std::map<uint32_t, std::vector<MessageBuffer>> free_message_buffers;

    std::vector<message_id_t> future_message_indices;

    std::vector<std::optional<RDMCMessage>> next_sends;
    std::map<uint32_t, bool> pending_sst_sends;
    std::vector<std::queue<RDMCMessage>> pending_sends;
    std::vector<std::optional<RDMCMessage>> current_sends;

    std::map<std::pair<subgroup_id_t, node_id_t>, RDMCMessage> current_receives;
    std::map<subgroup_id_t, std::function<void(char*, size_t)>> singleton_shard_receive_handlers;

    std::map<subgroup_id_t, std::map<message_id_t, RDMCMessage>> locally_stable_rdmc_messages;
    std::map<subgroup_id_t, std::map<message_id_t, SSTMessage>> locally_stable_sst_messages;
    std::map<subgroup_id_t, std::set<uint64_t>> pending_message_timestamps;
    std::map<subgroup_id_t, std::map<message_id_t, uint64_t>> pending_persistence;
    std::map<subgroup_id_t, std::map<message_id_t, RDMCMessage>> non_persistent_messages;
    std::map<subgroup_id_t, std::map<message_id_t, SSTMessage>> non_persistent_sst_messages;

    std::vector<message_id_t> next_message_to_deliver;
    std::recursive_mutex msg_state_mtx;
    std::condition_variable_any sender_cv;

    unsigned int sender_timeout;

    std::atomic<bool> thread_shutdown{false};
    std::thread sender_thread;

    std::thread timeout_thread;

    std::shared_ptr<DerechoSST> sst;

    std::vector<std::unique_ptr<sst::multicast_group<DerechoSST>>> sst_multicast_group_ptrs;

    using pred_handle = typename sst::Predicates<DerechoSST>::pred_handle;
    std::list<pred_handle> receiver_pred_handles;
    std::list<pred_handle> stability_pred_handles;
    std::list<pred_handle> delivery_pred_handles;
    std::list<pred_handle> persistence_pred_handles;
    std::list<pred_handle> sender_pred_handles;

    std::vector<bool> last_transfer_medium;

    subgroup_post_next_version_func_t post_next_version_callback;

    persistence_manager_callbacks_t persistence_manager_callbacks;

    void send_loop();

    uint64_t get_time();

    void check_failures_loop();

    bool create_rdmc_sst_groups();
    void initialize_sst_row();
    void register_predicates();

    void deliver_message(RDMCMessage& msg, const subgroup_id_t& subgroup_num,
        const persistent::version_t& version, const uint64_t& msg_timestamp);

    void deliver_message(SSTMessage& msg, const subgroup_id_t& subgroup_num,
        const persistent::version_t& version, const uint64_t& msg_timestamp);

    bool version_message(RDMCMessage& msg, const subgroup_id_t& subgroup_num,
        const persistent::version_t& version, const uint64_t& msg_timestamp);
    bool version_message(SSTMessage& msg, const subgroup_id_t& subgroup_num,
        const persistent::version_t& version, const uint64_t& msg_timestamp);

    uint32_t get_num_senders(const std::vector<int>& shard_senders) {
        uint32_t num = 0;
        for(const auto i : shard_senders) {
            if(i) {
                num++;
            }
        }
        return num;
    };

    int32_t resolve_num_received(int32_t index, uint32_t num_received_entry);

    /* Predicate functions for receiving and delivering messages, parameterized by subgroup.
     * register_predicates will create and bind one of these for each subgroup. */

    void delivery_trigger(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                          const uint32_t num_shard_members, DerechoSST& sst);

    void sst_receive_handler(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                             const std::map<uint32_t, uint32_t>& shard_ranks_by_sender_rank,
                             uint32_t num_shard_senders, uint32_t sender_rank,
                             volatile char* data, uint64_t size);

    bool receiver_predicate(const SubgroupSettings& subgroup_settings,
                            const std::map<uint32_t, uint32_t>& shard_ranks_by_sender_rank,
                            uint32_t num_shard_senders, const DerechoSST& sst);

    void receiver_function(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                           const std::map<uint32_t, uint32_t>& shard_ranks_by_sender_rank,
                           uint32_t num_shard_senders, DerechoSST& sst, unsigned int batch_size,
                           const std::function<void(uint32_t, volatile char*, uint32_t)>& sst_receive_handler_lambda);

    // Internally used to automatically send a NULL message
    void get_buffer_and_send_auto_null(subgroup_id_t subgroup_num);
    /* Get a pointer into the current buffer, to write data into it before sending
     * Now this is a private function, called by send internally */
    char* get_sendbuffer_ptr(subgroup_id_t subgroup_num, long long unsigned int payload_size, bool cooked_send);

public:
    MulticastGroup(
            std::vector<node_id_t> members, node_id_t my_node_id,
            std::shared_ptr<DerechoSST> sst,
            CallbackSet callbacks,
            uint32_t total_num_subgroups,
            const std::map<subgroup_id_t, SubgroupSettings>& subgroup_settings_by_id,
            unsigned int sender_timeout,
            const subgroup_post_next_version_func_t& post_next_version_callback,
            const persistence_manager_callbacks_t& persistence_manager_callbacks,
            std::vector<char> already_failed = {});
    MulticastGroup(
            std::vector<node_id_t> members, node_id_t my_node_id,
            std::shared_ptr<DerechoSST> sst,
            MulticastGroup&& old_group,
            uint32_t total_num_subgroups,
            const std::map<subgroup_id_t, SubgroupSettings>& subgroup_settings_by_id,
            const subgroup_post_next_version_func_t& post_next_version_callback,
            const persistence_manager_callbacks_t& persistence_manager_callbacks,
            std::vector<char> already_failed = {});

    ~MulticastGroup();

    void register_rpc_callback(rpc_handler_t handler) { rpc_callback = std::move(handler); }

    void deliver_messages_upto(const std::vector<int32_t>& max_indices_for_senders, subgroup_id_t subgroup_num, uint32_t num_shard_senders);
    bool send(subgroup_id_t subgroup_num, long long unsigned int payload_size,
              const std::function<void(char* buf)>& msg_generator, bool cooked_send);
    bool check_pending_sst_sends(subgroup_id_t subgroup_num);

    const uint64_t compute_global_stability_frontier(subgroup_id_t subgroup_num);

    void wedge();
    void debug_print();

    const std::map<subgroup_id_t, SubgroupSettings>& get_subgroup_settings() {
        return subgroup_settings_map;
    }
    std::vector<uint32_t> get_shard_sst_indices(subgroup_id_t subgroup_num);
};
}  // namespace derecho