qp.h

#pragma once
 
#include <atomic>
#include <coroutine>
#include <cstdint>
#include <exception>
#include <memory>
#include <optional>
#include <utility>
#include <vector>
 
#include <infiniband/verbs.h>
 
#include "rdmapp/cq.h"
#include "rdmapp/device.h"
#include "rdmapp/pd.h"
#include "rdmapp/srq.h"
 
#include "rdmapp/detail/noncopyable.h"
#include "rdmapp/detail/serdes.h"
 
namespace rdmapp {
 
struct deserialized_qp {
  struct qp_header {
    static constexpr size_t kSerializedSize =
        sizeof(uint16_t) + 3 * sizeof(uint32_t) + sizeof(union ibv_gid);
    uint16_t lid;
    uint32_t qp_num;
    uint32_t sq_psn;
    uint32_t user_data_size;
    union ibv_gid gid;
  } header;
  template <class It> static deserialized_qp deserialize(It it) {
    deserialized_qp des_qp;
    detail::deserialize(it, des_qp.header.lid);
    detail::deserialize(it, des_qp.header.qp_num);
    detail::deserialize(it, des_qp.header.sq_psn);
    detail::deserialize(it, des_qp.header.user_data_size);
    detail::deserialize(it, des_qp.header.gid);
    return des_qp;
  }
  std::vector<uint8_t> user_data;
};
 
class qp : public noncopyable, public std::enable_shared_from_this<qp> {
  static std::atomic<uint32_t> next_sq_psn;
  struct ibv_qp *qp_;
  struct ibv_srq *raw_srq_;
  uint32_t sq_psn_;
  void (qp::*post_recv_fn)(struct ibv_recv_wr const &recv_wr,
                           struct ibv_recv_wr *&bad_recv_wr) const;
 
  std::shared_ptr<pd> pd_;
  std::shared_ptr<cq> recv_cq_;
  std::shared_ptr<cq> send_cq_;
  std::shared_ptr<srq> srq_;
  std::vector<uint8_t> user_data_;
 
  void create();
 
  void init();
 
  void destroy();
 
public:
  class send_awaitable {
    std::shared_ptr<qp> qp_;
    std::shared_ptr<local_mr> local_mr_;
    std::exception_ptr exception_;
    remote_mr remote_mr_;
    uint64_t compare_add_;
    uint64_t swap_;
    uint32_t imm_;
    struct ibv_wc wc_;
    const enum ibv_wr_opcode opcode_;
 
  public:
    send_awaitable(std::shared_ptr<qp> qp, void *buffer, size_t length,
                   enum ibv_wr_opcode opcode);
    send_awaitable(std::shared_ptr<qp> qp, void *buffer, size_t length,
                   enum ibv_wr_opcode opcode, remote_mr const &remote_mr);
    send_awaitable(std::shared_ptr<qp> qp, void *buffer, size_t length,
                   enum ibv_wr_opcode opcode, remote_mr const &remote_mr,
                   uint32_t imm);
    send_awaitable(std::shared_ptr<qp> qp, void *buffer, size_t length,
                   enum ibv_wr_opcode opcode, remote_mr const &remote_mr,
                   uint64_t add);
    send_awaitable(std::shared_ptr<qp> qp, void *buffer, size_t length,
                   enum ibv_wr_opcode opcode, remote_mr const &remote_mr,
                   uint64_t compare, uint64_t swap);
    send_awaitable(std::shared_ptr<qp> qp, std::shared_ptr<local_mr> local_mr,
                   enum ibv_wr_opcode opcode);
    send_awaitable(std::shared_ptr<qp> qp, std::shared_ptr<local_mr> local_mr,
                   enum ibv_wr_opcode opcode, remote_mr const &remote_mr);
    send_awaitable(std::shared_ptr<qp> qp, std::shared_ptr<local_mr> local_mr,
                   enum ibv_wr_opcode opcode, remote_mr const &remote_mr,
                   uint32_t imm);
    send_awaitable(std::shared_ptr<qp> qp, std::shared_ptr<local_mr> local_mr,
                   enum ibv_wr_opcode opcode, remote_mr const &remote_mr,
                   uint64_t add);
    send_awaitable(std::shared_ptr<qp> qp, std::shared_ptr<local_mr> local_mr,
                   enum ibv_wr_opcode opcode, remote_mr const &remote_mr,
                   uint64_t compare, uint64_t swap);
    bool await_ready() const noexcept;
    bool await_suspend(std::coroutine_handle<> h) noexcept;
    uint32_t await_resume() const;
    constexpr bool is_rdma() const;
    constexpr bool is_atomic() const;
  };
 
