ORNL-25 process dir page hash collision

If dir page has hash collision with others, then remove such page
from cache after using to avoid to be found unexpectedly later.

Signed-off-by: Fan Yong <yong.fan@whamcloud.com>
Change-Id: I15ff85e5233248944d77a9d93292d8690e1a715f
Reviewed-on: http://review.whamcloud.com/1234
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/llite/dcache.c b/lustre/llite/dcache.c
index 87cc18f..e87f554 100644 (file)
--- a/lustre/llite/dcache.c
+++ b/lustre/llite/dcache.c
@@ -48,7 +48,6 @@
 #include <lustre_lite.h>
 #include <lustre/lustre_idl.h>
 #include <lustre_dlm.h>
-#include <lustre_mdc.h>
 //#include <lustre_ver.h>
 //#include <lustre_version.h>
 

diff --git a/lustre/llite/dir.c b/lustre/llite/dir.c
index 62f330f..99398d6 100644 (file)
--- a/lustre/llite/dir.c
+++ b/lustre/llite/dir.c
@@ -179,15 +179,7 @@ static int ll_dir_readpage(struct file *file, struct page *page0)
 
                 hash = fd->fd_dir.lfd_next;
         } else {
-                struct ll_inode_info *lli = ll_i2info(inode);
-
-                cfs_spin_lock(&lli->lli_sa_lock);
-                if (lli->lli_sai)
-                        LASSERT(lli->lli_sai->sai_pid == cfs_curproc_pid());
-                else
-                        LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
-                hash = lli->lli_sa_pos;
-                cfs_spin_unlock(&lli->lli_sa_lock);
+                hash = ll_i2info(inode)->lli_sa_pos;
         }
         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p) hash "LPU64"\n",
                inode->i_ino, inode->i_generation, inode, hash);
@@ -287,19 +279,14 @@ static void ll_check_page(struct inode *dir, struct page *page)
         SetPageChecked(page);
 }
 
