[ceph.git] / ceph / src / msg / async / rdma / RDMAStack.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2016 XSKY <haomai@xsky.com>
 *
 * Author: Haomai Wang <haomaiwang@gmail.com>
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation.  See file COPYING.
 *
 */

#ifndef CEPH_MSG_RDMASTACK_H
#define CEPH_MSG_RDMASTACK_H

#include <sys/eventfd.h>

#include <list>
#include <vector>
#include <thread>

#include "common/ceph_context.h"
#include "common/debug.h"
#include "common/errno.h"
#include "msg/async/Stack.h"
#include "Infiniband.h"

class RDMAConnectedSocketImpl;
class RDMAServerSocketImpl;
class RDMAStack;
class RDMAWorker;

class RDMADispatcher {
  typedef Infiniband::MemoryManager::Chunk Chunk;
  typedef Infiniband::QueuePair QueuePair;

  std::thread t;
  CephContext *cct;
  std::shared_ptr<Infiniband> ib;
  Infiniband::CompletionQueue* tx_cq = nullptr;
  Infiniband::CompletionQueue* rx_cq = nullptr;
  Infiniband::CompletionChannel *tx_cc = nullptr, *rx_cc = nullptr;
  bool done = false;
  std::atomic<uint64_t> num_qp_conn = {0};
  // protect `qp_conns`, `dead_queue_pairs`
  ceph::mutex lock = ceph::make_mutex("RDMADispatcher::lock");
  // qp_num -> InfRcConnection
  // The main usage of `qp_conns` is looking up connection by qp_num,
  // so the lifecycle of element in `qp_conns` is the lifecycle of qp.
  //// make qp queue into dead state
  /**
   * 1. Connection call mark_down
   * 2. Move the Queue Pair into the Error state(QueuePair::to_dead)
   * 3. Post a beacon
   * 4. Wait for beacon which indicates queues are drained
   * 5. Destroy the QP by calling ibv_destroy_qp()
   *
   * @param qp The qp needed to dead
   */
  ceph::unordered_map<uint32_t, std::pair<QueuePair*, RDMAConnectedSocketImpl*> > qp_conns;

  /// if a queue pair is closed when transmit buffers are active
  /// on it, the transmit buffers never get returned via tx_cq.  To
  /// work around this problem, don't delete queue pairs immediately. Instead,
  /// save them in this vector and delete them at a safe time, when there are
  /// no outstanding transmit buffers to be lost.
  std::vector<QueuePair*> dead_queue_pairs;

  std::atomic<uint64_t> num_pending_workers = {0};
  // protect pending workers
  ceph::mutex w_lock =
    ceph::make_mutex("RDMADispatcher::for worker pending list");
  // fixme: lockfree
  std::list<RDMAWorker*> pending_workers;
  void enqueue_dead_qp_lockless(uint32_t qp);
  void enqueue_dead_qp(uint32_t qpn);

 public:
  PerfCounters *perf_logger;

  explicit RDMADispatcher(CephContext* c, std::shared_ptr<Infiniband>& ib);
  virtual ~RDMADispatcher();
  void handle_async_event();

  void polling_start();
  void polling_stop();
  void polling();
  void register_qp(QueuePair *qp, RDMAConnectedSocketImpl* csi);
  void make_pending_worker(RDMAWorker* w) {
    std::lock_guard l{w_lock};
    auto it = std::find(pending_workers.begin(), pending_workers.end(), w);
    if (it != pending_workers.end())
      return;
    pending_workers.push_back(w);
    ++num_pending_workers;
  }
  RDMAConnectedSocketImpl* get_conn_lockless(uint32_t qp);
  QueuePair* get_qp_lockless(uint32_t qp);
  QueuePair* get_qp(uint32_t qp);
  void schedule_qp_destroy(uint32_t qp);
  Infiniband::CompletionQueue* get_tx_cq() const { return tx_cq; }
  Infiniband::CompletionQueue* get_rx_cq() const { return rx_cq; }
  void notify_pending_workers();
  void handle_tx_event(ibv_wc *cqe, int n);
  void post_tx_buffer(std::vector<Chunk*> &chunks);
  void handle_rx_event(ibv_wc *cqe, int rx_number);

