LU-991 llite: cleanup md_readpage interface

Use md_op_data to transfer the parameters of readpage on client
metadata stack.

Signed-off-by: Wang Di <di.wang@whamcloud.com>
Change-Id: I103a555094e678711aa2f2e5d73ec848a452aed0
Reviewed-on: http://review.whamcloud.com/2001
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Fan Yong <yong.fan@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 870c844..c46e99c 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -1302,6 +1302,10 @@ struct md_op_data {
 
         /* Operation type */
         __u32                   op_opc;
+
+        /* Used by readdir */
+        __u32                   op_npages;
+        __u64                   op_offset;
 };
 
 struct md_enqueue_info;
@@ -1572,9 +1576,8 @@ struct md_ops {
                          struct md_open_data **mod);
         int (*m_sync)(struct obd_export *, const struct lu_fid *,
                       struct obd_capa *, struct ptlrpc_request **);
-        int (*m_readpage)(struct obd_export *, const struct lu_fid *,
-                          struct obd_capa *, __u64, struct page **,
-                          unsigned, struct ptlrpc_request **);
+        int (*m_readpage)(struct obd_export *, struct md_op_data *,
+                          struct page **, struct ptlrpc_request **);
 
         int (*m_unlink)(struct obd_export *, struct md_op_data *,
                         struct ptlrpc_request **);

diff --git a/lustre/include/obd_class.h b/lustre/include/obd_class.h
index fca1291..309b29b 100644 (file)
--- a/lustre/include/obd_class.h
+++ b/lustre/include/obd_class.h
@@ -2047,17 +2047,15 @@ static inline int md_sync(struct obd_export *exp, const struct lu_fid *fid,
         RETURN(rc);
 }
 
-static inline int md_readpage(struct obd_export *exp, const struct lu_fid *fid,
-                              struct obd_capa *oc, __u64 offset,
-                              struct page **pages, unsigned npages,
+static inline int md_readpage(struct obd_export *exp, struct md_op_data *opdata,
+                              struct page **pages,
                               struct ptlrpc_request **request)
 {
         int rc;
         ENTRY;
         EXP_CHECK_MD_OP(exp, readpage);
         EXP_MD_COUNTER_INCREMENT(exp, readpage);
-        rc = MDP(exp->exp_obd, readpage)(exp, fid, oc, offset, pages, npages,
-                                         request);
+        rc = MDP(exp->exp_obd, readpage)(exp, opdata, pages, request);
         RETURN(rc);
 }
 

diff --git a/lustre/liblustre/dir.c b/lustre/liblustre/dir.c
index 846e228..5685b28 100644 (file)
--- a/lustre/liblustre/dir.c
+++ b/lustre/liblustre/dir.c
@@ -76,10 +76,10 @@ static int llu_dir_do_readpage(struct inode *inode, struct page *page)
         struct lookup_intent   it = { .it_op = IT_READDIR };
         struct md_op_data      op_data = {{ 0 }};
         ldlm_policy_data_t policy = { .l_inodebits = { MDS_INODELOCK_UPDATE } };
-        __u64 offset;
         int rc = 0;
         ENTRY;
 
+        llu_prep_md_op_data(&op_data, inode, NULL, NULL, 0, 0, LUSTRE_OPC_ANY);
         rc = md_lock_match(sbi->ll_md_exp, LDLM_FL_BLOCK_GRANTED,
                            &lli->lli_fid, LDLM_IBITS, &policy, LCK_CR, &lockh);
         if (!rc) {
@@ -87,9 +87,6 @@ static int llu_dir_do_readpage(struct inode *inode, struct page *page)
                         llu_md_blocking_ast, ldlm_completion_ast, NULL, NULL,
                         inode};
 
-                llu_prep_md_op_data(&op_data, inode, NULL, NULL, 0, 0,
-                                    LUSTRE_OPC_ANY);
-
                 rc = md_enqueue(sbi->ll_md_exp, &einfo, &it,
                                 &op_data, &lockh, NULL, 0, NULL,
                                 LDLM_FL_CANCEL_ON_BLOCK);
@@ -103,9 +100,9 @@ static int llu_dir_do_readpage(struct inode *inode, struct page *page)
         }
         ldlm_lock_dump_handle(D_OTHER, &lockh);
 
-        offset = (__u64)hash_x_index(page->index, 0);
-        rc = md_readpage(sbi->ll_md_exp, &lli->lli_fid, NULL,
-                         offset, &page, 1, &request);
+        op_data.op_offset = (__u64)hash_x_index(page->index, 0);
+        op_data.op_npages = 1;
+        rc = md_readpage(sbi->ll_md_exp, &op_data, &page, &request);
         if (!rc) {
                 body = req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY);
                 LASSERT(body != NULL);         /* checked by md_readpage() */

diff --git a/lustre/liblustre/file.c b/lustre/liblustre/file.c
index 29d8247..e00ac99 100644 (file)
--- a/lustre/liblustre/file.c
+++ b/lustre/liblustre/file.c
@@ -103,11 +103,6 @@ void llu_prep_md_op_data(struct md_op_data *op_data, struct inode *i1,
         op_data->op_data = NULL;
 }
 
