lustre/llite/dcache.c

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