-static void ll_release_page(struct page *page, __u64 hash,
-                            __u64 start, __u64 end)
+void ll_release_page(struct page *page, int remove)
 {
         kunmap(page);
-        lock_page(page);
-        if (likely(page->mapping != NULL)) {
-                ll_truncate_complete_page(page);
-                unlock_page(page);
-        } else {
+        if (remove) {
+                lock_page(page);
+                if (likely(page->mapping != NULL))
+                        ll_truncate_complete_page(page);
                 unlock_page(page);
-                CWARN("NULL mapping page %p, truncated by others: "
-                      "hash("LPX64") | start("LPX64") | end("LPX64")\n",
-                      page, hash, start, end);
         }
         page_cache_release(page);
 }
@@ -352,19 +339,18 @@ static struct page *ll_dir_page_locate(struct inode *dir, __u64 *hash,
                                  LPX64",hash = "LPX64"\n", *start, *end, *hash);
                         CDEBUG(D_VFSTRACE, "page %lu [%llu %llu], hash "LPU64"\n",
                                offset, *start, *end, *hash);
-                        if (*hash > *end || (*end != *start && *hash == *end)) {
+                        if (*hash > *end) {
+                                ll_release_page(page, 0);
+                                page = NULL;
+                        } else if (*end != *start && *hash == *end) {
                                 /*
                                  * upon hash collision, remove this page,
                                  * otherwise put page reference, and
                                  * ll_get_dir_page() will issue RPC to fetch
                                  * the page we want.
                                  */
-                                if (dp->ldp_flags & cpu_to_le32(LDF_COLLIDE)) {
-                                        ll_release_page(page, *hash, *start, *end);
-                                } else {
-                                        cfs_kunmap(page);
-                                        page_cache_release(page);
-                                }
+                                ll_release_page(page,
+                                    le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
                                 page = NULL;
                         }
                 } else {
@@ -380,7 +366,7 @@ static struct page *ll_dir_page_locate(struct inode *dir, __u64 *hash,
 }
 
 struct page *ll_get_dir_page(struct file *filp, struct inode *dir, __u64 hash,
-                             int exact, struct ll_dir_chain *chain)
+                             struct ll_dir_chain *chain)
 {
         ldlm_policy_data_t policy = {.l_inodebits = {MDS_INODELOCK_UPDATE} };
         struct address_space *mapping = dir->i_mapping;
@@ -438,9 +424,7 @@ struct page *ll_get_dir_page(struct file *filp, struct inode *dir, __u64 hash,
                 CERROR("dir page locate: "DFID" at "LPU64": rc %ld\n",
                        PFID(ll_inode2fid(dir)), lhash, PTR_ERR(page));
                 GOTO(out_unlock, page);
-        }
-
-        if (page != NULL) {
+        } else if (page != NULL) {
                 /*
                  * XXX nikita: not entirely correct handling of a corner case:
                  * suppose hash chain of entries with hash value HASH crosses
@@ -455,20 +439,7 @@ struct page *ll_get_dir_page(struct file *filp, struct inode *dir, __u64 hash,
                  * it as an "overflow" page. 1. invalidate all pages at
                  * once. 2. use HASH|1 as an index for P1.
                  */
-                if (exact && lhash != start) {
-                        /*
-                         * readdir asked for a page starting _exactly_ from
-                         * given hash, but cache contains stale page, with
-                         * entries with smaller hash values. Stale page should
-                         * be invalidated, and new one fetched.
-                         */
-                        CDEBUG(D_OTHER, "Stale readpage page %p: "
-                               "start = "LPX64",end = "LPX64"hash ="LPX64"\n",
-                               page, start, end, lhash);
-                        ll_release_page(page, lhash, start, end);
-                } else {
-                        GOTO(hash_collision, page);
-                }
+                GOTO(hash_collision, page);
         }
 
         page = read_cache_page(mapping, hash_x_index(hash, hash64),
@@ -525,7 +496,7 @@ out_unlock:
         return page;
 
 fail:
-        ll_put_page(page);
+        ll_release_page(page, 1);
         page = ERR_PTR(-EIO);
         goto out_unlock;
 }
@@ -562,7 +533,7 @@ int ll_readdir(struct file *filp, void *cookie, filldir_t filldir)
         ll_dir_chain_init(&chain);
 
         fd->fd_dir.lfd_next = pos;
-        page = ll_get_dir_page(filp, inode, pos, 0, &chain);
+        page = ll_get_dir_page(filp, inode, pos, &chain);
 
         while (rc == 0 && !done) {
                 struct lu_dirpage *dp;
@@ -619,7 +590,6 @@ int ll_readdir(struct file *filp, void *cookie, filldir_t filldir)
                                                lhash, ino, type);
                         }
                         next = le64_to_cpu(dp->ldp_hash_end);
-                        ll_put_page(page);
                         if (!done) {
                                 pos = next;
                                 if (pos == MDS_DIR_END_OFF) {
@@ -627,21 +597,29 @@ int ll_readdir(struct file *filp, void *cookie, filldir_t filldir)
                                          * End of directory reached.
                                          */
                                         done = 1;
+                                        ll_release_page(page, 0);
                                 } else if (1 /* chain is exhausted*/) {
                                         /*
                                          * Normal case: continue to the next
                                          * page.
                                          */
+                                        ll_release_page(page,
+                                            le32_to_cpu(dp->ldp_flags) &
+                                                        LDF_COLLIDE);
                                         fd->fd_dir.lfd_next = pos;
                                         page = ll_get_dir_page(filp, inode, pos,
-                                                               1, &chain);
+                                                               &chain);
                                 } else {
                                         /*
                                          * go into overflow page.
                                          */
+                                        LASSERT(le32_to_cpu(dp->ldp_flags) &
+                                                LDF_COLLIDE);
+                                        ll_release_page(page, 1);
                                 }
                         } else {
                                 pos = hash;
+                                ll_release_page(page, 0);
                         }
                 } else {
                         rc = PTR_ERR(page);

diff --git a/lustre/llite/file.c b/lustre/llite/file.c
index 7d35947..cb036bf 100644 (file)
--- a/lustre/llite/file.c
+++ b/lustre/llite/file.c
@@ -46,7 +46,6 @@
 #define DEBUG_SUBSYSTEM S_LLITE
 #include <lustre_dlm.h>
 #include <lustre_lite.h>
-#include <lustre_mdc.h>
 #include <linux/pagemap.h>
 #include <linux/file.h>
 #include "llite_internal.h"

diff --git a/lustre/llite/llite_close.c b/lustre/llite/llite_close.c
index d2a6b0b..b96081e 100644 (file)
--- a/lustre/llite/llite_close.c
+++ b/lustre/llite/llite_close.c
@@ -42,7 +42,6 @@
 
 #define DEBUG_SUBSYSTEM S_LLITE
 
-//#include <lustre_mdc.h>
 #include <lustre_lite.h>
 #include "llite_internal.h"
 

diff --git a/lustre/llite/llite_internal.h b/lustre/llite/llite_internal.h
index b488b2e..755c708 100644 (file)
--- a/lustre/llite/llite_internal.h
+++ b/lustre/llite/llite_internal.h
@@ -50,6 +50,7 @@
 /* for struct cl_lock_descr and struct cl_io */
 #include <cl_object.h>
 #include <lclient.h>
+#include <lustre_mdc.h>
 
 #ifndef FMODE_EXEC
 #define FMODE_EXEC 0
@@ -594,16 +595,11 @@ static void lprocfs_llite_init_vars(struct lprocfs_static_vars *lvars)
 
 
 /* llite/dir.c */
-static inline void ll_put_page(struct page *page)
-{
-        kunmap(page);
-        page_cache_release(page);
-}
-
+void ll_release_page(struct page *page, int remove);
 extern struct file_operations ll_dir_operations;
 extern struct inode_operations ll_dir_inode_operations;
 struct page *ll_get_dir_page(struct file *filp, struct inode *dir, __u64 hash,
-                             int exact, struct ll_dir_chain *chain);
+                             struct ll_dir_chain *chain);
 int ll_readdir(struct file *filp, void *cookie, filldir_t filldir);
 
 int ll_get_mdt_idx(struct inode *inode);
