sst.hpp

Go to the documentation of this file.

#pragma once

#include <atomic>
#include <bitset>
#include <cassert>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <list>
#include <memory>
#include <mutex>
#include <numeric>
#include <stdexcept>
#include <string.h>
#include <string>
#include <thread>
#include <vector>

#include "predicates.hpp"
#include <derecho/conf/conf.hpp>

#ifdef USE_VERBS_API
#include "detail/verbs.hpp"
#else  //LIBFABRIC
#include "detail/lf.hpp"
#endif

using sst::resources;

namespace sst {

const int alignTo = sizeof(long);

constexpr size_t padded_len(const size_t& len) {
    return (len < alignTo) ? alignTo : (len + alignTo) | (alignTo - 1);
}

class _SSTField {
public:
    volatile char* base;
    size_t rowLen;
    size_t field_len;

    _SSTField(const size_t field_len) : base(nullptr), rowLen(0), field_len(field_len) {}

    size_t set_base(volatile char* const base) {
        this->base = base;
        return padded_len(field_len);
    }

    char* get_base() {
        return const_cast<char*>(base);
    }

    void set_rowLen(const size_t& _rowLen) { rowLen = _rowLen; }
};

template <typename T>
class SSTField : public _SSTField {
public:
    using _SSTField::base;
    using _SSTField::field_len;
    using _SSTField::rowLen;

    SSTField() : _SSTField(sizeof(T)) {
    }

    // Tracks down the appropriate row
    volatile T& operator[](const size_t row_idx) const { return ((T&)base[row_idx * rowLen]); }

    // Getter
    volatile T const& operator()(const size_t row_idx) const {
        return *(T*)(base + row_idx * rowLen);
    }

    // Setter
    void operator()(const size_t row_idx, T const v) { *(T*)(base + row_idx * rowLen) = v; }
};

template <typename T>
class SSTFieldVector : public _SSTField {
private:
    const size_t length;

public:
    using _SSTField::base;
    using _SSTField::field_len;
    using _SSTField::rowLen;
    using value_type = T;

    SSTFieldVector(size_t num_elements) : _SSTField(num_elements * sizeof(T)), length(num_elements) {
    }

    // Tracks down the appropriate row
    volatile T* operator[](const size_t& idx) const { return (T*)(base + idx * rowLen); }

    size_t size() const { return length; }

    void __attribute__((noinline)) debug_print(size_t row_num) {
        volatile T* arr = (*this)[row_num];
        for(size_t j = 0; j < length; ++j) {
            std::cout << arr[j] << " ";
        }
        std::cout << std::endl;
    }
};

typedef std::function<void(uint32_t)> failure_upcall_t;

struct SSTParams {
    const std::vector<uint32_t>& members;
    const uint32_t my_node_id;
    const failure_upcall_t failure_upcall;
    const std::vector<char> already_failed;
    const bool start_predicate_thread;

    SSTParams(const std::vector<uint32_t>& _members,
              const uint32_t my_node_id,
              const failure_upcall_t failure_upcall = nullptr,
              const std::vector<char> already_failed = {},
              const bool start_predicate_thread = true)
            : members(_members),
              my_node_id(my_node_id),
              failure_upcall(failure_upcall),
              already_failed(already_failed),
              start_predicate_thread(start_predicate_thread) {}
};

template <class DerivedSST>
class SST {
private:
    template <typename... Fields>
    void init_SSTFields(Fields&... fields) {
        rowLen = 0;
        compute_rowLen(rowLen, fields...);
        rows = new char[rowLen * num_members];
        // snapshot = new char[rowLen * num_members];
        volatile char* base = rows;
        set_bases_and_rowLens(base, rowLen, fields...);
    }

    DerivedSST* derived_this;

    std::vector<std::thread> background_threads;
    std::atomic<bool> thread_shutdown;

    void detect();

public:
    Predicates<DerivedSST> predicates;
    friend class Predicates<DerivedSST>;

private:
    const uint32_t poll_cq_timeout_ms;
    volatile char* rows;
    // char* snapshot;
    size_t rowLen;
    const std::vector<uint32_t>& members;
    const unsigned int num_members;
    std::vector<uint32_t> all_indices;
    unsigned int my_index;
    std::map<uint32_t, int, std::greater<uint32_t>> members_by_id;
    uint32_t my_node_id;
    // std::map<int, int> qp_num_to_index;

    std::vector<bool> row_is_frozen;
    int num_frozen{0};
    failure_upcall_t failure_upcall;
    std::mutex freeze_mutex;

    std::vector<std::unique_ptr<resources>> res_vec;

    bool thread_start;
    std::mutex thread_start_mutex;
    std::condition_variable thread_start_cv;

public:
    SST(DerivedSST* derived_class_pointer, const SSTParams& params)
            : derived_this(derived_class_pointer),
              thread_shutdown(false),
              poll_cq_timeout_ms(derecho::getConfUInt32(CONF_DERECHO_SST_POLL_CQ_TIMEOUT_MS)),
              members(params.members),
              num_members(members.size()),
              all_indices(num_members),
              my_node_id(params.my_node_id),
              row_is_frozen(num_members),
              failure_upcall(params.failure_upcall),
              res_vec(num_members),
              thread_start(params.start_predicate_thread) {
        //Figure out my SST index
        my_index = (uint)-1;
        for(uint32_t i = 0; i < num_members; ++i) {
            if(members[i] == my_node_id) {
                my_index = i;
                break;
            }
        }
        assert(my_index != (uint)-1);

        std::iota(all_indices.begin(), all_indices.end(), 0);

        if(!params.already_failed.empty()) {
            assert(params.already_failed.size() == num_members);
            for(size_t index = 0; index < params.already_failed.size(); ++index) {
                if(params.already_failed[index]) {
                    row_is_frozen[index] = true;
                }
            }
        }

        // sort members descending by node ID, while keeping track of their
        // specified index in the SST
        for(unsigned int sst_index = 0; sst_index < num_members; ++sst_index) {
            members_by_id[members[sst_index]] = sst_index;
        }
    }

