LU-3531 mdc: release dir page cache after accessing

Release the dir page cache in llite/lmv, so the page
will be hold until entires was filled by filldir.

Signed-off-by: wang di <di.wang@intel.com>
Change-Id: I8b24bec74b14ff2b65130c02294821fc16ca1421
Reviewed-on: http://review.whamcloud.com/8935
Tested-by: Jenkins
Reviewed-by: John L. Hammond <john.hammond@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>
Tested-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lclient.h b/lustre/include/lclient.h
index 04a3744..3448891 100644 (file)
--- a/lustre/include/lclient.h
+++ b/lustre/include/lclient.h
@@ -462,7 +462,6 @@ enum op_cli_flags {
        CLI_RM_ENTRY    = 1 << 1,
        CLI_HASH64      = 1 << 2,
        CLI_API32       = 1 << 3,
-       CLI_READENT_END = 1 << 4,
 };
 
 #endif /*LCLIENT_H */

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 0f83c2a..e19aab1 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -1281,7 +1281,8 @@ struct md_ops {
                          struct page **, struct ptlrpc_request **);
 
        int (*m_read_entry)(struct obd_export *, struct md_op_data *,
-                           struct md_callback *cb_op, struct lu_dirent **ld);
+                           struct md_callback *cb_op, struct lu_dirent **ld,
+                           struct page **ppage);
 
        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 2207359..11917f3 100644 (file)
--- a/lustre/include/obd_class.h
+++ b/lustre/include/obd_class.h
@@ -2018,13 +2018,14 @@ static inline int md_readpage(struct obd_export *exp, struct md_op_data *opdata,
 static inline int md_read_entry(struct obd_export *exp,
                                struct md_op_data *op_data,
                                struct md_callback *cb_op,
-                               struct lu_dirent **ld)
+                               struct lu_dirent **ld,
+                               struct page **ppage)
 {
        int rc;
        ENTRY;
        EXP_CHECK_MD_OP(exp, read_entry);
        EXP_MD_COUNTER_INCREMENT(exp, read_entry);
-       rc = MDP(exp->exp_obd, read_entry)(exp, op_data, cb_op, ld);
+       rc = MDP(exp->exp_obd, read_entry)(exp, op_data, cb_op, ld, ppage);
        RETURN(rc);
 }
 