  std::atomic<uint64_t> inflight = {0};

  void post_chunk_to_pool(Chunk* chunk);
  int post_chunks_to_rq(int num, QueuePair *qp = nullptr);
};

class RDMAWorker : public Worker {
  typedef Infiniband::CompletionQueue CompletionQueue;
  typedef Infiniband::CompletionChannel CompletionChannel;
  typedef Infiniband::MemoryManager::Chunk Chunk;
  typedef Infiniband::MemoryManager MemoryManager;
  typedef std::vector<Chunk*>::iterator ChunkIter;
  std::shared_ptr<Infiniband> ib;
  EventCallbackRef tx_handler;
  std::list<RDMAConnectedSocketImpl*> pending_sent_conns;
  std::shared_ptr<RDMADispatcher> dispatcher;
  ceph::mutex lock = ceph::make_mutex("RDMAWorker::lock");

  class C_handle_cq_tx : public EventCallback {
    RDMAWorker *worker;
    public:
    explicit C_handle_cq_tx(RDMAWorker *w): worker(w) {}
    void do_request(uint64_t fd) {
      worker->handle_pending_message();
    }
  };

 public:
  PerfCounters *perf_logger;
  explicit RDMAWorker(CephContext *c, unsigned i);
  virtual ~RDMAWorker();
  virtual int listen(entity_addr_t &addr,
		     unsigned addr_slot,
		     const SocketOptions &opts, ServerSocket *) override;
  virtual int connect(const entity_addr_t &addr, const SocketOptions &opts, ConnectedSocket *socket) override;
  virtual void initialize() override;
  int get_reged_mem(RDMAConnectedSocketImpl *o, std::vector<Chunk*> &c, size_t bytes);
  void remove_pending_conn(RDMAConnectedSocketImpl *o) {
    ceph_assert(center.in_thread());
    pending_sent_conns.remove(o);
  }
  void handle_pending_message();
  void set_dispatcher(std::shared_ptr<RDMADispatcher>& dispatcher) { this->dispatcher = dispatcher; }
  void set_ib(std::shared_ptr<Infiniband> &ib) {this->ib = ib;}
  void notify_worker() {
    center.dispatch_event_external(tx_handler);
  }
};

struct RDMACMInfo {
  RDMACMInfo(rdma_cm_id *cid, rdma_event_channel *cm_channel_, uint32_t qp_num_)
    : cm_id(cid), cm_channel(cm_channel_), qp_num(qp_num_) {}
  rdma_cm_id *cm_id;
  rdma_event_channel *cm_channel;
  uint32_t qp_num;
};

class RDMAConnectedSocketImpl : public ConnectedSocketImpl {
 public:
  typedef Infiniband::MemoryManager::Chunk Chunk;
  typedef Infiniband::CompletionChannel CompletionChannel;
  typedef Infiniband::CompletionQueue CompletionQueue;

 protected:
  CephContext *cct;
  Infiniband::QueuePair *qp;
  uint32_t peer_qpn = 0;
  uint32_t local_qpn = 0;
  int connected;
  int error;
  std::shared_ptr<Infiniband> ib;
  std::shared_ptr<RDMADispatcher> dispatcher;
  RDMAWorker* worker;
  std::vector<Chunk*> buffers;
  int notify_fd = -1;
  ceph::buffer::list pending_bl;

  ceph::mutex lock = ceph::make_mutex("RDMAConnectedSocketImpl::lock");
  std::vector<ibv_wc> wc;
  bool is_server;
  EventCallbackRef read_handler;
  EventCallbackRef established_handler;
  int tcp_fd = -1;
  bool active;// qp is active ?
  bool pending;
  int post_backlog = 0;