@@ -702,17 +698,12 @@ int ll_put_grouplock(struct inode *inode, struct file *file, unsigned long arg);
 int ll_fid2path(struct obd_export *exp, void *arg);
 
 /* llite/dcache.c */
-/* llite/namei.c */
-/**
- * protect race ll_find_aliases vs ll_revalidate_it vs ll_unhash_aliases
- */
 int ll_dops_init(struct dentry *de, int block, int init_sa);
 extern cfs_spinlock_t ll_lookup_lock;
 extern struct dentry_operations ll_d_ops;
 void ll_intent_drop_lock(struct lookup_intent *);
 void ll_intent_release(struct lookup_intent *);
 int ll_drop_dentry(struct dentry *dentry);
-int ll_drop_dentry(struct dentry *dentry);
 void ll_unhash_aliases(struct inode *);
 void ll_frob_intent(struct lookup_intent **itp, struct lookup_intent *deft);
 void ll_lookup_finish_locks(struct lookup_intent *it, struct dentry *dentry);
@@ -1153,7 +1144,6 @@ struct ll_statahead_info {
         cfs_list_t              sai_entries_sent;     /* entries sent out */
         cfs_list_t              sai_entries_received; /* entries returned */
         cfs_list_t              sai_entries_stated;   /* entries stated */
-        pid_t                   sai_pid;        /* pid of statahead itself */
 };
 
 int do_statahead_enter(struct inode *dir, struct dentry **dentry, int lookup);

diff --git a/lustre/llite/llite_mmap.c b/lustre/llite/llite_mmap.c
index 4424382..981c7d7 100644 (file)
--- a/lustre/llite/llite_mmap.c
+++ b/lustre/llite/llite_mmap.c
@@ -60,7 +60,6 @@
 
 #define DEBUG_SUBSYSTEM S_LLITE
 
-//#include <lustre_mdc.h>
 #include <lustre_lite.h>
 #include "llite_internal.h"
 #include <linux/lustre_compat25.h>

diff --git a/lustre/llite/namei.c b/lustre/llite/namei.c
index 3e12e5c..783283c 100644 (file)
--- a/lustre/llite/namei.c
+++ b/lustre/llite/namei.c
@@ -52,7 +52,6 @@
 #include <lustre_lite.h>
 #include <lustre_dlm.h>
 #include <lustre_ver.h>
