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