  void notify();
  void buffer_prefetch(void);
  ssize_t read_buffers(char* buf, size_t len);
  int post_work_request(std::vector<Chunk*>&);
  size_t tx_copy_chunk(std::vector<Chunk*> &tx_buffers, size_t req_copy_len,
      decltype(std::cbegin(pending_bl.buffers()))& start,
      const decltype(std::cbegin(pending_bl.buffers()))& end);

 public:
  RDMAConnectedSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
			  std::shared_ptr<RDMADispatcher>& rdma_dispatcher, RDMAWorker *w);
  virtual ~RDMAConnectedSocketImpl();

  void pass_wc(std::vector<ibv_wc> &&v);
  void get_wc(std::vector<ibv_wc> &w);
  virtual int is_connected() override { return connected; }

  virtual ssize_t read(char* buf, size_t len) override;
  virtual ssize_t send(ceph::buffer::list &bl, bool more) override;
  virtual void shutdown() override;
  virtual void close() override;
  virtual int fd() const override { return notify_fd; }
  virtual void set_priority(int sd, int prio, int domain) override;
  void fault();
  const char* get_qp_state() { return Infiniband::qp_state_string(qp->get_state()); }
  uint32_t get_peer_qpn () const { return peer_qpn; }
  uint32_t get_local_qpn () const { return local_qpn; }
  Infiniband::QueuePair* get_qp () const { return qp; }
  ssize_t submit(bool more);
  int activate();
  void fin();
  void handle_connection();
  int handle_connection_established(bool need_set_fault = true);
  void cleanup();
  void set_accept_fd(int sd);
  virtual int try_connect(const entity_addr_t&, const SocketOptions &opt);
  bool is_pending() {return pending;}
  void set_pending(bool val) {pending = val;}
  void post_chunks_to_rq(int num);
  void update_post_backlog();
};

enum RDMA_CM_STATUS {
  IDLE = 1,
  RDMA_ID_CREATED,
  CHANNEL_FD_CREATED,
  RESOURCE_ALLOCATED,
  ADDR_RESOLVED,
  ROUTE_RESOLVED,
  CONNECTED,
  DISCONNECTED,
  ERROR
};

class RDMAIWARPConnectedSocketImpl : public RDMAConnectedSocketImpl {
  public:
  RDMAIWARPConnectedSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
			       std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
			       RDMAWorker *w, RDMACMInfo *info = nullptr);
    ~RDMAIWARPConnectedSocketImpl();
    virtual int try_connect(const entity_addr_t&, const SocketOptions &opt) override;
    virtual void close() override;
    virtual void shutdown() override;
    virtual void handle_cm_connection();
    void activate();
    int alloc_resource();
    void close_notify();

  private:
    rdma_cm_id *cm_id = nullptr;
    rdma_event_channel *cm_channel = nullptr;
    EventCallbackRef cm_con_handler;
    std::mutex close_mtx;
    std::condition_variable close_condition;
    bool closed = false;
    RDMA_CM_STATUS status = IDLE;


  class C_handle_cm_connection : public EventCallback {
    RDMAIWARPConnectedSocketImpl *csi;
    public:
      C_handle_cm_connection(RDMAIWARPConnectedSocketImpl *w): csi(w) {}
      void do_request(uint64_t fd) {
        csi->handle_cm_connection();
      }
  };
};

class RDMAServerSocketImpl : public ServerSocketImpl {
  protected:
    CephContext *cct;
    ceph::NetHandler net;
    int server_setup_socket;
    std::shared_ptr<Infiniband> ib;
    std::shared_ptr<RDMADispatcher> dispatcher;
    RDMAWorker *worker;
    entity_addr_t sa;

 public:
  RDMAServerSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
                       std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
		       RDMAWorker *w, entity_addr_t& a, unsigned slot);

  virtual int listen(entity_addr_t &sa, const SocketOptions &opt);
  virtual int accept(ConnectedSocket *s, const SocketOptions &opts, entity_addr_t *out, Worker *w) override;
  virtual void abort_accept() override;
  virtual int fd() const override { return server_setup_socket; }
};

