Whamcloud - gitweb
b=19151
[fs/lustre-release.git] / lustre / llite / dcache.c
index 019ff46..cf1f734 100644 (file)
@@ -1,22 +1,37 @@
 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
  * vim:expandtab:shiftwidth=8:tabstop=8:
  *
- *  Copyright (c) 2001-2003 Cluster File Systems, Inc.
+ * GPL HEADER START
  *
- *   This file is part of Lustre, http://www.lustre.org.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
- *   Lustre is free software; you can redistribute it and/or
- *   modify it under the terms of version 2 of the GNU General Public
- *   License as published by the Free Software Foundation.
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 only,
+ * as published by the Free Software Foundation.
  *
- *   Lustre is distributed in the hope that it will be useful,
- *   but WITHOUT ANY WARRANTY; without even the implied warranty of
- *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *   GNU General Public License for more details.
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License version 2 for more details (a copy is included
+ * in the LICENSE file that accompanied this code).
  *
- *   You should have received a copy of the GNU General Public License
- *   along with Lustre; if not, write to the Free Software
- *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ * You should have received a copy of the GNU General Public License
+ * version 2 along with this program; If not, see
+ * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
+ *
+ * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
+ * CA 95054 USA or visit www.sun.com if you need additional information or
+ * have any questions.
+ *
+ * GPL HEADER END
+ */
+/*
+ * Copyright  2008 Sun Microsystems, Inc. All rights reserved
+ * Use is subject to license terms.
+ */
+/*
+ * This file is part of Lustre, http://www.lustre.org/
+ * Lustre is a trademark of Sun Microsystems, Inc.
  */
 
 #include <linux/fs.h>
 #include <lustre_lite.h>
 #include <lustre/lustre_idl.h>
 #include <lustre_dlm.h>
-#include <linux/lustre_version.h>
+#include <lustre_mdc.h>
+//#include <lustre_ver.h>
+//#include <lustre_version.h>
 
 #include "llite_internal.h"
 
+spinlock_t ll_lookup_lock = SPIN_LOCK_UNLOCKED;
+
 /* should NOT be called with the dcache lock, see fs/dcache.c */
 static void ll_release(struct dentry *de)
 {
@@ -45,7 +64,7 @@ static void ll_release(struct dentry *de)
                 EXIT;
                 return;
         }
-#ifndef LUSTRE_KERNEL_VERSION
+#ifndef HAVE_VFS_INTENT_PATCHES
         if (lld->lld_it) {
                 ll_intent_release(lld->lld_it);
                 OBD_FREE(lld->lld_it, sizeof(*lld->lld_it));
@@ -58,7 +77,7 @@ static void ll_release(struct dentry *de)
         EXIT;
 }
 
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef DCACHE_LUSTRE_INVALID
 /* Compare if two dentries are the same.  Don't match if the existing dentry
  * is marked DCACHE_LUSTRE_INVALID.  Returns 1 if different, 0 if the same.
  *
@@ -118,11 +137,20 @@ void ll_set_dd(struct dentry *de)
         CDEBUG(D_DENTRY, "ldd on dentry %.*s (%p) parent %p inode %p refc %d\n",
                de->d_name.len, de->d_name.name, de, de->d_parent, de->d_inode,
                atomic_read(&de->d_count));
-        lock_kernel();
+
         if (de->d_fsdata == NULL) {
-                OBD_ALLOC(de->d_fsdata, sizeof(struct ll_dentry_data));
+                struct ll_dentry_data *lld;
+
+                OBD_ALLOC_PTR(lld);
+                if (likely(lld != NULL)) {
+                        lock_dentry(de);
+                        if (likely(de->d_fsdata == NULL))
+                                de->d_fsdata = lld;
+                        else
+                                OBD_FREE_PTR(lld);
+                        unlock_dentry(de);
+                }
         }
-        unlock_kernel();
 
         EXIT;
 }
@@ -147,14 +175,15 @@ void ll_intent_release(struct lookup_intent *it)
 {
         ENTRY;
 
+        CDEBUG(D_INFO, "intent %p released\n", it);
         ll_intent_drop_lock(it);
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef HAVE_VFS_INTENT_PATCHES
         it->it_magic = 0;
         it->it_op_release = 0;
 #endif
         /* We are still holding extra reference on a request, need to free it */
