lustre/llite/dcache.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, Whamcloud, Inc.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #include <linux/fs.h>
  38 #include <linux/sched.h>
  39 #include <linux/smp_lock.h>
  40 #include <linux/quotaops.h>
  41
  42 #define DEBUG_SUBSYSTEM S_LLITE
  43
  44 #include <obd_support.h>
  45 #include <lustre_lite.h>
  46 #include <lustre/lustre_idl.h>
  47 #include <lustre_dlm.h>
  48
  49 #include "llite_internal.h"
  50
  51 /* should NOT be called with the dcache lock, see fs/dcache.c */
  52 static void ll_release(struct dentry *de)
  53 {
  54         struct ll_dentry_data *lld;
  55         ENTRY;
  56         LASSERT(de != NULL);
  57         lld = ll_d2d(de);
  58         if (lld == NULL) /* NFS copies the de->d_op methods (bug 4655) */
  59                 RETURN_EXIT;
  60
  61         if (lld->lld_it) {
  62                 ll_intent_release(lld->lld_it);
  63                 OBD_FREE(lld->lld_it, sizeof(*lld->lld_it));
  64         }
  65         LASSERT(lld->lld_cwd_count == 0);
  66         LASSERT(lld->lld_mnt_count == 0);
  67         OBD_FREE(de->d_fsdata, sizeof(*lld));
  68
  69         EXIT;
  70 }
  71
  72 /* Compare if two dentries are the same.  Don't match if the existing dentry
  73  * is marked DCACHE_LUSTRE_INVALID.  Returns 1 if different, 0 if the same.
  74  *
  75  * This avoids a race where ll_lookup_it() instantiates a dentry, but we get
  76  * an AST before calling d_revalidate_it().  The dentry still exists (marked
  77  * INVALID) so d_lookup() matches it, but we have no lock on it (so
  78  * lock_match() fails) and we spin around real_lookup(). */
  79 #ifdef HAVE_D_COMPARE_7ARGS
  80 int ll_dcompare(const struct dentry *parent, const struct inode *pinode,
  81                 const struct dentry *dentry, const struct inode *inode,
  82                 unsigned int len, const char *str, const struct qstr *name)
  83 #else
  84 int ll_dcompare(struct dentry *parent, struct qstr *d_name, struct qstr *name)
  85 #endif
  86 {
  87 #ifdef HAVE_D_COMPARE_7ARGS
  88         ENTRY;
  89
  90         if (len != name->len)
  91                 RETURN(1);
  92
  93         if (memcmp(str, name->name, len))
  94                 RETURN(1);
  95 #else
  96         struct dentry *dentry;
  97         ENTRY;
  98
  99         if (d_name->len != name->len)
 100                 RETURN(1);
 101
 102         if (memcmp(d_name->name, name->name, name->len))
 103                 RETURN(1);
 104
 105         /* XXX: d_name must be in-dentry structure */
 106         dentry = container_of(d_name, struct dentry, d_name); /* ugh */
 107 #endif
 108
 109         CDEBUG(D_DENTRY, "found name %.*s(%p) flags %#x refc %d\n",
 110                name->len, name->name, dentry, dentry->d_flags,
 111                d_refcount(dentry));
 112
 113         /* mountpoint is always valid */
 114         if (d_mountpoint((struct dentry *)dentry))
 115                 RETURN(0);
 116
 117         if (d_lustre_invalid(dentry))
 118                 RETURN(1);
 119
 120         RETURN(0);
 121 }
 122
 123 static inline int return_if_equal(struct ldlm_lock *lock, void *data)
 124 {
 125         if ((lock->l_flags &
 126              (LDLM_FL_CANCELING | LDLM_FL_DISCARD_DATA)) ==
 127             (LDLM_FL_CANCELING | LDLM_FL_DISCARD_DATA))
 128                 return LDLM_ITER_CONTINUE;
 129         return LDLM_ITER_STOP;
 130 }
 131
 132 /* find any ldlm lock of the inode in mdc and lov
 133  * return 0    not find
 134  *        1    find one
 135  *      < 0    error */
 136 static int find_cbdata(struct inode *inode)
 137 {
 138         struct ll_sb_info *sbi = ll_i2sbi(inode);
 139         struct lov_stripe_md *lsm;
 140         int rc = 0;
 141         ENTRY;
 142
 143         LASSERT(inode);
 144         rc = md_find_cbdata(sbi->ll_md_exp, ll_inode2fid(inode),
 145                             return_if_equal, NULL);
 146         if (rc != 0)
 147                  RETURN(rc);
 148
 149         lsm = ccc_inode_lsm_get(inode);
 150         if (lsm == NULL)
 151                 RETURN(rc);
 152
 153         rc = obd_find_cbdata(sbi->ll_dt_exp, lsm, return_if_equal, NULL);
 154         ccc_inode_lsm_put(inode, lsm);
 155
 156         RETURN(rc);
 157 }
 158
 159 /**
 160  * Called when last reference to a dentry is dropped and dcache wants to know
 161  * whether or not it should cache it:
 162  * - return 1 to delete the dentry immediately
 163  * - return 0 to cache the dentry
 164  * Should NOT be called with the dcache lock, see fs/dcache.c
 165  */
 166 static int ll_ddelete(HAVE_D_DELETE_CONST struct dentry *de)
 167 {
 168         ENTRY;
 169         LASSERT(de);
 170
 171         CDEBUG(D_DENTRY, "%s dentry %.*s (%p, parent %p, inode %p) %s%s\n",
 172                d_lustre_invalid((struct dentry *)de) ? "deleting" : "keeping",
 173                de->d_name.len, de->d_name.name, de, de->d_parent, de->d_inode,
 174                d_unhashed((struct dentry *)de) ? "" : "hashed,",
 175                list_empty(&de->d_subdirs) ? "" : "subdirs");
 176
 177 #ifdef HAVE_DCACHE_LOCK
 178         LASSERT(d_refcount(de) == 0);
 179 #else
 180         /* kernel >= 2.6.38 last refcount is decreased after this function. */
 181         LASSERT(d_refcount(de) == 1);
 182 #endif
 183
 184         /* Disable this piece of code temproarily because this is called
 185          * inside dcache_lock so it's not appropriate to do lots of work
 186          * here. */
 187 #if 0
 188         /* if not ldlm lock for this inode, set i_nlink to 0 so that
 189          * this inode can be recycled later b=20433 */
 190         if (de->d_inode && !find_cbdata(de->d_inode))
 191                 de->d_inode->i_nlink = 0;
 192 #endif
 193
 194         if (d_lustre_invalid((struct dentry *)de))
 195                 RETURN(1);
 196         RETURN(0);
 197 }
 198
 199 static int ll_set_dd(struct dentry *de)
 200 {
 201         ENTRY;
 202         LASSERT(de != NULL);
 203
 204         CDEBUG(D_DENTRY, "ldd on dentry %.*s (%p) parent %p inode %p refc %d\n",
 205                 de->d_name.len, de->d_name.name, de, de->d_parent, de->d_inode,
 206                 d_refcount(de));
 207
 208         if (de->d_fsdata == NULL) {
 209                 struct ll_dentry_data *lld;
 210
 211                 OBD_ALLOC_PTR(lld);
 212                 if (likely(lld != NULL)) {
 213                         spin_lock(&de->d_lock);
 214                         if (likely(de->d_fsdata == NULL))
 215                                 de->d_fsdata = lld;
 216                         else
 217                                 OBD_FREE_PTR(lld);
 218                         spin_unlock(&de->d_lock);
 219                 } else {
 220                         RETURN(-ENOMEM);
 221                 }
 222         }
 223
 224         RETURN(0);
 225 }
 226
 227 int ll_dops_init(struct dentry *de, int block, int init_sa)
 228 {
 229         struct ll_dentry_data *lld = ll_d2d(de);
 230         int rc = 0;
 231
 232         if (lld == NULL && block != 0) {
 233                 rc = ll_set_dd(de);
 234                 if (rc)
 235                         return rc;
 236
 237                 lld = ll_d2d(de);
 238         }
 239
 240         if (lld != NULL && init_sa != 0)
 241                 lld->lld_sa_generation = 0;
 242
 243 #ifdef HAVE_DCACHE_LOCK
 244         de->d_op = &ll_d_ops;
 245 #else
 246         /* kernel >= 2.6.38 d_op is set in d_alloc() */
 247         LASSERT(de->d_op == &ll_d_ops);
 248 #endif
 249         return rc;
 250 }
 251
 252 void ll_intent_drop_lock(struct lookup_intent *it)
 253 {
 254         struct lustre_handle *handle;
 255
 256         if (it->it_op && it->d.lustre.it_lock_mode) {
 257                 handle = (struct lustre_handle *)&it->d.lustre.it_lock_handle;
 258                 CDEBUG(D_DLMTRACE, "releasing lock with cookie "LPX64
 259                        " from it %p\n", handle->cookie, it);
 260                 ldlm_lock_decref(handle, it->d.lustre.it_lock_mode);
 261
 262                 /* bug 494: intent_release may be called multiple times, from
 263                  * this thread and we don't want to double-decref this lock */
 264                 it->d.lustre.it_lock_mode = 0;
 265         }
 266 }
 267
 268 void ll_intent_release(struct lookup_intent *it)
 269 {
 270         ENTRY;
 271
 272         CDEBUG(D_INFO, "intent %p released\n", it);
 273         ll_intent_drop_lock(it);
 274         /* We are still holding extra reference on a request, need to free it */
 275         if (it_disposition(it, DISP_ENQ_OPEN_REF))
 276                  ptlrpc_req_finished(it->d.lustre.it_data); /* ll_file_open */
 277         if (it_disposition(it, DISP_ENQ_CREATE_REF)) /* create rec */
 278                 ptlrpc_req_finished(it->d.lustre.it_data);
 279         if (it_disposition(it, DISP_ENQ_COMPLETE)) /* saved req from revalidate
 280                                                     * to lookup */
 281                 ptlrpc_req_finished(it->d.lustre.it_data);
 282
 283         it->d.lustre.it_disposition = 0;
 284         it->d.lustre.it_data = NULL;
 285         EXIT;
 286 }
 287
 288 void ll_invalidate_aliases(struct inode *inode)
 289 {
 290         struct dentry *dentry;
 291         ENTRY;
 292
 293         LASSERT(inode != NULL);
 294
 295         CDEBUG(D_INODE, "marking dentries for ino %lu/%u(%p) invalid\n",
 296                inode->i_ino, inode->i_generation, inode);
 297
 298         ll_lock_dcache(inode);
 299         cfs_list_for_each_entry(dentry, &inode->i_dentry, d_alias) {
 300                 CDEBUG(D_DENTRY, "dentry in drop %.*s (%p) parent %p "
 301                        "inode %p flags %d\n", dentry->d_name.len,
 302                        dentry->d_name.name, dentry, dentry->d_parent,
 303                        dentry->d_inode, dentry->d_flags);
 304
 305                 if (dentry->d_name.len == 1 && dentry->d_name.name[0] == '/') {
 306                         CERROR("called on root (?) dentry=%p, inode=%p "
 307                                "ino=%lu\n", dentry, inode, inode->i_ino);
 308                         lustre_dump_dentry(dentry, 1);
 309                         libcfs_debug_dumpstack(NULL);
 310                 }
 311
 312                 d_lustre_invalidate(dentry);
 313         }
 314         ll_unlock_dcache(inode);
 315
 316         EXIT;
 317 }
 318
 319 int ll_revalidate_it_finish(struct ptlrpc_request *request,
 320                             struct lookup_intent *it,
 321                             struct dentry *de)
 322 {
 323         int rc = 0;
 324         ENTRY;
 325
 326         if (!request)
 327                 RETURN(0);
 328
 329         if (it_disposition(it, DISP_LOOKUP_NEG))
 330                 RETURN(-ENOENT);
 331
 332         rc = ll_prep_inode(&de->d_inode, request, NULL);
 333
 334         RETURN(rc);
 335 }
 336
 337 void ll_lookup_finish_locks(struct lookup_intent *it, struct dentry *dentry)
 338 {
 339         LASSERT(it != NULL);
 340         LASSERT(dentry != NULL);
 341
 342         if (it->d.lustre.it_lock_mode && dentry->d_inode != NULL) {
 343                 struct inode *inode = dentry->d_inode;
 344                 struct ll_sb_info *sbi = ll_i2sbi(dentry->d_inode);
 345
 346                 CDEBUG(D_DLMTRACE, "setting l_data to inode %p (%lu/%u)\n",
 347                        inode, inode->i_ino, inode->i_generation);
 348                 ll_set_lock_data(sbi->ll_md_exp, inode, it, NULL);
 349         }
 350
 351         /* drop lookup or getattr locks immediately */
 352         if (it->it_op == IT_LOOKUP || it->it_op == IT_GETATTR) {
 353                 /* on 2.6 there are situation when several lookups and
 354                  * revalidations may be requested during single operation.
 355                  * therefore, we don't release intent here -bzzz */
 356                 ll_intent_drop_lock(it);
 357         }
 358 }
 359
 360 void ll_frob_intent(struct lookup_intent **itp, struct lookup_intent *deft)
 361 {
 362         struct lookup_intent *it = *itp;
 363
 364         if (!it || it->it_op == IT_GETXATTR)
 365                 it = *itp = deft;
 366
 367 }
 368
 369 int ll_revalidate_it(struct dentry *de, int lookup_flags,
 370                      struct lookup_intent *it)
 371 {
 372         struct md_op_data *op_data;
 373         struct ptlrpc_request *req = NULL;
 374         struct lookup_intent lookup_it = { .it_op = IT_LOOKUP };
 375         struct obd_export *exp;
 376         struct inode *parent = de->d_parent->d_inode;
 377         int rc;
 378
 379         ENTRY;
 380         CDEBUG(D_VFSTRACE, "VFS Op:name=%s,intent=%s\n", de->d_name.name,
 381                LL_IT2STR(it));
 382
 383         if (de->d_inode == NULL) {
 384                 __u64 ibits;
 385
 386                 /* We can only use negative dentries if this is stat or lookup,
 387                    for opens and stuff we do need to query server. */
 388                 /* If there is IT_CREAT in intent op set, then we must throw
 389                    away this negative dentry and actually do the request to
 390                    kernel to create whatever needs to be created (if possible)*/
 391                 if (it && (it->it_op & IT_CREAT))
 392                         RETURN(0);
 393
 394                 if (d_lustre_invalid(de))
 395                         RETURN(0);
 396
 397                 ibits = MDS_INODELOCK_UPDATE;
 398                 rc = ll_have_md_lock(parent, &ibits, LCK_MINMODE);
 399                 GOTO(out_sa, rc);
 400         }
 401
 402         /* Never execute intents for mount points.
 403          * Attributes will be fixed up in ll_inode_revalidate_it */
 404         if (d_mountpoint(de))
 405                 GOTO(out_sa, rc = 1);
 406
 407         /* need to get attributes in case root got changed from other client */
 408         if (de == de->d_sb->s_root) {
 409                 rc = __ll_inode_revalidate_it(de, it, MDS_INODELOCK_LOOKUP);
 410                 if (rc == 0)
 411                         rc = 1;
 412                 GOTO(out_sa, rc);
 413         }
 414
 415         exp = ll_i2mdexp(de->d_inode);
 416
 417         OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_REVALIDATE_PAUSE, 5);
 418         ll_frob_intent(&it, &lookup_it);
 419         LASSERT(it);
 420
 421         if (it->it_op == IT_LOOKUP && !d_lustre_invalid(de))
 422                 RETURN(1);
 423
 424         if ((it->it_op == IT_OPEN) && de->d_inode) {
 425                 struct inode *inode = de->d_inode;
 426                 struct ll_inode_info *lli = ll_i2info(inode);
 427                 struct obd_client_handle **och_p;
 428                 __u64 *och_usecount;
 429                 __u64 ibits;
 430
 431                 /*
 432                  * We used to check for MDS_INODELOCK_OPEN here, but in fact
 433                  * just having LOOKUP lock is enough to justify inode is the
 434                  * same. And if inode is the same and we have suitable
 435                  * openhandle, then there is no point in doing another OPEN RPC
 436                  * just to throw away newly received openhandle.  There are no
 437                  * security implications too, if file owner or access mode is
 438                  * change, LOOKUP lock is revoked.
 439                  */
 440
 441
 442                 if (it->it_flags & FMODE_WRITE) {
 443                         och_p = &lli->lli_mds_write_och;
 444                         och_usecount = &lli->lli_open_fd_write_count;
 445                 } else if (it->it_flags & FMODE_EXEC) {
 446                         och_p = &lli->lli_mds_exec_och;
 447                         och_usecount = &lli->lli_open_fd_exec_count;
 448                 } else {
 449                         och_p = &lli->lli_mds_read_och;
 450                         och_usecount = &lli->lli_open_fd_read_count;
 451                 }
 452                 /* Check for the proper lock. */
 453                 ibits = MDS_INODELOCK_LOOKUP;
 454                 if (!ll_have_md_lock(inode, &ibits, LCK_MINMODE))
 455                         goto do_lock;
 456                 cfs_mutex_lock(&lli->lli_och_mutex);
 457                 if (*och_p) { /* Everything is open already, do nothing */
 458                         /*(*och_usecount)++;  Do not let them steal our open
 459                           handle from under us */
 460                         SET_BUT_UNUSED(och_usecount);
 461                         /* XXX The code above was my original idea, but in case
 462                            we have the handle, but we cannot use it due to later
 463                            checks (e.g. O_CREAT|O_EXCL flags set), nobody
 464                            would decrement counter increased here. So we just
 465                            hope the lock won't be invalidated in between. But
 466                            if it would be, we'll reopen the open request to
 467                            MDS later during file open path */
 468                         cfs_mutex_unlock(&lli->lli_och_mutex);
 469                         RETURN(1);
 470                 } else {
 471                         cfs_mutex_unlock(&lli->lli_och_mutex);
 472                 }
 473         }
 474
 475         if (it->it_op == IT_GETATTR) {
 476                 rc = ll_statahead_enter(parent, &de, 0);
 477                 if (rc == 1)
 478                         goto mark;
 479                 else if (rc != -EAGAIN && rc != 0)
 480                         GOTO(out, rc = 0);
 481         }
 482
 483 do_lock:
 484         op_data = ll_prep_md_op_data(NULL, parent, de->d_inode,
 485                                      de->d_name.name, de->d_name.len,
 486                                      0, LUSTRE_OPC_ANY, NULL);
 487         if (IS_ERR(op_data))
 488                 RETURN(PTR_ERR(op_data));
 489
 490         it->it_create_mode &= ~cfs_curproc_umask();
 491         it->it_create_mode |= M_CHECK_STALE;
 492         rc = md_intent_lock(exp, op_data, NULL, 0, it,
 493                             lookup_flags,
 494                             &req, ll_md_blocking_ast, 0);
 495         it->it_create_mode &= ~M_CHECK_STALE;
 496         ll_finish_md_op_data(op_data);
 497
 498         /* If req is NULL, then md_intent_lock only tried to do a lock match;
 499          * if all was well, it will return 1 if it found locks, 0 otherwise. */
 500         if (req == NULL && rc >= 0) {
 501                 if (!rc)
 502                         goto do_lookup;
 503                 GOTO(out, rc);
 504         }
 505
 506         if (rc < 0) {
 507                 if (rc != -ESTALE) {
 508                         CDEBUG(D_INFO, "ll_intent_lock: rc %d : it->it_status "
 509                                "%d\n", rc, it->d.lustre.it_status);
 510                 }
 511                 GOTO(out, rc = 0);
 512         }
 513
 514 revalidate_finish:
 515         rc = ll_revalidate_it_finish(req, it, de);
 516         if (rc != 0) {
 517                 if (rc != -ESTALE && rc != -ENOENT)
 518                         ll_intent_release(it);
 519                 GOTO(out, rc = 0);
 520         }
 521
 522         if ((it->it_op & IT_OPEN) && de->d_inode &&
 523             !S_ISREG(de->d_inode->i_mode) &&
 524             !S_ISDIR(de->d_inode->i_mode)) {
 525                 ll_release_openhandle(de, it);
 526         }
 527         rc = 1;
 528
 529 out:
 530         /* We do not free request as it may be reused during following lookup
 531          * (see comment in mdc/mdc_locks.c::mdc_intent_lock()), request will
 532          * be freed in ll_lookup_it or in ll_intent_release. But if
 533          * request was not completed, we need to free it. (bug 5154, 9903) */
 534         if (req != NULL && !it_disposition(it, DISP_ENQ_COMPLETE))
 535                 ptlrpc_req_finished(req);
 536         if (rc == 0) {
 537                 ll_invalidate_aliases(de->d_inode);
 538         } else {
 539                 __u64 bits = 0;
 540
 541                 CDEBUG(D_DENTRY, "revalidated dentry %.*s (%p) parent %p "
 542                        "inode %p refc %d\n", de->d_name.len,
 543                        de->d_name.name, de, de->d_parent, de->d_inode,
 544                        d_refcount(de));
 545                 ll_set_lock_data(exp, de->d_inode, it, &bits);
 546                 if ((bits & MDS_INODELOCK_LOOKUP) && d_lustre_invalid(de))
 547                         d_lustre_revalidate(de);
 548                 ll_lookup_finish_locks(it, de);
 549         }
 550
 551 mark:
 552         if (it != NULL && it->it_op == IT_GETATTR && rc > 0)
 553                 ll_statahead_mark(parent, de);
 554         RETURN(rc);
 555
 556         /*
 557          * This part is here to combat evil-evil race in real_lookup on 2.6
 558          * kernels.  The race details are: We enter do_lookup() looking for some
 559          * name, there is nothing in dcache for this name yet and d_lookup()
 560          * returns NULL.  We proceed to real_lookup(), and while we do this,
 561          * another process does open on the same file we looking up (most simple
 562          * reproducer), open succeeds and the dentry is added. Now back to
 563          * us. In real_lookup() we do d_lookup() again and suddenly find the
 564          * dentry, so we call d_revalidate on it, but there is no lock, so
 565          * without this code we would return 0, but unpatched real_lookup just
 566          * returns -ENOENT in such a case instead of retrying the lookup. Once
 567          * this is dealt with in real_lookup(), all of this ugly mess can go and
 568          * we can just check locks in ->d_revalidate without doing any RPCs
 569          * ever.
 570          */
 571 do_lookup:
 572         if (it != &lookup_it) {
 573                 /* MDS_INODELOCK_UPDATE needed for IT_GETATTR case. */
 574                 if (it->it_op == IT_GETATTR)
 575                         lookup_it.it_op = IT_GETATTR;
 576                 ll_lookup_finish_locks(it, de);
 577                 it = &lookup_it;
 578         }
 579
 580         /* Do real lookup here. */
 581         op_data = ll_prep_md_op_data(NULL, parent, NULL, de->d_name.name,
 582                                      de->d_name.len, 0, (it->it_op & IT_CREAT ?
 583                                                          LUSTRE_OPC_CREATE :
 584                                                          LUSTRE_OPC_ANY), NULL);
 585         if (IS_ERR(op_data))
 586                 RETURN(PTR_ERR(op_data));
 587
 588         rc = md_intent_lock(exp, op_data, NULL, 0,  it, 0, &req,
 589                             ll_md_blocking_ast, 0);
 590         if (rc >= 0) {
 591                 struct mdt_body *mdt_body;
 592                 struct lu_fid fid = {.f_seq = 0, .f_oid = 0, .f_ver = 0};
 593                 mdt_body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 594
 595                 if (de->d_inode)
 596                         fid = *ll_inode2fid(de->d_inode);
 597
 598                 /* see if we got same inode, if not - return error */
 599                 if (lu_fid_eq(&fid, &mdt_body->fid1)) {
 600                         ll_finish_md_op_data(op_data);
 601                         op_data = NULL;
 602                         goto revalidate_finish;
 603                 }
 604                 ll_intent_release(it);
 605         }
 606         ll_finish_md_op_data(op_data);
 607         GOTO(out, rc = 0);
 608
 609 out_sa:
 610         /*
 611          * For rc == 1 case, should not return directly to prevent losing
 612          * statahead windows; for rc == 0 case, the "lookup" will be done later.
 613          */
 614         if (it != NULL && it->it_op == IT_GETATTR && rc == 1)
 615                 ll_statahead_enter(parent, &de, 1);
 616         goto mark;
 617 }
 618
 619 int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
 620 {
 621         int rc;
 622         ENTRY;
 623
 624 #ifndef HAVE_DCACHE_LOCK
 625         /* kernel >= 2.6.38 supports rcu-walk, but lustre doesn't. */
 626         if (nd->flags & LOOKUP_RCU)
 627                 return -ECHILD;
 628 #endif
 629
 630         if (nd && !(nd->flags & (LOOKUP_CONTINUE|LOOKUP_PARENT))) {
 631                 struct lookup_intent *it;
 632
 633                 it = ll_convert_intent(&nd->intent.open, nd->flags);
 634                 if (IS_ERR(it))
 635                         RETURN(0);
 636
 637                 if (it->it_op == (IT_OPEN|IT_CREAT) &&
 638                     nd->intent.open.flags & O_EXCL) {
 639                         CDEBUG(D_VFSTRACE, "create O_EXCL, returning 0\n");
 640                         rc = 0;
 641                         goto out_it;
 642                 }
 643
 644                 rc = ll_revalidate_it(dentry, nd->flags, it);
 645
 646                 if (rc && (nd->flags & LOOKUP_OPEN) &&
 647                     it_disposition(it, DISP_OPEN_OPEN)) {/*Open*/
 648 // XXX Code duplication with ll_lookup_nd
 649                         if (S_ISFIFO(dentry->d_inode->i_mode)) {
 650                                 // We cannot call open here as it would
 651                                 // deadlock.
 652                                 ptlrpc_req_finished(
 653                                                (struct ptlrpc_request *)
 654                                                   it->d.lustre.it_data);
 655                         } else {
 656                                 struct file *filp;
 657
 658                                 nd->intent.open.file->private_data = it;
 659                                 filp = lookup_instantiate_filp(nd, dentry,NULL);
 660                                 if (IS_ERR(filp))
 661                                         rc = PTR_ERR(filp);
 662                         }
 663                 }
 664                 if (!rc && (nd->flags & LOOKUP_CREATE) &&
 665                     it_disposition(it, DISP_OPEN_CREATE)) {
 666                         /* We created something but we may only return
 667                          * negative dentry here, so save request in dentry,
 668                          * if lookup will be called later on, it will
 669                          * pick the request, otherwise it would be freed
 670                          * with dentry */
 671                         ll_d2d(dentry)->lld_it = it;
 672                         it = NULL; /* avoid freeing */
 673                 }
 674
 675 out_it:
 676                 if (it) {
 677                         ll_intent_release(it);
 678                         OBD_FREE(it, sizeof(*it));
 679                 }
 680         } else {
 681                 rc = ll_revalidate_it(dentry, 0, NULL);
 682         }
 683
 684         RETURN(rc);
 685 }
 686
 687 void ll_d_iput(struct dentry *de, struct inode *inode)
 688 {
 689         LASSERT(inode);
 690         if (!find_cbdata(inode))
 691                 inode->i_nlink = 0;
 692         iput(inode);
 693 }
 694
 695 struct dentry_operations ll_d_ops = {
 696         .d_revalidate = ll_revalidate_nd,
 697         .d_release = ll_release,
 698         .d_delete  = ll_ddelete,
 699         .d_iput    = ll_d_iput,
 700         .d_compare = ll_dcompare,
 701 };