lustre/llite/statahead.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  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/highmem.h>
  41 #include <linux/pagemap.h>
  42
  43 #define DEBUG_SUBSYSTEM S_LLITE
  44
  45 #include <obd_support.h>
  46 #include <lustre_lite.h>
  47 #include <lustre_dlm.h>
  48 #include "llite_internal.h"
  49
  50 #define SA_OMITTED_ENTRY_MAX 8ULL
  51
  52 typedef enum {
  53         /** negative values are for error cases */
  54         SA_ENTRY_INIT = 0,      /** init entry */
  55         SA_ENTRY_SUCC = 1,      /** stat succeed */
  56         SA_ENTRY_INVA = 2,      /** invalid entry */
  57         SA_ENTRY_DEST = 3,      /** entry to be destroyed */
  58 } se_stat_t;
  59
  60 struct ll_sa_entry {
  61         /* link into sai->sai_entries_{sent,received,stated} */
  62         cfs_list_t              se_list;
  63         /* link into sai hash table locally */
  64         cfs_list_t              se_hash;
  65         /* entry reference count */
  66         cfs_atomic_t            se_refcount;
  67         /* entry index in the sai */
  68         __u64                   se_index;
  69         /* low layer ldlm lock handle */
  70         __u64                   se_handle;
  71         /* entry status */
  72         se_stat_t               se_stat;
  73         /* entry size, contains name */
  74         int                     se_size;
  75         /* pointer to async getattr enqueue info */
  76         struct md_enqueue_info *se_minfo;
  77         /* pointer to the async getattr request */
  78         struct ptlrpc_request  *se_req;
  79         /* pointer to the target inode */
  80         struct inode           *se_inode;
  81         /* entry name */
  82         struct qstr             se_qstr;
  83 };
  84
  85 static unsigned int sai_generation = 0;
  86 static cfs_spinlock_t sai_generation_lock = CFS_SPIN_LOCK_UNLOCKED;
  87
  88 static inline int ll_sa_entry_unlinked(struct ll_sa_entry *entry)
  89 {
  90         return cfs_list_empty(&entry->se_list);
  91 }
  92
  93 static inline int ll_sa_entry_unhashed(struct ll_sa_entry *entry)
  94 {
  95         return cfs_list_empty(&entry->se_hash);
  96 }
  97
  98 /*
  99  * The entry only can be released by the caller, it is necessary to hold lock.
 100  */
 101 static inline int ll_sa_entry_stated(struct ll_sa_entry *entry)
 102 {
 103         smp_rmb();
 104         return (entry->se_stat != SA_ENTRY_INIT);
 105 }
 106
 107 static inline int ll_sa_entry_hash(int val)
 108 {
 109         return val & LL_SA_CACHE_MASK;
 110 }
 111
 112 /*
 113  * Insert entry to hash SA table.
 114  */
 115 static inline void
 116 ll_sa_entry_enhash(struct ll_statahead_info *sai, struct ll_sa_entry *entry)
 117 {
 118         int i = ll_sa_entry_hash(entry->se_qstr.hash);
 119
 120         cfs_spin_lock(&sai->sai_cache_lock[i]);
 121         cfs_list_add_tail(&entry->se_hash, &sai->sai_cache[i]);
 122         cfs_spin_unlock(&sai->sai_cache_lock[i]);
 123 }
 124
 125 /*
 126  * Remove entry from SA table.
 127  */
 128 static inline void
 129 ll_sa_entry_unhash(struct ll_statahead_info *sai, struct ll_sa_entry *entry)
 130 {
 131         int i = ll_sa_entry_hash(entry->se_qstr.hash);
 132
 133         cfs_spin_lock(&sai->sai_cache_lock[i]);
 134         cfs_list_del_init(&entry->se_hash);
 135         cfs_spin_unlock(&sai->sai_cache_lock[i]);
 136 }
 137
 138 static inline int agl_should_run(struct ll_statahead_info *sai,
 139                                  struct inode *inode)
 140 {
 141         if (inode != NULL && S_ISREG(inode->i_mode) &&
 142             ll_i2info(inode)->lli_has_smd && sai->sai_agl_valid)
 143                 return 1;
 144         return 0;
 145 }
 146
 147 static inline struct ll_sa_entry *
 148 sa_first_received_entry(struct ll_statahead_info *sai)
 149 {
 150         return cfs_list_entry(sai->sai_entries_received.next,
 151                               struct ll_sa_entry, se_list);
 152 }
 153
 154 static inline struct ll_inode_info *
 155 agl_first_entry(struct ll_statahead_info *sai)
 156 {
 157         return cfs_list_entry(sai->sai_entries_agl.next,
 158                               struct ll_inode_info, lli_agl_list);
 159 }
 160
 161 static inline int sa_sent_full(struct ll_statahead_info *sai)
 162 {
 163         return cfs_atomic_read(&sai->sai_cache_count) >= sai->sai_max;
 164 }
 165
 166 static inline int sa_received_empty(struct ll_statahead_info *sai)
 167 {
 168         return cfs_list_empty(&sai->sai_entries_received);
 169 }
 170
 171 static inline int agl_list_empty(struct ll_statahead_info *sai)
 172 {
 173         return cfs_list_empty(&sai->sai_entries_agl);
 174 }
 175
 176 /**
 177  * (1) hit ratio less than 80%
 178  * or
 179  * (2) consecutive miss more than 8
 180  * then means low hit.
 181  */
 182 static inline int sa_low_hit(struct ll_statahead_info *sai)
 183 {
 184         return ((sai->sai_hit > 7 && sai->sai_hit < 4 * sai->sai_miss) ||
 185                 (sai->sai_consecutive_miss > 8));
 186 }
 187
 188 /*
 189  * If the given index is behind of statahead window more than
 190  * SA_OMITTED_ENTRY_MAX, then it is old.
 191  */
 192 static inline int is_omitted_entry(struct ll_statahead_info *sai, __u64 index)
 193 {
 194         return ((__u64)sai->sai_max + index + SA_OMITTED_ENTRY_MAX <
 195                  sai->sai_index);
 196 }
 197
 198 /*
 199  * Insert it into sai_entries_sent tail when init.
 200  */
 201 static struct ll_sa_entry *
 202 ll_sa_entry_alloc(struct ll_statahead_info *sai, __u64 index,
 203                   const char *name, int len)
 204 {
 205         struct ll_inode_info *lli;
 206         struct ll_sa_entry   *entry;
 207         int                   entry_size;
 208         char                 *dname;
 209         ENTRY;
 210
 211         entry_size = sizeof(struct ll_sa_entry) + (len & ~3) + 4;
 212         OBD_ALLOC(entry, entry_size);
 213         if (unlikely(entry == NULL))
 214                 RETURN(ERR_PTR(-ENOMEM));
 215
 216         CDEBUG(D_READA, "alloc sai entry %.*s(%p) index "LPU64"\n",
 217                len, name, entry, index);
 218
 219         entry->se_index = index;
 220
 221         /*
 222          * Statahead entry reference rules:
 223          *
 224          * 1) When statahead entry is initialized, its reference is set as 2.
 225          *    One reference is used by the directory scanner. When the scanner
 226          *    searches the statahead cache for the given name, it can perform
 227          *    lockless hash lookup (only the scanner can remove entry from hash
 228          *    list), and once found, it needn't to call "atomic_inc()" for the
 229          *    entry reference. So the performance is improved. After using the
 230          *    statahead entry, the scanner will call "atomic_dec()" to drop the
 231          *    reference held when initialization. If it is the last reference,
 232          *    the statahead entry will be freed.
 233          *
 234          * 2) All other threads, including statahead thread and ptlrpcd thread,
 235          *    when they process the statahead entry, the reference for target
 236          *    should be held to guarantee the entry will not be released by the
 237          *    directory scanner. After processing the entry, these threads will
 238          *    drop the entry reference. If it is the last reference, the entry
 239          *    will be freed.
 240          *
 241          *    The second reference when initializes the statahead entry is used
 242          *    by the statahead thread, following the rule 2).
 243          */
 244         cfs_atomic_set(&entry->se_refcount, 2);
 245         entry->se_stat = SA_ENTRY_INIT;
 246         entry->se_size = entry_size;
 247         dname = (char *)entry + sizeof(struct ll_sa_entry);
 248         memcpy(dname, name, len);
 249         dname[len] = 0;
 250         entry->se_qstr.hash = full_name_hash(name, len);
 251         entry->se_qstr.len = len;
 252         entry->se_qstr.name = dname;
 253
 254         lli = ll_i2info(sai->sai_inode);
 255         cfs_spin_lock(&lli->lli_sa_lock);
 256         cfs_list_add_tail(&entry->se_list, &sai->sai_entries_sent);
 257         cfs_spin_unlock(&lli->lli_sa_lock);
 258
 259         cfs_atomic_inc(&sai->sai_cache_count);
 260         ll_sa_entry_enhash(sai, entry);
 261
 262         RETURN(entry);
 263 }
 264
 265 /*
 266  * Used by the directory scanner to search entry with name.
 267  *
 268  * Only the caller can remove the entry from hash, so it is unnecessary to hold
 269  * hash lock. It is caller's duty to release the init refcount on the entry, so
 270  * it is also unnecessary to increase refcount on the entry.
 271  */
 272 static struct ll_sa_entry *
 273 ll_sa_entry_get_byname(struct ll_statahead_info *sai, const struct qstr *qstr)
 274 {
 275         struct ll_sa_entry *entry;
 276         int i = ll_sa_entry_hash(qstr->hash);
 277
 278         cfs_list_for_each_entry(entry, &sai->sai_cache[i], se_hash) {
 279                 if (entry->se_qstr.hash == qstr->hash &&
 280                     entry->se_qstr.len == qstr->len &&
 281                     memcmp(entry->se_qstr.name, qstr->name, qstr->len) == 0)
 282                         return entry;
 283         }
 284         return NULL;
 285 }
 286
 287 /*
 288  * Used by the async getattr request callback to find entry with index.
 289  *
 290  * Inside lli_sa_lock to prevent others to change the list during the search.
 291  * It needs to increase entry refcount before returning to guarantee that the
 292  * entry cannot be freed by others.
 293  */
 294 static struct ll_sa_entry *
 295 ll_sa_entry_get_byindex(struct ll_statahead_info *sai, __u64 index)
 296 {
 297         struct ll_sa_entry *entry;
 298
 299         cfs_list_for_each_entry(entry, &sai->sai_entries_sent, se_list) {
 300                 if (entry->se_index == index) {
 301                         cfs_atomic_inc(&entry->se_refcount);
 302                         return entry;
 303                 }
 304                 if (entry->se_index > index)
 305                         break;
 306         }
 307         return NULL;
 308 }
 309
 310 static void ll_sa_entry_cleanup(struct ll_statahead_info *sai,
 311                                  struct ll_sa_entry *entry)
 312 {
 313         struct md_enqueue_info *minfo = entry->se_minfo;
 314         struct ptlrpc_request  *req   = entry->se_req;
 315
 316         if (minfo) {
 317                 entry->se_minfo = NULL;
 318                 ll_intent_release(&minfo->mi_it);
 319                 iput(minfo->mi_dir);
 320                 OBD_FREE_PTR(minfo);
 321         }
 322
 323         if (req) {
 324                 entry->se_req = NULL;
 325                 ptlrpc_req_finished(req);
 326         }
 327 }
 328
 329 static void ll_sa_entry_put(struct ll_statahead_info *sai,
 330                              struct ll_sa_entry *entry)
 331 {
 332         if (cfs_atomic_dec_and_test(&entry->se_refcount)) {
 333                 CDEBUG(D_READA, "free sai entry %.*s(%p) index "LPU64"\n",
 334                        entry->se_qstr.len, entry->se_qstr.name, entry,
 335                        entry->se_index);
 336
 337                 LASSERT(ll_sa_entry_unhashed(entry));
 338                 LASSERT(ll_sa_entry_unlinked(entry));
 339
 340                 ll_sa_entry_cleanup(sai, entry);
 341                 if (entry->se_inode)
 342                         iput(entry->se_inode);
 343
 344                 OBD_FREE(entry, entry->se_size);
 345                 cfs_atomic_dec(&sai->sai_cache_count);
 346         }
 347 }
 348
 349 static inline void
 350 do_sai_entry_fini(struct ll_statahead_info *sai, struct ll_sa_entry *entry)
 351 {
 352         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 353
 354         ll_sa_entry_unhash(sai, entry);
 355
 356         cfs_spin_lock(&lli->lli_sa_lock);
 357         entry->se_stat = SA_ENTRY_DEST;
 358         if (likely(!ll_sa_entry_unlinked(entry)))
 359                 cfs_list_del_init(&entry->se_list);
 360         cfs_spin_unlock(&lli->lli_sa_lock);
 361
 362         ll_sa_entry_put(sai, entry);
 363 }
 364
 365 /*
 366  * Delete it from sai_entries_stated list when fini.
 367  */
 368 static void
 369 ll_sa_entry_fini(struct ll_statahead_info *sai, struct ll_sa_entry *entry)
 370 {
 371         struct ll_sa_entry *pos, *next;
 372
 373         if (entry)
 374                 do_sai_entry_fini(sai, entry);
 375
 376         /* drop old entry from sent list */
 377         cfs_list_for_each_entry_safe(pos, next, &sai->sai_entries_sent,
 378                                      se_list) {
 379                 if (is_omitted_entry(sai, pos->se_index))
 380                         do_sai_entry_fini(sai, pos);
 381                 else
 382                         break;
 383         }
 384
 385         /* drop old entry from stated list */
 386         cfs_list_for_each_entry_safe(pos, next, &sai->sai_entries_stated,
 387                                      se_list) {
 388                 if (is_omitted_entry(sai, pos->se_index))
 389                         do_sai_entry_fini(sai, pos);
 390                 else
 391                         break;
 392         }
 393 }
 394
 395 /*
 396  * Inside lli_sa_lock.
 397  */
 398 static void
 399 do_sai_entry_to_stated(struct ll_statahead_info *sai,
 400                        struct ll_sa_entry *entry, int rc)
 401 {
 402         struct ll_sa_entry *se;
 403         cfs_list_t         *pos = &sai->sai_entries_stated;
 404
 405         if (!ll_sa_entry_unlinked(entry))
 406                 cfs_list_del_init(&entry->se_list);
 407
 408         cfs_list_for_each_entry_reverse(se, &sai->sai_entries_stated, se_list) {
 409                 if (se->se_index < entry->se_index) {
 410                         pos = &se->se_list;
 411                         break;
 412                 }
 413         }
 414
 415         cfs_list_add(&entry->se_list, pos);
 416         entry->se_stat = rc;
 417 }
 418
 419 /*
 420  * Move entry to sai_entries_stated and sort with the index.
 421  * \retval 1    -- entry to be destroyed.
 422  * \retval 0    -- entry is inserted into stated list.
 423  */
 424 static int
 425 ll_sa_entry_to_stated(struct ll_statahead_info *sai,
 426                        struct ll_sa_entry *entry, int rc)
 427 {
 428         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 429         int                   ret = 1;
 430
 431         ll_sa_entry_cleanup(sai, entry);
 432
 433         cfs_spin_lock(&lli->lli_sa_lock);
 434         if (likely(entry->se_stat != SA_ENTRY_DEST)) {
 435                 do_sai_entry_to_stated(sai, entry, rc);
 436                 ret = 0;
 437         }
 438         cfs_spin_unlock(&lli->lli_sa_lock);
 439
 440         return ret;
 441 }
 442
 443 /*
 444  * Insert inode into the list of sai_entries_agl.
 445  */
 446 static void ll_agl_add(struct ll_statahead_info *sai,
 447                        struct inode *inode, int index)
 448 {
 449         struct ll_inode_info *child  = ll_i2info(inode);
 450         struct ll_inode_info *parent = ll_i2info(sai->sai_inode);
 451         int                   added  = 0;
 452
 453         cfs_spin_lock(&child->lli_agl_lock);
 454         if (child->lli_agl_index == 0) {
 455                 child->lli_agl_index = index;
 456                 cfs_spin_unlock(&child->lli_agl_lock);
 457
 458                 LASSERT(cfs_list_empty(&child->lli_agl_list));
 459
 460                 igrab(inode);
 461                 cfs_spin_lock(&parent->lli_agl_lock);
 462                 if (agl_list_empty(sai))
 463                         added = 1;
 464                 cfs_list_add_tail(&child->lli_agl_list, &sai->sai_entries_agl);
 465                 cfs_spin_unlock(&parent->lli_agl_lock);
 466         } else {
 467                 cfs_spin_unlock(&child->lli_agl_lock);
 468         }
 469
 470         if (added > 0)
 471                 cfs_waitq_signal(&sai->sai_agl_thread.t_ctl_waitq);
 472 }
 473
 474 static struct ll_statahead_info *ll_sai_alloc(void)
 475 {
 476         struct ll_statahead_info *sai;
 477         int                       i;
 478         ENTRY;
 479
 480         OBD_ALLOC_PTR(sai);
 481         if (!sai)
 482                 RETURN(NULL);
 483
 484         cfs_atomic_set(&sai->sai_refcount, 1);
 485
 486         cfs_spin_lock(&sai_generation_lock);
 487         sai->sai_generation = ++sai_generation;
 488         if (unlikely(sai_generation == 0))
 489                 sai->sai_generation = ++sai_generation;
 490         cfs_spin_unlock(&sai_generation_lock);
 491
 492         sai->sai_max = LL_SA_RPC_MIN;
 493         sai->sai_index = 1;
 494         cfs_waitq_init(&sai->sai_waitq);
 495         cfs_waitq_init(&sai->sai_thread.t_ctl_waitq);
 496         cfs_waitq_init(&sai->sai_agl_thread.t_ctl_waitq);
 497
 498         CFS_INIT_LIST_HEAD(&sai->sai_entries_sent);
 499         CFS_INIT_LIST_HEAD(&sai->sai_entries_received);
 500         CFS_INIT_LIST_HEAD(&sai->sai_entries_stated);
 501         CFS_INIT_LIST_HEAD(&sai->sai_entries_agl);
 502
 503         for (i = 0; i < LL_SA_CACHE_SIZE; i++) {
 504                 CFS_INIT_LIST_HEAD(&sai->sai_cache[i]);
 505                 cfs_spin_lock_init(&sai->sai_cache_lock[i]);
 506         }
 507         cfs_atomic_set(&sai->sai_cache_count, 0);
 508
 509         RETURN(sai);
 510 }
 511
 512 static inline struct ll_statahead_info *
 513 ll_sai_get(struct ll_statahead_info *sai)
 514 {
 515         cfs_atomic_inc(&sai->sai_refcount);
 516         return sai;
 517 }
 518
 519 static void ll_sai_put(struct ll_statahead_info *sai)
 520 {
 521         struct inode         *inode = sai->sai_inode;
 522         struct ll_inode_info *lli   = ll_i2info(inode);
 523         ENTRY;
 524
 525         if (cfs_atomic_dec_and_lock(&sai->sai_refcount, &lli->lli_sa_lock)) {
 526                 struct ll_sa_entry *entry, *next;
 527
 528                 if (unlikely(cfs_atomic_read(&sai->sai_refcount) > 0)) {
 529                         /* It is race case, the interpret callback just hold
 530                          * a reference count */
 531                         cfs_spin_unlock(&lli->lli_sa_lock);
 532                         RETURN_EXIT;
 533                 }
 534
 535                 LASSERT(lli->lli_opendir_key == NULL);
 536                 LASSERT(thread_is_stopped(&sai->sai_thread));
 537                 LASSERT(thread_is_stopped(&sai->sai_agl_thread));
 538
 539                 lli->lli_sai = NULL;
 540                 lli->lli_opendir_pid = 0;
 541                 cfs_spin_unlock(&lli->lli_sa_lock);
 542
 543                 if (sai->sai_sent > sai->sai_replied)
 544                         CDEBUG(D_READA,"statahead for dir "DFID" does not "
 545                               "finish: [sent:"LPU64"] [replied:"LPU64"]\n",
 546                               PFID(&lli->lli_fid),
 547                               sai->sai_sent, sai->sai_replied);
 548
 549                 cfs_list_for_each_entry_safe(entry, next,
 550                                              &sai->sai_entries_sent, se_list)
 551                         do_sai_entry_fini(sai, entry);
 552
 553                 LASSERT(sa_received_empty(sai));
 554
 555                 cfs_list_for_each_entry_safe(entry, next,
 556                                              &sai->sai_entries_stated, se_list)
 557                         do_sai_entry_fini(sai, entry);
 558
 559                 LASSERT(cfs_atomic_read(&sai->sai_cache_count) == 0);
 560                 LASSERT(agl_list_empty(sai));
 561
 562                 iput(inode);
 563                 OBD_FREE_PTR(sai);
 564         }
 565
 566         EXIT;
 567 }
 568
 569 /* Do NOT forget to drop inode refcount when into sai_entries_agl. */
 570 static void ll_agl_trigger(struct inode *inode, struct ll_statahead_info *sai)
 571 {
 572         struct ll_inode_info *lli   = ll_i2info(inode);
 573         __u64                 index = lli->lli_agl_index;
 574         int                   rc;
 575         ENTRY;
 576
 577         LASSERT(cfs_list_empty(&lli->lli_agl_list));
 578
 579         /* AGL maybe fall behind statahead with one entry */
 580         if (is_omitted_entry(sai, index + 1)) {
 581                 lli->lli_agl_index = 0;
 582                 iput(inode);
 583                 RETURN_EXIT;
 584         }
 585
 586         /* Someone is in glimpse (sync or async), do nothing. */
 587         rc = cfs_down_write_trylock(&lli->lli_glimpse_sem);
 588         if (rc == 0) {
 589                 lli->lli_agl_index = 0;
 590                 iput(inode);
 591                 RETURN_EXIT;
 592         }
 593
 594         /*
 595          * Someone triggered glimpse within 1 sec before.
 596          * 1) The former glimpse succeeded with glimpse lock granted by OST, and
 597          *    if the lock is still cached on client, AGL needs to do nothing. If
 598          *    it is cancelled by other client, AGL maybe cannot obtaion new lock
 599          *    for no glimpse callback triggered by AGL.
 600          * 2) The former glimpse succeeded, but OST did not grant glimpse lock.
 601          *    Under such case, it is quite possible that the OST will not grant
 602          *    glimpse lock for AGL also.
 603          * 3) The former glimpse failed, compared with other two cases, it is
 604          *    relative rare. AGL can ignore such case, and it will not muchly
 605          *    affect the performance.
 606          */
 607         if (lli->lli_glimpse_time != 0 &&
 608             cfs_time_before(cfs_time_shift(-1), lli->lli_glimpse_time)) {
 609                 cfs_up_write(&lli->lli_glimpse_sem);
 610                 lli->lli_agl_index = 0;
 611                 iput(inode);
 612                 RETURN_EXIT;
 613         }
 614
 615         CDEBUG(D_READA, "Handling (init) async glimpse: inode = "
 616                DFID", idx = "LPU64"\n", PFID(&lli->lli_fid), index);
 617
 618         cl_agl(inode);
 619         lli->lli_agl_index = 0;
 620         lli->lli_glimpse_time = cfs_time_current();
 621         cfs_up_write(&lli->lli_glimpse_sem);
 622
 623         CDEBUG(D_READA, "Handled (init) async glimpse: inode= "
 624                DFID", idx = "LPU64", rc = %d\n",
 625                PFID(&lli->lli_fid), index, rc);
 626
 627         iput(inode);
 628
 629         EXIT;
 630 }
 631
 632 static void do_statahead_interpret(struct ll_statahead_info *sai,
 633                                    struct ll_sa_entry *target)
 634 {
 635         struct inode           *dir   = sai->sai_inode;
 636         struct inode           *child;
 637         struct ll_inode_info   *lli   = ll_i2info(dir);
 638         struct ll_sa_entry     *entry;
 639         struct md_enqueue_info *minfo;
 640         struct lookup_intent   *it;
 641         struct ptlrpc_request  *req;
 642         struct mdt_body        *body;
 643         int                     rc    = 0;
 644         ENTRY;
 645
 646         cfs_spin_lock(&lli->lli_sa_lock);
 647         if (target != NULL && target->se_req != NULL &&
 648             !cfs_list_empty(&target->se_list)) {
 649                 entry = target;
 650         } else if (unlikely(sa_received_empty(sai))) {
 651                 cfs_spin_unlock(&lli->lli_sa_lock);
 652                 RETURN_EXIT;
 653         } else {
 654                 entry = sa_first_received_entry(sai);
 655         }
 656
 657         cfs_atomic_inc(&entry->se_refcount);
 658         cfs_list_del_init(&entry->se_list);
 659         cfs_spin_unlock(&lli->lli_sa_lock);
 660
 661         LASSERT(entry->se_handle != 0);
 662
 663         minfo = entry->se_minfo;
 664         it = &minfo->mi_it;
 665         req = entry->se_req;
 666         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 667         if (body == NULL)
 668                 GOTO(out, rc = -EFAULT);
 669
 670         child = entry->se_inode;
 671         if (child == NULL) {
 672                 /*
 673                  * lookup.
 674                  */
 675                 LASSERT(fid_is_zero(&minfo->mi_data.op_fid2));
 676
 677                 /* XXX: No fid in reply, this is probaly cross-ref case.
 678                  * SA can't handle it yet. */
 679                 if (body->valid & OBD_MD_MDS)
 680                         GOTO(out, rc = -EAGAIN);
 681         } else {
 682                 /*
 683                  * revalidate.
 684                  */
 685                 /* unlinked and re-created with the same name */
 686                 if (unlikely(!lu_fid_eq(&minfo->mi_data.op_fid2, &body->fid1))){
 687                         entry->se_inode = NULL;
 688                         iput(child);
 689                         child = NULL;
 690                 }
 691         }
 692
 693         it->d.lustre.it_lock_handle = entry->se_handle;
 694         rc = md_revalidate_lock(ll_i2mdexp(dir), it, NULL, NULL);
 695         if (rc != 1)
 696                 GOTO(out, rc = -EAGAIN);
 697
 698         rc = ll_prep_inode(&child, req, dir->i_sb);
 699         if (rc)
 700                 GOTO(out, rc);
 701
 702         CDEBUG(D_DLMTRACE, "setting l_data to inode %p (%lu/%u)\n",
 703                child, child->i_ino, child->i_generation);
 704         ll_set_lock_data(ll_i2sbi(dir)->ll_md_exp, child, it, NULL);
 705
 706         entry->se_inode = child;
 707
 708         if (agl_should_run(sai, child))
 709                 ll_agl_add(sai, child, entry->se_index);
 710
 711         EXIT;
 712
 713 out:
 714         /* The "ll_sa_entry_to_stated()" will drop related ldlm ibits lock
 715          * reference count by calling "ll_intent_drop_lock()" in spite of the
 716          * above operations failed or not. Do not worry about calling
 717          * "ll_intent_drop_lock()" more than once. */
 718         rc = ll_sa_entry_to_stated(sai, entry, rc < 0 ? rc : SA_ENTRY_SUCC);
 719         if (rc == 0 && entry->se_index == sai->sai_index_wait && target == NULL)
 720                 cfs_waitq_signal(&sai->sai_waitq);
 721         ll_sa_entry_put(sai, entry);
 722 }
 723
 724 static int ll_statahead_interpret(struct ptlrpc_request *req,
 725                                   struct md_enqueue_info *minfo, int rc)
 726 {
 727         struct lookup_intent     *it  = &minfo->mi_it;
 728         struct inode             *dir = minfo->mi_dir;
 729         struct ll_inode_info     *lli = ll_i2info(dir);
 730         struct ll_statahead_info *sai = NULL;
 731         struct ll_sa_entry       *entry;
 732         int                       wakeup;
 733         ENTRY;
 734
 735         if (it_disposition(it, DISP_LOOKUP_NEG))
 736                 rc = -ENOENT;
 737
 738         cfs_spin_lock(&lli->lli_sa_lock);
 739         /* stale entry */
 740         if (unlikely(lli->lli_sai == NULL ||
 741                      lli->lli_sai->sai_generation != minfo->mi_generation)) {
 742                 cfs_spin_unlock(&lli->lli_sa_lock);
 743                 GOTO(out, rc = -ESTALE);
 744         } else {
 745                 sai = ll_sai_get(lli->lli_sai);
 746                 if (unlikely(!thread_is_running(&sai->sai_thread))) {
 747                         sai->sai_replied++;
 748                         cfs_spin_unlock(&lli->lli_sa_lock);
 749                         GOTO(out, rc = -EBADFD);
 750                 }
 751
 752                 entry = ll_sa_entry_get_byindex(sai, minfo->mi_cbdata);
 753                 if (entry == NULL) {
 754                         sai->sai_replied++;
 755                         cfs_spin_unlock(&lli->lli_sa_lock);
 756                         GOTO(out, rc = -EIDRM);
 757                 }
 758
 759                 cfs_list_del_init(&entry->se_list);
 760                 if (rc != 0) {
 761                         sai->sai_replied++;
 762                         do_sai_entry_to_stated(sai, entry, rc);
 763                         cfs_spin_unlock(&lli->lli_sa_lock);
 764                         if (entry->se_index == sai->sai_index_wait)
 765                                 cfs_waitq_signal(&sai->sai_waitq);
 766                 } else {
 767                         entry->se_minfo = minfo;
 768                         entry->se_req = ptlrpc_request_addref(req);
 769                         /* Release the async ibits lock ASAP to avoid deadlock
 770                          * when statahead thread tries to enqueue lock on parent
 771                          * for readpage and other tries to enqueue lock on child
 772                          * with parent's lock held, for example: unlink. */
 773                         entry->se_handle = it->d.lustre.it_lock_handle;
 774                         ll_intent_drop_lock(it);
 775                         wakeup = sa_received_empty(sai);
 776                         cfs_list_add_tail(&entry->se_list,
 777                                           &sai->sai_entries_received);
 778                         sai->sai_replied++;
 779                         cfs_spin_unlock(&lli->lli_sa_lock);
 780                         if (wakeup)
 781                                 cfs_waitq_signal(&sai->sai_thread.t_ctl_waitq);
 782                 }
 783                 ll_sa_entry_put(sai, entry);
 784         }
 785
 786         EXIT;
 787
 788 out:
 789         if (rc != 0) {
 790                 ll_intent_release(it);
 791                 iput(dir);
 792                 OBD_FREE_PTR(minfo);
 793         }
 794         if (sai != NULL)
 795                 ll_sai_put(sai);
 796         return rc;
 797 }
 798
 799 static void sa_args_fini(struct md_enqueue_info *minfo,
 800                          struct ldlm_enqueue_info *einfo)
 801 {
 802         LASSERT(minfo && einfo);
 803         iput(minfo->mi_dir);
 804         capa_put(minfo->mi_data.op_capa1);
 805         capa_put(minfo->mi_data.op_capa2);
 806         OBD_FREE_PTR(minfo);
 807         OBD_FREE_PTR(einfo);
 808 }
 809
 810 /**
 811  * There is race condition between "capa_put" and "ll_statahead_interpret" for
 812  * accessing "op_data.op_capa[1,2]" as following:
 813  * "capa_put" releases "op_data.op_capa[1,2]"'s reference count after calling
 814  * "md_intent_getattr_async". But "ll_statahead_interpret" maybe run first, and
 815  * fill "op_data.op_capa[1,2]" as POISON, then cause "capa_put" access invalid
 816  * "ocapa". So here reserve "op_data.op_capa[1,2]" in "pcapa" before calling
 817  * "md_intent_getattr_async".
 818  */
 819 static int sa_args_init(struct inode *dir, struct inode *child,
 820                         struct ll_sa_entry *entry, struct md_enqueue_info **pmi,
 821                         struct ldlm_enqueue_info **pei,
 822                         struct obd_capa **pcapa)
 823 {
 824         struct qstr              *qstr = &entry->se_qstr;
 825         struct ll_inode_info     *lli  = ll_i2info(dir);
 826         struct md_enqueue_info   *minfo;
 827         struct ldlm_enqueue_info *einfo;
 828         struct md_op_data        *op_data;
 829
 830         OBD_ALLOC_PTR(einfo);
 831         if (einfo == NULL)
 832                 return -ENOMEM;
 833
 834         OBD_ALLOC_PTR(minfo);
 835         if (minfo == NULL) {
 836                 OBD_FREE_PTR(einfo);
 837                 return -ENOMEM;
 838         }
 839
 840         op_data = ll_prep_md_op_data(&minfo->mi_data, dir, child, qstr->name,
 841                                      qstr->len, 0, LUSTRE_OPC_ANY, NULL);
 842         if (IS_ERR(op_data)) {
 843                 OBD_FREE_PTR(einfo);
 844                 OBD_FREE_PTR(minfo);
 845                 return PTR_ERR(op_data);
 846         }
 847
 848         minfo->mi_it.it_op = IT_GETATTR;
 849         minfo->mi_dir = igrab(dir);
 850         minfo->mi_cb = ll_statahead_interpret;
 851         minfo->mi_generation = lli->lli_sai->sai_generation;
 852         minfo->mi_cbdata = entry->se_index;
 853
 854         einfo->ei_type   = LDLM_IBITS;
 855         einfo->ei_mode   = it_to_lock_mode(&minfo->mi_it);
 856         einfo->ei_cb_bl  = ll_md_blocking_ast;
 857         einfo->ei_cb_cp  = ldlm_completion_ast;
 858         einfo->ei_cb_gl  = NULL;
 859         einfo->ei_cbdata = NULL;
 860
 861         *pmi = minfo;
 862         *pei = einfo;
 863         pcapa[0] = op_data->op_capa1;
 864         pcapa[1] = op_data->op_capa2;
 865
 866         return 0;
 867 }
 868
 869 static int do_sa_lookup(struct inode *dir, struct ll_sa_entry *entry)
 870 {
 871         struct md_enqueue_info   *minfo;
 872         struct ldlm_enqueue_info *einfo;
 873         struct obd_capa          *capas[2];
 874         int                       rc;
 875         ENTRY;
 876
 877         rc = sa_args_init(dir, NULL, entry, &minfo, &einfo, capas);
 878         if (rc)
 879                 RETURN(rc);
 880
 881         rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo);
 882         if (!rc) {
 883                 capa_put(capas[0]);
 884                 capa_put(capas[1]);
 885         } else {
 886                 sa_args_fini(minfo, einfo);
 887         }
 888
 889         RETURN(rc);
 890 }
 891
 892 /**
 893  * similar to ll_revalidate_it().
 894  * \retval      1 -- dentry valid
 895  * \retval      0 -- will send stat-ahead request
 896  * \retval others -- prepare stat-ahead request failed
 897  */
 898 static int do_sa_revalidate(struct inode *dir, struct ll_sa_entry *entry,
 899                             struct dentry *dentry)
 900 {
 901         struct inode             *inode = dentry->d_inode;
 902         struct lookup_intent      it = { .it_op = IT_GETATTR,
 903                                          .d.lustre.it_lock_handle = 0 };
 904         struct md_enqueue_info   *minfo;
 905         struct ldlm_enqueue_info *einfo;
 906         struct obd_capa          *capas[2];
 907         int rc;
 908         ENTRY;
 909
 910         if (unlikely(inode == NULL))
 911                 RETURN(1);
 912
 913         if (d_mountpoint(dentry))
 914                 RETURN(1);
 915
 916         if (unlikely(dentry == dentry->d_sb->s_root))
 917                 RETURN(1);
 918
 919         entry->se_inode = igrab(inode);
 920         rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode),NULL);
 921         if (rc == 1) {
 922                 entry->se_handle = it.d.lustre.it_lock_handle;
 923                 ll_intent_release(&it);
 924                 RETURN(1);
 925         }
 926
 927         rc = sa_args_init(dir, inode, entry, &minfo, &einfo, capas);
 928         if (rc) {
 929                 entry->se_inode = NULL;
 930                 iput(inode);
 931                 RETURN(rc);
 932         }
 933
 934         rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo);
 935         if (!rc) {
 936                 capa_put(capas[0]);
 937                 capa_put(capas[1]);
 938         } else {
 939                 entry->se_inode = NULL;
 940                 iput(inode);
 941                 sa_args_fini(minfo, einfo);
 942         }
 943
 944         RETURN(rc);
 945 }
 946
 947 static void ll_statahead_one(struct dentry *parent, const char* entry_name,
 948                              int entry_name_len)
 949 {
 950         struct inode             *dir    = parent->d_inode;
 951         struct ll_inode_info     *lli    = ll_i2info(dir);
 952         struct ll_statahead_info *sai    = lli->lli_sai;
 953         struct dentry            *dentry = NULL;
 954         struct ll_sa_entry       *entry;
 955         int                       rc;
 956         int                       rc1;
 957         ENTRY;
 958
 959         entry = ll_sa_entry_alloc(sai, sai->sai_index, entry_name,
 960                                   entry_name_len);
 961         if (IS_ERR(entry))
 962                 RETURN_EXIT;
 963
 964         dentry = d_lookup(parent, &entry->se_qstr);
 965         if (!dentry) {
 966                 rc = do_sa_lookup(dir, entry);
 967         } else {
 968                 rc = do_sa_revalidate(dir, entry, dentry);
 969                 if (rc == 1 && agl_should_run(sai, dentry->d_inode))
 970                         ll_agl_add(sai, dentry->d_inode, entry->se_index);
 971         }
 972
 973         if (dentry != NULL)
 974                 dput(dentry);
 975
 976         if (rc) {
 977                 rc1 = ll_sa_entry_to_stated(sai, entry,
 978                                         rc < 0 ? SA_ENTRY_INVA : SA_ENTRY_SUCC);
 979                 if (rc1 == 0 && entry->se_index == sai->sai_index_wait)
 980                         cfs_waitq_signal(&sai->sai_waitq);
 981         } else {
 982                 sai->sai_sent++;
 983         }
 984
 985         sai->sai_index++;
 986         /* drop one refcount on entry by ll_sa_entry_alloc */
 987         ll_sa_entry_put(sai, entry);
 988
 989         EXIT;
 990 }
 991
 992 static int ll_agl_thread(void *arg)
 993 {
 994         struct dentry            *parent = (struct dentry *)arg;
 995         struct inode             *dir    = parent->d_inode;
 996         struct ll_inode_info     *plli   = ll_i2info(dir);
 997         struct ll_inode_info     *clli;
 998         struct ll_sb_info        *sbi    = ll_i2sbi(dir);
 999         struct ll_statahead_info *sai    = ll_sai_get(plli->lli_sai);
1000         struct ptlrpc_thread     *thread = &sai->sai_agl_thread;
1001         struct l_wait_info        lwi    = { 0 };
1002         ENTRY;
1003
1004         {
1005                 char pname[16];
1006                 snprintf(pname, 15, "ll_agl_%u", plli->lli_opendir_pid);
1007                 cfs_daemonize(pname);
1008         }
1009
1010         CDEBUG(D_READA, "agl thread started: [pid %d] [parent %.*s]\n",
1011                cfs_curproc_pid(), parent->d_name.len, parent->d_name.name);
1012
1013         atomic_inc(&sbi->ll_agl_total);
1014         cfs_spin_lock(&plli->lli_agl_lock);
1015         sai->sai_agl_valid = 1;
1016         thread_set_flags(thread, SVC_RUNNING);
1017         cfs_spin_unlock(&plli->lli_agl_lock);
1018         cfs_waitq_signal(&thread->t_ctl_waitq);
1019
1020         while (1) {
1021                 l_wait_event(thread->t_ctl_waitq,
1022                              !agl_list_empty(sai) ||
1023                              !thread_is_running(thread),
1024                              &lwi);
1025
1026                 if (!thread_is_running(thread))
1027                         break;
1028
1029                 cfs_spin_lock(&plli->lli_agl_lock);
1030                 /* The statahead thread maybe help to process AGL entries,
1031                  * so check whether list empty again. */
1032                 if (!agl_list_empty(sai)) {
1033                         clli = agl_first_entry(sai);
1034                         cfs_list_del_init(&clli->lli_agl_list);
1035                         cfs_spin_unlock(&plli->lli_agl_lock);
1036                         ll_agl_trigger(&clli->lli_vfs_inode, sai);
1037                 } else {
1038                         cfs_spin_unlock(&plli->lli_agl_lock);
1039                 }
1040         }
1041
1042         cfs_spin_lock(&plli->lli_agl_lock);
1043         sai->sai_agl_valid = 0;
1044         while (!agl_list_empty(sai)) {
1045                 clli = agl_first_entry(sai);
1046                 cfs_list_del_init(&clli->lli_agl_list);
1047                 cfs_spin_unlock(&plli->lli_agl_lock);
1048                 clli->lli_agl_index = 0;
1049                 iput(&clli->lli_vfs_inode);
1050                 cfs_spin_lock(&plli->lli_agl_lock);
1051         }
1052         thread_set_flags(thread, SVC_STOPPED);
1053         cfs_spin_unlock(&plli->lli_agl_lock);
1054         cfs_waitq_signal(&thread->t_ctl_waitq);
1055         ll_sai_put(sai);
1056         CDEBUG(D_READA, "agl thread stopped: [pid %d] [parent %.*s]\n",
1057                cfs_curproc_pid(), parent->d_name.len, parent->d_name.name);
1058         RETURN(0);
1059 }
1060
1061 static void ll_start_agl(struct dentry *parent, struct ll_statahead_info *sai)
1062 {
1063         struct ptlrpc_thread *thread = &sai->sai_agl_thread;
1064         struct l_wait_info    lwi    = { 0 };
1065         int                   rc;
1066         ENTRY;
1067
1068         CDEBUG(D_READA, "start agl thread: [pid %d] [parent %.*s]\n",
1069                cfs_curproc_pid(), parent->d_name.len, parent->d_name.name);
1070
1071         rc = cfs_create_thread(ll_agl_thread, parent, 0);
1072         if (rc < 0) {
1073                 CERROR("can't start ll_agl thread, rc: %d\n", rc);
1074                 thread_set_flags(thread, SVC_STOPPED);
1075                 RETURN_EXIT;
1076         }
1077
1078         l_wait_event(thread->t_ctl_waitq,
1079                      thread_is_running(thread) || thread_is_stopped(thread),
1080                      &lwi);
1081         EXIT;
1082 }
1083
1084 static int ll_statahead_thread(void *arg)
1085 {
1086         struct dentry            *parent = (struct dentry *)arg;
1087         struct inode             *dir    = parent->d_inode;
1088         struct ll_inode_info     *plli   = ll_i2info(dir);
1089         struct ll_inode_info     *clli;
1090         struct ll_sb_info        *sbi    = ll_i2sbi(dir);
1091         struct ll_statahead_info *sai    = ll_sai_get(plli->lli_sai);
1092         struct ptlrpc_thread     *thread = &sai->sai_thread;
1093         struct ptlrpc_thread *agl_thread = &sai->sai_agl_thread;
1094         struct page              *page;
1095         __u64                     pos    = 0;
1096         int                       first  = 0;
1097         int                       rc     = 0;
1098         struct ll_dir_chain       chain;
1099         struct l_wait_info        lwi    = { 0 };
1100         ENTRY;
1101
1102         {
1103                 char pname[16];
1104                 snprintf(pname, 15, "ll_sa_%u", plli->lli_opendir_pid);
1105                 cfs_daemonize(pname);
1106         }
1107
1108         CDEBUG(D_READA, "statahead thread started: [pid %d] [parent %.*s]\n",
1109                cfs_curproc_pid(), parent->d_name.len, parent->d_name.name);
1110
1111         if (sbi->ll_flags & LL_SBI_AGL_ENABLED)
1112                 ll_start_agl(parent, sai);
1113
1114         atomic_inc(&sbi->ll_sa_total);
1115         cfs_spin_lock(&plli->lli_sa_lock);
1116         thread_set_flags(thread, SVC_RUNNING);
1117         cfs_spin_unlock(&plli->lli_sa_lock);
1118         cfs_waitq_signal(&thread->t_ctl_waitq);
1119
1120         plli->lli_sa_pos = 0;
1121         ll_dir_chain_init(&chain);
1122         page = ll_get_dir_page(NULL, dir, pos, &chain);
1123
1124         while (1) {
1125                 struct lu_dirpage *dp;
1126                 struct lu_dirent  *ent;
1127
1128                 if (IS_ERR(page)) {
1129                         rc = PTR_ERR(page);
1130                         CDEBUG(D_READA, "error reading dir "DFID" at "LPU64
1131                                "/"LPU64": [rc %d] [parent %u]\n",
1132                                PFID(ll_inode2fid(dir)), pos, sai->sai_index,
1133                                rc, plli->lli_opendir_pid);
1134                         GOTO(out, rc);
1135                 }
1136
1137                 dp = page_address(page);
1138                 for (ent = lu_dirent_start(dp); ent != NULL;
1139                      ent = lu_dirent_next(ent)) {
1140                         __u64 hash;
1141                         int namelen;
1142                         char *name;
1143
1144                         hash = le64_to_cpu(ent->lde_hash);
1145                         if (unlikely(hash < pos))
1146                                 /*
1147                                  * Skip until we find target hash value.
1148                                  */
1149                                 continue;
1150
1151                         namelen = le16_to_cpu(ent->lde_namelen);
1152                         if (unlikely(namelen == 0))
1153                                 /*
1154                                  * Skip dummy record.
1155                                  */
1156                                 continue;
1157
1158                         name = ent->lde_name;
1159                         if (name[0] == '.') {
1160                                 if (namelen == 1) {
1161                                         /*
1162                                          * skip "."
1163                                          */
1164                                         continue;
1165                                 } else if (name[1] == '.' && namelen == 2) {
1166                                         /*
1167                                          * skip ".."
1168                                          */
1169                                         continue;
1170                                 } else if (!sai->sai_ls_all) {
1171                                         /*
1172                                          * skip hidden files.
1173                                          */
1174                                         sai->sai_skip_hidden++;
1175                                         continue;
1176                                 }
1177                         }
1178
1179                         /*
1180                          * don't stat-ahead first entry.
1181                          */
1182                         if (unlikely(++first == 1))
1183                                 continue;
1184
1185 keep_it:
1186                         l_wait_event(thread->t_ctl_waitq,
1187                                      !sa_sent_full(sai) ||
1188                                      !sa_received_empty(sai) ||
1189                                      !agl_list_empty(sai) ||
1190                                      !thread_is_running(thread),
1191                                      &lwi);
1192
1193 interpret_it:
1194                         while (!sa_received_empty(sai))
1195                                 do_statahead_interpret(sai, NULL);
1196
1197                         if (unlikely(!thread_is_running(thread))) {
1198                                 ll_release_page(page, 0);
1199                                 GOTO(out, rc = 0);
1200                         }
1201
1202                         /* If no window for metadata statahead, but there are
1203                          * some AGL entries to be triggered, then try to help
1204                          * to process the AGL entries. */
1205                         if (sa_sent_full(sai)) {
1206                                 cfs_spin_lock(&plli->lli_agl_lock);
1207                                 while (!agl_list_empty(sai)) {
1208                                         clli = agl_first_entry(sai);
1209                                         cfs_list_del_init(&clli->lli_agl_list);
1210                                         cfs_spin_unlock(&plli->lli_agl_lock);
1211                                         ll_agl_trigger(&clli->lli_vfs_inode,
1212                                                        sai);
1213
1214                                         if (!sa_received_empty(sai))
1215                                                 goto interpret_it;
1216
1217                                         if (unlikely(
1218                                                 !thread_is_running(thread))) {
1219                                                 ll_release_page(page, 0);
1220                                                 GOTO(out, rc = 0);
1221                                         }
1222
1223                                         if (!sa_sent_full(sai))
1224                                                 goto do_it;
1225
1226                                         cfs_spin_lock(&plli->lli_agl_lock);
1227                                 }
1228                                 cfs_spin_unlock(&plli->lli_agl_lock);
1229
1230                                 goto keep_it;
1231                         }
1232
1233 do_it:
1234                         ll_statahead_one(parent, name, namelen);
1235                 }
1236                 pos = le64_to_cpu(dp->ldp_hash_end);
1237                 if (pos == MDS_DIR_END_OFF) {
1238                         /*
1239                          * End of directory reached.
1240                          */
1241                         ll_release_page(page, 0);
1242                         while (1) {
1243                                 l_wait_event(thread->t_ctl_waitq,
1244                                              !sa_received_empty(sai) ||
1245                                              sai->sai_sent == sai->sai_replied||
1246                                              !thread_is_running(thread),
1247                                              &lwi);
1248
1249                                 while (!sa_received_empty(sai))
1250                                         do_statahead_interpret(sai, NULL);
1251
1252                                 if (unlikely(!thread_is_running(thread)))
1253                                         GOTO(out, rc = 0);
1254
1255                                 if (sai->sai_sent == sai->sai_replied &&
1256                                     sa_received_empty(sai))
1257                                         break;
1258                         }
1259
1260                         cfs_spin_lock(&plli->lli_agl_lock);
1261                         while (!agl_list_empty(sai) &&
1262                                thread_is_running(thread)) {
1263                                 clli = agl_first_entry(sai);
1264                                 cfs_list_del_init(&clli->lli_agl_list);
1265                                 cfs_spin_unlock(&plli->lli_agl_lock);
1266                                 ll_agl_trigger(&clli->lli_vfs_inode, sai);
1267                                 cfs_spin_lock(&plli->lli_agl_lock);
1268                         }
1269                         cfs_spin_unlock(&plli->lli_agl_lock);
1270
1271                         GOTO(out, rc = 0);
1272                 } else if (1) {
1273                         /*
1274                          * chain is exhausted.
1275                          * Normal case: continue to the next page.
1276                          */
1277                         ll_release_page(page, le32_to_cpu(dp->ldp_flags) &
1278                                               LDF_COLLIDE);
1279                         plli->lli_sa_pos = pos;
1280                         sai->sai_in_readpage = 1;
1281                         page = ll_get_dir_page(NULL, dir, pos, &chain);
1282                         sai->sai_in_readpage = 0;
1283                 } else {
1284                         LASSERT(le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
1285                         ll_release_page(page, 1);
1286                         /*
1287                          * go into overflow page.
1288                          */
1289                 }
1290         }
1291         EXIT;
1292
1293 out:
1294         if (sai->sai_agl_valid) {
1295                 cfs_spin_lock(&plli->lli_agl_lock);
1296                 thread_set_flags(agl_thread, SVC_STOPPING);
1297                 cfs_spin_unlock(&plli->lli_agl_lock);
1298                 cfs_waitq_signal(&agl_thread->t_ctl_waitq);
1299
1300                 CDEBUG(D_READA, "stop agl thread: [pid %d]\n",
1301                        cfs_curproc_pid());
1302                 l_wait_event(agl_thread->t_ctl_waitq,
1303                              thread_is_stopped(agl_thread),
1304                              &lwi);
1305         } else {
1306                 /* Set agl_thread flags anyway. */
1307                 thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED);
1308         }
1309         ll_dir_chain_fini(&chain);
1310         cfs_spin_lock(&plli->lli_sa_lock);
1311         if (!sa_received_empty(sai)) {
1312                 thread_set_flags(thread, SVC_STOPPING);
1313                 cfs_spin_unlock(&plli->lli_sa_lock);
1314
1315                 /* To release the resources held by received entries. */
1316                 while (!sa_received_empty(sai))
1317                         do_statahead_interpret(sai, NULL);
1318
1319                 cfs_spin_lock(&plli->lli_sa_lock);
1320         }
1321         thread_set_flags(thread, SVC_STOPPED);
1322         cfs_spin_unlock(&plli->lli_sa_lock);
1323         cfs_waitq_signal(&sai->sai_waitq);
1324         cfs_waitq_signal(&thread->t_ctl_waitq);
1325         ll_sai_put(sai);
1326         dput(parent);
1327         CDEBUG(D_READA, "statahead thread stopped: [pid %d] [parent %.*s]\n",
1328                cfs_curproc_pid(), parent->d_name.len, parent->d_name.name);
1329         return rc;
1330 }
1331
1332 /**
1333  * called in ll_file_release().
1334  */
1335 void ll_stop_statahead(struct inode *dir, void *key)
1336 {
1337         struct ll_inode_info *lli = ll_i2info(dir);
1338
1339         if (unlikely(key == NULL))
1340                 return;
1341
1342         cfs_spin_lock(&lli->lli_sa_lock);
1343         if (lli->lli_opendir_key != key || lli->lli_opendir_pid == 0) {
1344                 cfs_spin_unlock(&lli->lli_sa_lock);
1345                 return;
1346         }
1347
1348         lli->lli_opendir_key = NULL;
1349
1350         if (lli->lli_sai) {
1351                 struct l_wait_info lwi = { 0 };
1352                 struct ptlrpc_thread *thread = &lli->lli_sai->sai_thread;
1353
1354                 if (!thread_is_stopped(thread)) {
1355                         thread_set_flags(thread, SVC_STOPPING);
1356                         cfs_spin_unlock(&lli->lli_sa_lock);
1357                         cfs_waitq_signal(&thread->t_ctl_waitq);
1358
1359                         CDEBUG(D_READA, "stop statahead thread: [pid %d]\n",
1360                                cfs_curproc_pid());
1361                         l_wait_event(thread->t_ctl_waitq,
1362                                      thread_is_stopped(thread),
1363                                      &lwi);
1364                 } else {
1365                         cfs_spin_unlock(&lli->lli_sa_lock);
1366                 }
1367
1368                 /*
1369                  * Put the ref which was held when first statahead_enter.
1370                  * It maybe not the last ref for some statahead requests
1371                  * maybe inflight.
1372                  */
1373                 ll_sai_put(lli->lli_sai);
1374         } else {
1375                 lli->lli_opendir_pid = 0;
1376                 cfs_spin_unlock(&lli->lli_sa_lock);
1377         }
1378 }
1379
1380 enum {
1381         /**
1382          * not first dirent, or is "."
1383          */
1384         LS_NONE_FIRST_DE = 0,
1385         /**
1386          * the first non-hidden dirent
1387          */
1388         LS_FIRST_DE,
1389         /**
1390          * the first hidden dirent, that is "."
1391          */
1392         LS_FIRST_DOT_DE
1393 };
1394
1395 static int is_first_dirent(struct inode *dir, struct dentry *dentry)
1396 {
1397         struct ll_inode_info *lli    = ll_i2info(dir);
1398         struct ll_dir_chain   chain;
1399         struct qstr          *target = &dentry->d_name;
1400         struct page          *page;
1401         __u64                 pos    = 0;
1402         int                   dot_de;
1403         int                   rc     = LS_NONE_FIRST_DE;
1404         ENTRY;
1405
1406         lli->lli_sa_pos = 0;
1407         ll_dir_chain_init(&chain);
1408         page = ll_get_dir_page(NULL, dir, pos, &chain);
1409
1410         while (1) {
1411                 struct lu_dirpage *dp;
1412                 struct lu_dirent  *ent;
1413
1414                 if (IS_ERR(page)) {
1415                         struct ll_inode_info *lli = ll_i2info(dir);
1416
1417                         rc = PTR_ERR(page);
1418                         CERROR("error reading dir "DFID" at "LPU64": "
1419                                "[rc %d] [parent %u]\n",
1420                                PFID(ll_inode2fid(dir)), pos,
1421                                rc, lli->lli_opendir_pid);
1422                         break;
1423                 }
1424
1425                 dp = page_address(page);
1426                 for (ent = lu_dirent_start(dp); ent != NULL;
1427                      ent = lu_dirent_next(ent)) {
1428                         __u64 hash;
1429                         int namelen;
1430                         char *name;
1431
1432                         hash = le64_to_cpu(ent->lde_hash);
1433                         /* The ll_get_dir_page() can return any page containing
1434                          * the given hash which may be not the start hash. */
1435                         if (unlikely(hash < pos))
1436                                 continue;
1437
1438                         namelen = le16_to_cpu(ent->lde_namelen);
1439                         if (unlikely(namelen == 0))
1440                                 /*
1441                                  * skip dummy record.
1442                                  */
1443                                 continue;
1444
1445                         name = ent->lde_name;
1446                         if (name[0] == '.') {
1447                                 if (namelen == 1)
1448                                         /*
1449                                          * skip "."
1450                                          */
1451                                         continue;
1452                                 else if (name[1] == '.' && namelen == 2)
1453                                         /*
1454                                          * skip ".."
1455                                          */
1456                                         continue;
1457                                 else
1458                                         dot_de = 1;
1459                         } else {
1460                                 dot_de = 0;
1461                         }
1462
1463                         if (dot_de && target->name[0] != '.') {
1464                                 CDEBUG(D_READA, "%.*s skip hidden file %.*s\n",
1465                                        target->len, target->name,
1466                                        namelen, name);
1467                                 continue;
1468                         }
1469
1470                         if (target->len != namelen ||
1471                             memcmp(target->name, name, namelen) != 0)
1472                                 rc = LS_NONE_FIRST_DE;
1473                         else if (!dot_de)
1474                                 rc = LS_FIRST_DE;
1475                         else
1476                                 rc = LS_FIRST_DOT_DE;
1477
1478                         ll_release_page(page, 0);
1479                         GOTO(out, rc);
1480                 }
1481                 pos = le64_to_cpu(dp->ldp_hash_end);
1482                 if (pos == MDS_DIR_END_OFF) {
1483                         /*
1484                          * End of directory reached.
1485                          */
1486                         ll_release_page(page, 0);
1487                         break;
1488                 } else if (1) {
1489                         /*
1490                          * chain is exhausted
1491                          * Normal case: continue to the next page.
1492                          */
1493                         ll_release_page(page, le32_to_cpu(dp->ldp_flags) &
1494                                               LDF_COLLIDE);
1495                         lli->lli_sa_pos = pos;
1496                         page = ll_get_dir_page(NULL, dir, pos, &chain);
1497                 } else {
1498                         /*
1499                          * go into overflow page.
1500                          */
1501                         LASSERT(le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
1502                         ll_release_page(page, 1);
1503                 }
1504         }
1505         EXIT;
1506
1507 out:
1508         ll_dir_chain_fini(&chain);
1509         return rc;
1510 }
1511
1512 static void
1513 ll_sai_unplug(struct ll_statahead_info *sai, struct ll_sa_entry *entry)
1514 {
1515         struct ptlrpc_thread *thread = &sai->sai_thread;
1516         struct ll_sb_info    *sbi    = ll_i2sbi(sai->sai_inode);
1517         int                   hit;
1518         ENTRY;
1519
1520         if (entry != NULL && entry->se_stat == SA_ENTRY_SUCC)
1521                 hit = 1;
1522         else
1523                 hit = 0;
1524
1525         ll_sa_entry_fini(sai, entry);
1526         if (hit) {
1527                 sai->sai_hit++;
1528                 sai->sai_consecutive_miss = 0;
1529                 sai->sai_max = min(2 * sai->sai_max, sbi->ll_sa_max);
1530         } else {
1531                 struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
1532
1533                 sai->sai_miss++;
1534                 sai->sai_consecutive_miss++;
1535                 if (sa_low_hit(sai) && thread_is_running(thread)) {
1536                         atomic_inc(&sbi->ll_sa_wrong);
1537                         CDEBUG(D_READA, "Statahead for dir "DFID" hit "
1538                                "ratio too low: hit/miss "LPU64"/"LPU64
1539                                ", sent/replied "LPU64"/"LPU64", stopping "
1540                                "statahead thread: pid %d\n",
1541                                PFID(&lli->lli_fid), sai->sai_hit,
1542                                sai->sai_miss, sai->sai_sent,
1543                                sai->sai_replied, cfs_curproc_pid());
1544                         cfs_spin_lock(&lli->lli_sa_lock);
1545                         if (!thread_is_stopped(thread))
1546                                 thread_set_flags(thread, SVC_STOPPING);
1547                         cfs_spin_unlock(&lli->lli_sa_lock);
1548                 }
1549         }
1550
1551         if (!thread_is_stopped(thread))
1552                 cfs_waitq_signal(&thread->t_ctl_waitq);
1553
1554         EXIT;
1555 }
1556
1557 /**
1558  * Start statahead thread if this is the first dir entry.
1559  * Otherwise if a thread is started already, wait it until it is ahead of me.
1560  * \retval 1       -- find entry with lock in cache, the caller needs to do
1561  *                    nothing.
1562  * \retval 0       -- find entry in cache, but without lock, the caller needs
1563  *                    refresh from MDS.
1564  * \retval others  -- the caller need to process as non-statahead.
1565  */
1566 int do_statahead_enter(struct inode *dir, struct dentry **dentryp,
1567                        int only_unplug)
1568 {
1569         struct ll_inode_info     *lli   = ll_i2info(dir);
1570         struct ll_statahead_info *sai   = lli->lli_sai;
1571         struct dentry            *parent;
1572         struct ll_sa_entry       *entry;
1573         struct ptlrpc_thread     *thread;
1574         struct l_wait_info        lwi   = { 0 };
1575         int                       rc    = 0;
1576         ENTRY;
1577
1578         LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
1579
1580         if (sai) {
1581                 thread = &sai->sai_thread;
1582                 if (unlikely(thread_is_stopped(thread) &&
1583                              cfs_list_empty(&sai->sai_entries_stated))) {
1584                         /* to release resource */
1585                         ll_stop_statahead(dir, lli->lli_opendir_key);
1586                         RETURN(-EAGAIN);
1587                 }
1588
1589                 if ((*dentryp)->d_name.name[0] == '.') {
1590                         if (sai->sai_ls_all ||
1591                             sai->sai_miss_hidden >= sai->sai_skip_hidden) {
1592                                 /*
1593                                  * Hidden dentry is the first one, or statahead
1594                                  * thread does not skip so many hidden dentries
1595                                  * before "sai_ls_all" enabled as below.
1596                                  */
1597                         } else {
1598                                 if (!sai->sai_ls_all)
1599                                         /*
1600                                          * It maybe because hidden dentry is not
1601                                          * the first one, "sai_ls_all" was not
1602                                          * set, then "ls -al" missed. Enable
1603                                          * "sai_ls_all" for such case.
1604                                          */
1605                                         sai->sai_ls_all = 1;
1606
1607                                 /*
1608                                  * Such "getattr" has been skipped before
1609                                  * "sai_ls_all" enabled as above.
1610                                  */
1611                                 sai->sai_miss_hidden++;
1612                                 RETURN(-EAGAIN);
1613                         }
1614                 }
1615
1616                 entry = ll_sa_entry_get_byname(sai, &(*dentryp)->d_name);
1617                 if (entry == NULL || only_unplug) {
1618                         ll_sai_unplug(sai, entry);
1619                         RETURN(entry ? 1 : -EAGAIN);
1620                 }
1621
1622                 while (!ll_sa_entry_stated(entry) &&
1623                        sai->sai_in_readpage &&
1624                        !sa_received_empty(sai))
1625                         do_statahead_interpret(sai, entry);
1626
1627                 if (!ll_sa_entry_stated(entry)) {
1628                         sai->sai_index_wait = entry->se_index;
1629                         lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(30), NULL,
1630                                                LWI_ON_SIGNAL_NOOP, NULL);
1631                         rc = l_wait_event(sai->sai_waitq,
1632                                           ll_sa_entry_stated(entry) ||
1633                                           thread_is_stopped(thread),
1634                                           &lwi);
1635                         if (rc < 0) {
1636                                 ll_sai_unplug(sai, entry);
1637                                 RETURN(-EAGAIN);
1638                         }
1639                 }
1640
1641                 if (entry->se_stat == SA_ENTRY_SUCC &&
1642                     entry->se_inode != NULL) {
1643                         struct inode *inode = entry->se_inode;
1644                         struct lookup_intent it = { .it_op = IT_GETATTR,
1645                                                     .d.lustre.it_lock_handle =
1646                                                      entry->se_handle };
1647                         __u64 bits;
1648
1649                         rc = md_revalidate_lock(ll_i2mdexp(dir), &it,
1650                                                 ll_inode2fid(inode), &bits);
1651                         if (rc == 1) {
1652                                 if ((*dentryp)->d_inode == NULL) {
1653                                         *dentryp = ll_splice_alias(inode,
1654                                                                    *dentryp);
1655                                 } else if ((*dentryp)->d_inode != inode) {
1656                                         /* revalidate, but inode is recreated */
1657                                         CDEBUG(D_READA,
1658                                               "stale dentry %.*s inode %lu/%u, "
1659                                               "statahead inode %lu/%u\n",
1660                                               (*dentryp)->d_name.len,
1661                                               (*dentryp)->d_name.name,
1662                                               (*dentryp)->d_inode->i_ino,
1663                                               (*dentryp)->d_inode->i_generation,
1664                                               inode->i_ino,
1665                                               inode->i_generation);
1666                                         ll_sai_unplug(sai, entry);
1667                                         RETURN(-ESTALE);
1668                                 } else {
1669                                         iput(inode);
1670                                 }
1671                                 entry->se_inode = NULL;
1672
1673                                 if ((bits & MDS_INODELOCK_LOOKUP) &&
1674                                     d_lustre_invalid(*dentryp))
1675                                         d_lustre_revalidate(*dentryp);
1676                                 ll_intent_release(&it);
1677                         }
1678                 }
1679
1680                 ll_sai_unplug(sai, entry);
1681                 RETURN(rc);
1682         }
1683
1684         /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */
1685         rc = is_first_dirent(dir, *dentryp);
1686         if (rc == LS_NONE_FIRST_DE)
1687                 /* It is not "ls -{a}l" operation, no need statahead for it. */
1688                 GOTO(out, rc = -EAGAIN);
1689
1690         sai = ll_sai_alloc();
1691         if (sai == NULL)
1692                 GOTO(out, rc = -ENOMEM);
1693
1694         sai->sai_ls_all = (rc == LS_FIRST_DOT_DE);
1695         sai->sai_inode = igrab(dir);
1696         if (unlikely(sai->sai_inode == NULL)) {
1697                 CWARN("Do not start stat ahead on dying inode "DFID"\n",
1698                       PFID(&lli->lli_fid));
1699                 GOTO(out, rc = -ESTALE);
1700         }
1701
1702         /* get parent reference count here, and put it in ll_statahead_thread */
1703         parent = dget((*dentryp)->d_parent);
1704         if (unlikely(sai->sai_inode != parent->d_inode)) {
1705                 struct ll_inode_info *nlli = ll_i2info(parent->d_inode);
1706
1707                 CWARN("Race condition, someone changed %.*s just now: "
1708                       "old parent "DFID", new parent "DFID"\n",
1709                       (*dentryp)->d_name.len, (*dentryp)->d_name.name,
1710                       PFID(&lli->lli_fid), PFID(&nlli->lli_fid));
1711                 dput(parent);
1712                 iput(sai->sai_inode);
1713                 GOTO(out, rc = -EAGAIN);
1714         }
1715
1716         CDEBUG(D_READA, "start statahead thread: [pid %d] [parent %.*s]\n",
1717                cfs_curproc_pid(), parent->d_name.len, parent->d_name.name);
1718
1719         lli->lli_sai = sai;
1720         rc = cfs_create_thread(ll_statahead_thread, parent, 0);
1721         thread = &sai->sai_thread;
1722         if (rc < 0) {
1723                 CERROR("can't start ll_sa thread, rc: %d\n", rc);
1724                 dput(parent);
1725                 lli->lli_opendir_key = NULL;
1726                 thread_set_flags(thread, SVC_STOPPED);
1727                 thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED);
1728                 ll_sai_put(sai);
1729                 LASSERT(lli->lli_sai == NULL);
1730                 RETURN(-EAGAIN);
1731         }
1732
1733         l_wait_event(thread->t_ctl_waitq,
1734                      thread_is_running(thread) || thread_is_stopped(thread),
1735                      &lwi);
1736
1737         /*
1738          * We don't stat-ahead for the first dirent since we are already in
1739          * lookup.
1740          */
1741         RETURN(-EAGAIN);
1742
1743 out:
1744         if (sai != NULL)
1745                 OBD_FREE_PTR(sai);
1746         cfs_spin_lock(&lli->lli_sa_lock);
1747         lli->lli_opendir_key = NULL;
1748         lli->lli_opendir_pid = 0;
1749         cfs_spin_unlock(&lli->lli_sa_lock);
1750         return rc;
1751 }