-        if (it_disposition(it, DISP_ENQ_OPEN_REF)) /* open req for llfile_open*/
-                ptlrpc_req_finished(it->d.lustre.it_data);
+        if (it_disposition(it, DISP_ENQ_OPEN_REF))
+                 ptlrpc_req_finished(it->d.lustre.it_data); /* ll_file_open */
         if (it_disposition(it, DISP_ENQ_CREATE_REF)) /* create rec */
                 ptlrpc_req_finished(it->d.lustre.it_data);
         if (it_disposition(it, DISP_ENQ_COMPLETE)) /* saved req from revalidate
@@ -181,12 +210,20 @@ int ll_drop_dentry(struct dentry *dentry)
                 __d_drop(dentry);
                 unlock_dentry(dentry);
                 spin_unlock(&dcache_lock);
+                spin_unlock(&ll_lookup_lock);
                 dput(dentry);
+                spin_lock(&ll_lookup_lock);
                 spin_lock(&dcache_lock);
                 return 1;
         }
+        /* disconected dentry can not be find without lookup, because we
+         * not need his to unhash or mark invalid. */
+        if (dentry->d_flags & DCACHE_DISCONNECTED) {
+                unlock_dentry(dentry);
+                RETURN (0);
+        }
 
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef DCACHE_LUSTRE_INVALID
         if (!(dentry->d_flags & DCACHE_LUSTRE_INVALID)) {
 #else
         if (!d_unhashed(dentry)) {
@@ -198,20 +235,11 @@ int ll_drop_dentry(struct dentry *dentry)
                 /* actually we don't unhash the dentry, rather just
                  * mark it inaccessible for to __d_lookup(). otherwise
                  * sys_getcwd() could return -ENOENT -bzzz */
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef DCACHE_LUSTRE_INVALID
                 dentry->d_flags |= DCACHE_LUSTRE_INVALID;
-#if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
-                __d_drop(dentry);
-                if (dentry->d_inode) {
-                        /* Put positive dentries to orphan list */
-                        list_add(&dentry->d_hash,
-                                 &ll_i2sbi(dentry->d_inode)->ll_orphan_dentry_list);
-                }
 #endif
-#else
                 if (!dentry->d_inode || !S_ISDIR(dentry->d_inode->i_mode))
                         __d_drop(dentry);
-#endif
 
         }
         unlock_dentry(dentry);
@@ -232,12 +260,18 @@ void ll_unhash_aliases(struct inode *inode)
                inode->i_ino, inode->i_generation, inode);
 
         head = &inode->i_dentry;
+        spin_lock(&ll_lookup_lock);
         spin_lock(&dcache_lock);
 restart:
         tmp = head;
         while ((tmp = tmp->next) != head) {
                 struct dentry *dentry = list_entry(tmp, struct dentry, d_alias);
 
+                CDEBUG(D_DENTRY, "dentry in drop %.*s (%p) parent %p "
+                       "inode %p flags %d\n", dentry->d_name.len,
+                       dentry->d_name.name, dentry, dentry->d_parent,
+                       dentry->d_inode, dentry->d_flags);
+
                 if (dentry->d_name.len == 1 && dentry->d_name.name[0] == '/') {
                         CERROR("called on root (?) dentry=%p, inode=%p "
                                "ino=%lu\n", dentry, inode, inode->i_ino);
@@ -255,16 +289,19 @@ restart:
 
                         continue;
                 }
-                
+
                 if (ll_drop_dentry(dentry))
                           goto restart;
         }
         spin_unlock(&dcache_lock);
+        spin_unlock(&ll_lookup_lock);
+
         EXIT;
 }
 