class RDMAIWARPServerSocketImpl : public RDMAServerSocketImpl {
  public:
    RDMAIWARPServerSocketImpl(
      CephContext *cct, std::shared_ptr<Infiniband>& ib,
      std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
      RDMAWorker* w, entity_addr_t& addr, unsigned addr_slot);
    virtual int listen(entity_addr_t &sa, const SocketOptions &opt) override;
    virtual int accept(ConnectedSocket *s, const SocketOptions &opts, entity_addr_t *out, Worker *w) override;
    virtual void abort_accept() override;
  private:
    rdma_cm_id *cm_id = nullptr;
    rdma_event_channel *cm_channel = nullptr;
};

class RDMAStack : public NetworkStack {
  std::vector<std::thread> threads;
  PerfCounters *perf_counter;
  std::shared_ptr<Infiniband> ib;
  std::shared_ptr<RDMADispatcher> rdma_dispatcher;

  std::atomic<bool> fork_finished = {false};

  virtual Worker* create_worker(CephContext *c, unsigned worker_id) override;

 public:
  explicit RDMAStack(CephContext *cct);
  virtual ~RDMAStack();
  virtual bool nonblock_connect_need_writable_event() const override { return false; }

  virtual void spawn_worker(std::function<void ()> &&func) override;
  virtual void join_worker(unsigned i) override;
  virtual bool is_ready() override { return fork_finished.load(); };
  virtual void ready() override { fork_finished = true; };
};


#endif
Commit	Line	Data
7c673cae FG	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab
	3	/*
	4	* Ceph - scalable distributed file system
	5	*
	6	* Copyright (C) 2016 XSKY <haomai@xsky.com>
	7	*
	8	* Author: Haomai Wang <haomaiwang@gmail.com>
	9	*
	10	* This is free software; you can redistribute it and/or
	11	* modify it under the terms of the GNU Lesser General Public
	12	* License version 2.1, as published by the Free Software
	13	* Foundation. See file COPYING.
	14	*
	15	*/
	16
	17	#ifndef CEPH_MSG_RDMASTACK_H
	18	#define CEPH_MSG_RDMASTACK_H
	19
	20	#include <sys/eventfd.h>
	21
	22	#include <list>
	23	#include <vector>
	24	#include <thread>
	25
	26	#include "common/ceph_context.h"
	27	#include "common/debug.h"
	28	#include "common/errno.h"
	29	#include "msg/async/Stack.h"
	30	#include "Infiniband.h"
7c673cae FG	31
	32	class RDMAConnectedSocketImpl;
	33	class RDMAServerSocketImpl;
	34	class RDMAStack;
	35	class RDMAWorker;
	36
