lf_helper.hpp

Go to the documentation of this file.

#ifndef LF_HELPER_HPP
#define LF_HELPER_HPP

#include <cstdint>
#include <optional>
#include <functional>
#include <map>
#include <memory>
#include <string>
#include <vector>
#include <rdma/fabric.h>

#include <derecho/core/derecho_type_definitions.hpp>

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

struct fid_mr;
struct fid_ep;
struct fid_cq;

namespace rdma {
class exception {};
class invalid_args : public exception {};
class configuration_failure : public exception {};
class connection_broken : public exception {};
class creation_failure : public exception {};
class mr_creation_failure : public creation_failure {};
class ep_creation_failure : public creation_failure {};
class cq_creation_failure : public creation_failure {};
class message_types_exhausted : public exception {};
class unsupported_feature : public exception {};

//template<typename fi_struct_type>
//struct close { 
//  void operator() (fi_struct_type* fi_struct) const {fi_close(fi_struct->fid);} 
//};

class memory_region {
    std::unique_ptr<fid_mr, std::function<void(fid_mr *)>> mr;
    std::unique_ptr<char[]> allocated_buffer;

    friend class endpoint;
    friend class task;

public:
    memory_region(size_t size);
    memory_region(char* buffer, size_t size);
    uint64_t get_key() const;

    char* const buffer;
    const size_t size;
};

class remote_memory_region {
public:
    remote_memory_region(uint64_t remote_address, size_t length,
                         uint64_t remote_key)
        : buffer(remote_address), size(length), rkey(remote_key) {}

    const uint64_t buffer;
    const size_t size;
    const uint64_t rkey;
};

class completion_queue {
    std::unique_ptr<fid_cq, std::function<void(fid_cq *)>> cq;

    friend class managed_endpoint;
    friend class task;

public:
    explicit completion_queue();
};

typedef std::function<void(uint64_t tag, uint32_t immediate, size_t length)> 
        completion_handler;

class message_type {
public:
    typedef uint8_t tag_type;
    static constexpr unsigned int shift_bits = 64 - 8 * sizeof(tag_type);

private:
    std::optional<tag_type> tag;
    message_type(tag_type t) : tag(t) {}

    friend class endpoint;
    friend class task;

public:
    message_type(const std::string& name, completion_handler send_handler,
                 completion_handler recv_handler,
                 completion_handler write_handler = nullptr);
    message_type() {}

    static message_type ignored();
};

class endpoint {
protected:
    std::unique_ptr<fid_eq, std::function<void(fid_eq *)>> eq;
    std::unique_ptr<fid_ep, std::function<void(fid_ep *)>> ep;

    explicit endpoint() {}

    friend class task;
public:
    virtual ~endpoint();
    
    explicit endpoint(size_t remote_index, bool is_lf_server);
    endpoint(size_t remote_index, bool is_lf_server,
             std::function<void(endpoint*)> post_recvs);
    endpoint(endpoint&&) = default;
    int init(struct fi_info *fi);
    void connect(size_t remote_index, bool is_lf_server,
                 std::function<void(endpoint *)> post_recvs);

    bool post_send(const memory_region& mr, size_t offset, 
                   size_t size, uint64_t wr_id, uint32_t immediate, 
                   const message_type& type);
    bool post_recv(const memory_region& mr, size_t offset, 
                   size_t size, uint64_t wr_id, 
                   const message_type& type);

    bool post_empty_send(uint64_t wr_id, uint32_t immediate,
                         const message_type& type);
    bool post_empty_recv(uint64_t wr_id, const message_type& type);

    bool post_write(const memory_region& mr, size_t offset, size_t size,
                    uint64_t wr_id, remote_memory_region remote_mr,
                    size_t remote_offset, const message_type& type,
                    bool signaled = false, bool send_inline = false);

    fi_addr_t remote_fi_addr;
};

class managed_endpoint : public endpoint {
public:
    completion_queue scq, rcq;
    managed_endpoint(size_t remote_index,
                       std::function<void(managed_endpoint*)> post_recvs) {}
};

class manager_endpoint : public endpoint {
public:
    explicit manager_endpoint() {}
};

class task {
protected:
    struct task_impl;
    std::unique_ptr<task_impl> impl;
    std::shared_ptr<manager_endpoint> mep;

public:
    task(std::shared_ptr<manager_endpoint> manager_ep);
    virtual ~task();

    void append_wait(const completion_queue& cq, int count, bool signaled,
                     bool last, uint64_t wr_id, const message_type& type);
    void append_enable_send(const managed_endpoint& ep, int count);
    void append_send(const managed_endpoint& ep, const memory_region& mr, 
                     size_t offset, size_t length, uint32_t immediate);
    void append_recv(const managed_endpoint& ep, const memory_region& mr, 
                     size_t offset, size_t length);
    bool post() __attribute__((warn_unused_result));
};

namespace impl {
  bool lf_initialize(const std::map<uint32_t, std::pair<ip_addr_t, uint16_t>>& ip_addrs_and_ports,
                   uint32_t node_rank);
bool lf_add_connection(uint32_t new_id, const std::pair<ip_addr_t, uint16_t> &new_ip_addr_and_port);
bool lf_remove_connection(uint32_t node_id);
bool lf_destroy();

std::map<uint32_t, remote_memory_region> lf_exchange_memory_regions(
         const std::vector<uint32_t>& members, uint32_t node_rank,
         const memory_region& mr);

bool set_interrupt_mode(bool enabled);
} /* namespace impl */
} /* namespace rdma */

#endif /* LF_HELPER_HPP */