-void llu_finish_md_op_data(struct md_op_data *op_data)
-{
-        OBD_FREE_PTR(op_data);
-}
-
 void obdo_refresh_inode(struct inode *dst,
                         struct obdo *src,
                         obd_flag valid)

diff --git a/lustre/liblustre/llite_lib.h b/lustre/liblustre/llite_lib.h
index ae3e550..55675c0 100644 (file)
--- a/lustre/liblustre/llite_lib.h
+++ b/lustre/liblustre/llite_lib.h
@@ -261,7 +261,6 @@ extern struct fssw_ops llu_fssw_ops;
 void llu_prep_md_op_data(struct md_op_data *op_data, struct inode *i1,
                          struct inode *i2, const char *name, int namelen,
                          int mode, __u32 opc);
-void llu_finish_md_op_data(struct md_op_data *op_data);
 int llu_create(struct inode *dir, struct pnode_base *pnode, int mode);
 int llu_local_open(struct llu_inode_info *lli, struct lookup_intent *it);
 int llu_iop_open(struct pnode *pnode, int flags, mode_t mode);

diff --git a/lustre/llite/dir.c b/lustre/llite/dir.c
index 27edd96..5ff49d9 100644 (file)
--- a/lustre/llite/dir.c
+++ b/lustre/llite/dir.c
@@ -159,7 +159,7 @@ static int ll_dir_readpage(struct file *file, struct page *page0)
         struct obd_export *exp = ll_i2sbi(inode)->ll_md_exp;
         struct ptlrpc_request *request;
         struct mdt_body *body;
-        struct obd_capa *oc;
+        struct md_op_data *op_data;
         __u64 hash;
         struct page **page_pool;
         struct page *page;
@@ -200,10 +200,12 @@ static int ll_dir_readpage(struct file *file, struct page *page0)
                 page_pool[npages] = page;
         }
 
-        oc = ll_mdscapa_get(inode);
-        rc = md_readpage(exp, ll_inode2fid(inode), oc, hash, page_pool, npages,
-                         &request);
-        capa_put(oc);
+        op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
+                                     LUSTRE_OPC_ANY, NULL);
+        op_data->op_npages = npages;
+        op_data->op_offset = hash;
+        rc = md_readpage(exp, op_data, page_pool, &request);
+        ll_finish_md_op_data(op_data);
         if (rc == 0) {
                 body = req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY);
                 /* Checked by mdc_readpage() */

diff --git a/lustre/lmv/lmv_obd.c b/lustre/lmv/lmv_obd.c
index d897826..2e2b5bf 100644 (file)
--- a/lustre/lmv/lmv_obd.c
+++ b/lustre/lmv/lmv_obd.c
@@ -2342,16 +2342,14 @@ static __u32 lmv_node_rank(struct obd_export *exp, const struct lu_fid *fid)
         return id ^ (id >> 32);
 }
 
