lustre/llite/statahead.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) 2008, 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/mm.h>
  40 #include <linux/smp_lock.h>
  41 #include <linux/highmem.h>
  42 #include <linux/pagemap.h>
  43
  44 #define DEBUG_SUBSYSTEM S_LLITE
  45
  46 #include <obd_support.h>
  47 #include <lustre_lite.h>
  48 #include <lustre_dlm.h>
  49 #include <linux/lustre_version.h>
  50 #include "llite_internal.h"
  51
  52 struct ll_sai_entry {
  53         cfs_list_t              se_list;
  54         unsigned int            se_index;
  55         int                     se_stat;
  56         struct ptlrpc_request  *se_req;
  57         struct md_enqueue_info *se_minfo;
  58         struct dentry          *se_dentry;
  59         struct inode           *se_inode;
  60 };
  61
  62 enum {
  63         SA_ENTRY_UNSTATED = 0,
  64         SA_ENTRY_STATED
  65 };
  66
  67 static unsigned int sai_generation = 0;
  68 static cfs_spinlock_t sai_generation_lock = CFS_SPIN_LOCK_UNLOCKED;
  69
  70 /**
  71  * Check whether first entry was stated already or not.
  72  * No need to hold lli_sa_lock, for:
  73  * (1) it is me that remove entry from the list
  74  * (2) the statahead thread only add new entry to the list
  75  */
  76 static int ll_sai_entry_stated(struct ll_statahead_info *sai)
  77 {
  78         struct ll_sai_entry  *entry;
  79         int                   rc = 0;
  80
  81         if (!cfs_list_empty(&sai->sai_entries_stated)) {
  82                 entry = cfs_list_entry(sai->sai_entries_stated.next,
  83                                        struct ll_sai_entry, se_list);
  84                 if (entry->se_index == sai->sai_index_next)
  85                         rc = 1;
  86         }
  87         return rc;
  88 }
  89
  90 static inline int sa_received_empty(struct ll_statahead_info *sai)
  91 {
  92         return cfs_list_empty(&sai->sai_entries_received);
  93 }
  94
  95 static inline int sa_not_full(struct ll_statahead_info *sai)
  96 {
  97         return (sai->sai_index < sai->sai_hit + sai->sai_miss + sai->sai_max);
  98 }
  99
 100 static inline int sa_is_running(struct ll_statahead_info *sai)
 101 {
 102         return !!(sai->sai_thread.t_flags & SVC_RUNNING);
 103 }
 104
 105 static inline int sa_is_stopping(struct ll_statahead_info *sai)
 106 {
 107         return !!(sai->sai_thread.t_flags & SVC_STOPPING);
 108 }
 109
 110 static inline int sa_is_stopped(struct ll_statahead_info *sai)
 111 {
 112         return !!(sai->sai_thread.t_flags & SVC_STOPPED);
 113 }
 114
 115 /**
 116  * (1) hit ratio less than 80%
 117  * or
 118  * (2) consecutive miss more than 8
 119  */
 120 static inline int sa_low_hit(struct ll_statahead_info *sai)
 121 {
 122         return ((sai->sai_hit > 7 && sai->sai_hit < 4 * sai->sai_miss) ||
 123                 (sai->sai_consecutive_miss > 8));
 124 }
 125
 126 static inline int sa_skip_nolock(struct ll_statahead_info *sai)
 127 {
 128         return (sai->sai_nolock >= 3);
 129 }
 130
 131 static void ll_sai_entry_free(struct ll_sai_entry *entry)
 132 {
 133         struct dentry *dentry = entry->se_dentry;
 134         struct inode  *inode  = entry->se_inode;
 135
 136         if (dentry) {
 137                 struct ll_dentry_data *lld = ll_d2d(dentry);
 138                 struct ll_inode_info *lli;
 139
 140                 entry->se_dentry = NULL;
 141                 LASSERT(inode != NULL);
 142                 lli = ll_i2info(inode);
 143                 if (!cfs_list_empty(&lli->lli_sa_dentry)) {
 144                         cfs_spin_lock(&lli->lli_sa_lock);
 145                         cfs_list_del_init(&lld->lld_sa_alias);
 146                         cfs_spin_unlock(&lli->lli_sa_lock);
 147                 }
 148                 dput(dentry);
 149         }
 150         if (inode) {
 151                 entry->se_inode = NULL;
 152                 iput(inode);
 153         }
 154         LASSERT(cfs_list_empty(&entry->se_list));
 155         OBD_FREE_PTR(entry);
 156 }
 157
 158 /**
 159  * process the deleted entry's member and free the entry.
 160  * (1) release intent
 161  * (2) free md_enqueue_info
 162  * (3) drop dentry's ref count
 163  * (4) release request's ref count
 164  */
 165 static void ll_sai_entry_cleanup(struct ll_sai_entry *entry, int free)
 166 {
 167         struct md_enqueue_info *minfo = entry->se_minfo;
 168         struct ptlrpc_request  *req   = entry->se_req;
 169         ENTRY;
 170
 171         if (minfo) {
 172                 entry->se_minfo = NULL;
 173                 ll_intent_release(&minfo->mi_it);
 174                 dput(minfo->mi_dentry);
 175                 iput(minfo->mi_dir);
 176                 OBD_FREE_PTR(minfo);
 177         }
 178         if (req) {
 179                 entry->se_req = NULL;
 180                 ptlrpc_req_finished(req);
 181         }
 182         if (free)
 183                 ll_sai_entry_free(entry);
 184
 185         EXIT;
 186 }
 187
 188 static struct ll_statahead_info *ll_sai_alloc(void)
 189 {
 190         struct ll_statahead_info *sai;
 191
 192         OBD_ALLOC_PTR(sai);
 193         if (!sai)
 194                 return NULL;
 195
 196         cfs_spin_lock(&sai_generation_lock);
 197         sai->sai_generation = ++sai_generation;
 198         if (unlikely(sai_generation == 0))
 199                 sai->sai_generation = ++sai_generation;
 200         cfs_spin_unlock(&sai_generation_lock);
 201         cfs_atomic_set(&sai->sai_refcount, 1);
 202         sai->sai_max = LL_SA_RPC_MIN;
 203         cfs_waitq_init(&sai->sai_waitq);
 204         cfs_waitq_init(&sai->sai_thread.t_ctl_waitq);
 205         CFS_INIT_LIST_HEAD(&sai->sai_entries_sent);
 206         CFS_INIT_LIST_HEAD(&sai->sai_entries_received);
 207         CFS_INIT_LIST_HEAD(&sai->sai_entries_stated);
 208         return sai;
 209 }
 210
 211 static inline
 212 struct ll_statahead_info *ll_sai_get(struct ll_statahead_info *sai)
 213 {
 214         LASSERT(sai);
 215         cfs_atomic_inc(&sai->sai_refcount);
 216         return sai;
 217 }
 218
 219 static void ll_sai_put(struct ll_statahead_info *sai)
 220 {
 221         struct inode         *inode = sai->sai_inode;
 222         struct ll_inode_info *lli;
 223         ENTRY;
 224
 225         LASSERT(inode != NULL);
 226         lli = ll_i2info(inode);
 227         LASSERT(lli->lli_sai == sai);
 228
 229         if (cfs_atomic_dec_and_test(&sai->sai_refcount)) {
 230                 struct ll_sai_entry *entry, *next;
 231
 232                 cfs_spin_lock(&lli->lli_sa_lock);
 233                 if (unlikely(cfs_atomic_read(&sai->sai_refcount) > 0)) {
 234                         /* It is race case, the interpret callback just hold
 235                          * a reference count */
 236                         cfs_spin_unlock(&lli->lli_sa_lock);
 237                         EXIT;
 238                         return;
 239                 }
 240
 241                 LASSERT(lli->lli_opendir_key == NULL);
 242                 lli->lli_sai = NULL;
 243                 lli->lli_opendir_pid = 0;
 244                 cfs_spin_unlock(&lli->lli_sa_lock);
 245
 246                 LASSERT(sa_is_stopped(sai));
 247
 248                 if (sai->sai_sent > sai->sai_replied)
 249                         CDEBUG(D_READA,"statahead for dir "DFID" does not "
 250                               "finish: [sent:%u] [replied:%u]\n",
 251                               PFID(&lli->lli_fid),
 252                               sai->sai_sent, sai->sai_replied);
 253
 254                 cfs_list_for_each_entry_safe(entry, next,
 255                                              &sai->sai_entries_sent, se_list) {
 256                         cfs_list_del_init(&entry->se_list);
 257                         ll_sai_entry_cleanup(entry, 1);
 258                 }
 259                 cfs_list_for_each_entry_safe(entry, next,
 260                                              &sai->sai_entries_received,
 261                                              se_list) {
 262                         cfs_list_del_init(&entry->se_list);
 263                         ll_sai_entry_cleanup(entry, 1);
 264                 }
 265                 cfs_list_for_each_entry_safe(entry, next,
 266                                              &sai->sai_entries_stated,
 267                                              se_list) {
 268                         cfs_list_del_init(&entry->se_list);
 269                         ll_sai_entry_cleanup(entry, 1);
 270                 }
 271                 iput(inode);
 272                 OBD_FREE_PTR(sai);
 273         }
 274         EXIT;
 275 }
 276
 277 /**
 278  * insert it into sai_entries_sent tail when init.
 279  */
 280 static struct ll_sai_entry *
 281 ll_sai_entry_init(struct ll_statahead_info *sai, unsigned int index)
 282 {
 283         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 284         struct ll_sai_entry  *entry;
 285         ENTRY;
 286
 287         OBD_ALLOC_PTR(entry);
 288         if (entry == NULL)
 289                 RETURN(ERR_PTR(-ENOMEM));
 290
 291         CDEBUG(D_READA, "alloc sai entry %p index %u\n",
 292                entry, index);
 293         entry->se_index = index;
 294         entry->se_stat = SA_ENTRY_UNSTATED;
 295
 296         cfs_spin_lock(&lli->lli_sa_lock);
 297         cfs_list_add_tail(&entry->se_list, &sai->sai_entries_sent);
 298         cfs_spin_unlock(&lli->lli_sa_lock);
 299
 300         RETURN(entry);
 301 }
 302
 303 /**
 304  * delete it from sai_entries_stated head when fini, it need not
 305  * to process entry's member.
 306  */
 307 static int ll_sai_entry_fini(struct ll_statahead_info *sai)
 308 {
 309         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 310         struct ll_sai_entry  *entry;
 311         int rc = 0;
 312         ENTRY;
 313
 314         cfs_spin_lock(&lli->lli_sa_lock);
 315         sai->sai_index_next++;
 316         if (likely(!cfs_list_empty(&sai->sai_entries_stated))) {
 317                 entry = cfs_list_entry(sai->sai_entries_stated.next,
 318                                        struct ll_sai_entry, se_list);
 319                 if (entry->se_index < sai->sai_index_next) {
 320                         cfs_list_del_init(&entry->se_list);
 321                         rc = entry->se_stat;
 322                         ll_sai_entry_free(entry);
 323                 }
 324         } else {
 325                 LASSERT(sa_is_stopped(sai));
 326         }
 327         cfs_spin_unlock(&lli->lli_sa_lock);
 328
 329         RETURN(rc);
 330 }
 331
 332 /**
 333  * inside lli_sa_lock.
 334  * \retval NULL : can not find the entry in sai_entries_sent with the index
 335  * \retval entry: find the entry in sai_entries_sent with the index
 336  */
 337 static struct ll_sai_entry *
 338 ll_sai_entry_set(struct ll_statahead_info *sai, unsigned int index, int stat,
 339                  struct ptlrpc_request *req, struct md_enqueue_info *minfo)
 340 {
 341         struct ll_sai_entry *entry;
 342         ENTRY;
 343
 344         if (!cfs_list_empty(&sai->sai_entries_sent)) {
 345                 cfs_list_for_each_entry(entry, &sai->sai_entries_sent,
 346                                         se_list) {
 347                         if (entry->se_index == index) {
 348                                 entry->se_stat = stat;
 349                                 entry->se_req = ptlrpc_request_addref(req);
 350                                 entry->se_minfo = minfo;
 351                                 RETURN(entry);
 352                         } else if (entry->se_index > index) {
 353                                 RETURN(NULL);
 354                         }
 355                 }
 356         }
 357         RETURN(NULL);
 358 }
 359
 360 /**
 361  * inside lli_sa_lock.
 362  * Move entry to sai_entries_received and
 363  * insert it into sai_entries_received tail.
 364  */
 365 static inline void
 366 ll_sai_entry_to_received(struct ll_statahead_info *sai, struct ll_sai_entry *entry)
 367 {
 368         if (!cfs_list_empty(&entry->se_list))
 369                 cfs_list_del_init(&entry->se_list);
 370         cfs_list_add_tail(&entry->se_list, &sai->sai_entries_received);
 371 }
 372
 373 /**
 374  * Move entry to sai_entries_stated and
 375  * sort with the index.
 376  */
 377 static int
 378 ll_sai_entry_to_stated(struct ll_statahead_info *sai, struct ll_sai_entry *entry)
 379 {
 380         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 381         struct ll_sai_entry  *se;
 382         ENTRY;
 383
 384         ll_sai_entry_cleanup(entry, 0);
 385
 386         cfs_spin_lock(&lli->lli_sa_lock);
 387         if (!cfs_list_empty(&entry->se_list))
 388                 cfs_list_del_init(&entry->se_list);
 389
 390         /* stale entry */
 391         if (unlikely(entry->se_index < sai->sai_index_next)) {
 392                 cfs_spin_unlock(&lli->lli_sa_lock);
 393                 ll_sai_entry_free(entry);
 394                 RETURN(0);
 395         }
 396
 397         cfs_list_for_each_entry_reverse(se, &sai->sai_entries_stated, se_list) {
 398                 if (se->se_index < entry->se_index) {
 399                         cfs_list_add(&entry->se_list, &se->se_list);
 400                         cfs_spin_unlock(&lli->lli_sa_lock);
 401                         RETURN(1);
 402                 }
 403         }
 404
 405         /*
 406          * I am the first entry.
 407          */
 408         cfs_list_add(&entry->se_list, &sai->sai_entries_stated);
 409         cfs_spin_unlock(&lli->lli_sa_lock);
 410         RETURN(1);
 411 }
 412
 413 /**
 414  * finish lookup/revalidate.
 415  */
 416 static int do_statahead_interpret(struct ll_statahead_info *sai)
 417 {
 418         struct ll_inode_info   *lli = ll_i2info(sai->sai_inode);
 419         struct ll_sai_entry    *entry;
 420         struct ptlrpc_request  *req;
 421         struct md_enqueue_info *minfo;
 422         struct lookup_intent   *it;
 423         struct dentry          *dentry;
 424         int                     rc = 0;
 425         struct mdt_body        *body;
 426         ENTRY;
 427
 428         cfs_spin_lock(&lli->lli_sa_lock);
 429         LASSERT(!sa_received_empty(sai));
 430         entry = cfs_list_entry(sai->sai_entries_received.next,
 431                                struct ll_sai_entry, se_list);
 432         cfs_list_del_init(&entry->se_list);
 433         cfs_spin_unlock(&lli->lli_sa_lock);
 434
 435         if (unlikely(entry->se_index < sai->sai_index_next)) {
 436                 CWARN("Found stale entry: [index %u] [next %u]\n",
 437                       entry->se_index, sai->sai_index_next);
 438                 ll_sai_entry_cleanup(entry, 1);
 439                 RETURN(0);
 440         }
 441
 442         if (entry->se_stat != SA_ENTRY_STATED)
 443                 GOTO(out, rc = entry->se_stat);
 444
 445         req = entry->se_req;
 446         minfo = entry->se_minfo;
 447         it = &minfo->mi_it;
 448         dentry = minfo->mi_dentry;
 449
 450         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 451         if (body == NULL)
 452                 GOTO(out, rc = -EFAULT);
 453
 454         if (dentry->d_inode == NULL) {
 455                 /*
 456                  * lookup.
 457                  */
 458                 struct dentry    *save = dentry;
 459                 __u32 bits             = 0;
 460                 struct it_cb_data icbd = {
 461                         .icbd_parent   = minfo->mi_dir,
 462                         .icbd_childp   = &dentry,
 463                         .icbd_alias    = &entry->se_inode,
 464                         .bits          = &bits
 465                 };
 466
 467                 LASSERT(fid_is_zero(&minfo->mi_data.op_fid2));
 468
 469                 /* XXX: No fid in reply, this is probaly cross-ref case.
 470                  * SA can't handle it yet. */
 471                 if (body->valid & OBD_MD_MDS)
 472                         GOTO(out, rc = -EAGAIN);
 473
 474                 /* Here dentry->d_inode might be NULL, because the entry may
 475                  * have been removed before we start doing stat ahead. */
 476
 477                 /* BUG 15962, 21739: since statahead thread does not hold
 478                  * parent's i_mutex, it can not alias the dentry to inode.
 479                  * Here we just create/update inode in memory, and let the
 480                  * main "ls -l" thread to alias such dentry to the inode with
 481                  * parent's i_mutex held.
 482                  * On the other hand, we hold ldlm ibits lock for the inode
 483                  * yet, to allow other operations to cancel such lock in time,
 484                  * we should drop the ldlm lock reference count, then the main
 485                  * "ls -l" thread should check/get such ldlm ibits lock before
 486                  * aliasing such dentry to the inode later. If we don't do such
 487                  * drop here, it maybe cause deadlock with i_muext held by
 488                  * others, just like bug 21739. */
 489                 rc = ll_lookup_it_finish(req, it, &icbd);
 490                 if (entry->se_inode != NULL) {
 491                         struct ll_dentry_data *lld = ll_d2d(dentry);
 492                         struct ll_inode_info *sei = ll_i2info(entry->se_inode);
 493
 494                         /* For statahead lookup case, both MDS_INODELOCK_LOOKUP
 495                          * and MDS_INODELOCK_UPDATE should be granted */
 496                         if (likely(bits & MDS_INODELOCK_LOOKUP &&
 497                                    bits & MDS_INODELOCK_UPDATE)) {
 498                                 /* the first dentry ref_count will be dropped by
 499                                  * ll_sai_entry_to_stated(), so hold another ref
 500                                  * in advance */
 501                                 entry->se_dentry = dget(dentry);
 502                                 cfs_spin_lock(&sei->lli_sa_lock);
 503                                 cfs_list_add(&lld->lld_sa_alias,
 504                                              &sei->lli_sa_dentry);
 505                                 cfs_spin_unlock(&sei->lli_sa_lock);
 506                                 sai->sai_nolock = 0;
 507                         } else {
 508                                 iput(entry->se_inode);
 509                                 entry->se_inode = NULL;
 510                                 sai->sai_nolock++;
 511                         }
 512                 }
 513                 LASSERT(dentry == save);
 514                 ll_intent_drop_lock(it);
 515         } else {
 516                 /*
 517                  * revalidate.
 518                  */
 519                 if (!lu_fid_eq(&minfo->mi_data.op_fid2, &body->fid1)) {
 520                         ll_unhash_aliases(dentry->d_inode);
 521                         GOTO(out, rc = -EAGAIN);
 522                 }
 523
 524                 rc = ll_revalidate_it_finish(req, it, dentry);
 525                 if (rc) {
 526                         ll_unhash_aliases(dentry->d_inode);
 527                         GOTO(out, rc);
 528                 }
 529
 530                 cfs_spin_lock(&ll_lookup_lock);
 531                 spin_lock(&dcache_lock);
 532                 lock_dentry(dentry);
 533                 __d_drop(dentry);
 534                 dentry->d_flags &= ~DCACHE_LUSTRE_INVALID;
 535                 unlock_dentry(dentry);
 536                 d_rehash_cond(dentry, 0);
 537                 spin_unlock(&dcache_lock);
 538                 cfs_spin_unlock(&ll_lookup_lock);
 539
 540                 ll_lookup_finish_locks(it, dentry);
 541         }
 542         EXIT;
 543
 544 out:
 545         /* The "ll_sai_entry_to_stated()" will drop related ldlm ibits lock
 546          * reference count with ll_intent_drop_lock() called in spite of the
 547          * above operations failed or not. Do not worry about calling
 548          * "ll_intent_drop_lock()" more than once. */
 549         if (likely(ll_sai_entry_to_stated(sai, entry)))
 550                 cfs_waitq_signal(&sai->sai_waitq);
 551         return rc;
 552 }
 553
 554 static int ll_statahead_interpret(struct ptlrpc_request *req,
 555                                   struct md_enqueue_info *minfo,
 556                                   int rc)
 557 {
 558         struct lookup_intent     *it = &minfo->mi_it;
 559         struct dentry            *dentry = minfo->mi_dentry;
 560         struct inode             *dir = minfo->mi_dir;
 561         struct ll_inode_info     *lli = ll_i2info(dir);
 562         struct ll_statahead_info *sai;
 563         struct ll_sai_entry      *entry;
 564         ENTRY;
 565
 566         CDEBUG(D_READA, "interpret statahead %.*s rc %d\n",
 567                dentry->d_name.len, dentry->d_name.name, rc);
 568
 569         cfs_spin_lock(&lli->lli_sa_lock);
 570         /* stale entry */
 571         if (unlikely(lli->lli_sai == NULL ||
 572             lli->lli_sai->sai_generation != minfo->mi_generation)) {
 573                 cfs_spin_unlock(&lli->lli_sa_lock);
 574                 ll_intent_release(it);
 575                 dput(dentry);
 576                 iput(dir);
 577                 OBD_FREE_PTR(minfo);
 578                 RETURN(-ESTALE);
 579         } else {
 580                 sai = ll_sai_get(lli->lli_sai);
 581                 entry = ll_sai_entry_set(sai,
 582                                          (unsigned int)(long)minfo->mi_cbdata,
 583                                          rc < 0 ? rc : SA_ENTRY_STATED, req,
 584                                          minfo);
 585                 LASSERT(entry != NULL);
 586                 if (likely(sa_is_running(sai))) {
 587                         ll_sai_entry_to_received(sai, entry);
 588                         sai->sai_replied++;
 589                         cfs_spin_unlock(&lli->lli_sa_lock);
 590                         cfs_waitq_signal(&sai->sai_thread.t_ctl_waitq);
 591                 } else {
 592                         if (!cfs_list_empty(&entry->se_list))
 593                                 cfs_list_del_init(&entry->se_list);
 594                         sai->sai_replied++;
 595                         cfs_spin_unlock(&lli->lli_sa_lock);
 596                         ll_sai_entry_cleanup(entry, 1);
 597                 }
 598                 ll_sai_put(sai);
 599                 RETURN(rc);
 600         }
 601 }
 602
 603 static void sa_args_fini(struct md_enqueue_info *minfo,
 604                          struct ldlm_enqueue_info *einfo)
 605 {
 606         LASSERT(minfo && einfo);
 607         iput(minfo->mi_dir);
 608         capa_put(minfo->mi_data.op_capa1);
 609         capa_put(minfo->mi_data.op_capa2);
 610         OBD_FREE_PTR(minfo);
 611         OBD_FREE_PTR(einfo);
 612 }
 613
 614 /**
 615  * There is race condition between "capa_put" and "ll_statahead_interpret" for
 616  * accessing "op_data.op_capa[1,2]" as following:
 617  * "capa_put" releases "op_data.op_capa[1,2]"'s reference count after calling
 618  * "md_intent_getattr_async". But "ll_statahead_interpret" maybe run first, and
 619  * fill "op_data.op_capa[1,2]" as POISON, then cause "capa_put" access invalid
 620  * "ocapa". So here reserve "op_data.op_capa[1,2]" in "pcapa" before calling
 621  * "md_intent_getattr_async".
 622  */
 623 static int sa_args_init(struct inode *dir, struct dentry *dentry,
 624                         struct md_enqueue_info **pmi,
 625                         struct ldlm_enqueue_info **pei,
 626                         struct obd_capa **pcapa)
 627 {
 628         struct ll_inode_info     *lli = ll_i2info(dir);
 629         struct md_enqueue_info   *minfo;
 630         struct ldlm_enqueue_info *einfo;
 631         struct md_op_data        *op_data;
 632
 633         OBD_ALLOC_PTR(einfo);
 634         if (einfo == NULL)
 635                 return -ENOMEM;
 636
 637         OBD_ALLOC_PTR(minfo);
 638         if (minfo == NULL) {
 639                 OBD_FREE_PTR(einfo);
 640                 return -ENOMEM;
 641         }
 642
 643         op_data = ll_prep_md_op_data(&minfo->mi_data, dir, dentry->d_inode,
 644                                      dentry->d_name.name, dentry->d_name.len,
 645                                      0, LUSTRE_OPC_ANY, NULL);
 646         if (IS_ERR(op_data)) {
 647                 OBD_FREE_PTR(einfo);
 648                 OBD_FREE_PTR(minfo);
 649                 return PTR_ERR(op_data);
 650         }
 651
 652         minfo->mi_it.it_op = IT_GETATTR;
 653         minfo->mi_dentry = dentry;
 654         minfo->mi_dir = igrab(dir);
 655         minfo->mi_cb = ll_statahead_interpret;
 656         minfo->mi_generation = lli->lli_sai->sai_generation;
 657         minfo->mi_cbdata = (void *)(long)lli->lli_sai->sai_index;
 658
 659         einfo->ei_type   = LDLM_IBITS;
 660         einfo->ei_mode   = it_to_lock_mode(&minfo->mi_it);
 661         einfo->ei_cb_bl  = ll_md_blocking_ast;
 662         einfo->ei_cb_cp  = ldlm_completion_ast;
 663         einfo->ei_cb_gl  = NULL;
 664         einfo->ei_cbdata = NULL;
 665
 666         *pmi = minfo;
 667         *pei = einfo;
 668         pcapa[0] = op_data->op_capa1;
 669         pcapa[1] = op_data->op_capa2;
 670
 671         return 0;
 672 }
 673
 674 /**
 675  * similar to ll_lookup_it().
 676  */
 677 static int do_sa_lookup(struct inode *dir, struct dentry *dentry)
 678 {
 679         struct md_enqueue_info   *minfo;
 680         struct ldlm_enqueue_info *einfo;
 681         struct obd_capa          *capas[2];
 682         int                       rc;
 683         ENTRY;
 684
 685         rc = sa_args_init(dir, dentry, &minfo, &einfo, capas);
 686         if (rc)
 687                 RETURN(rc);
 688
 689         rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo);
 690         if (!rc) {
 691                 capa_put(capas[0]);
 692                 capa_put(capas[1]);
 693         } else {
 694                 sa_args_fini(minfo, einfo);
 695         }
 696
 697         RETURN(rc);
 698 }
 699
 700 /**
 701  * similar to ll_revalidate_it().
 702  * \retval      1 -- dentry valid
 703  * \retval      0 -- will send stat-ahead request
 704  * \retval others -- prepare stat-ahead request failed
 705  */
 706 static int do_sa_revalidate(struct inode *dir, struct dentry *dentry)
 707 {
 708         struct inode             *inode = dentry->d_inode;
 709         struct lookup_intent      it = { .it_op = IT_GETATTR };
 710         struct md_enqueue_info   *minfo;
 711         struct ldlm_enqueue_info *einfo;
 712         struct obd_capa          *capas[2];
 713         int rc;
 714         ENTRY;
 715
 716         if (unlikely(inode == NULL))
 717                 RETURN(1);
 718
 719         if (d_mountpoint(dentry))
 720                 RETURN(1);
 721
 722         if (unlikely(dentry == dentry->d_sb->s_root))
 723                 RETURN(1);
 724
 725         rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode),
 726                                 NULL);
 727         if (rc == 1) {
 728                 ll_intent_release(&it);
 729                 RETURN(1);
 730         }
 731
 732         rc = sa_args_init(dir, dentry, &minfo, &einfo, capas);
 733         if (rc)
 734                 RETURN(rc);
 735
 736         rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo);
 737         if (!rc) {
 738                 capa_put(capas[0]);
 739                 capa_put(capas[1]);
 740         } else {
 741                 sa_args_fini(minfo, einfo);
 742         }
 743
 744         RETURN(rc);
 745 }
 746
 747 static inline void ll_name2qstr(struct qstr *q, const char *name, int namelen)
 748 {
 749         q->name = name;
 750         q->len  = namelen;
 751         q->hash = full_name_hash(name, namelen);
 752 }
 753
 754 static int ll_statahead_one(struct dentry *parent, const char* entry_name,
 755                             int entry_name_len)
 756 {
 757         struct inode             *dir = parent->d_inode;
 758         struct ll_inode_info     *lli = ll_i2info(dir);
 759         struct ll_statahead_info *sai = lli->lli_sai;
 760         struct qstr               name;
 761         struct dentry            *dentry;
 762         struct ll_sai_entry      *se;
 763         int                       rc;
 764         ENTRY;
 765
 766         if (parent->d_flags & DCACHE_LUSTRE_INVALID) {
 767                 CDEBUG(D_READA, "parent dentry@%p %.*s is "
 768                        "invalid, skip statahead\n",
 769                        parent, parent->d_name.len, parent->d_name.name);
 770                 RETURN(-EINVAL);
 771         }
 772
 773         se = ll_sai_entry_init(sai, sai->sai_index);
 774         if (IS_ERR(se))
 775                 RETURN(PTR_ERR(se));
 776
 777         ll_name2qstr(&name, entry_name, entry_name_len);
 778         dentry = d_lookup(parent, &name);
 779         if (!dentry) {
 780                 if (unlikely(sa_skip_nolock(sai))) {
 781                         CWARN("can not obtain lookup lock, skip the succeedent "
 782                               "lookup cases, will cause statahead miss, and "
 783                               "statahead maybe exit for that.\n");
 784                         GOTO(out, rc = -EAGAIN);
 785                 }
 786
 787                 dentry = d_alloc(parent, &name);
 788                 if (dentry) {
 789                         rc = ll_dops_init(dentry, 1);
 790                         if (!rc)
 791                                 rc = do_sa_lookup(dir, dentry);
 792                         if (rc)
 793                                 dput(dentry);
 794                 } else {
 795                         GOTO(out, rc = -ENOMEM);
 796                 }
 797         } else {
 798                 rc = do_sa_revalidate(dir, dentry);
 799                 if (rc)
 800                         dput(dentry);
 801         }
 802
 803         EXIT;
 804
 805 out:
 806         if (rc) {
 807                 CDEBUG(D_READA, "set sai entry %p index %u stat %d rc %d\n",
 808                        se, se->se_index, se->se_stat, rc);
 809                 se->se_stat = rc < 0 ? rc : SA_ENTRY_STATED;
 810                 if (ll_sai_entry_to_stated(sai, se))
 811                         cfs_waitq_signal(&sai->sai_waitq);
 812         } else {
 813                 sai->sai_sent++;
 814         }
 815
 816         sai->sai_index++;
 817         return rc;
 818 }
 819
 820 static int ll_statahead_thread(void *arg)
 821 {
 822         struct dentry            *parent = (struct dentry *)arg;
 823         struct inode             *dir = parent->d_inode;
 824         struct ll_inode_info     *lli = ll_i2info(dir);
 825         struct ll_sb_info        *sbi = ll_i2sbi(dir);
 826         struct ll_statahead_info *sai = ll_sai_get(lli->lli_sai);
 827         struct ptlrpc_thread     *thread = &sai->sai_thread;
 828         struct page              *page;
 829         __u64                     pos = 0;
 830         int                       first = 0;
 831         int                       rc = 0;
 832         struct ll_dir_chain       chain;
 833         ENTRY;
 834
 835         {
 836                 char pname[16];
 837                 snprintf(pname, 15, "ll_sa_%u", lli->lli_opendir_pid);
 838                 cfs_daemonize(pname);
 839         }
 840
 841         atomic_inc(&sbi->ll_sa_total);
 842         cfs_spin_lock(&lli->lli_sa_lock);
 843         thread->t_flags = SVC_RUNNING;
 844         cfs_spin_unlock(&lli->lli_sa_lock);
 845         cfs_waitq_signal(&thread->t_ctl_waitq);
 846         CDEBUG(D_READA, "start doing statahead for %s\n", parent->d_name.name);
 847
 848         ll_dir_chain_init(&chain);
 849         page = ll_get_dir_page(dir, pos, 0, &chain);
 850
 851         while (1) {
 852                 struct l_wait_info lwi = { 0 };
 853                 struct lu_dirpage *dp;
 854                 struct lu_dirent  *ent;
 855
 856                 if (IS_ERR(page)) {
 857                         rc = PTR_ERR(page);
 858                         CDEBUG(D_READA, "error reading dir "DFID" at "LPU64
 859                                "/%u: [rc %d] [parent %u]\n",
 860                                PFID(ll_inode2fid(dir)), pos, sai->sai_index,
 861                                rc, lli->lli_opendir_pid);
 862                         break;
 863                 }
 864
 865                 dp = page_address(page);
 866                 for (ent = lu_dirent_start(dp); ent != NULL;
 867                      ent = lu_dirent_next(ent)) {
 868                         char *name = ent->lde_name;
 869                         int namelen = le16_to_cpu(ent->lde_namelen);
 870
 871                         if (unlikely(namelen == 0))
 872                                 /*
 873                                  * Skip dummy record.
 874                                  */
 875                                 continue;
 876
 877                         if (name[0] == '.') {
 878                                 if (namelen == 1) {
 879                                         /*
 880                                          * skip "."
 881                                          */
 882                                         continue;
 883                                 } else if (name[1] == '.' && namelen == 2) {
 884                                         /*
 885                                          * skip ".."
 886                                          */
 887                                         continue;
 888                                 } else if (!sai->sai_ls_all) {
 889                                         /*
 890                                          * skip hidden files.
 891                                          */
 892                                         sai->sai_skip_hidden++;
 893                                         continue;
 894                                 }
 895                         }
 896
 897                         /*
 898                          * don't stat-ahead first entry.
 899                          */
 900                         if (unlikely(!first)) {
 901                                 first++;
 902                                 continue;
 903                         }
 904
 905 keep_de:
 906                         l_wait_event(thread->t_ctl_waitq,
 907                                      !sa_is_running(sai) || sa_not_full(sai) ||
 908                                      !sa_received_empty(sai),
 909                                      &lwi);
 910
 911                         while (!sa_received_empty(sai) && sa_is_running(sai))
 912                                 do_statahead_interpret(sai);
 913
 914                         if (unlikely(!sa_is_running(sai))) {
 915                                 ll_put_page(page);
 916                                 GOTO(out, rc);
 917                         }
 918
 919                         if (!sa_not_full(sai))
 920                                 /*
 921                                  * do not skip the current de.
 922                                  */
 923                                 goto keep_de;
 924
 925                         rc = ll_statahead_one(parent, name, namelen);
 926                         if (rc < 0) {
 927                                 ll_put_page(page);
 928                                 GOTO(out, rc);
 929                         }
 930                 }
 931                 pos = le64_to_cpu(dp->ldp_hash_end);
 932                 ll_put_page(page);
 933                 if (pos == DIR_END_OFF) {
 934                         /*
 935                          * End of directory reached.
 936                          */
 937                         while (1) {
 938                                 l_wait_event(thread->t_ctl_waitq,
 939                                              !sa_is_running(sai) ||
 940                                              !sa_received_empty(sai) ||
 941                                              sai->sai_sent == sai->sai_replied,
 942                                              &lwi);
 943                                 if (!sa_received_empty(sai) &&
 944                                     sa_is_running(sai))
 945                                         do_statahead_interpret(sai);
 946                                 else
 947                                         GOTO(out, rc);
 948                         }
 949                 } else if (1) {
 950                         /*
 951                          * chain is exhausted.
 952                          * Normal case: continue to the next page.
 953                          */
 954                         page = ll_get_dir_page(dir, pos, 1, &chain);
 955                 } else {
 956                         /*
 957                          * go into overflow page.
 958                          */
 959                 }
 960         }
 961         EXIT;
 962
 963 out:
 964         ll_dir_chain_fini(&chain);
 965         cfs_spin_lock(&lli->lli_sa_lock);
 966         thread->t_flags = SVC_STOPPED;
 967         cfs_spin_unlock(&lli->lli_sa_lock);
 968         cfs_waitq_signal(&sai->sai_waitq);
 969         cfs_waitq_signal(&thread->t_ctl_waitq);
 970         ll_sai_put(sai);
 971         dput(parent);
 972         CDEBUG(D_READA, "statahead thread stopped, pid %d\n",
 973                cfs_curproc_pid());
 974         return rc;
 975 }
 976
 977 /**
 978  * called in ll_file_release().
 979  */
 980 void ll_stop_statahead(struct inode *inode, void *key)
 981 {
 982         struct ll_inode_info *lli = ll_i2info(inode);
 983
 984         if (unlikely(key == NULL))
 985                 return;
 986
 987         cfs_spin_lock(&lli->lli_sa_lock);
 988         if (lli->lli_opendir_key != key || lli->lli_opendir_pid == 0) {
 989                 cfs_spin_unlock(&lli->lli_sa_lock);
 990                 return;
 991         }
 992
 993         lli->lli_opendir_key = NULL;
 994
 995         if (lli->lli_sai) {
 996                 struct l_wait_info lwi = { 0 };
 997                 struct ptlrpc_thread *thread = &lli->lli_sai->sai_thread;
 998
 999                 if (!sa_is_stopped(lli->lli_sai)) {
1000                         thread->t_flags = SVC_STOPPING;
1001                         cfs_spin_unlock(&lli->lli_sa_lock);
1002                         cfs_waitq_signal(&thread->t_ctl_waitq);
1003
1004                         CDEBUG(D_READA, "stopping statahead thread, pid %d\n",
1005                                cfs_curproc_pid());
1006                         l_wait_event(thread->t_ctl_waitq,
1007                                      sa_is_stopped(lli->lli_sai),
1008                                      &lwi);
1009                 } else {
1010                         cfs_spin_unlock(&lli->lli_sa_lock);
1011                 }
1012
1013                 /*
1014                  * Put the ref which was held when first statahead_enter.
1015                  * It maybe not the last ref for some statahead requests
1016                  * maybe inflight.
1017                  */
1018                 ll_sai_put(lli->lli_sai);
1019         } else {
1020                 lli->lli_opendir_pid = 0;
1021                 cfs_spin_unlock(&lli->lli_sa_lock);
1022         }
1023 }
1024
1025 enum {
1026         /**
1027          * not first dirent, or is "."
1028          */
1029         LS_NONE_FIRST_DE = 0,
1030         /**
1031          * the first non-hidden dirent
1032          */
1033         LS_FIRST_DE,
1034         /**
1035          * the first hidden dirent, that is "."
1036          */
1037         LS_FIRST_DOT_DE
1038 };
1039
1040 static int is_first_dirent(struct inode *dir, struct dentry *dentry)
1041 {
1042         struct ll_dir_chain chain;
1043         struct qstr        *target = &dentry->d_name;
1044         struct page        *page;
1045         __u64               pos = 0;
1046         int                 dot_de;
1047         int                 rc = LS_NONE_FIRST_DE;
1048         ENTRY;
1049
1050         ll_dir_chain_init(&chain);
1051         page = ll_get_dir_page(dir, pos, 0, &chain);
1052
1053         while (1) {
1054                 struct lu_dirpage *dp;
1055                 struct lu_dirent  *ent;
1056
1057                 if (IS_ERR(page)) {
1058                         struct ll_inode_info *lli = ll_i2info(dir);
1059
1060                         rc = PTR_ERR(page);
1061                         CERROR("error reading dir "DFID" at "LPU64": "
1062                                "[rc %d] [parent %u]\n",
1063                                PFID(ll_inode2fid(dir)), pos,
1064                                rc, lli->lli_opendir_pid);
1065                         break;
1066                 }
1067
1068                 dp = page_address(page);
1069                 for (ent = lu_dirent_start(dp); ent != NULL;
1070                      ent = lu_dirent_next(ent)) {
1071                         char *name = ent->lde_name;
1072                         int namelen = le16_to_cpu(ent->lde_namelen);
1073
1074                         if (namelen == 0)
1075                                 /*
1076                                  * skip dummy record.
1077                                  */
1078                                 continue;
1079
1080                         if (name[0] == '.') {
1081                                 if (namelen == 1)
1082                                         /*
1083                                          * skip "."
1084                                          */
1085                                         continue;
1086                                 else if (name[1] == '.' && namelen == 2)
1087                                         /*
1088                                          * skip ".."
1089                                          */
1090                                         continue;
1091                                 else
1092                                         dot_de = 1;
1093                         } else {
1094                                 dot_de = 0;
1095                         }
1096
1097                         if (dot_de && target->name[0] != '.') {
1098                                 CDEBUG(D_READA, "%.*s skip hidden file %.*s\n",
1099                                        target->len, target->name,
1100                                        namelen, name);
1101                                 continue;
1102                         }
1103
1104                         if (target->len != namelen ||
1105                             memcmp(target->name, name, namelen) != 0)
1106                                 rc = LS_NONE_FIRST_DE;
1107                         else if (!dot_de)
1108                                 rc = LS_FIRST_DE;
1109                         else
1110                                 rc = LS_FIRST_DOT_DE;
1111
1112                         ll_put_page(page);
1113                         GOTO(out, rc);
1114                 }
1115                 pos = le64_to_cpu(dp->ldp_hash_end);
1116                 ll_put_page(page);
1117                 if (pos == DIR_END_OFF) {
1118                         /*
1119                          * End of directory reached.
1120                          */
1121                         break;
1122                 } else if (1) {
1123                         /*
1124                          * chain is exhausted
1125                          * Normal case: continue to the next page.
1126                          */
1127                         page = ll_get_dir_page(dir, pos, 1, &chain);
1128                 } else {
1129                         /*
1130                          * go into overflow page.
1131                          */
1132                 }
1133         }
1134         EXIT;
1135
1136 out:
1137         ll_dir_chain_fini(&chain);
1138         return rc;
1139 }
1140
1141 /*
1142  * tgt: the dentry to be revalidate or lookup
1143  * new: the dentry created by statahead
1144  */
1145 static int is_same_dentry(struct dentry *tgt, struct dentry *new, int lookup)
1146 {
1147         if (tgt == new) {
1148                 LASSERT(lookup == 0);
1149                 return 1;
1150         }
1151         if (tgt->d_parent != new->d_parent)
1152                 return 0;
1153         if (tgt->d_name.hash != new->d_name.hash)
1154                 return 0;
1155         if (tgt->d_name.len != new->d_name.len)
1156                 return 0;
1157         if (memcmp(tgt->d_name.name, new->d_name.name, tgt->d_name.len) != 0)
1158                 return 0;
1159         if (tgt->d_inode == NULL && lookup)
1160                 return 1;
1161         if (tgt->d_inode)
1162                 LASSERTF(tgt->d_flags & DCACHE_LUSTRE_INVALID,
1163                          "[%.*s/%.*s] [%x %p "DFID"] [%x %p "DFID"]\n",
1164                          tgt->d_parent->d_name.len, tgt->d_parent->d_name.name,
1165                          tgt->d_name.len, tgt->d_name.name,
1166                          tgt->d_flags, tgt, PFID(ll_inode2fid(tgt->d_inode)),
1167                          new->d_flags, new, PFID(ll_inode2fid(new->d_inode)));
1168         else
1169                 LASSERTF(tgt->d_flags & DCACHE_LUSTRE_INVALID,
1170                          "[%.*s/%.*s] [%x %p 0] [%x %p "DFID"]\n",
1171                          tgt->d_parent->d_name.len, tgt->d_parent->d_name.name,
1172                          tgt->d_name.len, tgt->d_name.name,
1173                          tgt->d_flags, tgt,
1174                          new->d_flags, new, PFID(ll_inode2fid(new->d_inode)));
1175         return 0;
1176 }
1177
1178 /**
1179  * Start statahead thread if this is the first dir entry.
1180  * Otherwise if a thread is started already, wait it until it is ahead of me.
1181  * \retval 0       -- stat ahead thread process such dentry, miss for lookup
1182  * \retval 1       -- stat ahead thread process such dentry, hit for any case
1183  * \retval -EEXIST -- stat ahead thread started, and this is the first dentry
1184  * \retval -EBADFD -- statahead thread exit and not dentry available
1185  * \retval -EAGAIN -- try to stat by caller
1186  * \retval others  -- error
1187  */
1188 int do_statahead_enter(struct inode *dir, struct dentry **dentryp, int lookup)
1189 {
1190         struct ll_inode_info     *lli;
1191         struct ll_statahead_info *sai;
1192         struct dentry            *parent;
1193         struct l_wait_info        lwi = { 0 };
1194         int                       rc = 0;
1195         ENTRY;
1196
1197         LASSERT(dir != NULL);
1198         lli = ll_i2info(dir);
1199         LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
1200         sai = lli->lli_sai;
1201
1202         if (sai) {
1203                 if (unlikely(sa_is_stopped(sai) &&
1204                              cfs_list_empty(&sai->sai_entries_stated)))
1205                         RETURN(-EBADFD);
1206
1207                 if ((*dentryp)->d_name.name[0] == '.') {
1208                         if (likely(sai->sai_ls_all ||
1209                             sai->sai_miss_hidden >= sai->sai_skip_hidden)) {
1210                                 /*
1211                                  * Hidden dentry is the first one, or statahead
1212                                  * thread does not skip so many hidden dentries
1213                                  * before "sai_ls_all" enabled as below.
1214                                  */
1215                         } else {
1216                                 if (!sai->sai_ls_all)
1217                                         /*
1218                                          * It maybe because hidden dentry is not
1219                                          * the first one, "sai_ls_all" was not
1220                                          * set, then "ls -al" missed. Enable
1221                                          * "sai_ls_all" for such case.
1222                                          */
1223                                         sai->sai_ls_all = 1;
1224
1225                                 /*
1226                                  * Such "getattr" has been skipped before
1227                                  * "sai_ls_all" enabled as above.
1228                                  */
1229                                 sai->sai_miss_hidden++;
1230                                 RETURN(-ENOENT);
1231                         }
1232                 }
1233
1234                 if (!ll_sai_entry_stated(sai)) {
1235                         /*
1236                          * thread started already, avoid double-stat.
1237                          */
1238                         lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
1239                         rc = l_wait_event(sai->sai_waitq,
1240                                           ll_sai_entry_stated(sai) ||
1241                                           sa_is_stopped(sai),
1242                                           &lwi);
1243                         if (unlikely(rc == -EINTR))
1244                                 RETURN(rc);
1245                 }
1246
1247                 if (ll_sai_entry_stated(sai)) {
1248                         struct ll_sai_entry  *entry;
1249
1250                         entry = cfs_list_entry(sai->sai_entries_stated.next,
1251                                                struct ll_sai_entry, se_list);
1252                         /* This is for statahead lookup */
1253                         if (entry->se_inode != NULL) {
1254                                 struct lookup_intent it = {.it_op = IT_LOOKUP};
1255                                 struct dentry *dchild = entry->se_dentry;
1256                                 struct inode *ichild = entry->se_inode;
1257                                 struct ll_dentry_data *lld = ll_d2d(dchild);
1258                                 struct ll_inode_info *sei = ll_i2info(ichild);
1259                                 struct dentry *save = dchild;
1260                                 int invalid = 0;
1261                                 __u32 bits = MDS_INODELOCK_LOOKUP |
1262                                              MDS_INODELOCK_UPDATE;
1263                                 int found = 0;
1264
1265                                 LASSERT(dchild != *dentryp);
1266
1267                                 if (!lookup)
1268                                         mutex_lock(&dir->i_mutex);
1269
1270                                 /*
1271                                  * Make sure dentry is still valid.
1272                                  * For statahead lookup case, we need both
1273                                  * LOOKUP lock and UPDATE lock which obtained
1274                                  * by statahead thread originally.
1275                                  *
1276                                  * Consider following racer case:
1277                                  * 1. statahead thread on client1 get lock with
1278                                  *    both LOOKUK and UPDATE bits for "aaa"
1279                                  * 2. rename thread on client2 cancel such lock
1280                                  *    from client1, then rename "aaa" to "bbb"
1281                                  * 3. ls thread on client1 obtain LOOKUP lock
1282                                  *    for "bbb" again
1283                                  * 4. here the dentry "aaa" created by statahead
1284                                  *    thread should be invalid even related
1285                                  *    LOOKUP lock valid for the same inode
1286                                  */
1287                                 rc = md_revalidate_lock(ll_i2mdexp(dir), &it,
1288                                                         ll_inode2fid(ichild),
1289                                                         &bits);
1290                                 cfs_spin_lock(&sei->lli_sa_lock);
1291                                 if (!cfs_list_empty(&lld->lld_sa_alias))
1292                                         cfs_list_del_init(&lld->lld_sa_alias);
1293                                 else
1294                                         invalid = 1;
1295                                 cfs_spin_unlock(&sei->lli_sa_lock);
1296                                 if (rc != 1)
1297                                         /* Someone has cancelled the original
1298                                          * lock before the real "revalidate"
1299                                          * using it. Drop it. */
1300                                         goto out_mutex;
1301
1302                                 if (invalid) {
1303                                         /* Someone has cancelled the original
1304                                          * lock, and reobtained it, the dentry
1305                                          * maybe invalid anymore, Drop it. */
1306                                         ll_intent_drop_lock(&it);
1307                                         goto out_mutex;
1308                                 }
1309
1310                                 ll_lookup_it_alias(&dchild, ichild, bits);
1311                                 found = is_same_dentry(*dentryp, dchild, lookup);
1312                                 ll_lookup_finish_locks(&it, dchild);
1313                                 if (dchild != save)
1314                                         dput(save);
1315                                 ichild = NULL;
1316
1317 out_mutex:
1318                                 if (!lookup)
1319                                         mutex_unlock(&dir->i_mutex);
1320                                 /* Drop the inode reference count held by
1321                                  * interpreter. */
1322                                 if (ichild != NULL)
1323                                         iput(ichild);
1324
1325                                 entry->se_dentry = NULL;
1326                                 entry->se_inode = NULL;
1327                                 if (found) {
1328                                         if (lookup) {
1329                                                 LASSERT(*dentryp != dchild);
1330                                                 /* VFS will drop the reference
1331                                                  * count for dchild and *dentryp
1332                                                  * by itself. */
1333                                                 *dentryp = dchild;
1334                                         } else {
1335                                                 LASSERT(*dentryp == dchild);
1336                                                 /* Drop the dentry reference
1337                                                  * count held by statahead. */
1338                                                 dput(dchild);
1339                                         }
1340                                         RETURN(1);
1341                                 } else {
1342                                         /* Drop the dentry reference count held
1343                                          * by statahead. */
1344                                         dput(dchild);
1345                                 }
1346                         }
1347                 }
1348
1349                 if (lookup) {
1350                         struct dentry *result;
1351
1352                         result = d_lookup((*dentryp)->d_parent,
1353                                           &(*dentryp)->d_name);
1354                         if (result) {
1355                                 LASSERT(result != *dentryp);
1356                                 /* BUG 16303: do not drop reference count for
1357                                  * "*dentryp", VFS will do that by itself. */
1358                                 *dentryp = result;
1359                                 RETURN(1);
1360                         }
1361                 }
1362                 /*
1363                  * do nothing for revalidate.
1364                  */
1365                 RETURN(0);
1366         }
1367
1368         /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */
1369         rc = is_first_dirent(dir, *dentryp);
1370         if (rc == LS_NONE_FIRST_DE)
1371                 /* It is not "ls -{a}l" operation, no need statahead for it. */
1372                 GOTO(out, rc = -EAGAIN);
1373
1374         sai = ll_sai_alloc();
1375         if (sai == NULL)
1376                 GOTO(out, rc = -ENOMEM);
1377
1378         sai->sai_ls_all = (rc == LS_FIRST_DOT_DE);
1379         sai->sai_inode = igrab(dir);
1380         if (unlikely(sai->sai_inode == NULL)) {
1381                 CWARN("Do not start stat ahead on dying inode "DFID"\n",
1382                       PFID(&lli->lli_fid));
1383                 OBD_FREE_PTR(sai);
1384                 GOTO(out, rc = -ESTALE);
1385         }
1386
1387         /* get parent reference count here, and put it in ll_statahead_thread */
1388         parent = dget((*dentryp)->d_parent);
1389         if (unlikely(sai->sai_inode != parent->d_inode)) {
1390                 struct ll_inode_info *nlli = ll_i2info(parent->d_inode);
1391
1392                 CWARN("Race condition, someone changed %.*s just now: "
1393                       "old parent "DFID", new parent "DFID"\n",
1394                       (*dentryp)->d_name.len, (*dentryp)->d_name.name,
1395                       PFID(&lli->lli_fid), PFID(&nlli->lli_fid));
1396                 dput(parent);
1397                 iput(sai->sai_inode);
1398                 OBD_FREE_PTR(sai);
1399                 RETURN(-EAGAIN);
1400         }
1401
1402         lli->lli_sai = sai;
1403         rc = cfs_kernel_thread(ll_statahead_thread, parent, 0);
1404         if (rc < 0) {
1405                 CERROR("can't start ll_sa thread, rc: %d\n", rc);
1406                 dput(parent);
1407                 lli->lli_opendir_key = NULL;
1408                 sai->sai_thread.t_flags = SVC_STOPPED;
1409                 ll_sai_put(sai);
1410                 LASSERT(lli->lli_sai == NULL);
1411                 RETURN(-EAGAIN);
1412         }
1413
1414         l_wait_event(sai->sai_thread.t_ctl_waitq,
1415                      sa_is_running(sai) || sa_is_stopped(sai),
1416                      &lwi);
1417
1418         /*
1419          * We don't stat-ahead for the first dirent since we are already in
1420          * lookup, and -EEXIST also indicates that this is the first dirent.
1421          */
1422         RETURN(-EEXIST);
1423
1424 out:
1425         cfs_spin_lock(&lli->lli_sa_lock);
1426         lli->lli_opendir_key = NULL;
1427         lli->lli_opendir_pid = 0;
1428         cfs_spin_unlock(&lli->lli_sa_lock);
1429         return rc;
1430 }
1431
1432 /**
1433  * update hit/miss count.
1434  */
1435 void ll_statahead_exit(struct inode *dir, struct dentry *dentry, int result)
1436 {
1437         struct ll_inode_info     *lli;
1438         struct ll_statahead_info *sai;
1439         struct ll_sb_info        *sbi;
1440         struct ll_dentry_data    *ldd = ll_d2d(dentry);
1441         int                       rc;
1442         ENTRY;
1443
1444         LASSERT(dir != NULL);
1445         lli = ll_i2info(dir);
1446         LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
1447         sai = lli->lli_sai;
1448         LASSERT(sai != NULL);
1449         sbi = ll_i2sbi(dir);
1450
1451         rc = ll_sai_entry_fini(sai);
1452         /* rc == -ENOENT means such dentry was removed just between statahead
1453          * readdir and pre-fetched, count it as hit.
1454          *
1455          * result == -ENOENT has two meanings:
1456          * 1. such dentry was removed just between statahead pre-fetched and
1457          *    main process stat such dentry.
1458          * 2. main process stat non-exist dentry.
1459          * We can not distinguish such two cases, just count them as miss. */
1460         if (result >= 1 || unlikely(rc == -ENOENT)) {
1461                 sai->sai_hit++;
1462                 sai->sai_consecutive_miss = 0;
1463                 sai->sai_max = min(2 * sai->sai_max, sbi->ll_sa_max);
1464         } else {
1465                 sai->sai_miss++;
1466                 sai->sai_consecutive_miss++;
1467                 if (sa_low_hit(sai) && sa_is_running(sai)) {
1468                         atomic_inc(&sbi->ll_sa_wrong);
1469                         CDEBUG(D_READA, "Statahead for dir "DFID" hit ratio "
1470                                "too low: hit/miss %u/%u, sent/replied %u/%u, "
1471                                "stopping statahead thread: pid %d\n",
1472                                PFID(&lli->lli_fid), sai->sai_hit,
1473                                sai->sai_miss, sai->sai_sent,
1474                                sai->sai_replied, cfs_curproc_pid());
1475                         cfs_spin_lock(&lli->lli_sa_lock);
1476                         if (!sa_is_stopped(sai))
1477                                 sai->sai_thread.t_flags = SVC_STOPPING;
1478                         cfs_spin_unlock(&lli->lli_sa_lock);
1479                 }
1480         }
1481
1482         if (!sa_is_stopped(sai))
1483                 cfs_waitq_signal(&sai->sai_thread.t_ctl_waitq);
1484         if (likely(ldd != NULL))
1485                 ldd->lld_sa_generation = sai->sai_generation;
1486
1487         EXIT;
1488 }