lf.hpp

Go to the documentation of this file.

#ifndef LF_HPP
#define LF_HPP

#include <map>
#include <rdma/fabric.h>
#include <thread>

#include <derecho/core/derecho_type_definitions.hpp>

#ifndef LF_VERSION
#define LF_VERSION FI_VERSION(1, 5)
#endif

namespace sst {

struct lf_sender_ctxt {
    uint32_t ce_idx;     // index into the comepletion entry vector. - 0xFFFFFFFF for invalid
    uint32_t remote_id;  // thread id of the sender
};

class _resources {
private:
    void connect_endpoint(bool is_lf_server);
    int init_endpoint(struct fi_info* fi);

protected:
    int post_remote_send(struct lf_sender_ctxt* ctxt, const long long int offset, const long long int size,
                         const int op, const bool completion);

public:
    int remote_id;
    // struct fid_cq *txcq, *rxcq; - moved to g_ctxt
    struct fid_ep* ep;
    struct fid_mr* write_mr;
    struct fid_mr* read_mr;
    char* write_buf;
    char* read_buf;
    uint64_t mr_lrkey;
    uint64_t mr_lwkey;
    uint64_t mr_rwkey;
    fi_addr_t remote_fi_addr;
    struct fid_eq* eq;

    _resources(int r_id, char* write_addr, char* read_addr, int size_w,
               int size_r, int is_lf_server);
    virtual ~_resources();
};

class resources : public _resources {
public:
    resources(int r_id, char* write_addr, char* read_addr, int size_w,
              int size_r, int is_lf_server) : _resources(r_id, write_addr, read_addr, size_w, size_r, is_lf_server) {
    }

    /*
      wrapper functions that make up the user interface
      all call post_remote_send with different parameters
    */

    void post_remote_read(const long long int size);
    void post_remote_read(const long long int offset, const long long int size);
    void post_remote_write(const long long int size);
    void post_remote_write(const long long int offset, long long int size);
    void post_remote_write_with_completion(struct lf_sender_ctxt* ctxt, const long long int size);
    void post_remote_write_with_completion(struct lf_sender_ctxt* ctxt, const long long int offset, const long long int size);
};

class resources_two_sided : public _resources {
    int post_receive(struct lf_sender_ctxt* ctxt, const long long int offset, const long long int size);

public:
    resources_two_sided(int r_id, char* write_addr, char* read_addr, int size_w,
                        int size_r, int is_lf_server) : _resources(r_id, write_addr, read_addr, size_w, size_r, is_lf_server) {
    }

    void post_two_sided_send(const long long int size);
    void post_two_sided_send(const long long int offset, long long int size);
    void post_two_sided_send_with_completion(struct lf_sender_ctxt* ctxt, const long long int size);
    void post_two_sided_send_with_completion(struct lf_sender_ctxt* ctxt, const long long int offset, const long long int size);
    void post_two_sided_receive(struct lf_sender_ctxt* ctxt, const long long int size);
    void post_two_sided_receive(struct lf_sender_ctxt* ctxt, const long long int offset, const long long int size);
};

bool add_node(uint32_t new_id, const std::pair<ip_addr_t, uint16_t>& new_ip_addr_and_port);
bool remove_node(uint32_t node_id);
bool sync(uint32_t r_id);
void lf_initialize(const std::map<uint32_t, std::pair<ip_addr_t, uint16_t>>& ip_addrs_and_ports,
                   uint32_t node_rank);
std::pair<uint32_t, std::pair<int32_t, int32_t>> lf_poll_completion();
void shutdown_polling_thread();
void lf_destroy();
}  // namespace sst

#endif  // LF_HPP