7c673cae FG	37	class RDMADispatcher {
	38	typedef Infiniband::MemoryManager::Chunk Chunk;
	39	typedef Infiniband::QueuePair QueuePair;
	40
	41	std::thread t;
	42	CephContext *cct;
f67539c2	43	std::shared_ptr<Infiniband> ib;
11fdf7f2 TL	44	Infiniband::CompletionQueue* tx_cq = nullptr;
	45	Infiniband::CompletionQueue* rx_cq = nullptr;
	46	Infiniband::CompletionChannel tx_cc = nullptr, rx_cc = nullptr;
7c673cae	47	bool done = false;
7c673cae	48	std::atomic<uint64_t> num_qp_conn = {0};
9f95a23c TL	49	// protect `qp_conns`, `dead_queue_pairs`
9f95a23c TL	50	ceph::mutex lock = ceph::make_mutex("RDMADispatcher::lock");
7c673cae FG	51	// qp_num -> InfRcConnection
	52	// The main usage of `qp_conns` is looking up connection by qp_num,
	53	// so the lifecycle of element in `qp_conns` is the lifecycle of qp.
	54	//// make qp queue into dead state
	55	/**
	56	* 1. Connection call mark_down
	57	* 2. Move the Queue Pair into the Error state(QueuePair::to_dead)
9f95a23c TL	58	* 3. Post a beacon
	59	* 4. Wait for beacon which indicates queues are drained
	60	* 5. Destroy the QP by calling ibv_destroy_qp()
7c673cae FG	61	*
	62	* @param qp The qp needed to dead
	63	*/
	64	ceph::unordered_map<uint32_t, std::pair<QueuePair, RDMAConnectedSocketImpl> > qp_conns;
	65
	66	/// if a queue pair is closed when transmit buffers are active
	67	/// on it, the transmit buffers never get returned via tx_cq. To
	68	/// work around this problem, don't delete queue pairs immediately. Instead,
	69	/// save them in this vector and delete them at a safe time, when there are
	70	/// no outstanding transmit buffers to be lost.
	71	std::vector<QueuePair*> dead_queue_pairs;
	72
	73	std::atomic<uint64_t> num_pending_workers = {0};
9f95a23c TL	74	// protect pending workers
	75	ceph::mutex w_lock =
	76	ceph::make_mutex("RDMADispatcher::for worker pending list");
7c673cae FG	77	// fixme: lockfree
7c673cae FG	78	std::list<RDMAWorker*> pending_workers;
f67539c2 TL	79	void enqueue_dead_qp_lockless(uint32_t qp);
f67539c2 TL	80	void enqueue_dead_qp(uint32_t qpn);
31f18b77	81
7c673cae FG	82	public:
	83	PerfCounters *perf_logger;
	84
f67539c2	85	explicit RDMADispatcher(CephContext* c, std::shared_ptr<Infiniband>& ib);
7c673cae	86	virtual ~RDMADispatcher();
31f18b77	87	void handle_async_event();
7c673cae FG	88
	89	void polling_start();
	90	void polling_stop();
	91	void polling();
11fdf7f2	92	void register_qp(QueuePair qp, RDMAConnectedSocketImpl csi);
7c673cae	93	void make_pending_worker(RDMAWorker* w) {
9f95a23c	94	std::lock_guard l{w_lock};
224ce89b WB	95	auto it = std::find(pending_workers.begin(), pending_workers.end(), w);
	96	if (it != pending_workers.end())
	97	return;
	98	pending_workers.push_back(w);
	99	++num_pending_workers;
7c673cae	100	}
7c673cae	101	RDMAConnectedSocketImpl* get_conn_lockless(uint32_t qp);
9f95a23c	102	QueuePair* get_qp_lockless(uint32_t qp);
11fdf7f2	103	QueuePair* get_qp(uint32_t qp);
9f95a23c	104	void schedule_qp_destroy(uint32_t qp);
31f18b77 FG	105	Infiniband::CompletionQueue* get_tx_cq() const { return tx_cq; }
31f18b77 FG	106	Infiniband::CompletionQueue* get_rx_cq() const { return rx_cq; }
7c673cae	107	void notify_pending_workers();
31f18b77 FG	108	void handle_tx_event(ibv_wc *cqe, int n);
31f18b77 FG	109	void post_tx_buffer(std::vector<Chunk*> &chunks);
9f95a23c	110	void handle_rx_event(ibv_wc *cqe, int rx_number);
7c673cae FG	111
7c673cae FG	112	std::atomic<uint64_t> inflight = {0};
7c673cae	113
11fdf7f2	114	void post_chunk_to_pool(Chunk* chunk);
9f95a23c	115	int post_chunks_to_rq(int num, QueuePair *qp = nullptr);
7c673cae FG	116	};
	117
	118	class RDMAWorker : public Worker {
	119	typedef Infiniband::CompletionQueue CompletionQueue;
	120	typedef Infiniband::CompletionChannel CompletionChannel;
	121	typedef Infiniband::MemoryManager::Chunk Chunk;
	122	typedef Infiniband::MemoryManager MemoryManager;
	123	typedef std::vector<Chunk*>::iterator ChunkIter;
f67539c2	124	std::shared_ptr<Infiniband> ib;
7c673cae FG	125	EventCallbackRef tx_handler;
7c673cae FG	126	std::list<RDMAConnectedSocketImpl*> pending_sent_conns;
f67539c2	127	std::shared_ptr<RDMADispatcher> dispatcher;
9f95a23c	128	ceph::mutex lock = ceph::make_mutex("RDMAWorker::lock");
7c673cae FG	129
	130	class C_handle_cq_tx : public EventCallback {
	131	RDMAWorker *worker;
	132	public:
11fdf7f2 TL	133	explicit C_handle_cq_tx(RDMAWorker *w): worker(w) {}
11fdf7f2 TL	134	void do_request(uint64_t fd) {
7c673cae FG	135	worker->handle_pending_message();
	136	}
	137	};
	138
	139	public:
	140	PerfCounters *perf_logger;
	141	explicit RDMAWorker(CephContext *c, unsigned i);
	142	virtual ~RDMAWorker();
11fdf7f2 TL	143	virtual int listen(entity_addr_t &addr,
	144	unsigned addr_slot,
	145	const SocketOptions &opts, ServerSocket *) override;
7c673cae FG	146	virtual int connect(const entity_addr_t &addr, const SocketOptions &opts, ConnectedSocket *socket) override;
7c673cae FG	147	virtual void initialize() override;
7c673cae FG	148	int get_reged_mem(RDMAConnectedSocketImpl o, std::vector<Chunk> &c, size_t bytes);
7c673cae FG	149	void remove_pending_conn(RDMAConnectedSocketImpl *o) {
11fdf7f2	150	ceph_assert(center.in_thread());
7c673cae FG	151	pending_sent_conns.remove(o);
	152	}
	153	void handle_pending_message();
f67539c2 TL	154	void set_dispatcher(std::shared_ptr<RDMADispatcher>& dispatcher) { this->dispatcher = dispatcher; }
f67539c2 TL	155	void set_ib(std::shared_ptr<Infiniband> &ib) {this->ib = ib;}
7c673cae FG	156	void notify_worker() {
	157	center.dispatch_event_external(tx_handler);
	158	}
	159	};
	160