-#include <lustre_mdc.h>
 #include "llite_internal.h"
 
 /*

diff --git a/lustre/llite/rw.c b/lustre/llite/rw.c
index f8fab63..fcef728 100644 (file)
--- a/lustre/llite/rw.c
+++ b/lustre/llite/rw.c
@@ -61,7 +61,6 @@
 
 #define DEBUG_SUBSYSTEM S_LLITE
 
-//#include <lustre_mdc.h>
 #include <lustre_lite.h>
 #include <obd_cksum.h>
 #include "llite_internal.h"

diff --git a/lustre/llite/rw26.c b/lustre/llite/rw26.c
index 15df5fc..999fd27 100644 (file)
--- a/lustre/llite/rw26.c
+++ b/lustre/llite/rw26.c
@@ -66,7 +66,6 @@
 
 #define DEBUG_SUBSYSTEM S_LLITE
 
-//#include <lustre_mdc.h>
 #include <lustre_lite.h>
 #include "llite_internal.h"
 #include <linux/lustre_compat25.h>

diff --git a/lustre/llite/statahead.c b/lustre/llite/statahead.c
index 430a446..0d83b1e 100644 (file)
--- a/lustre/llite/statahead.c
+++ b/lustre/llite/statahead.c
@@ -767,10 +767,9 @@ static int ll_statahead_thread(void *arg)
         cfs_waitq_signal(&thread->t_ctl_waitq);
         CDEBUG(D_READA, "start doing statahead for %s\n", parent->d_name.name);
 
-        sai->sai_pid = cfs_curproc_pid();
         lli->lli_sa_pos = 0;
         ll_dir_chain_init(&chain);
-        page = ll_get_dir_page(NULL, dir, pos, 0, &chain);
+        page = ll_get_dir_page(NULL, dir, pos, &chain);
 
         while (1) {
                 struct l_wait_info lwi = { 0 };
@@ -846,7 +845,7 @@ keep_de:
                                 do_statahead_interpret(sai);
 
                         if (unlikely(!sa_is_running(sai))) {
-                                ll_put_page(page);
+                                ll_release_page(page, 0);
                                 GOTO(out, rc);
                         }
 
@@ -858,16 +857,16 @@ keep_de:
 
                         rc = ll_statahead_one(parent, name, namelen);
                         if (rc < 0) {
-                                ll_put_page(page);
+                                ll_release_page(page, 0);
                                 GOTO(out, rc);
                         }
                 }
                 pos = le64_to_cpu(dp->ldp_hash_end);
-                ll_put_page(page);
                 if (pos == MDS_DIR_END_OFF) {
                         /*
                          * End of directory reached.
                          */
+                        ll_release_page(page, 0);
                         while (1) {
                                 l_wait_event(thread->t_ctl_waitq,
                                              !sa_is_running(sai) ||
@@ -885,9 +884,13 @@ keep_de:
                          * chain is exhausted.
                          * Normal case: continue to the next page.
                          */
+                        ll_release_page(page, le32_to_cpu(dp->ldp_flags) &
+                                              LDF_COLLIDE);
                         lli->lli_sa_pos = pos;
-                        page = ll_get_dir_page(NULL, dir, pos, 1, &chain);
+                        page = ll_get_dir_page(NULL, dir, pos, &chain);
                 } else {
+                        LASSERT(le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
+                        ll_release_page(page, 1);
                         /*
                          * go into overflow page.
                          */
@@ -985,7 +988,7 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry)
 
         lli->lli_sa_pos = 0;
         ll_dir_chain_init(&chain);
-        page = ll_get_dir_page(NULL, dir, pos, 0, &chain);
+        page = ll_get_dir_page(NULL, dir, pos, &chain);
 
         while (1) {
                 struct lu_dirpage *dp;
@@ -1048,27 +1051,31 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry)
                         else
                                 rc = LS_FIRST_DOT_DE;
 
-                        ll_put_page(page);
+                        ll_release_page(page, 0);
                         GOTO(out, rc);
                 }
                 pos = le64_to_cpu(dp->ldp_hash_end);
-                ll_put_page(page);
                 if (pos == MDS_DIR_END_OFF) {
                         /*
                          * End of directory reached.
                          */
+                        ll_release_page(page, 0);
                         break;
                 } else if (1) {
                         /*
                          * chain is exhausted
                          * Normal case: continue to the next page.
                          */
+                        ll_release_page(page, le32_to_cpu(dp->ldp_flags) &
+                                              LDF_COLLIDE);
                         lli->lli_sa_pos = pos;
-                        page = ll_get_dir_page(NULL, dir, pos, 1, &chain);
+                        page = ll_get_dir_page(NULL, dir, pos, &chain);
                 } else {
                         /*
                          * go into overflow page.
                          */
+                        LASSERT(le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
+                        ll_release_page(page, 1);
                 }
         }
         EXIT;

diff --git a/lustre/llite/xattr.c b/lustre/llite/xattr.c
index 94b48bf..013a1b3 100644 (file)
--- a/lustre/llite/xattr.c
+++ b/lustre/llite/xattr.c
@@ -45,7 +45,6 @@
 #include <lustre_lite.h>
 #include <lustre_dlm.h>
 #include <lustre_ver.h>
-//#include <lustre_mdc.h>
 #include <lustre_acl.h>
 
 #include "llite_internal.h"