    template <typename... Fields>
    void SSTInit(Fields&... fields) {
        //Initialize rows and set the "base" field of each SSTField
        init_SSTFields(fields...);

        //Initialize res_vec with the correct offsets for each row
        unsigned int node_rank, sst_index;
        for(auto const& rank_index : members_by_id) {
            std::tie(node_rank, sst_index) = rank_index;
            char *write_addr, *read_addr;
            write_addr = const_cast<char*>(rows) + rowLen * sst_index;
            read_addr = const_cast<char*>(rows) + rowLen * my_index;
            if(sst_index != my_index) {
                if(row_is_frozen[sst_index]) {
                    continue;
                }
#ifdef USE_VERBS_API
                res_vec[sst_index] = std::make_unique<resources>(
                        node_rank, write_addr, read_addr, rowLen, rowLen);
#else  // use libfabric api by default
                res_vec[sst_index] = std::make_unique<resources>(
                        node_rank, write_addr, read_addr, rowLen, rowLen, (my_node_id < node_rank));
#endif
                // update qp_num_to_index
                // qp_num_to_index[res_vec[sst_index].get()->qp->qp_num] = sst_index;
            }
        }

        std::thread detector(&SST::detect, this);
        background_threads.push_back(std::move(detector));
    }

    ~SST();

    void start_predicate_evaluation();

    void sync_with_members() const;

    void sync_with_members(std::vector<uint32_t> row_indices) const;

    void freeze(int row_index);

    unsigned int get_num_rows() const { return num_members; }

    unsigned int get_local_index() const { return my_index; }

    const char* getBaseAddress() {
        return const_cast<char*>(rows);
    }

    void put() {
        put(all_indices, 0, rowLen);
    }

    void put_with_completion() {
        put_with_completion(all_indices, 0, rowLen);
    }

    void put(const std::vector<uint32_t> receiver_ranks) {
        put(receiver_ranks, 0, rowLen);
    }

    void put_with_completion(const std::vector<uint32_t> receiver_ranks) {
        put_with_completion(receiver_ranks, 0, rowLen);
    }

    void put(size_t offset, size_t size) {
        put(all_indices, offset, size);
    }

    template <typename T>
    void put(SSTField<T>& field) {
        put(all_indices, field.get_base() - getBaseAddress(), sizeof(field[0]));
    }

    template <typename T>
    void put(SSTFieldVector<T>& vec_field) {
        put(all_indices, vec_field.get_base() - getBaseAddress(),
            sizeof(vec_field[0][0]) * vec_field.size());
    }

    template <typename T>
    void put(SSTFieldVector<T>& vec_field, std::size_t index) {
        put(all_indices,
            const_cast<char*>(reinterpret_cast<volatile char*>(std::addressof(vec_field[0][index])))
                    - getBaseAddress(),
            sizeof(vec_field[0][index]));
    }

    template <typename T>
    void put(const std::vector<uint32_t> receiver_ranks,
             SSTFieldVector<T>& vec_field, std::size_t index) {
        put(receiver_ranks,
            const_cast<char*>(reinterpret_cast<volatile char*>(std::addressof(vec_field[0][index])))
                    - getBaseAddress(),
            sizeof(vec_field[0][index]));
    }

    void put_with_completion(size_t offset, size_t size) {
        put_with_completion(all_indices, offset, size);
    }

    void put(const std::vector<uint32_t> receiver_ranks, size_t offset, size_t size);

    void put_with_completion(const std::vector<uint32_t> receiver_ranks, size_t offset, size_t size);

private:
    using char_p = volatile char*;

    void compute_rowLen(size_t&) {}

    template <typename Field, typename... Fields>
    void compute_rowLen(size_t& rowLen, Field& f, Fields&... rest) {
        rowLen += padded_len(f.field_len);
        compute_rowLen(rowLen, rest...);
    }

    void set_bases_and_rowLens(char_p&, const size_t) {}

    template <typename Field, typename... Fields>
    void set_bases_and_rowLens(char_p& base, const size_t rlen, Field& f, Fields&... rest) {
        base += f.set_base(base);
        f.set_rowLen(rlen);
        set_bases_and_rowLens(base, rlen, rest...);
    }

    // void take_snapshot() {
    //   memcpy(snapshot, const_cast<char*>(rows), rowLen * num_members);
    // }

    // // returns snapshot == current
    // bool compare_snapshot_and_current() {
    //     int res = memcmp(const_cast<char*>(rows), snapshot, rowLen * num_members);
    //     if(res == 0) {
    //         return true;
    //     }
    //     return false;
    // }
};

} /* namespace sst */

#include "detail/sst_impl.hpp"