11fdf7f2 TL	161	struct RDMACMInfo {
	162	RDMACMInfo(rdma_cm_id cid, rdma_event_channel cm_channel_, uint32_t qp_num_)
	163	: cm_id(cid), cm_channel(cm_channel_), qp_num(qp_num_) {}
	164	rdma_cm_id *cm_id;
	165	rdma_event_channel *cm_channel;
	166	uint32_t qp_num;
	167	};
	168
31f18b77 FG	169	class RDMAConnectedSocketImpl : public ConnectedSocketImpl {
	170	public:
	171	typedef Infiniband::MemoryManager::Chunk Chunk;
	172	typedef Infiniband::CompletionChannel CompletionChannel;
	173	typedef Infiniband::CompletionQueue CompletionQueue;
	174
11fdf7f2	175	protected:
31f18b77 FG	176	CephContext *cct;
31f18b77 FG	177	Infiniband::QueuePair *qp;
9f95a23c TL	178	uint32_t peer_qpn = 0;
9f95a23c TL	179	uint32_t local_qpn = 0;
31f18b77 FG	180	int connected;
31f18b77 FG	181	int error;
f67539c2 TL	182	std::shared_ptr<Infiniband> ib;
f67539c2 TL	183	std::shared_ptr<RDMADispatcher> dispatcher;
31f18b77 FG	184	RDMAWorker* worker;
	185	std::vector<Chunk*> buffers;
	186	int notify_fd = -1;
f67539c2	187	ceph::buffer::list pending_bl;
31f18b77	188
9f95a23c	189	ceph::mutex lock = ceph::make_mutex("RDMAConnectedSocketImpl::lock");
31f18b77 FG	190	std::vector<ibv_wc> wc;
31f18b77 FG	191	bool is_server;
9f95a23c TL	192	EventCallbackRef read_handler;
9f95a23c TL	193	EventCallbackRef established_handler;
31f18b77 FG	194	int tcp_fd = -1;
31f18b77 FG	195	bool active;// qp is active ?
224ce89b	196	bool pending;
11fdf7f2	197	int post_backlog = 0;
31f18b77 FG	198
31f18b77 FG	199	void notify();
9f95a23c	200	void buffer_prefetch(void);
31f18b77 FG	201	ssize_t read_buffers(char* buf, size_t len);
31f18b77 FG	202	int post_work_request(std::vector<Chunk*>&);
9f95a23c TL	203	size_t tx_copy_chunk(std::vector<Chunk*> &tx_buffers, size_t req_copy_len,
	204	decltype(std::cbegin(pending_bl.buffers()))& start,
	205	const decltype(std::cbegin(pending_bl.buffers()))& end);
31f18b77 FG	206
31f18b77 FG	207	public:
f67539c2 TL	208	RDMAConnectedSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
f67539c2 TL	209	std::shared_ptr<RDMADispatcher>& rdma_dispatcher, RDMAWorker *w);
31f18b77 FG	210	virtual ~RDMAConnectedSocketImpl();
	211
	212	void pass_wc(std::vector<ibv_wc> &&v);
	213	void get_wc(std::vector<ibv_wc> &w);
	214	virtual int is_connected() override { return connected; }
	215
	216	virtual ssize_t read(char* buf, size_t len) override;
