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