rpc/rpc-clnt: don't use iobref passed from caller to hold iobufs required for building program and rpc headers.

- clustering translators like replicate assume iobrefs will be used as
    read-only by lower layers and there by use the same iobref while
    winding calls to multiple children. Since iobrefs can hold only limited
    number of iobufs, this limits the number of children these translators
    can have without overflowing iobrefs.

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

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

2 files changed, 55 insertions, 32 deletions

diff --git a/xlators/protocol/client/src/client.c b/xlators/protocol/client/src/client.c
index 56ac45a4d27..51c35ef8c70 100644
--- a/xlators/protocol/client/src/client.c
+++ b/xlators/protocol/client/src/client.c
@@ -49,12 +49,13 @@ client_submit_request (xlator_t *this, void *req, call_frame_t *frame,
                        struct iovec *rsp_payload, int rsp_payload_count,
                        struct iobref *rsp_iobref)
 {
-        int          ret         = -1;
-        clnt_conf_t *conf        = NULL;
-        struct iovec iov         = {0, };
-        struct iobuf *iobuf      = NULL;
-        int           count      = 0;
-        char          new_iobref = 0, start_ping = 0;
+        int            ret         = -1;
+        clnt_conf_t   *conf        = NULL;
+        struct iovec   iov         = {0, };
+        struct iobuf  *iobuf       = NULL;
+        int            count       = 0;
+        char           start_ping  = 0;
+        struct iobref *new_iobref  = NULL;
 
         if (!this || !prog || !frame)
                 goto out;
@@ -75,16 +76,26 @@ client_submit_request (xlator_t *this, void *req, call_frame_t *frame,
                 goto out;
         };
 
-        if (!iobref) {
-                iobref = iobref_new ();
-                if (!iobref) {
-                        goto out;
-                }
+        new_iobref = iobref_new ();
+        if (!new_iobref) {
+                goto out;
+        }
 
-                new_iobref = 1;
+        if (iobref != NULL) {
+                ret = iobref_merge (new_iobref, iobref);
+                if (ret != 0) {
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "cannot merge iobref passed from caller "
+                                "into new_iobref");
+                }
         }
 
-        iobref_add (iobref, iobuf);
+        ret = iobref_add (new_iobref, iobuf);
+        if (ret != 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "cannot add iobuf into iobref");
+                goto out;
+        }
 
         iov.iov_base = iobuf->ptr;
         iov.iov_len  = 128 * GF_UNIT_KB;
@@ -100,7 +111,7 @@ client_submit_request (xlator_t *this, void *req, call_frame_t *frame,
         }
         /* Send the msg */
         ret = rpc_clnt_submit (conf->rpc, prog, procnum, cbk, &iov, count, NULL,
-                               0, iobref, frame, rsphdr, rsphdr_count,
+                               0, new_iobref, frame, rsphdr, rsphdr_count,
                                rsp_payload, rsp_payload_count, rsp_iobref);
 
         if (ret == 0) {
@@ -118,8 +129,8 @@ client_submit_request (xlator_t *this, void *req, call_frame_t *frame,
 
         ret = 0;
 out:
-        if (new_iobref)
-                iobref_unref (iobref);
+        if (new_iobref != NULL)
+                iobref_unref (new_iobref);
 
         if (iobuf)
                 iobuf_unref (iobuf);
diff --git a/xlators/protocol/client/src/client3_1-fops.c b/xlators/protocol/client/src/client3_1-fops.c
index b1fe7fe1e99..6f7808ebb1b 100644
--- a/xlators/protocol/client/src/client3_1-fops.c
+++ b/xlators/protocol/client/src/client3_1-fops.c
@@ -37,13 +37,13 @@ client_submit_vec_request (xlator_t  *this, void *req, call_frame_t  *frame,
                            struct iovec  *payload, int payloadcnt,
                            struct iobref *iobref, gfs_serialize_t sfunc)
 {
-        int           ret        = 0;
-        clnt_conf_t  *conf       = NULL;
-        struct iovec  iov        = {0, };
-        struct iobuf *iobuf      = NULL;
-        int           count      = 0;
-        char          new_iobref = 0;
-        int           start_ping = 0;
+        int            ret        = 0;
+        clnt_conf_t   *conf       = NULL;
+        struct iovec   iov        = {0, };
+        struct iobuf  *iobuf      = NULL;
+        int            count      = 0;
+        int            start_ping = 0;
+        struct iobref *new_iobref = NULL;
 
         start_ping = 0;
 
@@ -54,16 +54,27 @@ client_submit_vec_request (xlator_t  *this, void *req, call_frame_t  *frame,
                 goto out;
         };
 
-        if (!iobref) {
-                iobref = iobref_new ();
-                if (!iobref) {
+        new_iobref = iobref_new ();
+        if (!new_iobref) {
+                goto out;
+        }
+
+        if (iobref != NULL) {
+                ret = iobref_merge (new_iobref, iobref);
+                if (ret != 0) {
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "cannot merge iobref passed from caller into "
+                                "new_iobref");
                         goto out;
                 }
-
-                new_iobref = 1;
         }
 
-        iobref_add (iobref, iobuf);
+        ret = iobref_add (new_iobref, iobuf);
+        if (ret != 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "cannot add iobuf into iobref");
+                goto out;
+        }
 
         iov.iov_base = iobuf->ptr;
         iov.iov_len  = 128 * GF_UNIT_KB;
@@ -74,12 +85,13 @@ client_submit_vec_request (xlator_t  *this, void *req, call_frame_t  *frame,
                 if (ret == -1) {
                         goto out;
                 }
+
                 iov.iov_len = ret;
                 count = 1;
         }
         /* Send the msg */
         ret = rpc_clnt_submit (conf->rpc, prog, procnum, cbk, &iov, count,
-                               payload, payloadcnt, iobref, frame, NULL, 0,
+                               payload, payloadcnt, new_iobref, frame, NULL, 0,
                                NULL, 0, NULL);
 
         if (ret == 0) {
@@ -96,8 +108,8 @@ client_submit_vec_request (xlator_t  *this, void *req, call_frame_t  *frame,
                 client_start_ping ((void *) this);
 
 out:
-        if (new_iobref) {
-                iobref_unref (iobref);
+        if (new_iobref != NULL) {
+                iobref_unref (new_iobref);
         }
 
         iobuf_unref (iobuf);