f67539c2	217	virtual ssize_t send(ceph::buffer::list &bl, bool more) override;
31f18b77 FG	218	virtual void shutdown() override;
	219	virtual void close() override;
	220	virtual int fd() const override { return notify_fd; }
1e59de90	221	virtual void set_priority(int sd, int prio, int domain) override;
31f18b77 FG	222	void fault();
31f18b77 FG	223	const char* get_qp_state() { return Infiniband::qp_state_string(qp->get_state()); }
9f95a23c TL	224	uint32_t get_peer_qpn () const { return peer_qpn; }
9f95a23c TL	225	uint32_t get_local_qpn () const { return local_qpn; }
f67539c2	226	Infiniband::QueuePair* get_qp () const { return qp; }
31f18b77 FG	227	ssize_t submit(bool more);
	228	int activate();
	229	void fin();
	230	void handle_connection();
9f95a23c	231	int handle_connection_established(bool need_set_fault = true);
31f18b77 FG	232	void cleanup();
31f18b77 FG	233	void set_accept_fd(int sd);
11fdf7f2	234	virtual int try_connect(const entity_addr_t&, const SocketOptions &opt);
224ce89b WB	235	bool is_pending() {return pending;}
224ce89b WB	236	void set_pending(bool val) {pending = val;}
11fdf7f2 TL	237	void post_chunks_to_rq(int num);
11fdf7f2 TL	238	void update_post_backlog();
31f18b77 FG	239	};
31f18b77 FG	240
11fdf7f2 TL	241	enum RDMA_CM_STATUS {
	242	IDLE = 1,
	243	RDMA_ID_CREATED,
	244	CHANNEL_FD_CREATED,
	245	RESOURCE_ALLOCATED,
	246	ADDR_RESOLVED,
	247	ROUTE_RESOLVED,
	248	CONNECTED,
	249	DISCONNECTED,
	250	ERROR
	251	};
	252
	253	class RDMAIWARPConnectedSocketImpl : public RDMAConnectedSocketImpl {
	254	public:
f67539c2 TL	255	RDMAIWARPConnectedSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
	256	std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
	257	RDMAWorker w, RDMACMInfo info = nullptr);
11fdf7f2 TL	258	~RDMAIWARPConnectedSocketImpl();
	259	virtual int try_connect(const entity_addr_t&, const SocketOptions &opt) override;
	260	virtual void close() override;
	261	virtual void shutdown() override;
	262	virtual void handle_cm_connection();
11fdf7f2 TL	263	void activate();
	264	int alloc_resource();
	265	void close_notify();
	266
	267	private:
