core: get xattrs also as part of readdirp

readdirp_req() call sends a dict_t * as an argument, which
contains all the xattr keys for which the entries got in
readdirp_rsp() are having xattr value filled dictionary.

Change-Id: I8b7e1290740ea3e884e67d19156ce849227167c0
Signed-off-by: Amar Tumballi <amar@gluster.com>
BUG: 765785
Reviewed-on: http://review.gluster.com/771
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@gluster.com>

7 files changed, 126 insertions, 26 deletions

diff --git a/xlators/protocol/client/src/client-helpers.c b/xlators/protocol/client/src/client-helpers.c
index 2a12573e2bb..5759f30624e 100644
--- a/xlators/protocol/client/src/client-helpers.c
+++ b/xlators/protocol/client/src/client-helpers.c
@@ -154,15 +154,21 @@ out:
 }
 
 int
-unserialize_rsp_direntp (struct gfs3_readdirp_rsp *rsp, gf_dirent_t *entries)
+unserialize_rsp_direntp (xlator_t *this, fd_t *fd,
+                         struct gfs3_readdirp_rsp *rsp, gf_dirent_t *entries)
 {
         struct gfs3_dirplist *trav      = NULL;
+        char                 *buf       = NULL;
 	gf_dirent_t          *entry     = NULL;
+        inode_table_t        *itable    = NULL;
         int                   entry_len = 0;
         int                   ret       = -1;
 
         trav = rsp->reply;
 
+        if (fd)
+                itable = fd->inode->table;
+
         while (trav) {
                 entry_len = gf_dirent_size (trav->name);
                 entry = GF_CALLOC (1, entry_len, gf_common_mt_gf_dirent_t);
@@ -178,6 +184,28 @@ unserialize_rsp_direntp (struct gfs3_readdirp_rsp *rsp, gf_dirent_t *entries)
 
                 strcpy (entry->d_name, trav->name);
 
+                if (trav->dict.dict_val) {
+                        /* Dictionary is sent along with response */
+                        buf = memdup (trav->dict.dict_val, trav->dict.dict_len);
+                        if (!buf)
+                                goto out;
+
+                        ret = dict_unserialize (buf, trav->dict.dict_len,
+                                                &entry->dict);
+                        if (ret < 0) {
+                                gf_log (THIS->name, GF_LOG_WARNING,
+                                        "failed to unserialize xattr dict");
+                                errno = EINVAL;
+                                goto out;
+                        }
+                        entry->dict->extra_free = buf;
+                        buf = NULL;
+                }
+
+                entry->inode = inode_find (itable, entry->d_stat.ia_gfid);
+                if (!entry->inode)
+                        entry->inode = inode_new (itable);
+
 		list_add_tail (&entry->list, &entries->list);
 
                 trav = trav->nextentry;
diff --git a/xlators/protocol/client/src/client.c b/xlators/protocol/client/src/client.c
index 87a4603c191..8e823bec4e8 100644
--- a/xlators/protocol/client/src/client.c
+++ b/xlators/protocol/client/src/client.c
@@ -1671,7 +1671,7 @@ out:
 
 int32_t
 client_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd,
-                 size_t size, off_t off)
+                 size_t size, off_t off, dict_t *dict)
 {
         int          ret  = -1;
         clnt_conf_t *conf = NULL;
@@ -1685,6 +1685,7 @@ client_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd,
         args.fd = fd;
         args.size = size;
         args.offset = off;
+        args.xattr_req = dict;
 
         proc = &conf->fops->proctable[GF_FOP_READDIRP];
         if (!proc) {
diff --git a/xlators/protocol/client/src/client.h b/xlators/protocol/client/src/client.h
index 6d0b6491a10..00b743a3c70 100644
--- a/xlators/protocol/client/src/client.h
+++ b/xlators/protocol/client/src/client.h
@@ -184,7 +184,8 @@ int protocol_client_reopendir (xlator_t *this, clnt_fd_ctx_t *fdctx);
 int protocol_client_reopen (xlator_t *this, clnt_fd_ctx_t *fdctx);
 
 int unserialize_rsp_dirent (struct gfs3_readdir_rsp *rsp, gf_dirent_t *entries);
-int unserialize_rsp_direntp (struct gfs3_readdirp_rsp *rsp, gf_dirent_t *entries);
+int unserialize_rsp_direntp (xlator_t *this, fd_t *fd,
+                             struct gfs3_readdirp_rsp *rsp, gf_dirent_t *entries);
 
 int clnt_readdir_rsp_cleanup (gfs3_readdir_rsp *rsp);
 int clnt_readdirp_rsp_cleanup (gfs3_readdirp_rsp *rsp);
diff --git a/xlators/protocol/client/src/client3_1-fops.c b/xlators/protocol/client/src/client3_1-fops.c
index 036e297de38..d34ffc20031 100644
--- a/xlators/protocol/client/src/client3_1-fops.c
+++ b/xlators/protocol/client/src/client3_1-fops.c
@@ -1914,12 +1914,12 @@ int
 client3_1_readdirp_cbk (struct rpc_req *req, struct iovec *iov, int count,
                         void *myframe)
 {
-        call_frame_t           *frame = NULL;
-        gfs3_readdirp_rsp       rsp   = {0,};
-        int32_t                 ret   = 0;
-        clnt_local_t         *local = NULL;
-        gf_dirent_t             entries;
-        xlator_t         *this       = NULL;
+        call_frame_t      *frame = NULL;
+        gfs3_readdirp_rsp  rsp   = {0,};
+        int32_t            ret   = 0;
+        clnt_local_t      *local = NULL;
+        gf_dirent_t        entries;
+        xlator_t          *this  = NULL;
 
         this = THIS;
 
@@ -1943,7 +1943,7 @@ client3_1_readdirp_cbk (struct rpc_req *req, struct iovec *iov, int count,
 
         INIT_LIST_HEAD (&entries.list);
         if (rsp.op_ret > 0) {
-                unserialize_rsp_direntp (&rsp, &entries);
+                unserialize_rsp_direntp (this, local->fd, &rsp, &entries);
         }
 
 out:
@@ -5073,6 +5073,7 @@ client3_1_readdirp (call_frame_t *frame, xlator_t *this,
         struct iovec     *rsphdr            = NULL;
         struct iovec      vector[MAX_IOVEC] = {{0}, };
         clnt_local_t     *local             = NULL;
+        size_t            dict_len          = 0;
 
         if (!frame || !this || !data)
                 goto unwind;
@@ -5085,23 +5086,23 @@ client3_1_readdirp (call_frame_t *frame, xlator_t *this,
         readdirp_rsp_size = xdr_sizeof ((xdrproc_t) xdr_gfs3_readdirp_rsp, &rsp)
                 + args->size;
 
+        local = GF_CALLOC (1, sizeof (*local),
+                           gf_client_mt_clnt_local_t);
+        if (!local) {
+                op_errno = ENOMEM;
+                goto unwind;
+        }
+        frame->local = local;
+
         if ((readdirp_rsp_size + GLUSTERFS_RPC_REPLY_SIZE
              + GLUSTERFS_RDMA_MAX_HEADER_SIZE)
             > (GLUSTERFS_RDMA_INLINE_THRESHOLD)) {
-                local = GF_CALLOC (1, sizeof (*local),
-                                   gf_client_mt_clnt_local_t);
-                if (!local) {
-                        op_errno = ENOMEM;
-                        goto unwind;
-                }
-                frame->local = local;
-
                 rsp_iobref = iobref_new ();
                 if (rsp_iobref == NULL) {
                         goto unwind;
                 }
 
-                        /* TODO: what is the size we should send ? */
+                /* TODO: what is the size we should send ? */
                 rsp_iobuf = iobuf_get (this->ctx->iobuf_pool);
                 if (rsp_iobuf == NULL) {
                         goto unwind;
@@ -5111,19 +5112,33 @@ client3_1_readdirp (call_frame_t *frame, xlator_t *this,
                 iobuf_unref (rsp_iobuf);
                 rsphdr = &vector[0];
                 rsphdr->iov_base = iobuf_ptr (rsp_iobuf);
-                rsphdr->iov_len
-                        = iobuf_pagesize (rsp_iobuf);
+                rsphdr->iov_len  = iobuf_pagesize (rsp_iobuf);
                 count = 1;
                 rsp_iobuf = NULL;
                 local->iobref = rsp_iobref;
                 rsp_iobref = NULL;
         }
 
+        local->fd = fd_ref (args->fd);
+
         req.size = args->size;
         req.offset = args->offset;
         req.fd = remote_fd;
         memcpy (req.gfid, args->fd->inode->gfid, 16);
 
+        if (args->dict) {
+                ret = dict_allocate_and_serialize (args->dict,
+                                                   &req.dict.dict_val,
+                                                   &dict_len);
+                if (ret < 0) {
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "failed to get serialized dict");
+                        op_errno = EINVAL;
+                        goto unwind;
+                }
+                req.dict.dict_len = dict_len;
+        }
+
         ret = client_submit_request (this, &req, frame, conf->fops,
                                      GFS3_OP_READDIRP,
                                      client3_1_readdirp_cbk, NULL,
diff --git a/xlators/protocol/server/src/server-helpers.c b/xlators/protocol/server/src/server-helpers.c
index 43f60e0e2e4..4980424d350 100644
--- a/xlators/protocol/server/src/server-helpers.c
+++ b/xlators/protocol/server/src/server-helpers.c
@@ -1327,11 +1327,40 @@ serialize_rsp_direntp (gf_dirent_t *entries, gfs3_readdirp_rsp *rsp)
                 trav->d_off  = entry->d_off;
                 trav->d_len  = entry->d_len;
                 trav->d_type = entry->d_type;
-                //trav->name   = memdup (entry->d_name, entry->d_len + 1);
                 trav->name   = entry->d_name;
 
                 gf_stat_from_iatt (&trav->stat, &entry->d_stat);
 
+                /* if 'dict' is present, pack it */
+                if (entry->dict) {
+                        trav->dict.dict_len = dict_serialized_length (entry->dict);
+                        if (trav->dict.dict_len < 0) {
+                                gf_log (THIS->name, GF_LOG_ERROR,
+                                        "failed to get serialized length "
+                                        "of reply dict");
+                                errno = EINVAL;
+                                trav->dict.dict_len = 0;
+                                goto out;
+                        }
+
+                        trav->dict.dict_val = GF_CALLOC (1, trav->dict.dict_len,
+                                                         gf_server_mt_rsp_buf_t);
+                        if (!trav->dict.dict_val) {
+                                errno = ENOMEM;
+                                trav->dict.dict_len = 0;
+                                goto out;
+                        }
+
+                        ret = dict_serialize (entry->dict, trav->dict.dict_val);
+                        if (ret < 0) {
+                                gf_log (THIS->name, GF_LOG_ERROR,
+                                        "failed to serialize reply dict");
+                                errno = -ret;
+                                trav->dict.dict_len = 0;
+                                goto out;
+                        }
+                }
+
                 if (prev)
                         prev->nextentry = trav;
                 else
diff --git a/xlators/protocol/server/src/server.h b/xlators/protocol/server/src/server.h
index f261752171d..92785c5a9d6 100644
--- a/xlators/protocol/server/src/server.h
+++ b/xlators/protocol/server/src/server.h
@@ -154,7 +154,7 @@ struct _server_state {
 
         fd_t             *fd;
         dict_t           *params;
-        int               flags;
+        int32_t           flags;
         int               wbflags;
         struct iovec      payload_vector[MAX_IOVEC];
         int               payload_count;
diff --git a/xlators/protocol/server/src/server3_1-fops.c b/xlators/protocol/server/src/server3_1-fops.c
index c2ab3590ed2..482c242d5ce 100644
--- a/xlators/protocol/server/src/server3_1-fops.c
+++ b/xlators/protocol/server/src/server3_1-fops.c
@@ -1038,7 +1038,6 @@ server_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 gf_stat_from_iatt (&rsp.preparent, preparent);
                 gf_stat_from_iatt (&rsp.postparent, postparent);
 
-
                 link_inode = inode_link (inode, state->loc2.parent,
                                          state->loc2.name, stbuf);
                 inode_unref (link_inode);
@@ -1738,6 +1737,10 @@ server_readdirp_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         op_errno = ENOMEM;
                         goto out;
                 }
+
+                /* TODO: need more clear thoughts before calling this function. */
+                /* gf_link_inodes_from_dirent (this, state->fd->inode, entries); */
+
         } else {
                 /* (op_ret == 0) is valid, and means EOF, don't log for that */
                 gf_log (this->name, (op_ret) ? GF_LOG_INFO : GF_LOG_TRACE,
@@ -2132,7 +2135,7 @@ server_readdirp_resume (call_frame_t *frame, xlator_t *bound_xl)
 
         STACK_WIND (frame, server_readdirp_cbk, bound_xl,
                     bound_xl->fops->readdirp, state->fd, state->size,
-                    state->offset);
+                    state->offset, state->dict);
 
         return 0;
 err:
@@ -3979,10 +3982,10 @@ server_readdirp (rpcsvc_request_t *req)
 {
         server_state_t      *state        = NULL;
         call_frame_t        *frame        = NULL;
+        char                *buf          = NULL;
         gfs3_readdirp_req    args         = {{0,},};
         size_t               headers_size = 0;
         int                  ret          = -1;
-
         if (!req)
                 return ret;
 
@@ -4023,6 +4026,29 @@ server_readdirp (rpcsvc_request_t *req)
         state->offset = args.offset;
         memcpy (state->resolve.gfid, args.gfid, 16);
 
+        if (args.dict.dict_len) {
+                /* Unserialize the dictionary */
+                state->dict = dict_new ();
+
+                buf = memdup (args.dict.dict_val, args.dict.dict_len);
+                if (buf == NULL) {
+                        goto out;
+                }
+
+                ret = dict_unserialize (buf, args.dict.dict_len,
+                                        &state->dict);
+                if (ret < 0) {
+                        gf_log (state->conn->bound_xl->name, GF_LOG_ERROR,
+                                "%"PRId64": failed to unserialize req-buffer "
+                                " to dictionary", frame->root->unique);
+                        goto out;
+                }
+
+                state->dict->extra_free = buf;
+
+                buf = NULL;
+        }
+
         ret = 0;
         resolve_and_resume (frame, server_readdirp_resume);
 out: