refactor

Author: yeshengm

CMakeLists.txt

@@ -74,6 +74,7 @@ set(pkg_config_names_private "")
 
 option(WITH_OPENPGM "Build with support for OpenPGM" OFF)
 option(WITH_VMCI "Build with support for VMware VMCI socket" OFF)
+option(WITH_RDMA "Build with RDMA support" OFF)
 
 if(APPLE)
   option(ZMQ_BUILD_FRAMEWORK "Build as OS X framework" OFF)
@@ -634,7 +635,6 @@ set(cxx-sources
   epoll.cpp
   err.cpp
   fq.cpp
-  ib_res.cpp
   io_object.cpp
   io_thread.cpp
   ip.cpp
@@ -685,10 +685,6 @@ set(cxx-sources
   socks_connecter.cpp
   stream.cpp
   stream_engine.cpp
-  rdma_address.cpp
-  rdma_connecter.cpp
-  rdma_engine.cpp
-  rdma_listener.cpp
   sub.cpp
   tcp.cpp
   tcp_address.cpp
@@ -757,7 +753,6 @@ set(cxx-sources
   i_engine.hpp
   i_mailbox.hpp
   i_poll_events.hpp
-  ib_conf.hpp
   ib_res.hpp
   io_object.hpp
   io_thread.hpp
@@ -804,10 +799,6 @@ set(cxx-sources
   random.hpp
   raw_decoder.hpp
   raw_encoder.hpp
-  rdma_address.hpp
-  rdma_connecter.hpp
-  rdma_engine.hpp
-  rdma_listener.hpp
   reaper.hpp
   rep.hpp
   req.hpp
@@ -918,6 +909,14 @@ if(ZMQ_HAVE_TIPC)
   list(APPEND cxx-sources tipc_address.cpp tipc_connecter.cpp tipc_listener.cpp)
 endif()
 
+if(WITH_RDMA)
+  add_definitions(-DZMQ_HAVE_RDMA)
+  list(APPEND cxx-sources   rdma_address.hpp
+          rdma_connecter.hpp
+          rdma_engine.hpp
+          rdma_listener.hpp rdma_address.cpp rdma_connecter.cpp rdma_engine.cpp rdma_listener.cpp)
+endif()
+
 #-----------------------------------------------------------------------------
 # source generators
 

include/zmq.h

@@ -219,7 +219,8 @@ ZMQ_EXPORT void zmq_version (int *major_, int *minor_, int *patch_);
 #define ZMQ_THREAD_AFFINITY_CPU_ADD 7
 #define ZMQ_THREAD_AFFINITY_CPU_REMOVE 8
 #define ZMQ_THREAD_NAME_PREFIX 9
-#define ZMQ_ENABLE_RDMA 10
+#define ZMQ_IB_NUM_QPS 10
+#define ZMQ_IB_BUF_SIZE 11
 
 /*  Default for new contexts                                                  */
 #define ZMQ_IO_THREADS_DFLT 1

src/address.cpp

@@ -83,8 +83,10 @@ int zmq::address_t::to_string (std::string &addr_) const
 {
     if (protocol == protocol_name::tcp && resolved.tcp_addr)
         return resolved.tcp_addr->to_string (addr_);
+#ifdef ZMQ_HAVE_RDMA
     if (protocol == protocol_name::rdma && resolved.rdma_addr)
         return resolved.rdma_addr->to_string(addr_);
+#endif
     if (protocol == protocol_name::udp && resolved.udp_addr)
         return resolved.udp_addr->to_string (addr_);
 #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS                     \

src/address.hpp

@@ -31,13 +31,15 @@
 #define __ZMQ_ADDRESS_HPP_INCLUDED__
 
 #include <string>
-#include "rdma_address.hpp"
 
 namespace zmq
 {
 class ctx_t;
 class tcp_address_t;
 class udp_address_t;
+#ifdef ZMQ_HAVE_RDMA
+class rdma_address_t;
+#endif
 #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
 class ipc_address_t;
 #endif
@@ -84,7 +86,9 @@ struct address_t
     {
         void *dummy;
         tcp_address_t *tcp_addr;
+#ifdef ZMQ_HAVE_RDMA
         rdma_address_t *rdma_addr;
+#endif
         udp_address_t *udp_addr;
 #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS                     \
   && !defined ZMQ_HAVE_VXWORKS

src/ctx.cpp

@@ -85,7 +85,10 @@ zmq::ctx_t::ctx_t () :
     _vmci_fd = -1;
     _vmci_family = -1;
 #endif
-
+#ifdef ZMQ_HAVE_RDMA
+    _ib_num_qps = 1;
+    _ib_buf_size = 1024 * 1024;
+#endif
     //  Initialise crypto library, if needed.
     zmq::random_open ();
 }
@@ -219,6 +222,9 @@ int zmq::ctx_t::shutdown ()
             _sockets[i]->stop ();
         if (_sockets.empty ())
             _reaper->stop ();
+#ifdef ZMQ_HAVE_RDMA
+        close_ib();
+#endif
     }
 
     return 0;
@@ -246,8 +252,14 @@ int zmq::ctx_t::set (int option_, int optval_)
     } else if (option_ == ZMQ_ZERO_COPY_RECV && optval_ >= 0) {
         scoped_lock_t locker(_opt_sync);
         _zero_copy = (optval_ != 0);
-    } else if (option_ == ZMQ_ENABLE_RDMA && optval_ >= 0) {
-        _rdma_enabled = (optval_ != 0);
+#ifdef ZMQ_HAVE_RDMA
+    } else if (option_ == ZMQ_IB_NUM_QPS && optval_ >= 0) {
+        scoped_lock_t locker(_opt_sync);
+        _ib_num_qps = optval_;
+    } else if (option_ == ZMQ_IB_BUF_SIZE && optval_ >= 0) {
+        scoped_lock_t locker(_opt_sync);
+        _ib_buf_size = optval_;
+#endif
     } else {
         rc = thread_ctx_t::set (option_, optval_);
     }
@@ -271,13 +283,16 @@ int zmq::ctx_t::get (int option_)
         rc = _max_msgsz;
     else if (option_ == ZMQ_MSG_T_SIZE)
         rc = sizeof (zmq_msg_t);
-    else if (option_ == ZMQ_ZERO_COPY_RECV) {
+    else if (option_ == ZMQ_ZERO_COPY_RECV)
         rc = _zero_copy;
-    } else if (option_ == ZMQ_ENABLE_RDMA) {
-        rc = _rdma_enabled;
-    } else {
+#ifdef ZMQ_HAVE_RDMA
+    else if (option_ == ZMQ_IB_NUM_QPS)
+        rc = _ib_num_qps;
+    else if (option_ == ZMQ_IB_BUF_SIZE)
+        rc = _ib_buf_size;
+#endif
+    else
         rc = thread_ctx_t::get (option_);
-    }
     return rc;
 }
 
@@ -340,8 +355,10 @@ bool zmq::ctx_t::start ()
         _empty_slots.push_back (i);
     }
 
+#ifdef ZMQ_HAVE_RDMA
     // RDMA: init ib resources
-    setup_ib(_ib_res, _max_sockets, _max_msgsz);
+    setup_ib();
+#endif
 
     _starting = false;
     return true;
@@ -708,3 +725,105 @@ int zmq::ctx_t::get_vmci_socket_family ()
 //  is a global variable. Thus, even sockets created in different contexts have
 //  unique IDs.
 zmq::atomic_counter_t zmq::ctx_t::max_socket_id;
+
+#ifdef ZMQ_HAVE_RDMA
+#include <infiniband/verbs.h>
+
+int zmq::ctx_t::setup_ib() {
+    ibv_device **dev_list = NULL;
+    memset(&_ib_res, 0, sizeof(ib_res_t));
+
+    _ib_res.num_qps = get(ZMQ_IB_NUM_QPS);
+
+    dev_list =ibv_get_device_list(NULL);
+    assert(dev_list != NULL);
+
+    _ib_res.ctx = ibv_open_device(*dev_list);
+    assert(_ib_res.ctx != NULL);
+
+    _ib_res.pd = ibv_alloc_pd(_ib_res.ctx);
+    assert(_ib_res.pd != NULL);
+
+    int ret = ibv_query_port(_ib_res.ctx, 1, &_ib_res.port_attr);
+    assert(ret == 0);
+
+    _ib_res.ib_buf_size = get(ZMQ_IB_BUF_SIZE);
+    posix_memalign((void**)_ib_res.ib_buf, 4096, _ib_res.ib_buf_size);
+    assert(_ib_res.ib_buf != NULL);
+
+    _ib_res.mr = ibv_reg_mr (_ib_res.pd, (void*)_ib_res.ib_buf,
+                            _ib_res.ib_buf_size,
+                            IBV_ACCESS_LOCAL_WRITE|
+                            IBV_ACCESS_REMOTE_READ|
+                            IBV_ACCESS_REMOTE_WRITE);
+    assert(_ib_res.mr != NULL);
+
+    ret = ibv_query_device(_ib_res.ctx, &_ib_res.dev_attr);
+    assert(ret == 0);
+
+    _ib_res.cq = ibv_create_cq(_ib_res.ctx, _ib_res.dev_attr.max_cqe,
+                              NULL, NULL, 0);
+    assert(_ib_res.cq != NULL);
+
+    struct ibv_srq_init_attr srq_init_attr;
+
+    srq_init_attr.attr.max_wr = _ib_res.dev_attr.max_srq_wr;
+    srq_init_attr.attr.max_sge = 1;
+
+
+    _ib_res.srq = ibv_create_srq(_ib_res.pd, &srq_init_attr);
+
+    struct ibv_qp_init_attr qp_init_attr;
+
+    qp_init_attr.send_cq = _ib_res.cq;
+    qp_init_attr.recv_cq = _ib_res.cq;
+    qp_init_attr.srq = _ib_res.srq;
+    qp_init_attr.cap.max_send_wr = _ib_res.dev_attr.max_qp_wr;
+    qp_init_attr.cap.max_recv_wr = _ib_res.dev_attr.max_qp_wr;
+    qp_init_attr.cap.max_send_sge = 1;
+    qp_init_attr.cap.max_recv_sge = 1;
+    qp_init_attr.qp_type = IBV_QPT_RC;
+
+    _ib_res.qp = (struct ibv_qp **) calloc (_ib_res.num_qps,
+                                           sizeof(struct ibv_qp *));
+    assert(_ib_res.qp != NULL);
+
+    for (int i = 0; i < _ib_res.num_qps; i++) {
+        _ib_res.qp[i] = ibv_create_qp (_ib_res.pd, & qp_init_attr);
+        assert(_ib_res.qp[i] != NULL);
+    }
+
+    ibv_free_device_list (dev_list);
+    return 0;
+}
+
+void zmq::ctx_t::close_ib() {
+    if (_ib_res.qp != NULL) {
+        for (int i = 0; i < _ib_res.num_qps; i++) {
+            if (_ib_res.qp[i] != NULL) {
+                ibv_destroy_qp(_ib_res.qp[i]);
+            }
+        }
+        free(_ib_res.qp);
+    }
+
+    if (_ib_res.srq != NULL)
+        ibv_destroy_srq(_ib_res.srq);
+    if (_ib_res.cq != NULL) {
+        ibv_destroy_cq (_ib_res.cq);
+    }
+    if (_ib_res.mr != NULL) {
+        ibv_dereg_mr (_ib_res.mr);
+    }
+
+    if (_ib_res.pd != NULL) {
+        ibv_dealloc_pd (_ib_res.pd);
+    }
+    if (_ib_res.ctx != NULL) {
+        ibv_close_device (_ib_res.ctx);
+    }
+    if (_ib_res.ib_buf != NULL) {
+        free (_ib_res.ib_buf);
+    }
+}
+#endif

src/ctx.hpp

@@ -43,7 +43,9 @@
 #include "options.hpp"
 #include "atomic_counter.hpp"
 #include "thread.hpp"
+#ifdef ZMQ_HAVE_RDMA
 #include "ib_res.hpp"
+#endif
 
 namespace zmq
 {
@@ -238,6 +240,11 @@ class ctx_t : public thread_ctx_t
     // Should we use zero copy message decoding in this context?
     bool _zero_copy;
 
+#ifdef ZMQ_HAVE_RDMA
+    int _ib_num_qps;
+    int _ib_buf_size; // will it overflow?
+#endif
+
     ctx_t (const ctx_t &);
     const ctx_t &operator= (const ctx_t &);
 
@@ -262,9 +269,13 @@ class ctx_t : public thread_ctx_t
     mutex_t _vmci_sync;
 #endif
 
+#ifdef ZMQ_HAVE_RDMA
     // RDMA: Infiniband related resources
     ib_res_t _ib_res;
-    bool _rdma_enabled;
+    int setup_ib();
+    void close_ib();
+#endif
+
 };
 }