derecho_sst.hpp

Go to the documentation of this file.

#pragma once

#include <atomic>
#include <cstdint>
#include <cstring>
#include <mutex>
#include <sstream>
#include <string>

#include "../derecho_type_definitions.hpp"
#include "derecho_internal.hpp"
#include <derecho/sst/sst.hpp>

namespace derecho {

using sst::SSTField;
using sst::SSTFieldVector;
class DerechoSST : public sst::SST<DerechoSST> {
public:
    // MulticastGroup members, related only to tracking message delivery
    SSTFieldVector<message_id_t> seq_num;
    SSTFieldVector<message_id_t> delivered_num;
    SSTFieldVector<persistent::version_t> persisted_num;

    // Group management service members, related only to handling view changes
    SSTField<int32_t> vid;
    SSTFieldVector<bool> suspected;
    SSTFieldVector<node_id_t> changes;
    SSTFieldVector<uint32_t> joiner_ips;
    SSTFieldVector<uint16_t> joiner_gms_ports;
    SSTFieldVector<uint16_t> joiner_rpc_ports;
    SSTFieldVector<uint16_t> joiner_sst_ports;
    SSTFieldVector<uint16_t> joiner_rdmc_ports;
    SSTField<int> num_changes;
    SSTField<int> num_committed;
    SSTField<int> num_acked;
    SSTField<int> num_installed;
    SSTFieldVector<int32_t> num_received;
    SSTField<bool> wedged;
    SSTFieldVector<int> global_min;
    SSTFieldVector<bool> global_min_ready;
    SSTFieldVector<char> slots;
    SSTFieldVector<int32_t> num_received_sst;

    SSTFieldVector<uint64_t> local_stability_frontier;

    SSTField<bool> rip;
    DerechoSST(const sst::SSTParams& parameters, uint32_t num_subgroups, uint32_t num_received_size, uint64_t slot_size)
            : sst::SST<DerechoSST>(this, parameters),
              seq_num(num_subgroups),
              delivered_num(num_subgroups),
              persisted_num(num_subgroups),
              suspected(parameters.members.size()),
              changes(100 + parameters.members.size()),
              joiner_ips(100 + parameters.members.size()),
              joiner_gms_ports(100 + parameters.members.size()),
              joiner_rpc_ports(100 + parameters.members.size()),
              joiner_sst_ports(100 + parameters.members.size()),
              joiner_rdmc_ports(100 + parameters.members.size()),
              num_received(num_received_size),
              global_min(num_received_size),
              global_min_ready(num_subgroups),
              slots(slot_size),
              num_received_sst(num_received_size),
              local_stability_frontier(num_subgroups) {
        SSTInit(seq_num, delivered_num,
                persisted_num, vid, suspected, changes, joiner_ips,
                joiner_gms_ports, joiner_rpc_ports, joiner_sst_ports, joiner_rdmc_ports,
                num_changes, num_committed, num_acked, num_installed,
                num_received, wedged, global_min, global_min_ready,
                slots, num_received_sst, local_stability_frontier, rip);
        //Once superclass constructor has finished, table entries can be initialized
        for(unsigned int row = 0; row < get_num_rows(); ++row) {
            vid[row] = 0;
            for(size_t i = 0; i < suspected.size(); ++i) {
                suspected[row][i] = false;
            }
            for(size_t i = 0; i < changes.size(); ++i) {
                changes[row][i] = false;
            }
            for(size_t i = 0; i < global_min_ready.size(); ++i) {
                global_min_ready[row][i] = false;
            }
            for(size_t i = 0; i < global_min.size(); ++i) {
                global_min[row][i] = 0;
            }
            memset(const_cast<uint32_t*>(joiner_ips[row]), 0, joiner_ips.size());
            memset(const_cast<uint16_t*>(joiner_gms_ports[row]), 0, joiner_gms_ports.size());
            memset(const_cast<uint16_t*>(joiner_rpc_ports[row]), 0, joiner_rpc_ports.size());
            memset(const_cast<uint16_t*>(joiner_sst_ports[row]), 0, joiner_sst_ports.size());
            memset(const_cast<uint16_t*>(joiner_rdmc_ports[row]), 0, joiner_rdmc_ports.size());
            num_changes[row] = 0;
            num_committed[row] = 0;
            num_installed[row] = 0;
            num_acked[row] = 0;
            wedged[row] = false;
            // start off local_stability_frontier with the current time
            struct timespec start_time;
            clock_gettime(CLOCK_REALTIME, &start_time);
            auto current_time = start_time.tv_sec * 1e9 + start_time.tv_nsec;
            for(size_t i = 0; i < local_stability_frontier.size(); ++i) {
                local_stability_frontier[row][i] = current_time;
            }
            rip[row] = false;
        }
    }

    void init_local_row_from_previous(const DerechoSST& old_sst, const int row, const int num_changes_installed);

    void init_local_change_proposals(const int other_row);

    std::string to_string() const;
};

namespace gmssst {

template <typename Elem>
void set(volatile Elem& e, const Elem& value) {
    e = value;
    std::atomic_signal_fence(std::memory_order_acq_rel);
}

template <typename Elem>
void set(volatile Elem& e, volatile const Elem& value) {
    e = value;
    std::atomic_signal_fence(std::memory_order_acq_rel);
}

template <typename Elem>
void set(volatile Elem* array, volatile Elem* value, const size_t length) {
    static thread_local std::mutex set_mutex;
    {
        std::lock_guard<std::mutex> lock(set_mutex);
        memcpy(const_cast<Elem*>(array), const_cast<Elem*>(value),
               length * sizeof(Elem));
    }
    std::atomic_signal_fence(std::memory_order_acq_rel);
}
template <typename Arr, size_t Len>
void set(volatile Arr (&e)[Len], const volatile Arr (&value)[Len]) {
    static thread_local std::mutex set_mutex;
    {
        std::lock_guard<std::mutex> lock(set_mutex);
        memcpy(const_cast<Arr(&)[Len]>(e), const_cast<const Arr(&)[Len]>(value),
               Len * sizeof(Arr));
        // copy_n just plain doesn't work, claiming that its argument types are
        // "not assignable"
        //        std::copy_n(const_cast<const Arr (&)[Len]>(value), Len,
        //        const_cast<Arr (&)[Len]>(e));
    }
    std::atomic_signal_fence(std::memory_order_acq_rel);
}

template <size_t L1, size_t L2, typename Arr>
void set(volatile Arr (&dst)[L1], const volatile Arr (&src)[L2], const size_t& num) {
    static thread_local std::mutex set_mutex;
    {
        std::lock_guard<std::mutex> lock(set_mutex);
        memcpy(const_cast<Arr(&)[L2]>(dst), const_cast<const Arr(&)[L1]>(src),
               num * sizeof(Arr));
    }
    std::atomic_signal_fence(std::memory_order_acq_rel);
}

void set(volatile char* string_array, const std::string& value);

void increment(volatile int& member);

bool equals(const volatile char& string_array, const std::string& value);

}  // namespace gmssst

}  // namespace derecho