-int revalidate_it_finish(struct ptlrpc_request *request, int offset,
-                         struct lookup_intent *it, struct dentry *de)
+int ll_revalidate_it_finish(struct ptlrpc_request *request,
+                            struct lookup_intent *it,
+                            struct dentry *de)
 {
         int rc = 0;
         ENTRY;
@@ -275,8 +312,7 @@ int revalidate_it_finish(struct ptlrpc_request *request, int offset,
         if (it_disposition(it, DISP_LOOKUP_NEG))
                 RETURN(-ENOENT);
 
-        rc = ll_prep_inode(ll_i2sbi(de->d_inode)->ll_osc_exp, &de->d_inode,
-                           request, offset, NULL);
+        rc = ll_prep_inode(&de->d_inode, request, NULL);
 
         RETURN(rc);
 }
@@ -288,38 +324,38 @@ void ll_lookup_finish_locks(struct lookup_intent *it, struct dentry *dentry)
 
         if (it->d.lustre.it_lock_mode && dentry->d_inode != NULL) {
                 struct inode *inode = dentry->d_inode;
+                struct ll_sb_info *sbi = ll_i2sbi(dentry->d_inode);
+
                 CDEBUG(D_DLMTRACE, "setting l_data to inode %p (%lu/%u)\n",
                        inode, inode->i_ino, inode->i_generation);
-                mdc_set_lock_data(&it->d.lustre.it_lock_handle, inode);
+                md_set_lock_data(sbi->ll_md_exp, &it->d.lustre.it_lock_handle,
+                                 inode);
         }
 
         /* drop lookup or getattr locks immediately */
         if (it->it_op == IT_LOOKUP || it->it_op == IT_GETATTR) {
-#if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
                 /* on 2.6 there are situation when several lookups and
                  * revalidations may be requested during single operation.
                  * therefore, we don't release intent here -bzzz */
                 ll_intent_drop_lock(it);
-#else
-                ll_intent_release(it);
-#endif
         }
 }
 
 void ll_frob_intent(struct lookup_intent **itp, struct lookup_intent *deft)
 {
         struct lookup_intent *it = *itp;
-#if defined(LUSTRE_KERNEL_VERSION)&&(LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
+#ifdef HAVE_VFS_INTENT_PATCHES
         if (it) {
-                LASSERTF(it->it_magic == INTENT_MAGIC, "bad intent magic: %x\n",
-                         it->it_magic);
+                LASSERTF(it->it_magic == INTENT_MAGIC,
+                         "%p has bad intent magic: %x\n",
+                         it, it->it_magic);
         }
 #endif
 
         if (!it || it->it_op == IT_GETXATTR)
                 it = *itp = deft;
 
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef HAVE_VFS_INTENT_PATCHES
         it->it_op_release = ll_intent_release;
 #endif
 }