  class recv_awaitable {
    std::shared_ptr<qp> qp_;
    std::shared_ptr<local_mr> local_mr_;
    std::exception_ptr exception_;
    struct ibv_wc wc_;
    enum ibv_wr_opcode opcode_;
 
  public:
    recv_awaitable(std::shared_ptr<qp> qp, std::shared_ptr<local_mr> local_mr);
    recv_awaitable(std::shared_ptr<qp> qp, void *buffer, size_t length);
    bool await_ready() const noexcept;
    bool await_suspend(std::coroutine_handle<> h) noexcept;
    std::pair<uint32_t, std::optional<uint32_t>> await_resume() const;
  };
 
  qp(const uint16_t remote_lid, const uint32_t remote_qpn,
     const uint32_t remote_psn, const union ibv_gid remote_gid,
     std::shared_ptr<pd> pd, std::shared_ptr<cq> cq,
     std::shared_ptr<srq> srq = nullptr);
 
  qp(const uint16_t remote_lid, const uint32_t remote_qpn,
     const uint32_t remote_psn, const union ibv_gid remote_gid,
     std::shared_ptr<pd> pd, std::shared_ptr<cq> recv_cq,
     std::shared_ptr<cq> send_cq, std::shared_ptr<srq> srq = nullptr);
 
  qp(std::shared_ptr<pd> pd, std::shared_ptr<cq> cq,
     std::shared_ptr<srq> srq = nullptr);
 
  qp(std::shared_ptr<pd> pd, std::shared_ptr<cq> recv_cq,
     std::shared_ptr<cq> send_cq, std::shared_ptr<srq> srq = nullptr);
 
  void post_send(struct ibv_send_wr const &send_wr,
                 struct ibv_send_wr *&bad_send_wr);
 
  void post_recv(struct ibv_recv_wr const &recv_wr,
                 struct ibv_recv_wr *&bad_recv_wr) const;
 
  [[nodiscard]] send_awaitable send(void *buffer, size_t length);
 
  [[nodiscard]] send_awaitable write(remote_mr const &remote_mr, void *buffer,
                                     size_t length);
 
  [[nodiscard]] send_awaitable write_with_imm(remote_mr const &remote_mr,
                                              void *buffer, size_t length,
                                              uint32_t imm);
 
  [[nodiscard]] send_awaitable read(remote_mr const &remote_mr, void *buffer,
                                    size_t length);
 
  [[nodiscard]] send_awaitable fetch_and_add(remote_mr const &remote_mr,
                                             void *buffer, size_t length,
                                             uint64_t add);
 
  [[nodiscard]] send_awaitable compare_and_swap(remote_mr const &remote_mr,
                                                void *buffer, size_t length,
                                                uint64_t compare,
                                                uint64_t swap);
 
  [[nodiscard]] recv_awaitable recv(void *buffer, size_t length);
 
  [[nodiscard]] send_awaitable send(std::shared_ptr<local_mr> local_mr);
 
  [[nodiscard]] send_awaitable write(remote_mr const &remote_mr,
                                     std::shared_ptr<local_mr> local_mr);
 
  [[nodiscard]] send_awaitable
  write_with_imm(remote_mr const &remote_mr, std::shared_ptr<local_mr> local_mr,
                 uint32_t imm);
 
  [[nodiscard]] send_awaitable read(remote_mr const &remote_mr,
                                    std::shared_ptr<local_mr> local_mr);
 
  [[nodiscard]] send_awaitable fetch_and_add(remote_mr const &remote_mr,
                                             std::shared_ptr<local_mr> local_mr,
                                             uint64_t add);
 
  [[nodiscard]] send_awaitable
  compare_and_swap(remote_mr const &remote_mr,
                   std::shared_ptr<local_mr> local_mr, uint64_t compare,
                   uint64_t swap);
 
  [[nodiscard]] recv_awaitable recv(std::shared_ptr<local_mr> local_mr);
 
  std::vector<uint8_t> serialize() const;
 
  std::vector<uint8_t> &user_data();
 
  std::shared_ptr<pd> pd_ptr() const;
  ~qp();
 
  void rtr(uint16_t remote_lid, uint32_t remote_qpn, uint32_t remote_psn,
           union ibv_gid remote_gid);
 
  void rts();
 
private:
  void post_recv_rq(struct ibv_recv_wr const &recv_wr,
                    struct ibv_recv_wr *&bad_recv_wr) const;
 
  void post_recv_srq(struct ibv_recv_wr const &recv_wr,
                     struct ibv_recv_wr *&bad_recv_wr) const;
};
 
} // namespace rdmapp