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