rpc-transport: don't merge payload iobuf and iobuf which holds program header into single iobref.

- io-cache holds a reference on iobref passed from transport layer. Hence,
   two iobufs are accounted instead of one in calculated used cache size.

Signed-off-by: Raghavendra G <raghavendra@gluster.com>
Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

BUG: 2135 ()
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=2135

5 files changed, 39 insertions, 36 deletions

diff --git a/rpc/rpc-lib/src/rpc-transport.c b/rpc/rpc-lib/src/rpc-transport.c
index 2a10a3b7f80..bf0e4dfd6e9 100644
--- a/rpc/rpc-lib/src/rpc-transport.c
+++ b/rpc/rpc-lib/src/rpc-transport.c
@@ -632,6 +632,10 @@ rpc_transport_pollin_destroy (rpc_transport_pollin_t *pollin)
                 iobref_unref (pollin->iobref);
         }
  
+        if (pollin->hdr_iobuf) {
+                iobuf_unref (pollin->hdr_iobuf);
+        }
+
         if (pollin->private) {
                 /* */
                 GF_FREE (pollin->private);
@@ -645,7 +649,8 @@ out:
 
 rpc_transport_pollin_t *
 rpc_transport_pollin_alloc (rpc_transport_t *this, struct iovec *vector,
-                            int count, struct iobref *iobref, void *private)
+                            int count, struct iobuf *hdr_iobuf,
+                            struct iobref *iobref, void *private)
 {
         rpc_transport_pollin_t *msg = NULL;
         msg = GF_CALLOC (1, sizeof (*msg), gf_common_mt_rpc_trans_pollin_t);
@@ -662,6 +667,7 @@ rpc_transport_pollin_alloc (rpc_transport_t *this, struct iovec *vector,
         msg->count = count;
         msg->iobref = iobref_ref (iobref);
         msg->private = private;
+        msg->hdr_iobuf = iobuf_ref (hdr_iobuf);
 
 out:
         return msg;
diff --git a/rpc/rpc-lib/src/rpc-transport.h b/rpc/rpc-lib/src/rpc-transport.h
index c5e7755ff8e..9db24c09f4e 100644
--- a/rpc/rpc-lib/src/rpc-transport.h
+++ b/rpc/rpc-lib/src/rpc-transport.h
@@ -168,6 +168,7 @@ struct rpc_transport_pollin {
         char vectored;
         void *private;
         struct iobref *iobref;
+        struct iobuf  *hdr_iobuf;
         char is_reply;
 };
 typedef struct rpc_transport_pollin rpc_transport_pollin_t;
@@ -289,7 +290,8 @@ rpc_transport_get_myaddr (rpc_transport_t *this, char *peeraddr, int addrlen,
 
 rpc_transport_pollin_t *
 rpc_transport_pollin_alloc (rpc_transport_t *this, struct iovec *vector,
-                            int count, struct iobref *iobref, void *private);
+                            int count, struct iobuf *hdr_iobuf,
+                            struct iobref *iobref, void *private);
 void
 rpc_transport_pollin_destroy (rpc_transport_pollin_t *pollin);
 
diff --git a/rpc/rpc-transport/rdma/src/rdma.c b/rpc/rpc-transport/rdma/src/rdma.c
index 22f40e826b3..81416d98fd0 100644
--- a/rpc/rpc-transport/rdma/src/rdma.c
+++ b/rpc/rpc-transport/rdma/src/rdma.c
@@ -825,6 +825,10 @@ rdma_post_context_destroy (rdma_post_context_t *ctx)
                 iobref_unref (ctx->iobref);
         }
 
+        if (ctx->hdr_iobuf != NULL) {
+                iobuf_unref (ctx->hdr_iobuf);
+        }
+
         memset (ctx, 0, sizeof (*ctx));
 out:
         return;
@@ -2726,8 +2730,6 @@ rdma_decode_msg (rdma_peer_t *peer, rdma_post_t *post,
         char               *ptr        = NULL;
         rdma_write_array_t *write_ary  = NULL;
         size_t              header_len = 0; 
-        struct iobuf       *iobuf      = NULL;
-        struct iobref      *iobref     = NULL;
 
         header = (rdma_header_t *)post->buf;
 
@@ -2781,32 +2783,19 @@ rdma_decode_msg (rdma_peer_t *peer, rdma_post_t *post,
         /* skip terminator of reply chunk */
         ptr = ptr + sizeof (uint32_t);
         if (header->rm_type != RDMA_NOMSG) {
-                iobuf = iobuf_get (peer->trans->ctx->iobuf_pool);
-                if (iobuf == NULL) {
-                        gf_log (RDMA_LOG_NAME, GF_LOG_ERROR, "out of memory");
-                        ret = -1;
-                        goto out;
-                }
-
-                post->ctx.iobref = iobref = iobref_new ();
-                if (iobref == NULL) {
+                post->ctx.hdr_iobuf = iobuf_get (peer->trans->ctx->iobuf_pool);
+                if (post->ctx.hdr_iobuf == NULL) {
                         gf_log (RDMA_LOG_NAME, GF_LOG_ERROR, "out of memory");
                         ret = -1;
                         goto out;
                 }
 
-                iobref_add (iobref, iobuf);
-                iobuf_unref (iobuf);
-
                 header_len = (long)ptr - (long)post->buf;
-                post->ctx.vector[0].iov_base = iobuf_ptr (iobuf);
+                post->ctx.vector[0].iov_base = iobuf_ptr (post->ctx.hdr_iobuf);
                 post->ctx.vector[0].iov_len = bytes_in_post - header_len;
                 memcpy (post->ctx.vector[0].iov_base, ptr,
                         post->ctx.vector[0].iov_len);
                 post->ctx.count = 1;
-
-                iobuf = NULL;
-                iobref = NULL;
         }
 
         post->ctx.reply_info = reply_info;
@@ -2820,14 +2809,6 @@ out:
                 if (write_ary != NULL) {
                         GF_FREE (write_ary);
                 }
-
-                if (iobuf != NULL) {
-                        iobuf_unref (iobuf);
-                }
-
-                if (iobref != NULL) {
-                        iobref_unref (iobref);
-                }
         }
 
         return ret;
@@ -3024,9 +3005,25 @@ rdma_pollin_notify (rdma_peer_t *peer, rdma_post_t *post)
                 goto out;
         }
 
+        if (post->ctx.iobref == NULL) {
+                post->ctx.iobref = iobref_new ();
+                if (post->ctx.iobref == NULL) {
+                        gf_log (RDMA_LOG_NAME, GF_LOG_ERROR, "out of memory");
+                        goto out;
+                }
+
+                /* handling the case where both hdr and payload of 
+                 * GF_FOP_READ_CBK were recieved in a single iobuf
+                 * because of server sending entire msg as inline without
+                 * doing rdma writes.
+                 */
+                iobref_add (post->ctx.iobref, post->ctx.hdr_iobuf);
+        }
+
         pollin = rpc_transport_pollin_alloc (peer->trans,
                                              post->ctx.vector,
                                              post->ctx.count,
+                                             post->ctx.hdr_iobuf,
                                              post->ctx.iobref,
                                              post->ctx.reply_info);
         if (pollin == NULL) {
@@ -3152,9 +3149,7 @@ rdma_recv_reply (rdma_peer_t *peer, rdma_post_t *post)
         }
 
         ctx = rpc_req->conn_private;
-        if ((post->ctx.iobref != NULL) && (ctx->rsp_iobref != NULL)) {
-                iobref_merge (post->ctx.iobref, ctx->rsp_iobref);
-        } else if (post->ctx.iobref == NULL) {
+        if (post->ctx.iobref == NULL) {
                 post->ctx.iobref = iobref_ref (ctx->rsp_iobref);
         }
 
diff --git a/rpc/rpc-transport/rdma/src/rdma.h b/rpc/rpc-transport/rdma/src/rdma.h
index e24ce76ccf1..dbd15d6689e 100644
--- a/rpc/rpc-transport/rdma/src/rdma.h
+++ b/rpc/rpc-transport/rdma/src/rdma.h
@@ -264,6 +264,7 @@ struct __rdma_post_context {
         struct iovec       vector[MAX_IOVEC];
         int                count;
         struct iobref     *iobref;
+        struct iobuf      *hdr_iobuf;
         char               is_request;
         int                rdma_reads;
         rdma_reply_info_t *reply_info;
diff --git a/rpc/rpc-transport/socket/src/socket.c b/rpc/rpc-transport/socket/src/socket.c
index e5272f0d3d1..c121210c478 100644
--- a/rpc/rpc-transport/socket/src/socket.c
+++ b/rpc/rpc-transport/socket/src/socket.c
@@ -1531,11 +1531,6 @@ __socket_proto_state_machine (rpc_transport_t *this,
 
                                 iobref = priv->incoming.iobref;
 
-                                iobref_add (iobref,
-                                            priv->incoming.iobuf);
-                                iobuf_unref (priv->incoming.iobuf);
-                                priv->incoming.iobuf = NULL; 
-
                                 count++;
 
                                 if (priv->incoming.payload_vector.iov_base
@@ -1548,8 +1543,12 @@ __socket_proto_state_machine (rpc_transport_t *this,
                                 *pollin = rpc_transport_pollin_alloc (this,
                                                                       vector,
                                                                       count,
+                                                                      priv->incoming.iobuf,
                                                                       iobref,
                                                                       priv->incoming.request_info);
+                                iobuf_unref (priv->incoming.iobuf);
+                                priv->incoming.iobuf = NULL;
+
                                 if (*pollin == NULL) {
                                         ret = -1;
                                         goto out;