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 DEFINE_SPINLOCK(sai_generation_lock);
  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         ll_dir_chain_init(&chain);
1121         page = ll_get_dir_page(dir, pos, &chain);
1122
1123         while (1) {
1124                 struct lu_dirpage *dp;
1125                 struct lu_dirent  *ent;
1126
1127                 if (IS_ERR(page)) {
1128                         rc = PTR_ERR(page);
1129                         CDEBUG(D_READA, "error reading dir "DFID" at "LPU64
1130                                "/"LPU64": [rc %d] [parent %u]\n",
1131                                PFID(ll_inode2fid(dir)), pos, sai->sai_index,
1132                                rc, plli->lli_opendir_pid);
1133                         GOTO(out, rc);
1134                 }
1135
1136                 dp = page_address(page);
1137                 for (ent = lu_dirent_start(dp); ent != NULL;
1138                      ent = lu_dirent_next(ent)) {
1139                         __u64 hash;
1140                         int namelen;
1141                         char *name;
1142
1143                         hash = le64_to_cpu(ent->lde_hash);
1144                         if (unlikely(hash < pos))
1145                                 /*
1146                                  * Skip until we find target hash value.
1147                                  */
1148                                 continue;
1149
1150                         namelen = le16_to_cpu(ent->lde_namelen);
1151                         if (unlikely(namelen == 0))
1152                                 /*
1153                                  * Skip dummy record.
1154                                  */
1155                                 continue;
1156
1157                         name = ent->lde_name;
1158                         if (name[0] == '.') {
1159                                 if (namelen == 1) {
1160                                         /*
1161                                          * skip "."
1162                                          */
1163                                         continue;
1164                                 } else if (name[1] == '.' && namelen == 2) {
1165                                         /*
1166                                          * skip ".."
1167                                          */
1168                                         continue;
1169                                 } else if (!sai->sai_ls_all) {
1170                                         /*
1171                                          * skip hidden files.
1172                                          */
1173                                         sai->sai_skip_hidden++;
1174                                         continue;
1175                                 }
1176                         }
1177
1178                         /*
1179                          * don't stat-ahead first entry.
1180                          */
1181                         if (unlikely(++first == 1))
1182                                 continue;
1183
1184 keep_it:
1185                         l_wait_event(thread->t_ctl_waitq,
1186                                      !sa_sent_full(sai) ||
1187                                      !sa_received_empty(sai) ||
1188                                      !agl_list_empty(sai) ||
1189                                      !thread_is_running(thread),
1190                                      &lwi);
1191
1192 interpret_it:
1193                         while (!sa_received_empty(sai))
1194                                 do_statahead_interpret(sai, NULL);
1195
1196                         if (unlikely(!thread_is_running(thread))) {
1197                                 ll_release_page(page, 0);
1198                                 GOTO(out, rc = 0);
1199                         }
1200
1201                         /* If no window for metadata statahead, but there are
1202                          * some AGL entries to be triggered, then try to help
1203                          * to process the AGL entries. */
1204                         if (sa_sent_full(sai)) {
1205                                 cfs_spin_lock(&plli->lli_agl_lock);
1206                                 while (!agl_list_empty(sai)) {
1207                                         clli = agl_first_entry(sai);
1208                                         cfs_list_del_init(&clli->lli_agl_list);
1209                                         cfs_spin_unlock(&plli->lli_agl_lock);
1210                                         ll_agl_trigger(&clli->lli_vfs_inode,
1211                                                        sai);
1212
1213                                         if (!sa_received_empty(sai))
1214                                                 goto interpret_it;
1215
1216                                         if (unlikely(
1217                                                 !thread_is_running(thread))) {
1218                                                 ll_release_page(page, 0);
1219                                                 GOTO(out, rc = 0);
1220                                         }
1221
1222                                         if (!sa_sent_full(sai))
1223                                                 goto do_it;
1224
1225                                         cfs_spin_lock(&plli->lli_agl_lock);
1226                                 }
1227                                 cfs_spin_unlock(&plli->lli_agl_lock);
1228
1229                                 goto keep_it;
1230                         }
1231
1232 do_it:
1233                         ll_statahead_one(parent, name, namelen);
1234                 }
1235                 pos = le64_to_cpu(dp->ldp_hash_end);
1236                 if (pos == MDS_DIR_END_OFF) {
1237                         /*
1238                          * End of directory reached.
1239                          */
1240                         ll_release_page(page, 0);
1241                         while (1) {
1242                                 l_wait_event(thread->t_ctl_waitq,
1243                                              !sa_received_empty(sai) ||
1244                                              sai->sai_sent == sai->sai_replied||
1245                                              !thread_is_running(thread),
1246                                              &lwi);
1247
1248                                 while (!sa_received_empty(sai))
1249                                         do_statahead_interpret(sai, NULL);
1250
1251                                 if (unlikely(!thread_is_running(thread)))
1252                                         GOTO(out, rc = 0);
1253
1254                                 if (sai->sai_sent == sai->sai_replied &&
1255                                     sa_received_empty(sai))
1256                                         break;
1257                         }
1258
1259                         cfs_spin_lock(&plli->lli_agl_lock);
1260                         while (!agl_list_empty(sai) &&
1261                                thread_is_running(thread)) {
1262                                 clli = agl_first_entry(sai);
1263                                 cfs_list_del_init(&clli->lli_agl_list);
1264                                 cfs_spin_unlock(&plli->lli_agl_lock);
1265                                 ll_agl_trigger(&clli->lli_vfs_inode, sai);
1266                                 cfs_spin_lock(&plli->lli_agl_lock);
1267                         }
1268                         cfs_spin_unlock(&plli->lli_agl_lock);
1269
1270                         GOTO(out, rc = 0);
1271                 } else if (1) {
1272                         /*
1273                          * chain is exhausted.
1274                          * Normal case: continue to the next page.
1275                          */
1276                         ll_release_page(page, le32_to_cpu(dp->ldp_flags) &
1277                                               LDF_COLLIDE);
1278                         sai->sai_in_readpage = 1;
1279                         page = ll_get_dir_page(dir, pos, &chain);
1280                         sai->sai_in_readpage = 0;
1281                 } else {
1282                         LASSERT(le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
1283                         ll_release_page(page, 1);
1284                         /*
1285                          * go into overflow page.
1286                          */
1287                 }
1288         }
1289         EXIT;
1290
1291 out:
1292         if (sai->sai_agl_valid) {
1293                 cfs_spin_lock(&plli->lli_agl_lock);
1294                 thread_set_flags(agl_thread, SVC_STOPPING);
1295                 cfs_spin_unlock(&plli->lli_agl_lock);
1296                 cfs_waitq_signal(&agl_thread->t_ctl_waitq);
1297
1298                 CDEBUG(D_READA, "stop agl thread: [pid %d]\n",
1299                        cfs_curproc_pid());
1300                 l_wait_event(agl_thread->t_ctl_waitq,
1301                              thread_is_stopped(agl_thread),
1302                              &lwi);
1303         } else {
1304                 /* Set agl_thread flags anyway. */
1305                 thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED);
1306         }
1307         ll_dir_chain_fini(&chain);
1308         cfs_spin_lock(&plli->lli_sa_lock);
1309         if (!sa_received_empty(sai)) {
1310                 thread_set_flags(thread, SVC_STOPPING);
1311                 cfs_spin_unlock(&plli->lli_sa_lock);
1312
1313                 /* To release the resources held by received entries. */
1314                 while (!sa_received_empty(sai))
1315                         do_statahead_interpret(sai, NULL);
1316
1317                 cfs_spin_lock(&plli->lli_sa_lock);
1318         }
1319         thread_set_flags(thread, SVC_STOPPED);
1320         cfs_spin_unlock(&plli->lli_sa_lock);
1321         cfs_waitq_signal(&sai->sai_waitq);
1322         cfs_waitq_signal(&thread->t_ctl_waitq);
1323         ll_sai_put(sai);
1324         dput(parent);
1325         CDEBUG(D_READA, "statahead thread stopped: [pid %d] [parent %.*s]\n",
1326                cfs_curproc_pid(), parent->d_name.len, parent->d_name.name);
1327         return rc;
1328 }
1329
1330 /**
1331  * called in ll_file_release().
1332  */
1333 void ll_stop_statahead(struct inode *dir, void *key)
1334 {
1335         struct ll_inode_info *lli = ll_i2info(dir);
1336
1337         if (unlikely(key == NULL))
1338                 return;
1339
1340         cfs_spin_lock(&lli->lli_sa_lock);
1341         if (lli->lli_opendir_key != key || lli->lli_opendir_pid == 0) {
1342                 cfs_spin_unlock(&lli->lli_sa_lock);
1343                 return;
1344         }
1345
1346         lli->lli_opendir_key = NULL;
1347
1348         if (lli->lli_sai) {
1349                 struct l_wait_info lwi = { 0 };
1350                 struct ptlrpc_thread *thread = &lli->lli_sai->sai_thread;
1351
1352                 if (!thread_is_stopped(thread)) {
1353                         thread_set_flags(thread, SVC_STOPPING);
1354                         cfs_spin_unlock(&lli->lli_sa_lock);
1355                         cfs_waitq_signal(&thread->t_ctl_waitq);
1356
1357                         CDEBUG(D_READA, "stop statahead thread: [pid %d]\n",
1358                                cfs_curproc_pid());
1359                         l_wait_event(thread->t_ctl_waitq,
1360                                      thread_is_stopped(thread),
1361                                      &lwi);
1362                 } else {
1363                         cfs_spin_unlock(&lli->lli_sa_lock);
1364                 }
1365
1366                 /*
1367                  * Put the ref which was held when first statahead_enter.
1368                  * It maybe not the last ref for some statahead requests
1369                  * maybe inflight.
1370                  */
1371                 ll_sai_put(lli->lli_sai);
1372         } else {
1373                 lli->lli_opendir_pid = 0;
1374                 cfs_spin_unlock(&lli->lli_sa_lock);
1375         }
1376 }
1377
1378 enum {
1379         /**
1380          * not first dirent, or is "."
1381          */
1382         LS_NONE_FIRST_DE = 0,
1383         /**
1384          * the first non-hidden dirent
1385          */
1386         LS_FIRST_DE,
1387         /**
1388          * the first hidden dirent, that is "."
1389          */
1390         LS_FIRST_DOT_DE
1391 };
1392
1393 static int is_first_dirent(struct inode *dir, struct dentry *dentry)
1394 {
1395         struct ll_dir_chain   chain;
1396         struct qstr          *target = &dentry->d_name;
1397         struct page          *page;
1398         __u64                 pos    = 0;
1399         int                   dot_de;
1400         int                   rc     = LS_NONE_FIRST_DE;
1401         ENTRY;
1402
1403         ll_dir_chain_init(&chain);
1404         page = ll_get_dir_page(dir, pos, &chain);
1405
1406         while (1) {
1407                 struct lu_dirpage *dp;
1408                 struct lu_dirent  *ent;
1409
1410                 if (IS_ERR(page)) {
1411                         struct ll_inode_info *lli = ll_i2info(dir);
1412
1413                         rc = PTR_ERR(page);
1414                         CERROR("error reading dir "DFID" at "LPU64": "
1415                                "[rc %d] [parent %u]\n",
1416                                PFID(ll_inode2fid(dir)), pos,
1417                                rc, lli->lli_opendir_pid);
1418                         break;
1419                 }
1420
1421                 dp = page_address(page);
1422                 for (ent = lu_dirent_start(dp); ent != NULL;
1423                      ent = lu_dirent_next(ent)) {
1424                         __u64 hash;
1425                         int namelen;
1426                         char *name;
1427
1428                         hash = le64_to_cpu(ent->lde_hash);
1429                         /* The ll_get_dir_page() can return any page containing
1430                          * the given hash which may be not the start hash. */
1431                         if (unlikely(hash < pos))
1432                                 continue;
1433
1434                         namelen = le16_to_cpu(ent->lde_namelen);
1435                         if (unlikely(namelen == 0))
1436                                 /*
1437                                  * skip dummy record.
1438                                  */
1439                                 continue;
1440
1441                         name = ent->lde_name;
1442                         if (name[0] == '.') {
1443                                 if (namelen == 1)
1444                                         /*
1445                                          * skip "."
1446                                          */
1447                                         continue;
1448                                 else if (name[1] == '.' && namelen == 2)
1449                                         /*
1450                                          * skip ".."
1451                                          */
1452                                         continue;
1453                                 else
1454                                         dot_de = 1;
1455                         } else {
1456                                 dot_de = 0;
1457                         }
1458
1459                         if (dot_de && target->name[0] != '.') {
1460                                 CDEBUG(D_READA, "%.*s skip hidden file %.*s\n",
1461                                        target->len, target->name,
1462                                        namelen, name);
1463                                 continue;
1464                         }
1465
1466                         if (target->len != namelen ||
1467                             memcmp(target->name, name, namelen) != 0)
1468                                 rc = LS_NONE_FIRST_DE;
1469                         else if (!dot_de)
1470                                 rc = LS_FIRST_DE;
1471                         else
1472                                 rc = LS_FIRST_DOT_DE;
1473
1474                         ll_release_page(page, 0);
1475                         GOTO(out, rc);
1476                 }
1477                 pos = le64_to_cpu(dp->ldp_hash_end);
1478                 if (pos == MDS_DIR_END_OFF) {
1479                         /*
1480                          * End of directory reached.
1481                          */
1482                         ll_release_page(page, 0);
1483                         break;
1484                 } else if (1) {
1485                         /*
1486                          * chain is exhausted
1487                          * Normal case: continue to the next page.
1488                          */
1489                         ll_release_page(page, le32_to_cpu(dp->ldp_flags) &
1490                                               LDF_COLLIDE);
1491                         page = ll_get_dir_page(dir, pos, &chain);
1492                 } else {
1493                         /*
1494                          * go into overflow page.
1495                          */
1496                         LASSERT(le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
1497                         ll_release_page(page, 1);
1498                 }
1499         }
1500         EXIT;
1501
1502 out:
1503         ll_dir_chain_fini(&chain);
1504         return rc;
1505 }
1506
1507 static void
1508 ll_sai_unplug(struct ll_statahead_info *sai, struct ll_sa_entry *entry)
1509 {
1510         struct ptlrpc_thread *thread = &sai->sai_thread;
1511         struct ll_sb_info    *sbi    = ll_i2sbi(sai->sai_inode);
1512         int                   hit;
1513         ENTRY;
1514
1515         if (entry != NULL && entry->se_stat == SA_ENTRY_SUCC)
1516                 hit = 1;
1517         else
1518                 hit = 0;
1519
1520         ll_sa_entry_fini(sai, entry);
1521         if (hit) {
1522                 sai->sai_hit++;
1523                 sai->sai_consecutive_miss = 0;
1524                 sai->sai_max = min(2 * sai->sai_max, sbi->ll_sa_max);
1525         } else {
1526                 struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
1527
1528                 sai->sai_miss++;
1529                 sai->sai_consecutive_miss++;
1530                 if (sa_low_hit(sai) && thread_is_running(thread)) {
1531                         atomic_inc(&sbi->ll_sa_wrong);
1532                         CDEBUG(D_READA, "Statahead for dir "DFID" hit "
1533                                "ratio too low: hit/miss "LPU64"/"LPU64
1534                                ", sent/replied "LPU64"/"LPU64", stopping "
1535                                "statahead thread: pid %d\n",
1536                                PFID(&lli->lli_fid), sai->sai_hit,
1537                                sai->sai_miss, sai->sai_sent,
1538                                sai->sai_replied, cfs_curproc_pid());
1539                         cfs_spin_lock(&lli->lli_sa_lock);
1540                         if (!thread_is_stopped(thread))
1541                                 thread_set_flags(thread, SVC_STOPPING);
1542                         cfs_spin_unlock(&lli->lli_sa_lock);
1543                 }
1544         }
1545
1546         if (!thread_is_stopped(thread))
1547                 cfs_waitq_signal(&thread->t_ctl_waitq);
1548
1549         EXIT;
1550 }
1551
1552 /**
1553  * Start statahead thread if this is the first dir entry.
1554  * Otherwise if a thread is started already, wait it until it is ahead of me.
1555  * \retval 1       -- find entry with lock in cache, the caller needs to do
1556  *                    nothing.
1557  * \retval 0       -- find entry in cache, but without lock, the caller needs
1558  *                    refresh from MDS.
1559  * \retval others  -- the caller need to process as non-statahead.
1560  */
1561 int do_statahead_enter(struct inode *dir, struct dentry **dentryp,
1562                        int only_unplug)
1563 {
1564         struct ll_inode_info     *lli   = ll_i2info(dir);
1565         struct ll_statahead_info *sai   = lli->lli_sai;
1566         struct dentry            *parent;
1567         struct ll_sa_entry       *entry;
1568         struct ptlrpc_thread     *thread;
1569         struct l_wait_info        lwi   = { 0 };
1570         int                       rc    = 0;
1571         ENTRY;
1572
1573         LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
1574
1575         if (sai) {
1576                 thread = &sai->sai_thread;
1577                 if (unlikely(thread_is_stopped(thread) &&
1578                              cfs_list_empty(&sai->sai_entries_stated))) {
1579                         /* to release resource */
1580                         ll_stop_statahead(dir, lli->lli_opendir_key);
1581                         RETURN(-EAGAIN);
1582                 }
1583
1584                 if ((*dentryp)->d_name.name[0] == '.') {
1585                         if (sai->sai_ls_all ||
1586                             sai->sai_miss_hidden >= sai->sai_skip_hidden) {
1587                                 /*
1588                                  * Hidden dentry is the first one, or statahead
1589                                  * thread does not skip so many hidden dentries
1590                                  * before "sai_ls_all" enabled as below.
1591                                  */
1592                         } else {
1593                                 if (!sai->sai_ls_all)
1594                                         /*
1595                                          * It maybe because hidden dentry is not
1596                                          * the first one, "sai_ls_all" was not
1597                                          * set, then "ls -al" missed. Enable
1598                                          * "sai_ls_all" for such case.
1599                                          */
1600                                         sai->sai_ls_all = 1;
1601
1602                                 /*
1603                                  * Such "getattr" has been skipped before
1604                                  * "sai_ls_all" enabled as above.
1605                                  */
1606                                 sai->sai_miss_hidden++;
1607                                 RETURN(-EAGAIN);
1608                         }
1609                 }
1610
1611                 entry = ll_sa_entry_get_byname(sai, &(*dentryp)->d_name);
1612                 if (entry == NULL || only_unplug) {
1613                         ll_sai_unplug(sai, entry);
1614                         RETURN(entry ? 1 : -EAGAIN);
1615                 }
1616
1617                 while (!ll_sa_entry_stated(entry) &&
1618                        sai->sai_in_readpage &&
1619                        !sa_received_empty(sai))
1620                         do_statahead_interpret(sai, entry);
1621
1622                 if (!ll_sa_entry_stated(entry)) {
1623                         sai->sai_index_wait = entry->se_index;
1624                         lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(30), NULL,
1625                                                LWI_ON_SIGNAL_NOOP, NULL);
1626                         rc = l_wait_event(sai->sai_waitq,
1627                                           ll_sa_entry_stated(entry) ||
1628                                           thread_is_stopped(thread),
1629                                           &lwi);
1630                         if (rc < 0) {
1631                                 ll_sai_unplug(sai, entry);
1632                                 RETURN(-EAGAIN);
1633                         }
1634                 }
1635
1636                 if (entry->se_stat == SA_ENTRY_SUCC &&
1637                     entry->se_inode != NULL) {
1638                         struct inode *inode = entry->se_inode;
1639                         struct lookup_intent it = { .it_op = IT_GETATTR,
1640                                                     .d.lustre.it_lock_handle =
1641                                                      entry->se_handle };
1642                         __u64 bits;
1643
1644                         rc = md_revalidate_lock(ll_i2mdexp(dir), &it,
1645                                                 ll_inode2fid(inode), &bits);
1646                         if (rc == 1) {
1647                                 if ((*dentryp)->d_inode == NULL) {
1648                                         *dentryp = ll_splice_alias(inode,
1649                                                                    *dentryp);
1650                                 } else if ((*dentryp)->d_inode != inode) {
1651                                         /* revalidate, but inode is recreated */
1652                                         CDEBUG(D_READA,
1653                                               "stale dentry %.*s inode %lu/%u, "
1654                                               "statahead inode %lu/%u\n",
1655                                               (*dentryp)->d_name.len,
1656                                               (*dentryp)->d_name.name,
1657                                               (*dentryp)->d_inode->i_ino,
1658                                               (*dentryp)->d_inode->i_generation,
1659                                               inode->i_ino,
1660                                               inode->i_generation);
1661                                         ll_sai_unplug(sai, entry);
1662                                         RETURN(-ESTALE);
1663                                 } else {
1664                                         iput(inode);
1665                                 }
1666                                 entry->se_inode = NULL;
1667
1668                                 if ((bits & MDS_INODELOCK_LOOKUP) &&
1669                                     d_lustre_invalid(*dentryp))
1670                                         d_lustre_revalidate(*dentryp);
1671                                 ll_intent_release(&it);
1672                         }
1673                 }
1674
1675                 ll_sai_unplug(sai, entry);
1676                 RETURN(rc);
1677         }
1678
1679         /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */
1680         rc = is_first_dirent(dir, *dentryp);
1681         if (rc == LS_NONE_FIRST_DE)
1682                 /* It is not "ls -{a}l" operation, no need statahead for it. */
1683                 GOTO(out, rc = -EAGAIN);
1684
1685         sai = ll_sai_alloc();
1686         if (sai == NULL)
1687                 GOTO(out, rc = -ENOMEM);
1688
1689         sai->sai_ls_all = (rc == LS_FIRST_DOT_DE);
1690         sai->sai_inode = igrab(dir);
1691         if (unlikely(sai->sai_inode == NULL)) {
1692                 CWARN("Do not start stat ahead on dying inode "DFID"\n",
1693                       PFID(&lli->lli_fid));
1694                 GOTO(out, rc = -ESTALE);
1695         }
1696
1697         /* get parent reference count here, and put it in ll_statahead_thread */
1698         parent = dget((*dentryp)->d_parent);
1699         if (unlikely(sai->sai_inode != parent->d_inode)) {
1700                 struct ll_inode_info *nlli = ll_i2info(parent->d_inode);
1701
1702                 CWARN("Race condition, someone changed %.*s just now: "
1703                       "old parent "DFID", new parent "DFID"\n",
1704                       (*dentryp)->d_name.len, (*dentryp)->d_name.name,
1705                       PFID(&lli->lli_fid), PFID(&nlli->lli_fid));
1706                 dput(parent);
1707                 iput(sai->sai_inode);
1708                 GOTO(out, rc = -EAGAIN);
1709         }
1710
1711         CDEBUG(D_READA, "start statahead thread: [pid %d] [parent %.*s]\n",
1712                cfs_curproc_pid(), parent->d_name.len, parent->d_name.name);
1713
1714         lli->lli_sai = sai;
1715         rc = cfs_create_thread(ll_statahead_thread, parent, 0);
1716         thread = &sai->sai_thread;
1717         if (rc < 0) {
1718                 CERROR("can't start ll_sa thread, rc: %d\n", rc);
1719                 dput(parent);
1720                 lli->lli_opendir_key = NULL;
1721                 thread_set_flags(thread, SVC_STOPPED);
1722                 thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED);
1723                 ll_sai_put(sai);
1724                 LASSERT(lli->lli_sai == NULL);
1725                 RETURN(-EAGAIN);
1726         }
1727
1728         l_wait_event(thread->t_ctl_waitq,
1729                      thread_is_running(thread) || thread_is_stopped(thread),
1730                      &lwi);
1731
1732         /*
1733          * We don't stat-ahead for the first dirent since we are already in
1734          * lookup.
1735          */
1736         RETURN(-EAGAIN);
1737
1738 out:
1739         if (sai != NULL)
1740                 OBD_FREE_PTR(sai);
1741         cfs_spin_lock(&lli->lli_sa_lock);
1742         lli->lli_opendir_key = NULL;
1743         lli->lli_opendir_pid = 0;
1744         cfs_spin_unlock(&lli->lli_sa_lock);
1745         return rc;
1746 }