@@ -327,11 +363,12 @@ void ll_frob_intent(struct lookup_intent **itp, struct lookup_intent *deft)
 int ll_revalidate_it(struct dentry *de, int lookup_flags,
                      struct lookup_intent *it)
 {
-        int rc;
-        struct mdc_op_data op_data;
+        struct md_op_data *op_data;
         struct ptlrpc_request *req = NULL;
         struct lookup_intent lookup_it = { .it_op = IT_LOOKUP };
         struct obd_export *exp;
+        struct inode *parent;
+        int rc, first = 0;
 
         ENTRY;
         CDEBUG(D_VFSTRACE, "VFS Op:name=%s,intent=%s\n", de->d_name.name,
@@ -346,48 +383,58 @@ int ll_revalidate_it(struct dentry *de, int lookup_flags,
                 if (it && (it->it_op & IT_CREAT))
                         RETURN(0);
 
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef DCACHE_LUSTRE_INVALID
                 if (de->d_flags & DCACHE_LUSTRE_INVALID)
                         RETURN(0);
 #endif
 
-                rc = ll_have_md_lock(de->d_parent->d_inode, 
+                rc = ll_have_md_lock(de->d_parent->d_inode,
                                      MDS_INODELOCK_UPDATE);
-        
-                RETURN(rc);
+                GOTO(out_sa, rc);
         }
 
-        exp = ll_i2mdcexp(de->d_inode);
-
         /* Never execute intents for mount points.
          * Attributes will be fixed up in ll_inode_revalidate_it */
         if (d_mountpoint(de))
-                RETURN(1);
+                GOTO(out_sa, rc = 1);
+
+        /* need to get attributes in case root got changed from other client */
+        if (de == de->d_sb->s_root) {
+                rc = __ll_inode_revalidate_it(de, it, MDS_INODELOCK_LOOKUP);
+                if (rc == 0)
+                        rc = 1;
+                GOTO(out_sa, rc);
+        }
 
-        /* Root of the lustre tree. Always valid.
-         * Attributes will be fixed up in ll_inode_revalidate_it */
-        if (de->d_name.name[0] == '/' && de->d_name.len == 1)
-                RETURN(1);
+        exp = ll_i2mdexp(de->d_inode);
 
         OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_REVALIDATE_PAUSE, 5);
         ll_frob_intent(&it, &lookup_it);
         LASSERT(it);
+        parent = de->d_parent->d_inode;
 
-        ll_prepare_mdc_op_data(&op_data, de->d_parent->d_inode, de->d_inode,
-                               de->d_name.name, de->d_name.len, 0);
+        op_data = ll_prep_md_op_data(NULL, parent, de->d_inode,
+                                     de->d_name.name, de->d_name.len,
+                                     0, LUSTRE_OPC_ANY, NULL);
+        if (IS_ERR(op_data))
+                RETURN(PTR_ERR(op_data));
 
         if ((it->it_op == IT_OPEN) && de->d_inode) {
                 struct inode *inode = de->d_inode;
                 struct ll_inode_info *lli = ll_i2info(inode);
                 struct obd_client_handle **och_p;
                 __u64 *och_usecount;
-                /* We used to check for MDS_INODELOCK_OPEN here, but in fact
+
+                /*
+                 * We used to check for MDS_INODELOCK_OPEN here, but in fact
                  * just having LOOKUP lock is enough to justify inode is the
                  * same. And if inode is the same and we have suitable
                  * openhandle, then there is no point in doing another OPEN RPC
-                 * just to throw away newly received openhandle.
-                 * There are no security implications too, if file owner or
-                 * access mode is change, LOOKUP lock is revoked */
+                 * just to throw away newly received openhandle.  There are no
+                 * security implications too, if file owner or access mode is
+                 * change, LOOKUP lock is revoked.
+                 */
+
 
                 if (it->it_flags & FMODE_WRITE) {
                         och_p = &lli->lli_mds_write_och;
@@ -395,7 +442,7 @@ int ll_revalidate_it(struct dentry *de, int lookup_flags,
                 } else if (it->it_flags & FMODE_EXEC) {
                         och_p = &lli->lli_mds_exec_och;
                         och_usecount = &lli->lli_open_fd_exec_count;
-                 } else {
+                } else {
                         och_p = &lli->lli_mds_read_och;
                         och_usecount = &lli->lli_open_fd_read_count;
                 }
@@ -405,7 +452,7 @@ int ll_revalidate_it(struct dentry *de, int lookup_flags,
                 down(&lli->lli_och_sem);
                 if (*och_p) { /* Everything is open already, do nothing */
                         /*(*och_usecount)++;  Do not let them steal our open
-                                              handle from under us */
+                          handle from under us */
                         /* XXX The code above was my original idea, but in case
                            we have the handle, but we cannot use it due to later
                            checks (e.g. O_CREAT|O_EXCL flags set), nobody
@@ -414,18 +461,33 @@ int ll_revalidate_it(struct dentry *de, int lookup_flags,
                            if it would be, we'll reopen the open request to
                            MDS later during file open path */
                         up(&lli->lli_och_sem);
+                        ll_finish_md_op_data(op_data);
                         RETURN(1);
                 } else {
                         up(&lli->lli_och_sem);
                 }
         }
 
+        if (it->it_op == IT_GETATTR)
+                first = ll_statahead_enter(de->d_parent->d_inode, &de, 0);
+
 do_lock:
         it->it_create_mode &= ~current->fs->umask;
-
-        rc = mdc_intent_lock(exp, &op_data, NULL, 0, it, lookup_flags,
-                             &req, ll_mdc_blocking_ast, 0);
-        /* If req is NULL, then mdc_intent_lock only tried to do a lock match;
+        it->it_create_mode |= M_CHECK_STALE;
+        rc = md_intent_lock(exp, op_data, NULL, 0, it,
+                            lookup_flags,
+                            &req, ll_md_blocking_ast, 0);
+        it->it_create_mode &= ~M_CHECK_STALE;
+        ll_finish_md_op_data(op_data);
+        if (it->it_op == IT_GETATTR && !first)
+                /* If there are too many locks on client-side, then some
+                 * locks taken by statahead maybe dropped automatically
+                 * before the real "revalidate" using them. */
+                ll_statahead_exit(de, req == NULL ? rc : 0);
+        else if (first == -EEXIST)
+                ll_statahead_mark(de);
+
+        /* If req is NULL, then md_intent_lock only tried to do a lock match;
          * if all was well, it will return 1 if it found locks, 0 otherwise. */
         if (req == NULL && rc >= 0) {
                 if (!rc)
@@ -437,18 +499,27 @@ do_lock:
                 if (rc != -ESTALE) {
                         CDEBUG(D_INFO, "ll_intent_lock: rc %d : it->it_status "
                                "%d\n", rc, it->d.lustre.it_status);
+                } else {
+#ifndef HAVE_VFS_INTENT_PATCHES
+                        if (it_disposition(it, DISP_OPEN_OPEN) &&
+                            !it_open_error(DISP_OPEN_OPEN, it))
+                                /* server have valid open - close file first*/
+                                ll_release_openhandle(de, it);
+#endif
                 }
                 GOTO(out, rc = 0);
         }
 
 revalidate_finish:
-        rc = revalidate_it_finish(req, DLM_REPLY_REC_OFF, it, de);
+        rc = ll_revalidate_it_finish(req, it, de);
         if (rc != 0) {
-                ll_intent_release(it);
+                if (rc != -ESTALE && rc != -ENOENT)
+                        ll_intent_release(it);
                 GOTO(out, rc = 0);
         }
-        if ((it->it_op & IT_OPEN) && de->d_inode && 
-            !S_ISREG(de->d_inode->i_mode) && 
+
+        if ((it->it_op & IT_OPEN) && de->d_inode &&
+            !S_ISREG(de->d_inode->i_mode) &&
             !S_ISDIR(de->d_inode->i_mode)) {
                 ll_release_openhandle(de, it);
         }
@@ -456,14 +527,16 @@ revalidate_finish:
 
         /* unfortunately ll_intent_lock may cause a callback and revoke our
          * dentry */
+        spin_lock(&ll_lookup_lock);
         spin_lock(&dcache_lock);
         lock_dentry(de);
         __d_drop(de);
         unlock_dentry(de);
-        __d_rehash(de, 0);
+        d_rehash_cond(de, 0);
         spin_unlock(&dcache_lock);
+        spin_unlock(&ll_lookup_lock);
 
- out:
+out:
         /* We do not free request as it may be reused during following lookup
          * (see comment in mdc/mdc_locks.c::mdc_intent_lock()), request will
          * be freed in ll_lookup_it or in ll_intent_release. But if
@@ -471,60 +544,95 @@ revalidate_finish:
         if (req != NULL && !it_disposition(it, DISP_ENQ_COMPLETE))
                 ptlrpc_req_finished(req);
         if (rc == 0) {
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef DCACHE_LUSTRE_INVALID
                 ll_unhash_aliases(de->d_inode);
                 /* done in ll_unhash_aliases()
-                dentry->d_flags |= DCACHE_LUSTRE_INVALID; */
+                   dentry->d_flags |= DCACHE_LUSTRE_INVALID; */
 #else
                 /* We do not want d_invalidate to kill all child dentries too */
                 d_drop(de);
 #endif
         } else {
                 CDEBUG(D_DENTRY, "revalidated dentry %.*s (%p) parent %p "
-                               "inode %p refc %d\n", de->d_name.len,
-                               de->d_name.name, de, de->d_parent, de->d_inode,
-                               atomic_read(&de->d_count));
+                       "inode %p refc %d\n", de->d_name.len,
+                       de->d_name.name, de, de->d_parent, de->d_inode,
+                       atomic_read(&de->d_count));
                 ll_lookup_finish_locks(it, de);
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef DCACHE_LUSTRE_INVALID
                 lock_dentry(de);
                 de->d_flags &= ~DCACHE_LUSTRE_INVALID;
                 unlock_dentry(de);
 #endif
         }
         RETURN(rc);
-/* This part is here to combat evil-evil race in real_lookup on 2.6 kernels.
- * The race details are: We enter do_lookup() looking for some name,
- * there is nothing in dcache for this name yet and d_lookup() returns NULL.
- * We proceed to real_lookup(), and while we do this, another process does
- * open on the same file we looking up (most simple reproducer), open succeeds
- * and the dentry is added. Now back to us. In real_lookup() we do d_lookup()
- * again and suddenly find the dentry, so we call d_revalidate on it, but there
- * is no lock, so without this code we would return 0, but unpatched
- * real_lookup just returns -ENOENT in such a case instead of retrying the
- * lookup. Once this is dealt with in real_lookup(), all of this ugly mess
- * can go and we can just check locks in ->d_revalidate without doing any
- * RPCs ever. */
+
+        /*
+         * This part is here to combat evil-evil race in real_lookup on 2.6
+         * kernels.  The race details are: We enter do_lookup() looking for some
+         * name, there is nothing in dcache for this name yet and d_lookup()
+         * returns NULL.  We proceed to real_lookup(), and while we do this,
+         * another process does open on the same file we looking up (most simple
+         * reproducer), open succeeds and the dentry is added. Now back to
+         * us. In real_lookup() we do d_lookup() again and suddenly find the
+         * dentry, so we call d_revalidate on it, but there is no lock, so
+         * without this code we would return 0, but unpatched real_lookup just
+         * returns -ENOENT in such a case instead of retrying the lookup. Once
+         * this is dealt with in real_lookup(), all of this ugly mess can go and
+         * we can just check locks in ->d_revalidate without doing any RPCs
+         * ever.
+         */
 do_lookup:
         if (it != &lookup_it) {
+                /* MDS_INODELOCK_UPDATE needed for IT_GETATTR case. */
+                if (it->it_op == IT_GETATTR)
+                        lookup_it.it_op = IT_GETATTR;
                 ll_lookup_finish_locks(it, de);
                 it = &lookup_it;
         }
-        /*do real lookup here */
-        ll_prepare_mdc_op_data(&op_data, de->d_parent->d_inode, NULL,
-                               de->d_name.name, de->d_name.len, 0);
-        rc = mdc_intent_lock(exp, &op_data, NULL, 0,  it, 0, &req,
-                             ll_mdc_blocking_ast, 0);
+
+        /* Do real lookup here. */
+        op_data = ll_prep_md_op_data(NULL, parent, NULL, de->d_name.name,
+                                     de->d_name.len, 0, (it->it_op & IT_CREAT ?
+                                                         LUSTRE_OPC_CREATE :
+                                                         LUSTRE_OPC_ANY), NULL);
+        if (IS_ERR(op_data))
+                RETURN(PTR_ERR(op_data));
+
+        rc = md_intent_lock(exp, op_data, NULL, 0,  it, 0, &req,
+                            ll_md_blocking_ast, 0);
         if (rc >= 0) {
-                struct mds_body *mds_body = lustre_msg_buf(req->rq_repmsg,
-                                                           DLM_REPLY_REC_OFF,
-                                                           sizeof(*mds_body));
+                struct mdt_body *mdt_body;
+                struct lu_fid fid = {.f_seq = 0, .f_oid = 0, .f_ver = 0};
+                mdt_body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
+
+                if (de->d_inode)
+                        fid = *ll_inode2fid(de->d_inode);
+
                 /* see if we got same inode, if not - return error */
-                if(!memcmp(&op_data.fid2, &mds_body->fid1,
-                           sizeof(op_data.fid2)))
+                if (lu_fid_eq(&fid, &mdt_body->fid1)) {
+                        ll_finish_md_op_data(op_data);
+                        op_data = NULL;
                         goto revalidate_finish;
+                }
                 ll_intent_release(it);
         }
+        ll_finish_md_op_data(op_data);
         GOTO(out, rc = 0);
+
+out_sa:
+        /*
+         * For rc == 1 case, should not return directly to prevent losing
+         * statahead windows; for rc == 0 case, the "lookup" will be done later.
+         */
+        if (it && it->it_op == IT_GETATTR && rc == 1) {
+                first = ll_statahead_enter(de->d_parent->d_inode, &de, 0);
+                if (!first)
+                        ll_statahead_exit(de, 1);
+                else if (first == -EEXIST)
+                        ll_statahead_mark(de);
+        }
+
+        return rc;
 }
 
 /*static*/ void ll_pin(struct dentry *de, struct vfsmount *mnt, int flag)
@@ -533,6 +641,7 @@ do_lookup:
         struct ll_sb_info *sbi = ll_i2sbi(inode);
         struct ll_dentry_data *ldd = ll_d2d(de);
         struct obd_client_handle *handle;
+        struct obd_capa *oc;
         int rc = 0;
         ENTRY;
         LASSERT(ldd);
@@ -556,9 +665,9 @@ do_lookup:
         unlock_kernel();
 
         handle = (flag) ? &ldd->lld_mnt_och : &ldd->lld_cwd_och;
-        rc = obd_pin(sbi->ll_mdc_exp, inode->i_ino, inode->i_generation,
-                     inode->i_mode & S_IFMT, handle, flag);
-
+        oc = ll_mdscapa_get(inode);
+        rc = obd_pin(sbi->ll_md_exp, ll_inode2fid(inode), oc, handle, flag);
+        capa_put(oc);
         if (rc) {
                 lock_kernel();
                 memset(handle, 0, sizeof(*handle));
@@ -606,14 +715,13 @@ do_lookup:
                 return;
         }
 
-        rc = obd_unpin(sbi->ll_mdc_exp, &handle, flag);
+        rc = obd_unpin(sbi->ll_md_exp, &handle, flag);
         EXIT;
         return;
 }
 
-#if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
-#ifdef LUSTRE_KERNEL_VERSION
-static int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
+#ifdef HAVE_VFS_INTENT_PATCHES
+int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
 {
         int rc;
         ENTRY;
@@ -656,16 +764,20 @@ int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
                                                (struct ptlrpc_request *)
                                                   it->d.lustre.it_data);
                         } else {
-                                struct file *filp;
-
-                                nd->intent.open.file->private_data = it;
-                                filp = lookup_instantiate_filp(nd, dentry,NULL);
 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,6,17))
 /* 2.6.1[456] have a bug in open_namei() that forgets to check
  * nd->intent.open.file for error, so we need to return it as lookup's result
  * instead */