diff --git a/lustre/llite/dir.c b/lustre/llite/dir.c
index 1c849b8..dc5d623 100644 (file)
--- a/lustre/llite/dir.c
+++ b/lustre/llite/dir.c
@@ -159,14 +159,16 @@
  * ll_dir_entry_end: release the last page.
  **/
 struct lu_dirent *ll_dir_entry_start(struct inode *dir,
-                                    struct md_op_data *op_data)
+                                    struct md_op_data *op_data,
+                                    struct page **ppage)
 {
        struct lu_dirent *entry;
        struct md_callback cb_op;
        int rc;
 
+       LASSERT(*ppage == NULL);
        cb_op.md_blocking_ast = ll_md_blocking_ast;
-       rc = md_read_entry(ll_i2mdexp(dir), op_data, &cb_op, &entry);
+       rc = md_read_entry(ll_i2mdexp(dir), op_data, &cb_op, &entry, ppage);
        if (rc != 0)
                entry = ERR_PTR(rc);
        return entry;
@@ -174,32 +176,25 @@ struct lu_dirent *ll_dir_entry_start(struct inode *dir,
 
 struct lu_dirent *ll_dir_entry_next(struct inode *dir,
                                    struct md_op_data *op_data,
-                                   struct lu_dirent *ent)
+                                   struct lu_dirent *ent,
+                                   struct page **ppage)
 {
        struct lu_dirent *entry;
        struct md_callback cb_op;
        int rc;
 
+       LASSERT(*ppage != NULL);
        cb_op.md_blocking_ast = ll_md_blocking_ast;
        op_data->op_hash_offset = le64_to_cpu(ent->lde_hash);
-       rc = md_read_entry(ll_i2mdexp(dir), op_data, &cb_op, &entry);
+       kunmap(*ppage);
+       page_cache_release(*ppage);
+       *ppage = NULL;
+       rc = md_read_entry(ll_i2mdexp(dir), op_data, &cb_op, &entry, ppage);
        if (rc != 0)
                entry = ERR_PTR(rc);
        return entry;
 }
 
-void ll_dir_entry_end(struct inode *dir, struct md_op_data *op_data,
-                     struct lu_dirent *ent)
-{
-       struct lu_dirent *entry;
-       struct md_callback cb_op;
-
-       cb_op.md_blocking_ast = ll_md_blocking_ast;
-       op_data->op_cli_flags = CLI_READENT_END;
-       md_read_entry(ll_i2mdexp(dir), op_data, &cb_op, &entry);
-       return;
-}
-
 int ll_dir_read(struct inode *inode, struct md_op_data *op_data,
                void *cookie, filldir_t filldir)
 {
@@ -212,12 +207,13 @@ int ll_dir_read(struct inode *inode, struct md_op_data *op_data,
        int                     rc = 0;
        __u64                   hash = MDS_DIR_END_OFF;
        __u64                   last_hash = MDS_DIR_END_OFF;
+       struct page             *page = NULL;
        ENTRY;
 
         ll_dir_chain_init(&chain);
-       for (ent = ll_dir_entry_start(inode, op_data);
+       for (ent = ll_dir_entry_start(inode, op_data, &page);
             ent != NULL && !IS_ERR(ent) && !done;
-            ent = ll_dir_entry_next(inode, op_data, ent)) {
+            ent = ll_dir_entry_next(inode, op_data, ent, &page)) {
                __u16          type;
                int            namelen;
                struct lu_fid  fid;
@@ -262,8 +258,11 @@ int ll_dir_read(struct inode *inode, struct md_op_data *op_data,
 
        if (IS_ERR(ent))
                rc = PTR_ERR(ent);
-       else if (ent != NULL)
-               ll_dir_entry_end(inode, op_data, ent);
+
+       if (page != NULL) {
+               kunmap(page);
+               page_cache_release(page);
+       }
 
        ll_dir_chain_fini(&chain);
        RETURN(rc);

diff --git a/lustre/llite/llite_internal.h b/lustre/llite/llite_internal.h
index deee1ef..5c22795 100644 (file)
--- a/lustre/llite/llite_internal.h
+++ b/lustre/llite/llite_internal.h
@@ -723,13 +723,13 @@ int ll_dir_read(struct inode *inode, struct md_op_data *op_data,
 int ll_get_mdt_idx(struct inode *inode);
 
 struct lu_dirent *ll_dir_entry_start(struct inode *dir,
-                                    struct md_op_data *op_data);
+                                    struct md_op_data *op_data,
+                                    struct page **ppage);
 
 struct lu_dirent *ll_dir_entry_next(struct inode *dir,
                                    struct md_op_data *op_data,
-                                   struct lu_dirent *ent);
-void ll_dir_entry_end(struct inode *dir, struct md_op_data *op_data,
-                     struct lu_dirent *ent);
+                                   struct lu_dirent *ent,
+                                   struct page **ppage);
 /* llite/namei.c */
 int ll_objects_destroy(struct ptlrpc_request *request,
                        struct inode *dir);

diff --git a/lustre/llite/statahead.c b/lustre/llite/statahead.c
index 01521ef..011b448 100644 (file)
--- a/lustre/llite/statahead.c
+++ b/lustre/llite/statahead.c
@@ -1072,6 +1072,7 @@ static int ll_statahead_thread(void *arg)
        struct ll_dir_chain       chain;
        struct l_wait_info        lwi    = { 0 };
        struct lu_dirent         *ent;
+       struct page             *page = NULL;
        ENTRY;
 
        CDEBUG(D_READA, "statahead thread started: [pid %d] [parent %.*s]\n",
@@ -1096,9 +1097,9 @@ static int ll_statahead_thread(void *arg)
        wake_up(&thread->t_ctl_waitq);
 
        ll_dir_chain_init(&chain);
-       for (ent = ll_dir_entry_start(dir, op_data);
+       for (ent = ll_dir_entry_start(dir, op_data, &page);
             ent != NULL && !IS_ERR(ent);
-            ent = ll_dir_entry_next(dir, op_data, ent)) {
+            ent = ll_dir_entry_next(dir, op_data, ent, &page)) {
                __u64 hash;
                int namelen;
                char *name;
@@ -1192,8 +1193,10 @@ do_it:
                ll_statahead_one(parent, name, namelen);
        }
 
-       if (ent != NULL && !IS_ERR(ent))
-               ll_dir_entry_end(dir, op_data, ent);
+       if (page != NULL) {
+               kunmap(page);
+               page_cache_release(page);
+       }
 
         /*
         * End of directory reached.
@@ -1337,6 +1340,7 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry)
        struct md_op_data    *op_data;
        int                   dot_de;
        struct lu_dirent     *ent;
+       struct page          *page = NULL;
        int                   rc     = LS_NONE_FIRST_DE;
        ENTRY;
 
@@ -1354,9 +1358,9 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry)
        op_data->op_max_pages =
                ll_i2sbi(dir)->ll_md_brw_size >> PAGE_CACHE_SHIFT;
 
-       for (ent = ll_dir_entry_start(dir, op_data);
+       for (ent = ll_dir_entry_start(dir, op_data, &page);
             ent != NULL && !IS_ERR(ent);
-            ent = ll_dir_entry_next(dir, op_data, ent)) {
+            ent = ll_dir_entry_next(dir, op_data, ent, &page)) {
                __u64 hash;
                int namelen;
                char *name;
@@ -1410,8 +1414,11 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry)
                break;
        }
         EXIT;
-       if (ent != NULL && !IS_ERR(ent))
-               ll_dir_entry_end(dir, op_data, ent);
+
+       if (page != NULL) {
+               kunmap(page);
+               page_cache_release(page);
+       }
        ll_finish_md_op_data(op_data);
 out:
        ll_dir_chain_fini(&chain);

diff --git a/lustre/lmv/lmv_obd.c b/lustre/lmv/lmv_obd.c
index 2d24c6e..e0941c9 100644 (file)
--- a/lustre/lmv/lmv_obd.c
+++ b/lustre/lmv/lmv_obd.c
@@ -2288,7 +2288,8 @@ static void lmv_adjust_dirpages(struct page **pages, int ncfspgs, int nlupgs)
 
 #define NORMAL_MAX_STRIPES 4
 int lmv_read_entry(struct obd_export *exp, struct md_op_data *op_data,
-                  struct md_callback *cb_op, struct lu_dirent **ldp)
+                  struct md_callback *cb_op, struct lu_dirent **ldp,
+                  struct page **ppage)
 {
        struct obd_device       *obd = exp->exp_obd;
        struct lmv_obd          *lmv = &obd->u.lmv;
@@ -2298,6 +2299,7 @@ int lmv_read_entry(struct obd_export *exp, struct md_op_data *op_data,
        int                     stripe_count;
        __u64                   min_hash;
        int                     min_idx = 0;
+       struct page             *min_page = NULL;
        int                     i;
        int                     rc;
        ENTRY;
@@ -2323,6 +2325,7 @@ int lmv_read_entry(struct obd_export *exp, struct md_op_data *op_data,
        min_hash = MDS_DIR_END_OFF;
        for (i = 0; i < stripe_count; i++) {
                struct lmv_tgt_desc *tgt;
+               struct page *page = NULL;
 
                if (likely(lsm == NULL)) {
                        tgt = lmv_find_target(lmv, &op_data->op_fid1);
@@ -2338,12 +2341,16 @@ int lmv_read_entry(struct obd_export *exp, struct md_op_data *op_data,
                        op_data->op_stripe_offset = i;
                }
 
-               rc = md_read_entry(tgt->ltd_exp, op_data, cb_op, &ents[i]);
+               rc = md_read_entry(tgt->ltd_exp, op_data, cb_op, &ents[i],
+                                  &page);
                if (rc != 0)
                        GOTO(out, rc);
 
                if (ents[i] != NULL &&
                    le64_to_cpu(ents[i]->lde_hash) <= min_hash) {
+                       if (min_page != NULL)
+                               page_cache_release(min_page);
+                       min_page = page;
                        min_hash = le64_to_cpu(ents[i]->lde_hash);
                        min_idx = i;
                }
@@ -2357,6 +2364,13 @@ out:
        if (stripe_count > NORMAL_MAX_STRIPES && ents != NULL)
                OBD_FREE(ents, sizeof(ents[0]) * stripe_count);
 
+       if (rc != 0 && min_page != NULL) {
+               kunmap(min_page);
+               page_cache_release(min_page);
+       } else {
+               *ppage = min_page;
+       }
+
        RETURN(rc);
 }
 

diff --git a/lustre/mdc/mdc_request.c b/lustre/mdc/mdc_request.c
index fcc2217..e111015 100644 (file)
--- a/lustre/mdc/mdc_request.c
+++ b/lustre/mdc/mdc_request.c
@@ -1185,7 +1185,6 @@ restart_bulk:
 #ifdef __KERNEL__
 static void mdc_release_page(struct page *page, int remove)
 {
-       kunmap(page);
        if (remove) {
                lock_page(page);
                if (likely(page->mapping != NULL))
@@ -1239,6 +1238,7 @@ static struct page *mdc_page_locate(struct address_space *mapping, __u64 *hash,
                        CDEBUG(D_VFSTRACE, "page%lu [%llu %llu], hash"LPU64"\n",
                               offset, *start, *end, *hash);
                        if (*hash > *end) {
+                               kunmap(page);
                                mdc_release_page(page, 0);
                                page = NULL;
                        } else if (*end != *start && *hash == *end) {
@@ -1248,6 +1248,7 @@ static struct page *mdc_page_locate(struct address_space *mapping, __u64 *hash,
                                 * ll_get_dir_page() will issue RPC to fetch
                                 * the page we want.
                                 */
+                               kunmap(page);
                                mdc_release_page(page,
                                    le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
                                page = NULL;
@@ -1637,37 +1638,38 @@ fail:
  * Read one directory entry from the cache.
  */
 int mdc_read_entry(struct obd_export *exp, struct md_op_data *op_data,
-                  struct md_callback *cb_op, struct lu_dirent **entp)
+                  struct md_callback *cb_op, struct lu_dirent **entp,
+                  struct page **ppage)
 {
        struct page             *page = NULL;
        struct lu_dirpage       *dp;
        struct lu_dirent        *ent;
        int                     rc = 0;
-       int                     index = 0;
        ENTRY;
 
-       if (op_data->op_hash_offset == MDS_DIR_END_OFF) {
-               *entp = NULL;
+       CDEBUG(D_INFO, DFID "offset = "LPU64"\n", PFID(&op_data->op_fid1),
+              op_data->op_hash_offset);
+
+       *ppage = NULL;
+       *entp = NULL;
+
+       if (op_data->op_hash_offset == MDS_DIR_END_OFF)
                RETURN(0);
-       }
 
        rc = mdc_read_page(exp, op_data, cb_op, &page);
        if (rc != 0)
                RETURN(rc);
 
-       if (op_data->op_cli_flags & CLI_READENT_END) {
-               mdc_release_page(page, 0);
-               RETURN(0);
-       }
-
-       dp = kmap(page);
+       dp = page_address(page);
        for (ent = lu_dirent_start(dp); ent != NULL;
             ent = lu_dirent_next(ent)) {
-               index++;
-               if (ent->lde_hash > op_data->op_hash_offset)
+               /* Skip dummy entry */
+               if (le16_to_cpu(ent->lde_namelen) == 0)
+                       continue;
+
+               if (le64_to_cpu(ent->lde_hash) > op_data->op_hash_offset)
                        break;
        }
-       kunmap(page);
 
        /* If it can not find entry in current page, try next page. */
        if (ent == NULL) {
@@ -1686,14 +1688,14 @@ int mdc_read_entry(struct obd_export *exp, struct md_op_data *op_data,
                        RETURN(rc);
 
                if (page != NULL) {
-                       dp = kmap(page);
+                       dp = page_address(page);
                        ent = lu_dirent_start(dp);
-                       kunmap(page);
                }
 
                op_data->op_hash_offset = orig_offset;
        }
 
+       *ppage = page;
        *entp = ent;
 
        RETURN(rc);
@@ -1746,7 +1748,8 @@ static int mdc_read_page(struct obd_export *exp, struct md_op_data *op_data,
 }
 
 int mdc_read_entry(struct obd_export *exp, struct md_op_data *op_data,
-                  struct md_callback *cb_op, struct lu_dirent **entp)
+                  struct md_callback *cb_op, struct lu_dirent **entp,
+                  struct page **ppage)
 {
        struct page             *page = NULL;
        struct lu_dirpage       *dp;