lustre/llite/dcache.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see [sun.com URL with a
  20  * copy of GPLv2].
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  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 #include <lustre_mdc.h>
  49 //#include <lustre_ver.h>
  50 //#include <lustre_version.h>
  51
  52 #include "llite_internal.h"
  53
  54 /* should NOT be called with the dcache lock, see fs/dcache.c */
  55 void ll_release(struct dentry *de)
  56 {
  57         struct ll_dentry_data *lld;
  58         ENTRY;
  59         LASSERT(de != NULL);
  60         lld = ll_d2d(de);
  61         if (lld == NULL) { /* NFS copies the de->d_op methods (bug 4655) */
  62                 EXIT;
  63                 return;
  64         }
  65 #ifndef HAVE_VFS_INTENT_PATCHES
  66         if (lld->lld_it) {
  67                 ll_intent_release(lld->lld_it);
  68                 OBD_FREE(lld->lld_it, sizeof(*lld->lld_it));
  69         }
  70 #endif
  71         LASSERT(lld->lld_cwd_count == 0);
  72         LASSERT(lld->lld_mnt_count == 0);
  73         OBD_FREE(de->d_fsdata, sizeof(*lld));
  74
  75         EXIT;
  76 }
  77
  78 #ifdef DCACHE_LUSTRE_INVALID
  79 /* Compare if two dentries are the same.  Don't match if the existing dentry
  80  * is marked DCACHE_LUSTRE_INVALID.  Returns 1 if different, 0 if the same.
  81  *
  82  * This avoids a race where ll_lookup_it() instantiates a dentry, but we get
  83  * an AST before calling d_revalidate_it().  The dentry still exists (marked
  84  * INVALID) so d_lookup() matches it, but we have no lock on it (so
  85  * lock_match() fails) and we spin around real_lookup(). */
  86 int ll_dcompare(struct dentry *parent, struct qstr *d_name, struct qstr *name)
  87 {
  88         struct dentry *dchild;
  89         ENTRY;
  90
  91         if (d_name->len != name->len)
  92                 RETURN(1);
  93
  94         if (memcmp(d_name->name, name->name, name->len))
  95                 RETURN(1);
  96
  97         /* XXX: d_name must be in-dentry structure */
  98         dchild = container_of(d_name, struct dentry, d_name); /* ugh */
  99         if (dchild->d_flags & DCACHE_LUSTRE_INVALID) {
 100                 CDEBUG(D_DENTRY,"INVALID dentry %p not matched, was bug 3784\n",
 101                        dchild);
 102                 RETURN(1);
 103         }
 104
 105         RETURN(0);
 106 }
 107 #endif
 108
 109 /* should NOT be called with the dcache lock, see fs/dcache.c */
 110 static int ll_ddelete(struct dentry *de)
 111 {
 112         ENTRY;
 113         LASSERT(de);
 114 #ifndef DCACHE_LUSTRE_INVALID
 115 #define DCACHE_LUSTRE_INVALID 0
 116 #endif
 117
 118         CDEBUG(D_DENTRY, "%s dentry %.*s (%p, parent %p, inode %p) %s%s\n",
 119                (de->d_flags & DCACHE_LUSTRE_INVALID ? "deleting" : "keeping"),
 120                de->d_name.len, de->d_name.name, de, de->d_parent, de->d_inode,
 121                d_unhashed(de) ? "" : "hashed,",
 122                list_empty(&de->d_subdirs) ? "" : "subdirs");
 123 #if DCACHE_LUSTRE_INVALID == 0
 124 #undef DCACHE_LUSTRE_INVALID
 125 #endif
 126
 127         RETURN(0);
 128 }
 129
 130 void ll_set_dd(struct dentry *de)
 131 {
 132         ENTRY;
 133         LASSERT(de != NULL);
 134
 135         CDEBUG(D_DENTRY, "ldd on dentry %.*s (%p) parent %p inode %p refc %d\n",
 136                de->d_name.len, de->d_name.name, de, de->d_parent, de->d_inode,
 137                atomic_read(&de->d_count));
 138
 139         if (de->d_fsdata == NULL) {
 140                 struct ll_dentry_data *lld;
 141
 142                 OBD_ALLOC_PTR(lld);
 143                 if (likely(lld != NULL)) {
 144                         cfs_waitq_init(&lld->lld_waitq);
 145                         lock_dentry(de);
 146                         if (likely(de->d_fsdata == NULL))
 147                                 de->d_fsdata = lld;
 148                         else
 149                                 OBD_FREE_PTR(lld);
 150                         unlock_dentry(de);
 151                 }
 152         }
 153
 154         EXIT;
 155 }
 156
 157 void ll_intent_drop_lock(struct lookup_intent *it)
 158 {
 159         struct lustre_handle *handle;
 160
 161         if (it->it_op && it->d.lustre.it_lock_mode) {
 162                 handle = (struct lustre_handle *)&it->d.lustre.it_lock_handle;
 163                 CDEBUG(D_DLMTRACE, "releasing lock with cookie "LPX64
 164                        " from it %p\n", handle->cookie, it);
 165                 ldlm_lock_decref(handle, it->d.lustre.it_lock_mode);
 166
 167                 /* bug 494: intent_release may be called multiple times, from
 168                  * this thread and we don't want to double-decref this lock */
 169                 it->d.lustre.it_lock_mode = 0;
 170         }
 171 }
 172
 173 void ll_intent_release(struct lookup_intent *it)
 174 {
 175         ENTRY;
 176
 177         CDEBUG(D_INFO, "intent %p released\n", it);
 178         ll_intent_drop_lock(it);
 179 #ifdef HAVE_VFS_INTENT_PATCHES
 180         it->it_magic = 0;
 181         it->it_op_release = 0;
 182 #endif
 183         /* We are still holding extra reference on a request, need to free it */
 184         if (it_disposition(it, DISP_ENQ_OPEN_REF))
 185                  ptlrpc_req_finished(it->d.lustre.it_data); /* ll_file_open */
 186         if (it_disposition(it, DISP_ENQ_CREATE_REF)) /* create rec */
 187                 ptlrpc_req_finished(it->d.lustre.it_data);
 188         if (it_disposition(it, DISP_ENQ_COMPLETE)) /* saved req from revalidate
 189                                                     * to lookup */
 190                 ptlrpc_req_finished(it->d.lustre.it_data);
 191
 192         it->d.lustre.it_disposition = 0;
 193         it->d.lustre.it_data = NULL;
 194         EXIT;
 195 }
 196
 197 /* Drop dentry if it is not used already, unhash otherwise.
 198    Should be called with dcache lock held!
 199    Returns: 1 if dentry was dropped, 0 if unhashed. */
 200 int ll_drop_dentry(struct dentry *dentry)
 201 {
 202         lock_dentry(dentry);
 203         if (atomic_read(&dentry->d_count) == 0) {
 204                 CDEBUG(D_DENTRY, "deleting dentry %.*s (%p) parent %p "
 205                        "inode %p\n", dentry->d_name.len,
 206                        dentry->d_name.name, dentry, dentry->d_parent,
 207                        dentry->d_inode);
 208                 dget_locked(dentry);
 209                 __d_drop(dentry);
 210                 unlock_dentry(dentry);
 211                 spin_unlock(&dcache_lock);
 212                 dput(dentry);
 213                 spin_lock(&dcache_lock);
 214                 return 1;
 215         }
 216         /* disconected dentry can not be find without lookup, because we
 217          * not need his to unhash or mark invalid. */
 218         if (dentry->d_flags & DCACHE_DISCONNECTED) {
 219                 unlock_dentry(dentry);
 220                 RETURN (0);
 221         }
 222
 223 #ifdef DCACHE_LUSTRE_INVALID
 224         if (!(dentry->d_flags & DCACHE_LUSTRE_INVALID)) {
 225 #else
 226         if (!d_unhashed(dentry)) {
 227 #endif
 228                 CDEBUG(D_DENTRY, "unhashing dentry %.*s (%p) parent %p "
 229                        "inode %p refc %d\n", dentry->d_name.len,
 230                        dentry->d_name.name, dentry, dentry->d_parent,
 231                        dentry->d_inode, atomic_read(&dentry->d_count));
 232                 /* actually we don't unhash the dentry, rather just
 233                  * mark it inaccessible for to __d_lookup(). otherwise
 234                  * sys_getcwd() could return -ENOENT -bzzz */
 235 #ifdef DCACHE_LUSTRE_INVALID
 236                 dentry->d_flags |= DCACHE_LUSTRE_INVALID;
 237 #endif
 238                 if (!dentry->d_inode || !S_ISDIR(dentry->d_inode->i_mode))
 239                         __d_drop(dentry);
 240
 241         }
 242         unlock_dentry(dentry);
 243         return 0;
 244 }
 245
 246 void ll_unhash_aliases(struct inode *inode)
 247 {
 248         struct list_head *tmp, *head;
 249         ENTRY;
 250
 251         if (inode == NULL) {
 252                 CERROR("unexpected NULL inode, tell phil\n");
 253                 return;
 254         }
 255
 256         CDEBUG(D_INODE, "marking dentries for ino %lu/%u(%p) invalid\n",
 257                inode->i_ino, inode->i_generation, inode);
 258
 259         head = &inode->i_dentry;
 260         spin_lock(&dcache_lock);
 261 restart:
 262         tmp = head;
 263         while ((tmp = tmp->next) != head) {
 264                 struct dentry *dentry = list_entry(tmp, struct dentry, d_alias);
 265
 266                 CDEBUG(D_DENTRY, "dentry in drop %.*s (%p) parent %p "
 267                        "inode %p flags %d\n", dentry->d_name.len,
 268                        dentry->d_name.name, dentry, dentry->d_parent,
 269                        dentry->d_inode, dentry->d_flags);
 270
 271                 if (dentry->d_name.len == 1 && dentry->d_name.name[0] == '/') {
 272                         CERROR("called on root (?) dentry=%p, inode=%p "
 273                                "ino=%lu\n", dentry, inode, inode->i_ino);
 274                         lustre_dump_dentry(dentry, 1);
 275                         libcfs_debug_dumpstack(NULL);
 276                 } else if (d_mountpoint(dentry)) {
 277                         /* For mountpoints we skip removal of the dentry
 278                            which happens solely because we have a lock on it
 279                            obtained when this dentry was not a mountpoint yet */
 280                         CDEBUG(D_DENTRY, "Skippind mountpoint dentry removal "
 281                                          "%.*s (%p) parent %p\n",
 282                                           dentry->d_name.len,
 283                                           dentry->d_name.name,
 284                                           dentry, dentry->d_parent);
 285
 286                         continue;
 287                 }
 288
 289                 if (ll_drop_dentry(dentry))
 290                           goto restart;
 291         }
 292         spin_unlock(&dcache_lock);
 293         EXIT;
 294 }
 295
 296 int ll_revalidate_it_finish(struct ptlrpc_request *request,
 297                             struct lookup_intent *it,
 298                             struct dentry *de)
 299 {
 300         int rc = 0;
 301         ENTRY;
 302
 303         if (!request)
 304                 RETURN(0);
 305
 306         if (it_disposition(it, DISP_LOOKUP_NEG))
 307                 RETURN(-ENOENT);
 308
 309         rc = ll_prep_inode(&de->d_inode, request, NULL);
 310
 311         RETURN(rc);
 312 }
 313
 314 void ll_lookup_finish_locks(struct lookup_intent *it, struct dentry *dentry)
 315 {
 316         LASSERT(it != NULL);
 317         LASSERT(dentry != NULL);
 318
 319         if (it->d.lustre.it_lock_mode && dentry->d_inode != NULL) {
 320                 struct inode *inode = dentry->d_inode;
 321                 struct ll_sb_info *sbi = ll_i2sbi(dentry->d_inode);
 322
 323                 CDEBUG(D_DLMTRACE, "setting l_data to inode %p (%lu/%u)\n",
 324                        inode, inode->i_ino, inode->i_generation);
 325                 md_set_lock_data(sbi->ll_md_exp, &it->d.lustre.it_lock_handle,
 326                                  inode);
 327         }
 328
 329         /* drop lookup or getattr locks immediately */
 330         if (it->it_op == IT_LOOKUP || it->it_op == IT_GETATTR) {
 331                 /* on 2.6 there are situation when several lookups and
 332                  * revalidations may be requested during single operation.
 333                  * therefore, we don't release intent here -bzzz */
 334                 ll_intent_drop_lock(it);
 335         }
 336 }
 337
 338 void ll_frob_intent(struct lookup_intent **itp, struct lookup_intent *deft)
 339 {
 340         struct lookup_intent *it = *itp;
 341 #ifdef HAVE_VFS_INTENT_PATCHES
 342         if (it) {
 343                 LASSERTF(it->it_magic == INTENT_MAGIC,
 344                          "%p has bad intent magic: %x\n",
 345                          it, it->it_magic);
 346         }
 347 #endif
 348
 349         if (!it || it->it_op == IT_GETXATTR)
 350                 it = *itp = deft;
 351
 352 #ifdef HAVE_VFS_INTENT_PATCHES
 353         it->it_op_release = ll_intent_release;
 354 #endif
 355 }
 356
 357 int ll_revalidate_it(struct dentry *de, int lookup_flags,
 358                      struct lookup_intent *it)
 359 {
 360         struct md_op_data *op_data;
 361         struct ptlrpc_request *req = NULL;
 362         struct lookup_intent lookup_it = { .it_op = IT_LOOKUP };
 363         struct obd_export *exp;
 364         struct inode *parent;
 365         int rc, first = 0;
 366
 367         ENTRY;
 368         CDEBUG(D_VFSTRACE, "VFS Op:name=%s,intent=%s\n", de->d_name.name,
 369                LL_IT2STR(it));
 370
 371         if (de->d_inode == NULL) {
 372                 /* We can only use negative dentries if this is stat or lookup,
 373                    for opens and stuff we do need to query server. */
 374                 /* If there is IT_CREAT in intent op set, then we must throw
 375                    away this negative dentry and actually do the request to
 376                    kernel to create whatever needs to be created (if possible)*/
 377                 if (it && (it->it_op & IT_CREAT))
 378                         RETURN(0);
 379
 380 #ifdef DCACHE_LUSTRE_INVALID
 381                 if (de->d_flags & DCACHE_LUSTRE_INVALID)
 382                         RETURN(0);
 383 #endif
 384
 385                 rc = ll_have_md_lock(de->d_parent->d_inode,
 386                                      MDS_INODELOCK_UPDATE);
 387                 GOTO(out_sa, rc);
 388         }
 389
 390         exp = ll_i2mdexp(de->d_inode);
 391
 392         /* Never execute intents for mount points.
 393          * Attributes will be fixed up in ll_inode_revalidate_it */
 394         if (d_mountpoint(de))
 395                 GOTO(out_sa, rc = 1);
 396
 397         /* Root of the lustre tree. Always valid.
 398          * Attributes will be fixed up in ll_inode_revalidate_it */
 399         if (de == de->d_sb->s_root)
 400                 GOTO(out_sa, rc = 1);
 401
 402         OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_REVALIDATE_PAUSE, 5);
 403         ll_frob_intent(&it, &lookup_it);
 404         LASSERT(it);
 405         parent = de->d_parent->d_inode;
 406
 407         op_data = ll_prep_md_op_data(NULL, parent, de->d_inode,
 408                                      de->d_name.name, de->d_name.len,
 409                                      0, LUSTRE_OPC_ANY, NULL);
 410         if (IS_ERR(op_data))
 411                 RETURN(PTR_ERR(op_data));
 412
 413         if ((it->it_op == IT_OPEN) && de->d_inode) {
 414                 struct inode *inode = de->d_inode;
 415                 struct ll_inode_info *lli = ll_i2info(inode);
 416                 struct obd_client_handle **och_p;
 417                 __u64 *och_usecount;
 418
 419                 /*
 420                  * We used to check for MDS_INODELOCK_OPEN here, but in fact
 421                  * just having LOOKUP lock is enough to justify inode is the
 422                  * same. And if inode is the same and we have suitable
 423                  * openhandle, then there is no point in doing another OPEN RPC
 424                  * just to throw away newly received openhandle.  There are no
 425                  * security implications too, if file owner or access mode is
 426                  * change, LOOKUP lock is revoked.
 427                  */
 428
 429
 430                 if (it->it_flags & FMODE_WRITE) {
 431                         och_p = &lli->lli_mds_write_och;
 432                         och_usecount = &lli->lli_open_fd_write_count;
 433                 } else if (it->it_flags & FMODE_EXEC) {
 434                         och_p = &lli->lli_mds_exec_och;
 435                         och_usecount = &lli->lli_open_fd_exec_count;
 436                 } else {
 437                         och_p = &lli->lli_mds_read_och;
 438                         och_usecount = &lli->lli_open_fd_read_count;
 439                 }
 440                 /* Check for the proper lock. */
 441                 if (!ll_have_md_lock(inode, MDS_INODELOCK_LOOKUP))
 442                         goto do_lock;
 443                 down(&lli->lli_och_sem);
 444                 if (*och_p) { /* Everything is open already, do nothing */
 445                         /*(*och_usecount)++;  Do not let them steal our open
 446                           handle from under us */
 447                         /* XXX The code above was my original idea, but in case
 448                            we have the handle, but we cannot use it due to later
 449                            checks (e.g. O_CREAT|O_EXCL flags set), nobody
 450                            would decrement counter increased here. So we just
 451                            hope the lock won't be invalidated in between. But
 452                            if it would be, we'll reopen the open request to
 453                            MDS later during file open path */
 454                         up(&lli->lli_och_sem);
 455                         ll_finish_md_op_data(op_data);
 456                         RETURN(1);
 457                 } else {
 458                         up(&lli->lli_och_sem);
 459                 }
 460         }
 461
 462         if (it->it_op == IT_GETATTR)
 463                 first = ll_statahead_enter(de->d_parent->d_inode, &de, 0);
 464
 465 do_lock:
 466         it->it_create_mode &= ~current->fs->umask;
 467         it->it_flags |= O_CHECK_STALE;
 468         rc = md_intent_lock(exp, op_data, NULL, 0, it,
 469                             lookup_flags,
 470                             &req, ll_md_blocking_ast, 0);
 471         it->it_flags &= ~O_CHECK_STALE;
 472         ll_finish_md_op_data(op_data);
 473         if (it->it_op == IT_GETATTR && !first)
 474                 ll_statahead_exit(de, rc);
 475
 476         /* If req is NULL, then md_intent_lock only tried to do a lock match;
 477          * if all was well, it will return 1 if it found locks, 0 otherwise. */
 478         if (req == NULL && rc >= 0) {
 479                 if (!rc)
 480                         goto do_lookup;
 481                 GOTO(out, rc);
 482         }
 483
 484         if (rc < 0) {
 485                 if (rc != -ESTALE) {
 486                         CDEBUG(D_INFO, "ll_intent_lock: rc %d : it->it_status "
 487                                "%d\n", rc, it->d.lustre.it_status);
 488                 }
 489                 GOTO(out, rc = 0);
 490         }
 491
 492 revalidate_finish:
 493         rc = ll_revalidate_it_finish(req, it, de);
 494         if (rc != 0) {
 495                 if (rc != -ESTALE && rc != -ENOENT)
 496                         ll_intent_release(it);
 497                 GOTO(out, rc = 0);
 498         }
 499
 500         if ((it->it_op & IT_OPEN) && de->d_inode &&
 501             !S_ISREG(de->d_inode->i_mode) &&
 502             !S_ISDIR(de->d_inode->i_mode)) {
 503                 ll_release_openhandle(de, it);
 504         }
 505         rc = 1;
 506
 507         /* unfortunately ll_intent_lock may cause a callback and revoke our
 508          * dentry */
 509         spin_lock(&dcache_lock);
 510         lock_dentry(de);
 511         __d_drop(de);
 512         unlock_dentry(de);
 513         d_rehash_cond(de, 0);
 514         spin_unlock(&dcache_lock);
 515
 516 out:
 517         /* We do not free request as it may be reused during following lookup
 518          * (see comment in mdc/mdc_locks.c::mdc_intent_lock()), request will
 519          * be freed in ll_lookup_it or in ll_intent_release. But if
 520          * request was not completed, we need to free it. (bug 5154, 9903) */
 521         if (req != NULL && !it_disposition(it, DISP_ENQ_COMPLETE))
 522                 ptlrpc_req_finished(req);
 523         if (rc == 0) {
 524 #ifdef DCACHE_LUSTRE_INVALID
 525                 ll_unhash_aliases(de->d_inode);
 526                 /* done in ll_unhash_aliases()
 527                    dentry->d_flags |= DCACHE_LUSTRE_INVALID; */
 528 #else
 529                 /* We do not want d_invalidate to kill all child dentries too */
 530                 d_drop(de);
 531 #endif
 532         } else {
 533                 CDEBUG(D_DENTRY, "revalidated dentry %.*s (%p) parent %p "
 534                        "inode %p refc %d\n", de->d_name.len,
 535                        de->d_name.name, de, de->d_parent, de->d_inode,
 536                        atomic_read(&de->d_count));
 537                 ll_lookup_finish_locks(it, de);
 538 #ifdef DCACHE_LUSTRE_INVALID
 539                 lock_dentry(de);
 540                 de->d_flags &= ~DCACHE_LUSTRE_INVALID;
 541                 unlock_dentry(de);
 542 #endif
 543         }
 544         RETURN(rc);
 545
 546         /*
 547          * This part is here to combat evil-evil race in real_lookup on 2.6
 548          * kernels.  The race details are: We enter do_lookup() looking for some
 549          * name, there is nothing in dcache for this name yet and d_lookup()
 550          * returns NULL.  We proceed to real_lookup(), and while we do this,
 551          * another process does open on the same file we looking up (most simple
 552          * reproducer), open succeeds and the dentry is added. Now back to
 553          * us. In real_lookup() we do d_lookup() again and suddenly find the
 554          * dentry, so we call d_revalidate on it, but there is no lock, so
 555          * without this code we would return 0, but unpatched real_lookup just
 556          * returns -ENOENT in such a case instead of retrying the lookup. Once
 557          * this is dealt with in real_lookup(), all of this ugly mess can go and
 558          * we can just check locks in ->d_revalidate without doing any RPCs
 559          * ever.
 560          */
 561 do_lookup:
 562         if (it != &lookup_it) {
 563                 /* MDS_INODELOCK_UPDATE needed for IT_GETATTR case. */
 564                 if (it->it_op == IT_GETATTR)
 565                         lookup_it.it_op = IT_GETATTR;
 566                 ll_lookup_finish_locks(it, de);
 567                 it = &lookup_it;
 568         }
 569
 570         /* Do real lookup here. */
 571         op_data = ll_prep_md_op_data(NULL, parent, NULL, de->d_name.name,
 572                                      de->d_name.len, 0, (it->it_op & IT_CREAT ?
 573                                                          LUSTRE_OPC_CREATE :
 574                                                          LUSTRE_OPC_ANY), NULL);
 575         if (IS_ERR(op_data))
 576                 RETURN(PTR_ERR(op_data));
 577
 578         rc = md_intent_lock(exp, op_data, NULL, 0,  it, 0, &req,
 579                             ll_md_blocking_ast, 0);
 580         if (rc >= 0) {
 581                 struct mdt_body *mdt_body;
 582                 struct lu_fid fid = {.f_seq = 0, .f_oid = 0, .f_ver = 0};
 583                 mdt_body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 584
 585                 if (de->d_inode)
 586                         fid = *ll_inode2fid(de->d_inode);
 587
 588                 /* see if we got same inode, if not - return error */
 589                 if (lu_fid_eq(&fid, &mdt_body->fid1)) {
 590                         ll_finish_md_op_data(op_data);
 591                         op_data = NULL;
 592                         goto revalidate_finish;
 593                 }
 594                 ll_intent_release(it);
 595         }
 596         ll_finish_md_op_data(op_data);
 597         GOTO(out, rc = 0);
 598
 599 out_sa:
 600         /*
 601          * For rc == 1 case, should not return directly to prevent losing
 602          * statahead windows; for rc == 0 case, the "lookup" will be done later.
 603          */
 604         if (it && it->it_op == IT_GETATTR && rc == 1) {
 605                 first = ll_statahead_enter(de->d_parent->d_inode, &de, 0);
 606                 if (!first)
 607                         ll_statahead_exit(de, rc);
 608         }
 609
 610         return rc;
 611 }
 612
 613 /*static*/ void ll_pin(struct dentry *de, struct vfsmount *mnt, int flag)
 614 {
 615         struct inode *inode= de->d_inode;
 616         struct ll_sb_info *sbi = ll_i2sbi(inode);
 617         struct ll_dentry_data *ldd = ll_d2d(de);
 618         struct obd_client_handle *handle;
 619         struct obd_capa *oc;
 620         int rc = 0;
 621         ENTRY;
 622         LASSERT(ldd);
 623
 624         lock_kernel();
 625         /* Strictly speaking this introduces an additional race: the
 626          * increments should wait until the rpc has returned.
 627          * However, given that at present the function is void, this
 628          * issue is moot. */
 629         if (flag == 1 && (++ldd->lld_mnt_count) > 1) {
 630                 unlock_kernel();
 631                 EXIT;
 632                 return;
 633         }
 634
 635         if (flag == 0 && (++ldd->lld_cwd_count) > 1) {
 636                 unlock_kernel();
 637                 EXIT;
 638                 return;
 639         }
 640         unlock_kernel();
 641
 642         handle = (flag) ? &ldd->lld_mnt_och : &ldd->lld_cwd_och;
 643         oc = ll_mdscapa_get(inode);
 644         rc = obd_pin(sbi->ll_md_exp, ll_inode2fid(inode), oc, handle, flag);
 645         capa_put(oc);
 646         if (rc) {
 647                 lock_kernel();
 648                 memset(handle, 0, sizeof(*handle));
 649                 if (flag == 0)
 650                         ldd->lld_cwd_count--;
 651                 else
 652                         ldd->lld_mnt_count--;
 653                 unlock_kernel();
 654         }
 655
 656         EXIT;
 657         return;
 658 }
 659
 660 /*static*/ void ll_unpin(struct dentry *de, struct vfsmount *mnt, int flag)
 661 {
 662         struct ll_sb_info *sbi = ll_i2sbi(de->d_inode);
 663         struct ll_dentry_data *ldd = ll_d2d(de);
 664         struct obd_client_handle handle;
 665         int count, rc = 0;
 666         ENTRY;
 667         LASSERT(ldd);
 668
 669         lock_kernel();
 670         /* Strictly speaking this introduces an additional race: the
 671          * increments should wait until the rpc has returned.
 672          * However, given that at present the function is void, this
 673          * issue is moot. */
 674         handle = (flag) ? ldd->lld_mnt_och : ldd->lld_cwd_och;
 675         if (handle.och_magic != OBD_CLIENT_HANDLE_MAGIC) {
 676                 /* the "pin" failed */
 677                 unlock_kernel();
 678                 EXIT;
 679                 return;
 680         }
 681
 682         if (flag)
 683                 count = --ldd->lld_mnt_count;
 684         else
 685                 count = --ldd->lld_cwd_count;
 686         unlock_kernel();
 687
 688         if (count != 0) {
 689                 EXIT;
 690                 return;
 691         }
 692
 693         rc = obd_unpin(sbi->ll_md_exp, &handle, flag);
 694         EXIT;
 695         return;
 696 }
 697
 698 #ifdef HAVE_VFS_INTENT_PATCHES
 699 static int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
 700 {
 701         int rc;
 702         ENTRY;
 703
 704         if (nd && nd->flags & LOOKUP_LAST && !(nd->flags & LOOKUP_LINK_NOTLAST))
 705                 rc = ll_revalidate_it(dentry, nd->flags, &nd->intent);
 706         else
 707                 rc = ll_revalidate_it(dentry, 0, NULL);
 708
 709         RETURN(rc);
 710 }
 711 #else
 712 int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
 713 {
 714         int rc;
 715         ENTRY;
 716
 717         if (nd && !(nd->flags & (LOOKUP_CONTINUE|LOOKUP_PARENT))) {
 718                 struct lookup_intent *it;
 719                 it = ll_convert_intent(&nd->intent.open, nd->flags);
 720                 if (IS_ERR(it))
 721                         RETURN(0);
 722                 if (it->it_op == (IT_OPEN|IT_CREAT))
 723                         if (nd->intent.open.flags & O_EXCL) {
 724                                 CDEBUG(D_VFSTRACE, "create O_EXCL, returning 0\n");
 725                                 rc = 0;
 726                                 goto out_it;
 727                         }
 728
 729                 rc = ll_revalidate_it(dentry, nd->flags, it);
 730
 731                 if (rc && (nd->flags & LOOKUP_OPEN) &&
 732                     it_disposition(it, DISP_OPEN_OPEN)) {/*Open*/
 733 #ifdef HAVE_FILE_IN_STRUCT_INTENT
 734 // XXX Code duplication with ll_lookup_nd
 735                         if (S_ISFIFO(dentry->d_inode->i_mode)) {
 736                                 // We cannot call open here as it would
 737                                 // deadlock.
 738                                 ptlrpc_req_finished(
 739                                                (struct ptlrpc_request *)
 740                                                   it->d.lustre.it_data);
 741                         } else {
 742                                 struct file *filp;
 743
 744                                 nd->intent.open.file->private_data = it;
 745                                 filp = lookup_instantiate_filp(nd, dentry,NULL);
 746 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,6,17))
 747 /* 2.6.1[456] have a bug in open_namei() that forgets to check
 748  * nd->intent.open.file for error, so we need to return it as lookup's result
 749  * instead */
 750                                 if (IS_ERR(filp))
 751                                         rc = 0;
 752 #endif
 753                         }
 754 #else
 755                         ll_release_openhandle(dentry, it);
 756 #endif /* HAVE_FILE_IN_STRUCT_INTENT */
 757                 }
 758                 if (!rc && (nd->flags & LOOKUP_CREATE) &&
 759                     it_disposition(it, DISP_OPEN_CREATE)) {
 760                         /* We created something but we may only return
 761                          * negative dentry here, so save request in dentry,
 762                          * if lookup will be called later on, it will
 763                          * pick the request, otherwise it would be freed
 764                          * with dentry */
 765                         ll_d2d(dentry)->lld_it = it;
 766                         it = NULL; /* avoid freeing */
 767                 }
 768
 769 out_it:
 770                 if (it) {
 771                         ll_intent_release(it);
 772                         OBD_FREE(it, sizeof(*it));
 773                 }
 774         } else {
 775                 rc = ll_revalidate_it(dentry, 0, NULL);
 776         }
 777
 778         RETURN(rc);
 779 }
 780 #endif
 781
 782 struct dentry_operations ll_d_ops = {
 783         .d_revalidate = ll_revalidate_nd,
 784         .d_release = ll_release,
 785         .d_delete = ll_ddelete,
 786 #ifdef DCACHE_LUSTRE_INVALID
 787         .d_compare = ll_dcompare,
 788 #endif
 789 #if 0
 790         .d_pin = ll_pin,
 791         .d_unpin = ll_unpin,
 792 #endif
 793 };
 794
 795 static int ll_fini_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
 796 {
 797         ENTRY;
 798         /* need lookup */
 799         RETURN(0);
 800 }
 801
 802 struct dentry_operations ll_fini_d_ops = {
 803         .d_revalidate = ll_fini_revalidate_nd,
 804         .d_release = ll_release,
 805 };
 806
 807 /*
 808  * It is for the following race condition:
 809  * When someone (maybe statahead thread) adds the dentry to the dentry hash
 810  * table, the dentry's "d_op" maybe NULL, at the same time, another (maybe
 811  * "ls -l") process finds such dentry by "do_lookup()" without "do_revalidate()"
 812  * called. It causes statahead window lost, and maybe other issues. --Fan Yong
 813  */
 814 static int ll_init_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
 815 {
 816         struct l_wait_info lwi = { 0 };
 817         struct ll_dentry_data *lld;
 818         ENTRY;
 819
 820         ll_set_dd(dentry);
 821         lld = ll_d2d(dentry);
 822         if (unlikely(lld == NULL))
 823                 RETURN(-ENOMEM);
 824
 825         l_wait_event(lld->lld_waitq, dentry->d_op != &ll_init_d_ops, &lwi);
 826         if (likely(dentry->d_op == &ll_d_ops))
 827                 RETURN(ll_revalidate_nd(dentry, nd));
 828         else
 829                 RETURN(dentry->d_op == &ll_fini_d_ops ? 0 : -EINVAL);
 830 }
 831
 832 struct dentry_operations ll_init_d_ops = {
 833         .d_revalidate = ll_init_revalidate_nd,
 834         .d_release = ll_release,
 835 };