9f95a23c TL	268	rdma_cm_id *cm_id = nullptr;
9f95a23c TL	269	rdma_event_channel *cm_channel = nullptr;
11fdf7f2	270	EventCallbackRef cm_con_handler;
11fdf7f2 TL	271	std::mutex close_mtx;
11fdf7f2 TL	272	std::condition_variable close_condition;
9f95a23c TL	273	bool closed = false;
9f95a23c TL	274	RDMA_CM_STATUS status = IDLE;
11fdf7f2 TL	275
	276
	277	class C_handle_cm_connection : public EventCallback {
	278	RDMAIWARPConnectedSocketImpl *csi;
	279	public:
	280	C_handle_cm_connection(RDMAIWARPConnectedSocketImpl *w): csi(w) {}
	281	void do_request(uint64_t fd) {
	282	csi->handle_cm_connection();
	283	}
	284	};
	285	};
	286
31f18b77	287	class RDMAServerSocketImpl : public ServerSocketImpl {
11fdf7f2 TL	288	protected:
11fdf7f2 TL	289	CephContext *cct;
f67539c2	290	ceph::NetHandler net;
11fdf7f2	291	int server_setup_socket;
f67539c2 TL	292	std::shared_ptr<Infiniband> ib;
f67539c2 TL	293	std::shared_ptr<RDMADispatcher> dispatcher;
11fdf7f2 TL	294	RDMAWorker *worker;
11fdf7f2 TL	295	entity_addr_t sa;
31f18b77 FG	296
31f18b77 FG	297	public:
f67539c2 TL	298	RDMAServerSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
f67539c2 TL	299	std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
11fdf7f2	300	RDMAWorker *w, entity_addr_t& a, unsigned slot);
31f18b77	301
11fdf7f2	302	virtual int listen(entity_addr_t &sa, const SocketOptions &opt);
31f18b77 FG	303	virtual int accept(ConnectedSocket s, const SocketOptions &opts, entity_addr_t out, Worker *w) override;
	304	virtual void abort_accept() override;
	305	virtual int fd() const override { return server_setup_socket; }
31f18b77	306	};
7c673cae	307
11fdf7f2 TL	308	class RDMAIWARPServerSocketImpl : public RDMAServerSocketImpl {
	309	public:
	310	RDMAIWARPServerSocketImpl(
f67539c2 TL	311	CephContext *cct, std::shared_ptr<Infiniband>& ib,
f67539c2 TL	312	std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
9f95a23c	313	RDMAWorker* w, entity_addr_t& addr, unsigned addr_slot);
11fdf7f2 TL	314	virtual int listen(entity_addr_t &sa, const SocketOptions &opt) override;
	315	virtual int accept(ConnectedSocket s, const SocketOptions &opts, entity_addr_t out, Worker *w) override;
	316	virtual void abort_accept() override;
	317	private:
9f95a23c TL	318	rdma_cm_id *cm_id = nullptr;
9f95a23c TL	319	rdma_event_channel *cm_channel = nullptr;
11fdf7f2 TL	320	};
11fdf7f2 TL	321
7c673cae	322	class RDMAStack : public NetworkStack {
f67539c2	323	std::vector<std::thread> threads;
11fdf7f2	324	PerfCounters *perf_counter;
f67539c2 TL	325	std::shared_ptr<Infiniband> ib;
f67539c2 TL	326	std::shared_ptr<RDMADispatcher> rdma_dispatcher;
7c673cae FG	327
	328	std::atomic<bool> fork_finished = {false};
	329
20effc67	330	virtual Worker* create_worker(CephContext *c, unsigned worker_id) override;
f67539c2	331
7c673cae	332	public:
f67539c2	333	explicit RDMAStack(CephContext *cct);
7c673cae	334	virtual ~RDMAStack();
11fdf7f2	335	virtual bool nonblock_connect_need_writable_event() const override { return false; }
7c673cae	336
20effc67	337	virtual void spawn_worker(std::function<void ()> &&func) override;
7c673cae	338	virtual void join_worker(unsigned i) override;
7c673cae FG	339	virtual bool is_ready() override { return fork_finished.load(); };
	340	virtual void ready() override { fork_finished = true; };
	341	};
31f18b77	342
11fdf7f2	343
7c673cae	344	#endif