rpc: changes to glusterfs programs that can take an optional payload argument.

- The existing interface required the transport to separate the procedure
    header and procedure payload into two different buffers. Making this
    separation can prove cumbersome for transports like rdma wherein the header
    and payload can be received in a single buffer (For eg., header and payload
    of write fop sent as inline msg using rdma-send). This patch delegates the
    responsiblity of separating out header and payload to programs.

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

BUG: 875 (Implement a new protocol to provide proper backward/forward compatibility)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=875

2 files changed, 31 insertions, 23 deletions

diff --git a/xlators/protocol/server/src/server.h b/xlators/protocol/server/src/server.h
index 20853879f..a35eeb166 100644
--- a/xlators/protocol/server/src/server.h
+++ b/xlators/protocol/server/src/server.h
@@ -162,7 +162,7 @@ struct _server_state {
 	fd_t             *fd;
 	int               flags;
         int               wbflags;
-        struct iovec      payload_vector;
+        struct iovec      payload_vector[MAX_IOVEC];
         int               payload_count;
         struct iobuf     *iobuf;
         struct iobref    *iobref;
diff --git a/xlators/protocol/server/src/server3_1-fops.c b/xlators/protocol/server/src/server3_1-fops.c
index fb69cf8da..517525160 100644
--- a/xlators/protocol/server/src/server3_1-fops.c
+++ b/xlators/protocol/server/src/server3_1-fops.c
@@ -2438,20 +2438,16 @@ int
 server_writev_resume (call_frame_t *frame, xlator_t *bound_xl)
 {
         server_state_t   *state = NULL;
-        struct iovec      iov = {0, };
 
         state = CALL_STATE (frame);
 
         if (state->resolve.op_ret != 0)
                 goto err;
 
-        if (state->payload_count) {
-                iov = state->payload_vector;
-        }
-
         STACK_WIND (frame, server_writev_cbk,
                     bound_xl, bound_xl->fops->writev,
-                    state->fd, &iov, 1, state->offset, state->iobref);
+                    state->fd, state->payload_vector, state->payload_count,
+                    state->offset, state->iobref);
 
         return 0;
 err:
@@ -2978,24 +2974,17 @@ out:
 int
 server_writev (rpcsvc_request_t *req)
 {
-        /* TODO : */
-        assert (0);
-        return 0;
-}
-
-
-int
-server_writev_vec (rpcsvc_request_t *req, struct iovec *payload,
-                   int payload_count, struct iobref *iobref)
-{
         server_state_t      *state  = NULL;
         call_frame_t        *frame  = NULL;
         gfs3_write_req       args   = {0,};
+        ssize_t              len    = 0;
+        int                  i      = 0;
 
         if (!req)
                 return 0;
 
-        if (!xdr_to_writev_req (req->msg[0], &args)) {
+        len = xdr_to_writev_req (req->msg[0], &args);
+        if (len == 0) {
                 //failed to decode msg;
                 req->rpc_err = GARBAGE_ARGS;
                 goto out;
@@ -3019,12 +3008,23 @@ server_writev_vec (rpcsvc_request_t *req, struct iovec *payload,
         state->resolve.type  = RESOLVE_MUST;
         state->resolve.fd_no = args.fd;
         state->offset        = args.offset;
+        state->iobref        = iobref_ref (req->iobref);
 
-        if (payload_count != 0) {
-                state->iobref = iobref_ref (iobref);
-                state->size = req->msg[1].iov_len;
-                state->payload_count = payload_count;
-                state->payload_vector = *payload;
+        if (len < req->msg[0].iov_len) {
+                state->payload_vector[0].iov_base
+                        = (req->msg[0].iov_base + len);
+                state->payload_vector[0].iov_len
+                        = req->msg[0].iov_len - len;
+                state->payload_count = 1;
+        }
+
+        for (i = 1; i < req->count; i++) {
+                state->payload_vector[state->payload_count++]
+                        = req->msg[i];
+        }
+
+        for (i = 0; i < state->payload_count; i++) {
+                state->size += state->payload_vector[i].iov_len;
         }
 
         resolve_and_resume (frame, server_writev_resume);
@@ -3034,6 +3034,14 @@ out:
 
 
 int
+server_writev_vec (rpcsvc_request_t *req, struct iovec *payload,
+                   int payload_count, struct iobref *iobref)
+{
+        return server_writev (req);
+}
+
+
+int
 server_release (rpcsvc_request_t *req)
 {
         server_connection_t *conn = NULL;