-                                if (IS_ERR(filp))
-                                        rc = 0;
+                                struct file *filp;
+
+                                nd->intent.open.file->private_data = it;
+                                filp = lookup_instantiate_filp(nd, dentry,NULL);
+                                if (IS_ERR(filp)) {
+                                        rc = PTR_ERR(filp);
+                                }
+#else
+                                nd->intent.open.file->private_data = it;
+                                (void)lookup_instantiate_filp(nd, dentry,NULL);
 #endif
                         }
 #else
@@ -682,7 +794,7 @@ int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
                         ll_d2d(dentry)->lld_it = it;
                         it = NULL; /* avoid freeing */
                 }
-                        
+
 out_it:
                 if (it) {
                         ll_intent_release(it);
@@ -695,17 +807,12 @@ out_it:
         RETURN(rc);
 }
 #endif
-#endif
 
 struct dentry_operations ll_d_ops = {
-#if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
         .d_revalidate = ll_revalidate_nd,
-#else
-        .d_revalidate_it = ll_revalidate_it,
-#endif
         .d_release = ll_release,
         .d_delete = ll_ddelete,
-#ifdef LUSTRE_KERNEL_VERSION
+#ifdef DCACHE_LUSTRE_INVALID
         .d_compare = ll_dcompare,
 #endif
 #if 0