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 (c) 2002, 2010, Oracle and/or its affiliates. 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 cfs_spinlock_t ll_lookup_lock = CFS_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         RETURN(0);
 114 }
 115
 116 static inline int return_if_equal(struct ldlm_lock *lock, void *data)
 117 {
 118         if (lock->l_flags & LDLM_FL_CANCELING)
 119                 return LDLM_ITER_CONTINUE;
 120         return LDLM_ITER_STOP;
 121 }
 122
 123 /* find any ldlm lock of the inode in mdc and lov
 124  * return 0    not find
 125  *        1    find one
 126  *      < 0    error */
 127 static int find_cbdata(struct inode *inode)
 128 {
 129         struct ll_inode_info *lli = ll_i2info(inode);
 130         struct ll_sb_info *sbi = ll_i2sbi(inode);
 131         int rc = 0;
 132         ENTRY;
 133
 134         LASSERT(inode);
 135         rc = md_find_cbdata(sbi->ll_md_exp, ll_inode2fid(inode),
 136                             return_if_equal, NULL);
 137         if (rc != 0)
 138                  RETURN(rc);
 139
 140         if (lli->lli_smd)
 141                 rc = obd_find_cbdata(sbi->ll_dt_exp, lli->lli_smd,
 142                                      return_if_equal, NULL);
 143
 144         RETURN(rc);
 145 }
 146
 147 /* should NOT be called with the dcache lock, see fs/dcache.c */
 148 static int ll_ddelete(struct dentry *de)
 149 {
 150         ENTRY;
 151         LASSERT(de);
 152
 153         CDEBUG(D_DENTRY, "%s dentry %.*s (%p, parent %p, inode %p) %s%s\n",
 154                (de->d_flags & DCACHE_LUSTRE_INVALID ? "deleting" : "keeping"),
 155                de->d_name.len, de->d_name.name, de, de->d_parent, de->d_inode,
 156                d_unhashed(de) ? "" : "hashed,",
 157                list_empty(&de->d_subdirs) ? "" : "subdirs");
 158
 159         /* if not ldlm lock for this inode, set i_nlink to 0 so that
 160          * this inode can be recycled later b=20433 */
 161         LASSERT(atomic_read(&de->d_count) == 0);
 162         if (de->d_inode && !find_cbdata(de->d_inode))
 163                 de->d_inode->i_nlink = 0;
 164
 165         if (de->d_flags & DCACHE_LUSTRE_INVALID)
 166                 RETURN(1);
 167
 168         RETURN(0);
 169 }
 170
 171 static int ll_set_dd(struct dentry *de)
 172 {
 173         ENTRY;
 174         LASSERT(de != NULL);
 175
 176         CDEBUG(D_DENTRY, "ldd on dentry %.*s (%p) parent %p inode %p refc %d\n",
 177                de->d_name.len, de->d_name.name, de, de->d_parent, de->d_inode,
 178                atomic_read(&de->d_count));
 179
 180         if (de->d_fsdata == NULL) {
 181                 struct ll_dentry_data *lld;
 182
 183                 OBD_ALLOC_PTR(lld);
 184                 if (likely(lld != NULL)) {
 185                         lock_dentry(de);
 186                         if (likely(de->d_fsdata == NULL))
 187                                 de->d_fsdata = lld;
 188                         else
 189                                 OBD_FREE_PTR(lld);
 190                         unlock_dentry(de);
 191                 } else {
 192                         RETURN(-ENOMEM);
 193                 }
 194         }
 195
 196         RETURN(0);
 197 }
 198
 199 int ll_dops_init(struct dentry *de, int block)
 200 {
 201         struct ll_dentry_data *lld = ll_d2d(de);
 202         int rc = 0;
 203
 204         if (lld == NULL && block != 0) {
 205                 rc = ll_set_dd(de);
 206                 if (rc)
 207                         return rc;
 208
 209                 lld = ll_d2d(de);
 210         }
 211
 212         if (lld != NULL)
 213                 lld->lld_sa_generation = 0;
 214
 215         de->d_op = &ll_d_ops;
 216         return rc;
 217 }
 218
 219 void ll_intent_drop_lock(struct lookup_intent *it)
 220 {
 221         struct lustre_handle *handle;
 222
 223         if (it->it_op && it->d.lustre.it_lock_mode) {
 224                 handle = (struct lustre_handle *)&it->d.lustre.it_lock_handle;
 225                 CDEBUG(D_DLMTRACE, "releasing lock with cookie "LPX64
 226                        " from it %p\n", handle->cookie, it);
 227                 ldlm_lock_decref(handle, it->d.lustre.it_lock_mode);
 228
 229                 /* bug 494: intent_release may be called multiple times, from
 230                  * this thread and we don't want to double-decref this lock */
 231                 it->d.lustre.it_lock_mode = 0;
 232         }
 233 }
 234
 235 void ll_intent_release(struct lookup_intent *it)
 236 {
 237         ENTRY;
 238
 239         CDEBUG(D_INFO, "intent %p released\n", it);
 240         ll_intent_drop_lock(it);
 241 #ifdef HAVE_VFS_INTENT_PATCHES
 242         it->it_magic = 0;
 243         it->it_op_release = 0;
 244 #endif
 245         /* We are still holding extra reference on a request, need to free it */
 246         if (it_disposition(it, DISP_ENQ_OPEN_REF))
 247                  ptlrpc_req_finished(it->d.lustre.it_data); /* ll_file_open */
 248         if (it_disposition(it, DISP_ENQ_CREATE_REF)) /* create rec */
 249                 ptlrpc_req_finished(it->d.lustre.it_data);
 250         if (it_disposition(it, DISP_ENQ_COMPLETE)) /* saved req from revalidate
 251                                                     * to lookup */
 252                 ptlrpc_req_finished(it->d.lustre.it_data);
 253
 254         it->d.lustre.it_disposition = 0;
 255         it->d.lustre.it_data = NULL;
 256         EXIT;
 257 }
 258
 259 /* Drop dentry if it is not used already, unhash otherwise.
 260    Should be called with dcache lock held!
 261    Returns: 1 if dentry was dropped, 0 if unhashed. */
 262 int ll_drop_dentry(struct dentry *dentry)
 263 {
 264         lock_dentry(dentry);
 265         if (atomic_read(&dentry->d_count) == 0) {
 266                 CDEBUG(D_DENTRY, "deleting dentry %.*s (%p) parent %p "
 267                        "inode %p\n", dentry->d_name.len,
 268                        dentry->d_name.name, dentry, dentry->d_parent,
 269                        dentry->d_inode);
 270                 dget_locked(dentry);
 271                 __d_drop(dentry);
 272                 unlock_dentry(dentry);
 273                 spin_unlock(&dcache_lock);
 274                 cfs_spin_unlock(&ll_lookup_lock);
 275                 dput(dentry);
 276                 cfs_spin_lock(&ll_lookup_lock);
 277                 spin_lock(&dcache_lock);
 278                 return 1;
 279         }
 280         /* disconected dentry can not be find without lookup, because we
 281          * not need his to unhash or mark invalid. */
 282         if (dentry->d_flags & DCACHE_DISCONNECTED) {
 283                 unlock_dentry(dentry);
 284                 RETURN (0);
 285         }
 286
 287         if (!(dentry->d_flags & DCACHE_LUSTRE_INVALID)) {
 288                 CDEBUG(D_DENTRY, "unhashing dentry %.*s (%p) parent %p "
 289                        "inode %p refc %d\n", dentry->d_name.len,
 290                        dentry->d_name.name, dentry, dentry->d_parent,
 291                        dentry->d_inode, atomic_read(&dentry->d_count));
 292                 /* actually we don't unhash the dentry, rather just
 293                  * mark it inaccessible for to __d_lookup(). otherwise
 294                  * sys_getcwd() could return -ENOENT -bzzz */
 295                 dentry->d_flags |= DCACHE_LUSTRE_INVALID;
 296                 if (!dentry->d_inode || !S_ISDIR(dentry->d_inode->i_mode))
 297                         __d_drop(dentry);
 298         }
 299         unlock_dentry(dentry);
 300         return 0;
 301 }
 302
 303 void ll_unhash_aliases(struct inode *inode)
 304 {
 305         struct list_head *tmp, *head;
 306         ENTRY;
 307
 308         if (inode == NULL) {
 309                 CERROR("unexpected NULL inode, tell phil\n");
 310                 return;
 311         }
 312
 313         CDEBUG(D_INODE, "marking dentries for ino %lu/%u(%p) invalid\n",
 314                inode->i_ino, inode->i_generation, inode);
 315
 316         head = &inode->i_dentry;
 317         cfs_spin_lock(&ll_lookup_lock);
 318         spin_lock(&dcache_lock);
 319 restart:
 320         tmp = head;
 321         while ((tmp = tmp->next) != head) {
 322                 struct dentry *dentry = list_entry(tmp, struct dentry, d_alias);
 323
 324                 CDEBUG(D_DENTRY, "dentry in drop %.*s (%p) parent %p "
 325                        "inode %p flags %d\n", dentry->d_name.len,
 326                        dentry->d_name.name, dentry, dentry->d_parent,
 327                        dentry->d_inode, dentry->d_flags);
 328
 329                 if (dentry->d_name.len == 1 && dentry->d_name.name[0] == '/') {
 330                         CERROR("called on root (?) dentry=%p, inode=%p "
 331                                "ino=%lu\n", dentry, inode, inode->i_ino);
 332                         lustre_dump_dentry(dentry, 1);
 333                         libcfs_debug_dumpstack(NULL);
 334                 }
 335
 336                 if (ll_drop_dentry(dentry))
 337                           goto restart;
 338         }
 339         spin_unlock(&dcache_lock);
 340         cfs_spin_unlock(&ll_lookup_lock);
 341
 342         EXIT;
 343 }
 344
 345 int ll_revalidate_it_finish(struct ptlrpc_request *request,
 346                             struct lookup_intent *it,
 347                             struct dentry *de)
 348 {
 349         int rc = 0;
 350         ENTRY;
 351
 352         if (!request)
 353                 RETURN(0);
 354
 355         if (it_disposition(it, DISP_LOOKUP_NEG))
 356                 RETURN(-ENOENT);
 357
 358         rc = ll_prep_inode(&de->d_inode, request, NULL);
 359
 360         RETURN(rc);
 361 }
 362
 363 void ll_lookup_finish_locks(struct lookup_intent *it, struct dentry *dentry)
 364 {
 365         LASSERT(it != NULL);
 366         LASSERT(dentry != NULL);
 367
 368         if (it->d.lustre.it_lock_mode && dentry->d_inode != NULL) {
 369                 struct inode *inode = dentry->d_inode;
 370                 struct ll_sb_info *sbi = ll_i2sbi(dentry->d_inode);
 371
 372                 CDEBUG(D_DLMTRACE, "setting l_data to inode %p (%lu/%u)\n",
 373                        inode, inode->i_ino, inode->i_generation);
 374                 md_set_lock_data(sbi->ll_md_exp, &it->d.lustre.it_lock_handle,
 375                                  inode, NULL);
 376         }
 377
 378         /* drop lookup or getattr locks immediately */
 379         if (it->it_op == IT_LOOKUP || it->it_op == IT_GETATTR) {
 380                 /* on 2.6 there are situation when several lookups and
 381                  * revalidations may be requested during single operation.
 382                  * therefore, we don't release intent here -bzzz */
 383                 ll_intent_drop_lock(it);
 384         }
 385 }
 386
 387 void ll_frob_intent(struct lookup_intent **itp, struct lookup_intent *deft)
 388 {
 389         struct lookup_intent *it = *itp;
 390 #ifdef HAVE_VFS_INTENT_PATCHES
 391         if (it) {
 392                 LASSERTF(it->it_magic == INTENT_MAGIC,
 393                          "%p has bad intent magic: %x\n",
 394                          it, it->it_magic);
 395         }
 396 #endif
 397
 398         if (!it || it->it_op == IT_GETXATTR)
 399                 it = *itp = deft;
 400
 401 #ifdef HAVE_VFS_INTENT_PATCHES
 402         it->it_op_release = ll_intent_release;
 403 #endif
 404 }
 405
 406 int ll_revalidate_it(struct dentry *de, int lookup_flags,
 407                      struct lookup_intent *it)
 408 {
 409         struct md_op_data *op_data;
 410         struct ptlrpc_request *req = NULL;
 411         struct lookup_intent lookup_it = { .it_op = IT_LOOKUP };
 412         struct obd_export *exp;
 413         struct inode *parent = de->d_parent->d_inode;
 414         int rc, first = 0;
 415
 416         ENTRY;
 417         CDEBUG(D_VFSTRACE, "VFS Op:name=%s,intent=%s\n", de->d_name.name,
 418                LL_IT2STR(it));
 419
 420         if (de->d_inode == NULL) {
 421                 /* We can only use negative dentries if this is stat or lookup,
 422                    for opens and stuff we do need to query server. */
 423                 /* If there is IT_CREAT in intent op set, then we must throw
 424                    away this negative dentry and actually do the request to
 425                    kernel to create whatever needs to be created (if possible)*/
 426                 if (it && (it->it_op & IT_CREAT))
 427                         RETURN(0);
 428
 429                 if (de->d_flags & DCACHE_LUSTRE_INVALID)
 430                         RETURN(0);
 431
 432                 rc = ll_have_md_lock(parent, MDS_INODELOCK_UPDATE, LCK_MINMODE);
 433                 GOTO(out_sa, rc);
 434         }
 435
 436         /* Never execute intents for mount points.
 437          * Attributes will be fixed up in ll_inode_revalidate_it */
 438         if (d_mountpoint(de))
 439                 GOTO(out_sa, rc = 1);
 440
 441         /* need to get attributes in case root got changed from other client */
 442         if (de == de->d_sb->s_root) {
 443                 rc = __ll_inode_revalidate_it(de, it, MDS_INODELOCK_LOOKUP);
 444                 if (rc == 0)
 445                         rc = 1;
 446                 GOTO(out_sa, rc);
 447         }
 448
 449         exp = ll_i2mdexp(de->d_inode);
 450
 451         OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_REVALIDATE_PAUSE, 5);
 452         ll_frob_intent(&it, &lookup_it);
 453         LASSERT(it);
 454
 455         if (it->it_op == IT_LOOKUP && !(de->d_flags & DCACHE_LUSTRE_INVALID))
 456                 GOTO(out_sa, rc = 1);
 457
 458         op_data = ll_prep_md_op_data(NULL, parent, de->d_inode,
 459                                      de->d_name.name, de->d_name.len,
 460                                      0, LUSTRE_OPC_ANY, NULL);
 461         if (IS_ERR(op_data))
 462                 RETURN(PTR_ERR(op_data));
 463
 464         if ((it->it_op == IT_OPEN) && de->d_inode) {
 465                 struct inode *inode = de->d_inode;
 466                 struct ll_inode_info *lli = ll_i2info(inode);
 467                 struct obd_client_handle **och_p;
 468                 __u64 *och_usecount;
 469
 470                 /*
 471                  * We used to check for MDS_INODELOCK_OPEN here, but in fact
 472                  * just having LOOKUP lock is enough to justify inode is the
 473                  * same. And if inode is the same and we have suitable
 474                  * openhandle, then there is no point in doing another OPEN RPC
 475                  * just to throw away newly received openhandle.  There are no
 476                  * security implications too, if file owner or access mode is
 477                  * change, LOOKUP lock is revoked.
 478                  */
 479
 480
 481                 if (it->it_flags & FMODE_WRITE) {
 482                         och_p = &lli->lli_mds_write_och;
 483                         och_usecount = &lli->lli_open_fd_write_count;
 484                 } else if (it->it_flags & FMODE_EXEC) {
 485                         och_p = &lli->lli_mds_exec_och;
 486                         och_usecount = &lli->lli_open_fd_exec_count;
 487                 } else {
 488                         och_p = &lli->lli_mds_read_och;
 489                         och_usecount = &lli->lli_open_fd_read_count;
 490                 }
 491                 /* Check for the proper lock. */
 492                 if (!ll_have_md_lock(inode, MDS_INODELOCK_LOOKUP, LCK_MINMODE))
 493                         goto do_lock;
 494                 cfs_down(&lli->lli_och_sem);
 495                 if (*och_p) { /* Everything is open already, do nothing */
 496                         /*(*och_usecount)++;  Do not let them steal our open
 497                           handle from under us */
 498                         /* XXX The code above was my original idea, but in case
 499                            we have the handle, but we cannot use it due to later
 500                            checks (e.g. O_CREAT|O_EXCL flags set), nobody
 501                            would decrement counter increased here. So we just
 502                            hope the lock won't be invalidated in between. But
 503                            if it would be, we'll reopen the open request to
 504                            MDS later during file open path */
 505                         cfs_up(&lli->lli_och_sem);
 506                         ll_finish_md_op_data(op_data);
 507                         RETURN(1);
 508                 } else {
 509                         cfs_up(&lli->lli_och_sem);
 510                 }
 511         }
 512
 513         if (it->it_op == IT_GETATTR)
 514                 first = ll_statahead_enter(parent, &de, 0);
 515
 516 do_lock:
 517         it->it_create_mode &= ~cfs_curproc_umask();
 518         it->it_create_mode |= M_CHECK_STALE;
 519         rc = md_intent_lock(exp, op_data, NULL, 0, it,
 520                             lookup_flags,
 521                             &req, ll_md_blocking_ast, 0);
 522         it->it_create_mode &= ~M_CHECK_STALE;
 523         ll_finish_md_op_data(op_data);
 524         if (it->it_op == IT_GETATTR && !first)
 525                 /* If there are too many locks on client-side, then some
 526                  * locks taken by statahead maybe dropped automatically
 527                  * before the real "revalidate" using them. */
 528                 ll_statahead_exit(parent, de, req == NULL ? rc : 0);
 529         else if (first == -EEXIST)
 530                 ll_statahead_mark(parent, de);
 531
 532         /* If req is NULL, then md_intent_lock only tried to do a lock match;
 533          * if all was well, it will return 1 if it found locks, 0 otherwise. */
 534         if (req == NULL && rc >= 0) {
 535                 if (!rc)
 536                         goto do_lookup;
 537                 GOTO(out, rc);
 538         }
 539
 540         if (rc < 0) {
 541                 if (rc != -ESTALE) {
 542                         CDEBUG(D_INFO, "ll_intent_lock: rc %d : it->it_status "
 543                                "%d\n", rc, it->d.lustre.it_status);
 544                 }
 545                 GOTO(out, rc = 0);
 546         }
 547
 548 revalidate_finish:
 549         rc = ll_revalidate_it_finish(req, it, de);
 550         if (rc != 0) {
 551                 if (rc != -ESTALE && rc != -ENOENT)
 552                         ll_intent_release(it);
 553                 GOTO(out, rc = 0);
 554         }
 555
 556         if ((it->it_op & IT_OPEN) && de->d_inode &&
 557             !S_ISREG(de->d_inode->i_mode) &&
 558             !S_ISDIR(de->d_inode->i_mode)) {
 559                 ll_release_openhandle(de, it);
 560         }
 561         rc = 1;
 562
 563         /* unfortunately ll_intent_lock may cause a callback and revoke our
 564          * dentry */
 565         cfs_spin_lock(&ll_lookup_lock);
 566         spin_lock(&dcache_lock);
 567         lock_dentry(de);
 568         __d_drop(de);
 569         unlock_dentry(de);
 570         d_rehash_cond(de, 0);
 571         spin_unlock(&dcache_lock);
 572         cfs_spin_unlock(&ll_lookup_lock);
 573
 574 out:
 575         /* We do not free request as it may be reused during following lookup
 576          * (see comment in mdc/mdc_locks.c::mdc_intent_lock()), request will
 577          * be freed in ll_lookup_it or in ll_intent_release. But if
 578          * request was not completed, we need to free it. (bug 5154, 9903) */
 579         if (req != NULL && !it_disposition(it, DISP_ENQ_COMPLETE))
 580                 ptlrpc_req_finished(req);
 581         if (rc == 0) {
 582                 ll_unhash_aliases(de->d_inode);
 583                 /* done in ll_unhash_aliases()
 584                    dentry->d_flags |= DCACHE_LUSTRE_INVALID; */
 585         } else {
 586                 CDEBUG(D_DENTRY, "revalidated dentry %.*s (%p) parent %p "
 587                        "inode %p refc %d\n", de->d_name.len,
 588                        de->d_name.name, de, de->d_parent, de->d_inode,
 589                        atomic_read(&de->d_count));
 590                 if (de->d_flags & DCACHE_LUSTRE_INVALID) {
 591                         lock_dentry(de);
 592                         de->d_flags &= ~DCACHE_LUSTRE_INVALID;
 593                         unlock_dentry(de);
 594                 }
 595                 ll_lookup_finish_locks(it, de);
 596         }
 597         RETURN(rc);
 598
 599         /*
 600          * This part is here to combat evil-evil race in real_lookup on 2.6
 601          * kernels.  The race details are: We enter do_lookup() looking for some
 602          * name, there is nothing in dcache for this name yet and d_lookup()
 603          * returns NULL.  We proceed to real_lookup(), and while we do this,
 604          * another process does open on the same file we looking up (most simple
 605          * reproducer), open succeeds and the dentry is added. Now back to
 606          * us. In real_lookup() we do d_lookup() again and suddenly find the
 607          * dentry, so we call d_revalidate on it, but there is no lock, so
 608          * without this code we would return 0, but unpatched real_lookup just
 609          * returns -ENOENT in such a case instead of retrying the lookup. Once
 610          * this is dealt with in real_lookup(), all of this ugly mess can go and
 611          * we can just check locks in ->d_revalidate without doing any RPCs
 612          * ever.
 613          */
 614 do_lookup:
 615         if (it != &lookup_it) {
 616                 /* MDS_INODELOCK_UPDATE needed for IT_GETATTR case. */
 617                 if (it->it_op == IT_GETATTR)
 618                         lookup_it.it_op = IT_GETATTR;
 619                 ll_lookup_finish_locks(it, de);
 620                 it = &lookup_it;
 621         }
 622
 623         /* Do real lookup here. */
 624         op_data = ll_prep_md_op_data(NULL, parent, NULL, de->d_name.name,
 625                                      de->d_name.len, 0, (it->it_op & IT_CREAT ?
 626                                                          LUSTRE_OPC_CREATE :
 627                                                          LUSTRE_OPC_ANY), NULL);
 628         if (IS_ERR(op_data))
 629                 RETURN(PTR_ERR(op_data));
 630
 631         rc = md_intent_lock(exp, op_data, NULL, 0,  it, 0, &req,
 632                             ll_md_blocking_ast, 0);
 633         if (rc >= 0) {
 634                 struct mdt_body *mdt_body;
 635                 struct lu_fid fid = {.f_seq = 0, .f_oid = 0, .f_ver = 0};
 636                 mdt_body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 637
 638                 if (de->d_inode)
 639                         fid = *ll_inode2fid(de->d_inode);
 640
 641                 /* see if we got same inode, if not - return error */
 642                 if (lu_fid_eq(&fid, &mdt_body->fid1)) {
 643                         ll_finish_md_op_data(op_data);
 644                         op_data = NULL;
 645                         goto revalidate_finish;
 646                 }
 647                 ll_intent_release(it);
 648         }
 649         ll_finish_md_op_data(op_data);
 650         GOTO(out, rc = 0);
 651
 652 out_sa:
 653         /*
 654          * For rc == 1 case, should not return directly to prevent losing
 655          * statahead windows; for rc == 0 case, the "lookup" will be done later.
 656          */
 657         if (it && it->it_op == IT_GETATTR && rc == 1) {
 658                 first = ll_statahead_enter(parent, &de, 0);
 659                 if (first >= 0)
 660                         ll_statahead_exit(parent, de, 1);
 661                 else if (first == -EEXIST)
 662                         ll_statahead_mark(parent, de);
 663         }
 664
 665         return rc;
 666 }
 667
 668 #if 0
 669 static void ll_pin(struct dentry *de, struct vfsmount *mnt, int flag)
 670 {
 671         struct inode *inode= de->d_inode;
 672         struct ll_sb_info *sbi = ll_i2sbi(inode);
 673         struct ll_dentry_data *ldd = ll_d2d(de);
 674         struct obd_client_handle *handle;
 675         struct obd_capa *oc;
 676         int rc = 0;
 677         ENTRY;
 678         LASSERT(ldd);
 679
 680         cfs_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         if (flag == 1 && (++ldd->lld_mnt_count) > 1) {
 686                 cfs_unlock_kernel();
 687                 EXIT;
 688                 return;
 689         }
 690
 691         if (flag == 0 && (++ldd->lld_cwd_count) > 1) {
 692                 cfs_unlock_kernel();
 693                 EXIT;
 694                 return;
 695         }
 696         cfs_unlock_kernel();
 697
 698         handle = (flag) ? &ldd->lld_mnt_och : &ldd->lld_cwd_och;
 699         oc = ll_mdscapa_get(inode);
 700         rc = obd_pin(sbi->ll_md_exp, ll_inode2fid(inode), oc, handle, flag);
 701         capa_put(oc);
 702         if (rc) {
 703                 cfs_lock_kernel();
 704                 memset(handle, 0, sizeof(*handle));
 705                 if (flag == 0)
 706                         ldd->lld_cwd_count--;
 707                 else
 708                         ldd->lld_mnt_count--;
 709                 cfs_unlock_kernel();
 710         }
 711
 712         EXIT;
 713         return;
 714 }
 715
 716 static void ll_unpin(struct dentry *de, struct vfsmount *mnt, int flag)
 717 {
 718         struct ll_sb_info *sbi = ll_i2sbi(de->d_inode);
 719         struct ll_dentry_data *ldd = ll_d2d(de);
 720         struct obd_client_handle handle;
 721         int count, rc = 0;
 722         ENTRY;
 723         LASSERT(ldd);
 724
 725         cfs_lock_kernel();
 726         /* Strictly speaking this introduces an additional race: the
 727          * increments should wait until the rpc has returned.
 728          * However, given that at present the function is void, this
 729          * issue is moot. */
 730         handle = (flag) ? ldd->lld_mnt_och : ldd->lld_cwd_och;
 731         if (handle.och_magic != OBD_CLIENT_HANDLE_MAGIC) {
 732                 /* the "pin" failed */
 733                 cfs_unlock_kernel();
 734                 EXIT;
 735                 return;
 736         }
 737
 738         if (flag)
 739                 count = --ldd->lld_mnt_count;
 740         else
 741                 count = --ldd->lld_cwd_count;
 742         cfs_unlock_kernel();
 743
 744         if (count != 0) {
 745                 EXIT;
 746                 return;
 747         }
 748
 749         rc = obd_unpin(sbi->ll_md_exp, &handle, flag);
 750         EXIT;
 751         return;
 752 }
 753 #endif
 754
 755 #ifdef HAVE_VFS_INTENT_PATCHES
 756 int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
 757 {
 758         int rc;
 759         ENTRY;
 760
 761         if (nd && nd->flags & LOOKUP_LAST && !(nd->flags & LOOKUP_LINK_NOTLAST))
 762                 rc = ll_revalidate_it(dentry, nd->flags, &nd->intent);
 763         else
 764                 rc = ll_revalidate_it(dentry, 0, NULL);
 765
 766         RETURN(rc);
 767 }
 768 #else
 769 int ll_revalidate_nd(struct dentry *dentry, struct nameidata *nd)
 770 {
 771         int rc;
 772         ENTRY;
 773
 774         if (nd && !(nd->flags & (LOOKUP_CONTINUE|LOOKUP_PARENT))) {
 775                 struct lookup_intent *it;
 776
 777                 it = ll_convert_intent(&nd->intent.open, nd->flags);
 778                 if (IS_ERR(it))
 779                         RETURN(0);
 780
 781                 if (it->it_op == (IT_OPEN|IT_CREAT) &&
 782                     nd->intent.open.flags & O_EXCL) {
 783                         CDEBUG(D_VFSTRACE, "create O_EXCL, returning 0\n");
 784                         rc = 0;
 785                         goto out_it;
 786                 }
 787
 788                 rc = ll_revalidate_it(dentry, nd->flags, it);
 789
 790                 if (rc && (nd->flags & LOOKUP_OPEN) &&
 791                     it_disposition(it, DISP_OPEN_OPEN)) {/*Open*/
 792 #ifdef HAVE_FILE_IN_STRUCT_INTENT
 793 // XXX Code duplication with ll_lookup_nd
 794                         if (S_ISFIFO(dentry->d_inode->i_mode)) {
 795                                 // We cannot call open here as it would
 796                                 // deadlock.
 797                                 ptlrpc_req_finished(
 798                                                (struct ptlrpc_request *)
 799                                                   it->d.lustre.it_data);
 800                         } else {
 801 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,6,17))
 802 /* 2.6.1[456] have a bug in open_namei() that forgets to check
 803  * nd->intent.open.file for error, so we need to return it as lookup's result
 804  * instead */
 805                                 struct file *filp;
 806
 807                                 nd->intent.open.file->private_data = it;
 808                                 filp = lookup_instantiate_filp(nd, dentry,NULL);
 809                                 if (IS_ERR(filp)) {
 810                                         rc = PTR_ERR(filp);
 811                                 }
 812 #else
 813                                 nd->intent.open.file->private_data = it;
 814                                 (void)lookup_instantiate_filp(nd, dentry,NULL);
 815 #endif
 816                         }
 817 #else
 818                         ll_release_openhandle(dentry, it);
 819 #endif /* HAVE_FILE_IN_STRUCT_INTENT */
 820                 }
 821                 if (!rc && (nd->flags & LOOKUP_CREATE) &&
 822                     it_disposition(it, DISP_OPEN_CREATE)) {
 823                         /* We created something but we may only return
 824                          * negative dentry here, so save request in dentry,
 825                          * if lookup will be called later on, it will
 826                          * pick the request, otherwise it would be freed
 827                          * with dentry */
 828                         ll_d2d(dentry)->lld_it = it;
 829                         it = NULL; /* avoid freeing */
 830                 }
 831
 832 out_it:
 833                 if (it) {
 834                         ll_intent_release(it);
 835                         OBD_FREE(it, sizeof(*it));
 836                 }
 837         } else {
 838                 rc = ll_revalidate_it(dentry, 0, NULL);
 839         }
 840
 841         RETURN(rc);
 842 }
 843 #endif
 844
 845 void ll_d_iput(struct dentry *de, struct inode *inode)
 846 {
 847         LASSERT(inode);
 848         if (!find_cbdata(inode))
 849                 inode->i_nlink = 0;
 850         iput(inode);
 851 }
 852
 853 struct dentry_operations ll_d_ops = {
 854         .d_revalidate = ll_revalidate_nd,
 855         .d_release = ll_release,
 856         .d_delete  = ll_ddelete,
 857         .d_iput    = ll_d_iput,
 858         .d_compare = ll_dcompare,
 859 #if 0
 860         .d_pin = ll_pin,
 861         .d_unpin = ll_unpin,
 862 #endif
 863 };