rpc-transport/rdma: reap async events in a dedicated thread.

* Also sets srq_limit of srq to 10, so that we'll receive an event
  when we are about to empty the receive buffer list.

BUG: 765051
Change-Id: I5436166ea21fc963ee15088fc2df743ec4b96ba7
Signed-off-by: Raghavendra G <raghavendra@gluster.com>
Reviewed-on: http://review.gluster.org/4378
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>
Reviewed-on: http://review.gluster.org/5097

2 files changed, 52 insertions, 2 deletions

diff --git a/rpc/rpc-transport/rdma/src/rdma.c b/rpc/rpc-transport/rdma/src/rdma.c
index 135fbdf2868..8ef7d1e3afa 100644
--- a/rpc/rpc-transport/rdma/src/rdma.c
+++ b/rpc/rpc-transport/rdma/src/rdma.c
@@ -40,6 +40,9 @@ gf_rdma_send_completion_proc (void *data);
 static void *
 gf_rdma_recv_completion_proc (void *data);
 
+void *
+gf_rdma_async_event_thread (void *context);
+
 static int32_t
 gf_rdma_create_qp (rpc_transport_t *this);
 
@@ -591,7 +594,8 @@ gf_rdma_get_device (rpc_transport_t *this, struct ibv_context *ibctx,
                 struct ibv_srq_init_attr attr = {
                         .attr = {
                                 .max_wr = options->recv_count,
-                                .max_sge = 1
+                                .max_sge = 1,
+                                .srq_limit = 10
                         }
                 };
                 trav->srq = ibv_create_srq (trav->pd, &attr);
@@ -637,6 +641,16 @@ gf_rdma_get_device (rpc_transport_t *this, struct ibv_context *ibctx,
                         return NULL;
                 }
 
+                ret = pthread_create (&trav->async_event_thread,
+                                      NULL,
+                                      gf_rdma_async_event_thread,
+                                      ibctx);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "could not create async_event_thread");
+                        return NULL;
+                }
+
                 /* qpreg */
                 pthread_mutex_init (&trav->qpreg.lock, NULL);
                 for (i=0; i<42; i++) {
@@ -3743,6 +3757,42 @@ out:
         return;
 }
 
+void *
+gf_rdma_async_event_thread (void *context)
+{
+        struct ibv_async_event event;
+        int ret;
+
+        while (1) {
+                do {
+                        ret = ibv_get_async_event((struct ibv_context *)context,
+                                                  &event);
+
+                        if (ret && errno != EINTR) {
+                                gf_log (GF_RDMA_LOG_NAME, GF_LOG_WARNING,
+                                        "Error getting event (%s)",
+                                        strerror (errno));
+                        }
+                } while(ret && errno == EINTR);
+
+                switch (event.event_type) {
+                case IBV_EVENT_SRQ_LIMIT_REACHED:
+                        gf_log (GF_RDMA_LOG_NAME, GF_LOG_WARNING,
+                                "recieved srq_limit reached");
+                        break;
+
+                default:
+                        gf_log (GF_RDMA_LOG_NAME, GF_LOG_DEBUG,
+                                "event (%d) recieved", event.event_type);
+                        break;
+                }
+
+                ibv_ack_async_event(&event);
+        }
+
+        return 0;
+}
+
 
 static void *
 gf_rdma_recv_completion_proc (void *data)
diff --git a/rpc/rpc-transport/rdma/src/rdma.h b/rpc/rpc-transport/rdma/src/rdma.h
index 6a2951b897d..7f76244f071 100644
--- a/rpc/rpc-transport/rdma/src/rdma.h
+++ b/rpc/rpc-transport/rdma/src/rdma.h
@@ -324,7 +324,7 @@ struct __gf_rdma_device {
         struct ibv_comp_channel *send_chan, *recv_chan;
         struct ibv_cq *send_cq, *recv_cq;
         gf_rdma_queue_t sendq, recvq;
-        pthread_t send_thread, recv_thread;
+        pthread_t send_thread, recv_thread, async_event_thread;
         struct mem_pool *request_ctx_pool;
         struct mem_pool *ioq_pool;
         struct mem_pool *reply_info_pool;