-static int lmv_readpage(struct obd_export *exp, const struct lu_fid *fid,
-                        struct obd_capa *oc, __u64 offset64,
-                        struct page **pages, unsigned npages,
-                        struct ptlrpc_request **request)
+static int lmv_readpage(struct obd_export *exp, struct md_op_data *op_data,
+                        struct page **pages, struct ptlrpc_request **request)
 {
         struct obd_device       *obd = exp->exp_obd;
         struct lmv_obd          *lmv = &obd->u.lmv;
-        struct lu_fid            rid = *fid;
         struct lmv_object       *obj;
-        __u64                    offset;
+        struct lu_fid            rid = op_data->op_fid1;
+        __u64                    offset = op_data->op_offset;
         __u64                    hash_adj = 0;
         __u32                    rank = 0;
         __u64                    seg_size = 0;
@@ -2371,8 +2369,6 @@ static int lmv_readpage(struct obd_export *exp, const struct lu_fid *fid,
         struct lu_dirent        *ent;
         ENTRY;
 
-        offset = offset64;
-
         rc = lmv_check_connect(obd);
         if (rc)
                 RETURN(rc);
@@ -2399,7 +2395,7 @@ static int lmv_readpage(struct obd_export *exp, const struct lu_fid *fid,
          * [R*MAX_HASH/N ... (R + 1)*MAX_HASH/N] there for we do hash_adj
          * on hash  values that we get.
          */
-        obj = lmv_object_find_lock(obd, fid);
+        obj = lmv_object_find_lock(obd, &rid);
         if (obj) {
                 nr       = obj->lo_objcount;
                 LASSERT(nr > 0);
@@ -2407,7 +2403,7 @@ static int lmv_readpage(struct obd_export *exp, const struct lu_fid *fid,
                 do_div(seg_size, nr);
                 los      = obj->lo_stripes;
                 tgt      = lmv_get_target(lmv, los[0].ls_mds);
-                rank     = lmv_node_rank(tgt->ltd_exp, fid) % nr;
+                rank     = lmv_node_rank(tgt->ltd_exp, &rid) % nr;
                 tgt_tmp  = offset;
                 do_div(tgt_tmp, seg_size);
                 tgt0_idx = do_div(tgt_tmp,  nr);
@@ -2442,8 +2438,8 @@ static int lmv_readpage(struct obd_export *exp, const struct lu_fid *fid,
         if (IS_ERR(tgt))
                 GOTO(cleanup, rc = PTR_ERR(tgt));
 
-        rc = md_readpage(tgt->ltd_exp, &rid, oc, offset, pages, npages,
-                         request);
+        op_data->op_fid1 = rid;
+        rc = md_readpage(tgt->ltd_exp, op_data, pages, request);
         if (rc)
                 GOTO(cleanup, rc);
 
@@ -2451,9 +2447,10 @@ static int lmv_readpage(struct obd_export *exp, const struct lu_fid *fid,
                  >> CFS_PAGE_SHIFT;
         nlupgs = (*request)->rq_bulk->bd_nob_transferred >> LU_PAGE_SHIFT;
         LASSERT(!((*request)->rq_bulk->bd_nob_transferred & ~LU_PAGE_MASK));
-        LASSERT(nrdpgs > 0 && nrdpgs <= npages);
+        LASSERT(nrdpgs > 0 && nrdpgs <= op_data->op_npages);
 
-        CDEBUG(D_INODE, "read %d(%d)/%d pages\n", nrdpgs, nlupgs, npages);
+        CDEBUG(D_INODE, "read %d(%d)/%d pages\n", nrdpgs, nlupgs,
+               op_data->op_npages);
 
         for (i = 0; i < nrdpgs; i++) {
 #if CFS_PAGE_SIZE > LU_PAGE_SIZE
@@ -2467,7 +2464,8 @@ static int lmv_readpage(struct obd_export *exp, const struct lu_fid *fid,
                 if (obj) {
                         lmv_hash_adjust(&dp->ldp_hash_start, hash_adj);
                         lmv_hash_adjust(&dp->ldp_hash_end,   hash_adj);
-                        LASSERT(le64_to_cpu(dp->ldp_hash_start) <= offset64);
+                        LASSERT(le64_to_cpu(dp->ldp_hash_start) <=
+                                op_data->op_offset);
 
                         if ((tgt0_idx != nr - 1) &&
                             (le64_to_cpu(dp->ldp_hash_end) == MDS_DIR_END_OFF))

diff --git a/lustre/mdc/mdc_request.c b/lustre/mdc/mdc_request.c
index 1f2c6eb..a9033f7 100644 (file)
--- a/lustre/mdc/mdc_request.c
+++ b/lustre/mdc/mdc_request.c
@@ -1019,9 +1019,8 @@ out:
 EXPORT_SYMBOL(mdc_sendpage);
 #endif
 
-int mdc_readpage(struct obd_export *exp, const struct lu_fid *fid,
-                 struct obd_capa *oc, __u64 offset, struct page **pages,
-                 unsigned npages, struct ptlrpc_request **request)
+int mdc_readpage(struct obd_export *exp, struct md_op_data *op_data,
+                 struct page **pages, struct ptlrpc_request **request)
 {
         struct ptlrpc_request   *req;
         struct ptlrpc_bulk_desc *desc;
@@ -1040,7 +1039,7 @@ restart_bulk:
         if (req == NULL)
                 RETURN(-ENOMEM);
 
-        mdc_set_capa_size(req, &RMF_CAPA1, oc);
+        mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 
         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_READPAGE);
         if (rc) {
@@ -1051,7 +1050,7 @@ restart_bulk:
         req->rq_request_portal = MDS_READPAGE_PORTAL;
         ptlrpc_at_set_req_timeout(req);
 
-        desc = ptlrpc_prep_bulk_imp(req, npages, BULK_PUT_SINK,
+        desc = ptlrpc_prep_bulk_imp(req, op_data->op_npages, BULK_PUT_SINK,
                                     MDS_BULK_PORTAL);
         if (desc == NULL) {
                 ptlrpc_request_free(req);
@@ -1059,10 +1058,12 @@ restart_bulk:
         }
 
         /* NB req now owns desc and will free it when it gets freed */
-        for (i = 0; i < npages; i++)
+        for (i = 0; i < op_data->op_npages; i++)
                 ptlrpc_prep_bulk_page(desc, pages[i], 0, CFS_PAGE_SIZE);
 
-        mdc_readdir_pack(req, offset, CFS_PAGE_SIZE * npages, fid, oc);
+        mdc_readdir_pack(req, op_data->op_offset,
+                         CFS_PAGE_SIZE * op_data->op_npages,
+                         &op_data->op_fid1, op_data->op_capa1);
 
         ptlrpc_request_set_replen(req);
         rc = ptlrpc_queue_wait(req);
@@ -1092,7 +1093,7 @@ restart_bulk:
         if (req->rq_bulk->bd_nob_transferred & ~LU_PAGE_MASK) {
                 CERROR("Unexpected # bytes transferred: %d (%ld expected)\n",
                         req->rq_bulk->bd_nob_transferred,
-                        CFS_PAGE_SIZE * npages);
+                        CFS_PAGE_SIZE * op_data->op_npages);
                 ptlrpc_req_finished(req);
                 RETURN(-EPROTO);
         }