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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  */
  31
  32 #include <linux/fs.h>
  33 #include <linux/sched.h>
  34 #include <linux/kthread.h>
  35 #include <linux/mm.h>
  36 #include <linux/highmem.h>
  37 #include <linux/pagemap.h>
  38 #include <linux/delay.h>
  39
  40 #define DEBUG_SUBSYSTEM S_LLITE
  41
  42 #include <obd_support.h>
  43 #include <lustre_dlm.h>
  44 #include "llite_internal.h"
  45
  46 #define SA_OMITTED_ENTRY_MAX 8ULL
  47
  48 typedef enum {
  49         /** negative values are for error cases */
  50         SA_ENTRY_INIT = 0,      /** init entry */
  51         SA_ENTRY_SUCC = 1,      /** stat succeed */
  52         SA_ENTRY_INVA = 2,      /** invalid entry */
  53 } se_state_t;
  54
  55 /*
  56  * sa_entry is not refcounted: statahead thread allocates it and do async stat,
  57  * and in async stat callback ll_statahead_interpret() will prepare the inode
  58  * and set lock data in the ptlrpcd context. Then the scanner process will be
  59  * woken up if this entry is the waiting one, can access and free it.
  60  */
  61 struct sa_entry {
  62         /* link into sai_entries */
  63         struct list_head        se_list;
  64         /* link into sai hash table locally */
  65         struct list_head        se_hash;
  66         /* entry index in the sai */
  67         __u64                   se_index;
  68         /* low layer ldlm lock handle */
  69         __u64                   se_handle;
  70         /* entry status */
  71         se_state_t              se_state;
  72         /* entry size, contains name */
  73         int                     se_size;
  74         /* pointer to the target inode */
  75         struct inode           *se_inode;
  76         /* entry name */
  77         struct qstr             se_qstr;
  78         /* entry fid */
  79         struct lu_fid           se_fid;
  80 };
  81
  82 static unsigned int sai_generation;
  83 static DEFINE_SPINLOCK(sai_generation_lock);
  84
  85 static inline int sa_unhashed(struct sa_entry *entry)
  86 {
  87         return list_empty(&entry->se_hash);
  88 }
  89
  90 /* sa_entry is ready to use */
  91 static inline int sa_ready(struct sa_entry *entry)
  92 {
  93         /* Make sure sa_entry is updated and ready to use */
  94         smp_rmb();
  95         return (entry->se_state != SA_ENTRY_INIT);
  96 }
  97
  98 /* hash value to put in sai_cache */
  99 static inline int sa_hash(int val)
 100 {
 101         return val & LL_SA_CACHE_MASK;
 102 }
 103
 104 /* hash entry into sai_cache */
 105 static inline void
 106 sa_rehash(struct ll_statahead_info *sai, struct sa_entry *entry)
 107 {
 108         int i = sa_hash(entry->se_qstr.hash);
 109
 110         spin_lock(&sai->sai_cache_lock[i]);
 111         list_add_tail(&entry->se_hash, &sai->sai_cache[i]);
 112         spin_unlock(&sai->sai_cache_lock[i]);
 113 }
 114
 115 /* unhash entry from sai_cache */
 116 static inline void
 117 sa_unhash(struct ll_statahead_info *sai, struct sa_entry *entry)
 118 {
 119         int i = sa_hash(entry->se_qstr.hash);
 120
 121         spin_lock(&sai->sai_cache_lock[i]);
 122         list_del_init(&entry->se_hash);
 123         spin_unlock(&sai->sai_cache_lock[i]);
 124 }
 125
 126 static inline int agl_should_run(struct ll_statahead_info *sai,
 127                                  struct inode *inode)
 128 {
 129         return inode && S_ISREG(inode->i_mode) && sai->sai_agl_task;
 130 }
 131
 132 static inline struct ll_inode_info *
 133 agl_first_entry(struct ll_statahead_info *sai)
 134 {
 135         return list_first_entry(&sai->sai_agls, struct ll_inode_info,
 136                                 lli_agl_list);
 137 }
 138
 139 /* statahead window is full */
 140 static inline int sa_sent_full(struct ll_statahead_info *sai)
 141 {
 142         return atomic_read(&sai->sai_cache_count) >= sai->sai_max;
 143 }
 144
 145 static inline int agl_list_empty(struct ll_statahead_info *sai)
 146 {
 147         return list_empty(&sai->sai_agls);
 148 }
 149
 150 /**
 151  * (1) hit ratio less than 80%
 152  * or
 153  * (2) consecutive miss more than 8
 154  * then means low hit.
 155  */
 156 static inline int sa_low_hit(struct ll_statahead_info *sai)
 157 {
 158         return ((sai->sai_hit > 7 && sai->sai_hit < 4 * sai->sai_miss) ||
 159                 (sai->sai_consecutive_miss > 8));
 160 }
 161
 162 /*
 163  * if the given index is behind of statahead window more than
 164  * SA_OMITTED_ENTRY_MAX, then it is old.
 165  */
 166 static inline int is_omitted_entry(struct ll_statahead_info *sai, __u64 index)
 167 {
 168         return ((__u64)sai->sai_max + index + SA_OMITTED_ENTRY_MAX <
 169                 sai->sai_index);
 170 }
 171
 172 /* allocate sa_entry and hash it to allow scanner process to find it */
 173 static struct sa_entry *
 174 sa_alloc(struct dentry *parent, struct ll_statahead_info *sai, __u64 index,
 175          const char *name, int len, const struct lu_fid *fid)
 176 {
 177         struct ll_inode_info *lli;
 178         struct sa_entry *entry;
 179         int entry_size;
 180         char *dname;
 181
 182         ENTRY;
 183
 184         entry_size = sizeof(struct sa_entry) + (len & ~3) + 4;
 185         OBD_ALLOC(entry, entry_size);
 186         if (unlikely(!entry))
 187                 RETURN(ERR_PTR(-ENOMEM));
 188
 189         CDEBUG(D_READA, "alloc sa entry %.*s(%p) index %llu\n",
 190                len, name, entry, index);
 191
 192         entry->se_index = index;
 193
 194         entry->se_state = SA_ENTRY_INIT;
 195         entry->se_size = entry_size;
 196         dname = (char *)entry + sizeof(struct sa_entry);
 197         memcpy(dname, name, len);
 198         dname[len] = 0;
 199         entry->se_qstr.hash = ll_full_name_hash(parent, name, len);
 200         entry->se_qstr.len = len;
 201         entry->se_qstr.name = dname;
 202         entry->se_fid = *fid;
 203
 204         lli = ll_i2info(sai->sai_dentry->d_inode);
 205
 206         spin_lock(&lli->lli_sa_lock);
 207         INIT_LIST_HEAD(&entry->se_list);
 208         sa_rehash(sai, entry);
 209         spin_unlock(&lli->lli_sa_lock);
 210
 211         atomic_inc(&sai->sai_cache_count);
 212
 213         RETURN(entry);
 214 }
 215
 216 /* free sa_entry, which should have been unhashed and not in any list */
 217 static void sa_free(struct ll_statahead_info *sai, struct sa_entry *entry)
 218 {
 219         CDEBUG(D_READA, "free sa entry %.*s(%p) index %llu\n",
 220                entry->se_qstr.len, entry->se_qstr.name, entry,
 221                entry->se_index);
 222
 223         LASSERT(list_empty(&entry->se_list));
 224         LASSERT(sa_unhashed(entry));
 225
 226         OBD_FREE(entry, entry->se_size);
 227         atomic_dec(&sai->sai_cache_count);
 228 }
 229
 230 /*
 231  * find sa_entry by name, used by directory scanner, lock is not needed because
 232  * only scanner can remove the entry from cache.
 233  */
 234 static struct sa_entry *
 235 sa_get(struct ll_statahead_info *sai, const struct qstr *qstr)
 236 {
 237         struct sa_entry *entry;
 238         int i = sa_hash(qstr->hash);
 239
 240         list_for_each_entry(entry, &sai->sai_cache[i], se_hash) {
 241                 if (entry->se_qstr.hash == qstr->hash &&
 242                     entry->se_qstr.len == qstr->len &&
 243                     memcmp(entry->se_qstr.name, qstr->name, qstr->len) == 0)
 244                         return entry;
 245         }
 246         return NULL;
 247 }
 248
 249 /* unhash and unlink sa_entry, and then free it */
 250 static inline void
 251 sa_kill(struct ll_statahead_info *sai, struct sa_entry *entry)
 252 {
 253         struct ll_inode_info *lli = ll_i2info(sai->sai_dentry->d_inode);
 254
 255         LASSERT(!sa_unhashed(entry));
 256         LASSERT(!list_empty(&entry->se_list));
 257         LASSERT(sa_ready(entry));
 258
 259         sa_unhash(sai, entry);
 260
 261         spin_lock(&lli->lli_sa_lock);
 262         list_del_init(&entry->se_list);
 263         spin_unlock(&lli->lli_sa_lock);
 264
 265         iput(entry->se_inode);
 266
 267         sa_free(sai, entry);
 268 }
 269
 270 /* called by scanner after use, sa_entry will be killed */
 271 static void
 272 sa_put(struct ll_statahead_info *sai, struct sa_entry *entry)
 273 {
 274         struct sa_entry *tmp, *next;
 275
 276         if (entry && entry->se_state == SA_ENTRY_SUCC) {
 277                 struct ll_sb_info *sbi = ll_i2sbi(sai->sai_dentry->d_inode);
 278
 279                 sai->sai_hit++;
 280                 sai->sai_consecutive_miss = 0;
 281                 sai->sai_max = min(2 * sai->sai_max, sbi->ll_sa_max);
 282         } else {
 283                 sai->sai_miss++;
 284                 sai->sai_consecutive_miss++;
 285         }
 286
 287         if (entry)
 288                 sa_kill(sai, entry);
 289
 290         /*
 291          * kill old completed entries, only scanner process does this, no need
 292          * to lock
 293          */
 294         list_for_each_entry_safe(tmp, next, &sai->sai_entries, se_list) {
 295                 if (!is_omitted_entry(sai, tmp->se_index))
 296                         break;
 297                 sa_kill(sai, tmp);
 298         }
 299 }
 300
 301 /*
 302  * update state and sort add entry to sai_entries by index, return true if
 303  * scanner is waiting on this entry.
 304  */
 305 static bool
 306 __sa_make_ready(struct ll_statahead_info *sai, struct sa_entry *entry, int ret)
 307 {
 308         struct sa_entry *se;
 309         struct list_head *pos = &sai->sai_entries;
 310         __u64 index = entry->se_index;
 311
 312         LASSERT(!sa_ready(entry));
 313         LASSERT(list_empty(&entry->se_list));
 314
 315         list_for_each_entry_reverse(se, &sai->sai_entries, se_list) {
 316                 if (se->se_index < entry->se_index) {
 317                         pos = &se->se_list;
 318                         break;
 319                 }
 320         }
 321         list_add(&entry->se_list, pos);
 322         /*
 323          * LU-9210: ll_statahead_interpet must be able to see this before
 324          * we wake it up
 325          */
 326         smp_store_release(&entry->se_state,
 327                           ret < 0 ? SA_ENTRY_INVA : SA_ENTRY_SUCC);
 328
 329         return (index == sai->sai_index_wait);
 330 }
 331
 332 /* finish async stat RPC arguments */
 333 static void sa_fini_data(struct md_op_item *item)
 334 {
 335         struct md_op_data *op_data = &item->mop_data;
 336
 337         if (op_data->op_flags & MF_OPNAME_KMALLOCED)
 338                 /* allocated via ll_setup_filename called from sa_prep_data */
 339                 kfree(op_data->op_name);
 340         ll_unlock_md_op_lsm(&item->mop_data);
 341         iput(item->mop_dir);
 342         OBD_FREE_PTR(item);
 343 }
 344
 345 static int ll_statahead_interpret(struct req_capsule *pill,
 346                                   struct md_op_item *item, int rc);
 347
 348 /*
 349  * prepare arguments for async stat RPC.
 350  */
 351 static struct md_op_item *
 352 sa_prep_data(struct inode *dir, struct inode *child, struct sa_entry *entry)
 353 {
 354         struct md_op_item *item;
 355         struct ldlm_enqueue_info *einfo;
 356         struct md_op_data *op_data;
 357
 358         OBD_ALLOC_PTR(item);
 359         if (!item)
 360                 return ERR_PTR(-ENOMEM);
 361
 362         op_data = ll_prep_md_op_data(&item->mop_data, dir, child,
 363                                      entry->se_qstr.name, entry->se_qstr.len, 0,
 364                                      LUSTRE_OPC_ANY, NULL);
 365         if (IS_ERR(op_data)) {
 366                 OBD_FREE_PTR(item);
 367                 return (struct md_op_item *)op_data;
 368         }
 369
 370         if (!child)
 371                 op_data->op_fid2 = entry->se_fid;
 372
 373         item->mop_it.it_op = IT_GETATTR;
 374         item->mop_dir = igrab(dir);
 375         item->mop_cb = ll_statahead_interpret;
 376         item->mop_cbdata = entry;
 377
 378         einfo = &item->mop_einfo;
 379         einfo->ei_type = LDLM_IBITS;
 380         einfo->ei_mode = it_to_lock_mode(&item->mop_it);
 381         einfo->ei_cb_bl = ll_md_blocking_ast;
 382         einfo->ei_cb_cp = ldlm_completion_ast;
 383         einfo->ei_cb_gl = NULL;
 384         einfo->ei_cbdata = NULL;
 385         einfo->ei_req_slot = 1;
 386
 387         return item;
 388 }
 389
 390 /*
 391  * release resources used in async stat RPC, update entry state and wakeup if
 392  * scanner process it waiting on this entry.
 393  */
 394 static void
 395 sa_make_ready(struct ll_statahead_info *sai, struct sa_entry *entry, int ret)
 396 {
 397         struct ll_inode_info *lli = ll_i2info(sai->sai_dentry->d_inode);
 398         bool wakeup;
 399
 400         spin_lock(&lli->lli_sa_lock);
 401         wakeup = __sa_make_ready(sai, entry, ret);
 402         spin_unlock(&lli->lli_sa_lock);
 403
 404         if (wakeup)
 405                 wake_up(&sai->sai_waitq);
 406 }
 407
 408 /* insert inode into the list of sai_agls */
 409 static void ll_agl_add(struct ll_statahead_info *sai,
 410                        struct inode *inode, int index)
 411 {
 412         struct ll_inode_info *child  = ll_i2info(inode);
 413         struct ll_inode_info *parent = ll_i2info(sai->sai_dentry->d_inode);
 414
 415         spin_lock(&child->lli_agl_lock);
 416         if (child->lli_agl_index == 0) {
 417                 child->lli_agl_index = index;
 418                 spin_unlock(&child->lli_agl_lock);
 419
 420                 LASSERT(list_empty(&child->lli_agl_list));
 421
 422                 spin_lock(&parent->lli_agl_lock);
 423                 /* Re-check under the lock */
 424                 if (agl_should_run(sai, inode)) {
 425                         if (agl_list_empty(sai))
 426                                 wake_up_process(sai->sai_agl_task);
 427                         igrab(inode);
 428                         list_add_tail(&child->lli_agl_list, &sai->sai_agls);
 429                 } else
 430                         child->lli_agl_index = 0;
 431                 spin_unlock(&parent->lli_agl_lock);
 432         } else {
 433                 spin_unlock(&child->lli_agl_lock);
 434         }
 435 }
 436
 437 /* allocate sai */
 438 static struct ll_statahead_info *ll_sai_alloc(struct dentry *dentry)
 439 {
 440         struct ll_statahead_info *sai;
 441         struct ll_inode_info *lli = ll_i2info(dentry->d_inode);
 442         int i;
 443
 444         ENTRY;
 445
 446         OBD_ALLOC_PTR(sai);
 447         if (!sai)
 448                 RETURN(NULL);
 449
 450         sai->sai_dentry = dget(dentry);
 451         atomic_set(&sai->sai_refcount, 1);
 452         sai->sai_max = LL_SA_RPC_MIN;
 453         sai->sai_index = 1;
 454         init_waitqueue_head(&sai->sai_waitq);
 455
 456         INIT_LIST_HEAD(&sai->sai_entries);
 457         INIT_LIST_HEAD(&sai->sai_agls);
 458
 459         for (i = 0; i < LL_SA_CACHE_SIZE; i++) {
 460                 INIT_LIST_HEAD(&sai->sai_cache[i]);
 461                 spin_lock_init(&sai->sai_cache_lock[i]);
 462         }
 463         atomic_set(&sai->sai_cache_count, 0);
 464
 465         spin_lock(&sai_generation_lock);
 466         lli->lli_sa_generation = ++sai_generation;
 467         if (unlikely(sai_generation == 0))
 468                 lli->lli_sa_generation = ++sai_generation;
 469         spin_unlock(&sai_generation_lock);
 470
 471         RETURN(sai);
 472 }
 473
 474 /* free sai */
 475 static inline void ll_sai_free(struct ll_statahead_info *sai)
 476 {
 477         LASSERT(sai->sai_dentry != NULL);
 478         dput(sai->sai_dentry);
 479         OBD_FREE_PTR(sai);
 480 }
 481
 482 /*
 483  * take refcount of sai if sai for @dir exists, which means statahead is on for
 484  * this directory.
 485  */
 486 static inline struct ll_statahead_info *ll_sai_get(struct inode *dir)
 487 {
 488         struct ll_inode_info *lli = ll_i2info(dir);
 489         struct ll_statahead_info *sai = NULL;
 490
 491         spin_lock(&lli->lli_sa_lock);
 492         sai = lli->lli_sai;
 493         if (sai)
 494                 atomic_inc(&sai->sai_refcount);
 495         spin_unlock(&lli->lli_sa_lock);
 496
 497         return sai;
 498 }
 499
 500 /*
 501  * put sai refcount after use, if refcount reaches zero, free sai and sa_entries
 502  * attached to it.
 503  */
 504 static void ll_sai_put(struct ll_statahead_info *sai)
 505 {
 506         struct ll_inode_info *lli = ll_i2info(sai->sai_dentry->d_inode);
 507
 508         if (atomic_dec_and_lock(&sai->sai_refcount, &lli->lli_sa_lock)) {
 509                 struct sa_entry *entry, *next;
 510                 struct ll_sb_info *sbi = ll_i2sbi(sai->sai_dentry->d_inode);
 511
 512                 lli->lli_sai = NULL;
 513                 spin_unlock(&lli->lli_sa_lock);
 514
 515                 LASSERT(!sai->sai_task);
 516                 LASSERT(!sai->sai_agl_task);
 517                 LASSERT(sai->sai_sent == sai->sai_replied);
 518
 519                 list_for_each_entry_safe(entry, next, &sai->sai_entries,
 520                                          se_list)
 521                         sa_kill(sai, entry);
 522
 523                 LASSERT(atomic_read(&sai->sai_cache_count) == 0);
 524                 LASSERT(agl_list_empty(sai));
 525
 526                 ll_sai_free(sai);
 527                 atomic_dec(&sbi->ll_sa_running);
 528         }
 529 }
 530
 531 /* Do NOT forget to drop inode refcount when into sai_agls. */
 532 static void ll_agl_trigger(struct inode *inode, struct ll_statahead_info *sai)
 533 {
 534         struct ll_inode_info *lli = ll_i2info(inode);
 535         u64 index = lli->lli_agl_index;
 536         ktime_t expire;
 537         int rc;
 538
 539         ENTRY;
 540
 541         LASSERT(list_empty(&lli->lli_agl_list));
 542
 543         /* AGL maybe fall behind statahead with one entry */
 544         if (is_omitted_entry(sai, index + 1)) {
 545                 lli->lli_agl_index = 0;
 546                 iput(inode);
 547                 RETURN_EXIT;
 548         }
 549
 550         /*
 551          * In case of restore, the MDT has the right size and has already
 552          * sent it back without granting the layout lock, inode is up-to-date.
 553          * Then AGL (async glimpse lock) is useless.
 554          * Also to glimpse we need the layout, in case of a runninh restore
 555          * the MDT holds the layout lock so the glimpse will block up to the
 556          * end of restore (statahead/agl will block)
 557          */
 558         if (test_bit(LLIF_FILE_RESTORING, &lli->lli_flags)) {
 559                 lli->lli_agl_index = 0;
 560                 iput(inode);
 561                 RETURN_EXIT;
 562         }
 563
 564         /* Someone is in glimpse (sync or async), do nothing. */
 565         rc = down_write_trylock(&lli->lli_glimpse_sem);
 566         if (rc == 0) {
 567                 lli->lli_agl_index = 0;
 568                 iput(inode);
 569                 RETURN_EXIT;
 570         }
 571
 572         /*
 573          * Someone triggered glimpse within 1 sec before.
 574          * 1) The former glimpse succeeded with glimpse lock granted by OST, and
 575          *    if the lock is still cached on client, AGL needs to do nothing. If
 576          *    it is cancelled by other client, AGL maybe cannot obtaion new lock
 577          *    for no glimpse callback triggered by AGL.
 578          * 2) The former glimpse succeeded, but OST did not grant glimpse lock.
 579          *    Under such case, it is quite possible that the OST will not grant
 580          *    glimpse lock for AGL also.
 581          * 3) The former glimpse failed, compared with other two cases, it is
 582          *    relative rare. AGL can ignore such case, and it will not muchly
 583          *    affect the performance.
 584          */
 585         expire = ktime_sub_ns(ktime_get(), NSEC_PER_SEC);
 586         if (ktime_to_ns(lli->lli_glimpse_time) &&
 587             ktime_before(expire, lli->lli_glimpse_time)) {
 588                 up_write(&lli->lli_glimpse_sem);
 589                 lli->lli_agl_index = 0;
 590                 iput(inode);
 591                 RETURN_EXIT;
 592         }
 593
 594         CDEBUG(D_READA,
 595                "Handling (init) async glimpse: inode = " DFID", idx = %llu\n",
 596                PFID(&lli->lli_fid), index);
 597
 598         cl_agl(inode);
 599         lli->lli_agl_index = 0;
 600         lli->lli_glimpse_time = ktime_get();
 601         up_write(&lli->lli_glimpse_sem);
 602
 603         CDEBUG(D_READA,
 604                "Handled (init) async glimpse: inode= " DFID", idx = %llu, rc = %d\n",
 605                PFID(&lli->lli_fid), index, rc);
 606
 607         iput(inode);
 608
 609         EXIT;
 610 }
 611
 612 static int ll_statahead_interpret_common(struct inode *dir,
 613                                          struct ll_statahead_info *sai,
 614                                          struct req_capsule *pill,
 615                                          struct lookup_intent *it,
 616                                          struct sa_entry *entry,
 617                                          struct mdt_body *body)
 618 {
 619         struct inode *child;
 620         int rc;
 621
 622         ENTRY;
 623
 624         child = entry->se_inode;
 625         rc = ll_prep_inode(&child, pill, dir->i_sb, it);
 626         if (rc)
 627                 GOTO(out, rc);
 628
 629         /* If encryption context was returned by MDT, put it in
 630          * inode now to save an extra getxattr.
 631          */
 632         if (body->mbo_valid & OBD_MD_ENCCTX) {
 633                 void *encctx = req_capsule_server_get(pill, &RMF_FILE_ENCCTX);
 634                 __u32 encctxlen = req_capsule_get_size(pill, &RMF_FILE_ENCCTX,
 635                                                        RCL_SERVER);
 636
 637                 if (encctxlen) {
 638                         CDEBUG(D_SEC,
 639                                "server returned encryption ctx for "DFID"\n",
 640                                PFID(ll_inode2fid(child)));
 641                         rc = ll_xattr_cache_insert(child,
 642                                                    xattr_for_enc(child),
 643                                                    encctx, encctxlen);
 644                         if (rc)
 645                                 CWARN("%s: cannot set enc ctx for "DFID": rc = %d\n",
 646                                       ll_i2sbi(child)->ll_fsname,
 647                                       PFID(ll_inode2fid(child)), rc);
 648                 }
 649         }
 650
 651         CDEBUG(D_READA, "%s: setting %.*s"DFID" l_data to inode %p\n",
 652                ll_i2sbi(dir)->ll_fsname, entry->se_qstr.len,
 653                entry->se_qstr.name, PFID(ll_inode2fid(child)), child);
 654         ll_set_lock_data(ll_i2sbi(dir)->ll_md_exp, child, it, NULL);
 655
 656         entry->se_inode = child;
 657
 658         if (agl_should_run(sai, child))
 659                 ll_agl_add(sai, child, entry->se_index);
 660
 661 out:
 662         RETURN(rc);
 663 }
 664
 665 static void ll_statahead_interpret_fini(struct ll_inode_info *lli,
 666                                         struct ll_statahead_info *sai,
 667                                         struct md_op_item *item,
 668                                         struct sa_entry *entry,
 669                                         struct ptlrpc_request *req,
 670                                         int rc)
 671 {
 672         /*
 673          * First it will drop ldlm ibits lock refcount by calling
 674          * ll_intent_drop_lock() in spite of failures. Do not worry about
 675          * calling ll_intent_drop_lock() more than once.
 676          */
 677         ll_intent_release(&item->mop_it);
 678         sa_fini_data(item);
 679         if (req)
 680                 ptlrpc_req_finished(req);
 681         sa_make_ready(sai, entry, rc);
 682
 683         spin_lock(&lli->lli_sa_lock);
 684         sai->sai_replied++;
 685         spin_unlock(&lli->lli_sa_lock);
 686 }
 687
 688 static void ll_statahead_interpret_work(struct work_struct *data)
 689 {
 690         struct ll_interpret_work *work = container_of(data,
 691                                         struct ll_interpret_work, lpw_work);
 692         struct md_op_item *item = work->lpw_item;
 693         struct req_capsule *pill = work->lpw_pill;
 694         struct inode *dir = item->mop_dir;
 695         struct ll_inode_info *lli = ll_i2info(dir);
 696         struct ll_statahead_info *sai = lli->lli_sai;
 697         struct lookup_intent *it;
 698         struct sa_entry *entry;
 699         struct mdt_body *body;
 700         struct inode *child;
 701         int rc;
 702
 703         ENTRY;
 704
 705         entry = (struct sa_entry *)item->mop_cbdata;
 706         LASSERT(entry->se_handle != 0);
 707
 708         it = &item->mop_it;
 709         body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 710         if (!body)
 711                 GOTO(out, rc = -EFAULT);
 712
 713         child = entry->se_inode;
 714         /* revalidate; unlinked and re-created with the same name */
 715         if (unlikely(!lu_fid_eq(&item->mop_data.op_fid2, &body->mbo_fid1))) {
 716                 if (child) {
 717                         entry->se_inode = NULL;
 718                         iput(child);
 719                 }
 720                 /* The mdt_body is invalid. Skip this entry */
 721                 GOTO(out, rc = -EAGAIN);
 722         }
 723
 724         it->it_lock_handle = entry->se_handle;
 725         rc = md_revalidate_lock(ll_i2mdexp(dir), it, ll_inode2fid(dir), NULL);
 726         if (rc != 1)
 727                 GOTO(out, rc = -EAGAIN);
 728
 729         LASSERT(it->it_extra_rpc_check == 0);
 730         rc = ll_statahead_interpret_common(dir, sai, pill, it, entry, body);
 731 out:
 732         ll_statahead_interpret_fini(lli, sai, item, entry, pill->rc_req, rc);
 733         OBD_FREE_PTR(work);
 734 }
 735
 736 /*
 737  * Callback for async stat RPC, this is called in ptlrpcd context. It prepares
 738  * the inode and set lock data directly in the ptlrpcd context. It will wake up
 739  * the directory listing process if the dentry is the waiting one.
 740  */
 741 static int ll_statahead_interpret(struct req_capsule *pill,
 742                                   struct md_op_item *item, int rc)
 743 {
 744         struct lookup_intent *it = &item->mop_it;
 745         struct inode *dir = item->mop_dir;
 746         struct ll_inode_info *lli = ll_i2info(dir);
 747         struct ll_statahead_info *sai = lli->lli_sai;
 748         struct sa_entry *entry = (struct sa_entry *)item->mop_cbdata;
 749         struct mdt_body *body;
 750         struct inode *child;
 751         __u64 handle = 0;
 752
 753         ENTRY;
 754
 755         if (it_disposition(it, DISP_LOOKUP_NEG))
 756                 rc = -ENOENT;
 757
 758         /*
 759          * because statahead thread will wait for all inflight RPC to finish,
 760          * sai should be always valid, no need to refcount
 761          */
 762         LASSERT(sai != NULL);
 763         LASSERT(entry != NULL);
 764
 765         CDEBUG(D_READA, "sa_entry %.*s rc %d\n",
 766                entry->se_qstr.len, entry->se_qstr.name, rc);
 767
 768         if (rc != 0)
 769                 GOTO(out, rc);
 770
 771         body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 772         if (!body)
 773                 GOTO(out, rc = -EFAULT);
 774
 775         child = entry->se_inode;
 776         /* revalidate; unlinked and re-created with the same name */
 777         if (unlikely(!lu_fid_eq(&item->mop_data.op_fid2, &body->mbo_fid1))) {
 778                 if (child) {
 779                         entry->se_inode = NULL;
 780                         iput(child);
 781                 }
 782                 /* The mdt_body is invalid. Skip this entry */
 783                 GOTO(out, rc = -EAGAIN);
 784         }
 785
 786         entry->se_handle = it->it_lock_handle;
 787         /*
 788          * In ptlrpcd context, it is not allowed to generate new RPCs
 789          * especially for striped directories.
 790          */
 791         it->it_extra_rpc_check = 1;
 792         rc = ll_statahead_interpret_common(dir, sai, pill, it, entry, body);
 793         if (rc == -EAGAIN && it->it_extra_rpc_need) {
 794                 struct ll_interpret_work *work;
 795
 796                 /*
 797                  * release ibits lock ASAP to avoid deadlock when statahead
 798                  * thread enqueues lock on parent in readdir and another
 799                  * process enqueues lock on child with parent lock held, eg.
 800                  * unlink.
 801                  */
 802                 handle = it->it_lock_handle;
 803                 ll_intent_drop_lock(it);
 804                 ll_unlock_md_op_lsm(&item->mop_data);
 805                 it->it_extra_rpc_check = 0;
 806                 it->it_extra_rpc_need = 0;
 807
 808                 /*
 809                  * If the stat-ahead entry is a striped directory, there are two
 810                  * solutions:
 811                  * 1. It can drop the result, let the scanning process do stat()
 812                  * on the striped directory in synchronous way. By this way, it
 813                  * can avoid to generate new RPCs to obtain the attributes for
 814                  * slaves of the striped directory in the ptlrpcd context as it
 815                  * is dangerous of blocking in ptlrpcd thread.
 816                  * 2. Use work queue or the separate statahead thread to handle
 817                  * the extra RPCs (@ll_prep_inode->@lmv_revalidate_slaves).
 818                  * Here we adopt the second solution.
 819                  */
 820                 OBD_ALLOC_GFP(work, sizeof(*work), GFP_ATOMIC);
 821                 if (work == NULL)
 822                         GOTO(out, rc = -ENOMEM);
 823
 824                 INIT_WORK(&work->lpw_work, ll_statahead_interpret_work);
 825                 work->lpw_item = item;
 826                 work->lpw_pill = pill;
 827                 ptlrpc_request_addref(pill->rc_req);
 828                 schedule_work(&work->lpw_work);
 829                 RETURN(0);
 830         }
 831
 832 out:
 833         ll_statahead_interpret_fini(lli, sai, item, entry, NULL, rc);
 834         RETURN(rc);
 835 }
 836
 837 /* async stat for file not found in dcache */
 838 static int sa_lookup(struct inode *dir, struct sa_entry *entry)
 839 {
 840         struct md_op_item *item;
 841         int rc;
 842
 843         ENTRY;
 844
 845         item = sa_prep_data(dir, NULL, entry);
 846         if (IS_ERR(item))
 847                 RETURN(PTR_ERR(item));
 848
 849         rc = md_intent_getattr_async(ll_i2mdexp(dir), item);
 850         if (rc < 0)
 851                 sa_fini_data(item);
 852
 853         RETURN(rc);
 854 }
 855
 856 /**
 857  * async stat for file found in dcache, similar to .revalidate
 858  *
 859  * \retval      1 dentry valid, no RPC sent
 860  * \retval      0 dentry invalid, will send async stat RPC
 861  * \retval      negative number upon error
 862  */
 863 static int sa_revalidate(struct inode *dir, struct sa_entry *entry,
 864                          struct dentry *dentry)
 865 {
 866         struct inode *inode = dentry->d_inode;
 867         struct lookup_intent it = { .it_op = IT_GETATTR,
 868                                     .it_lock_handle = 0 };
 869         struct md_op_item *item;
 870         int rc;
 871
 872         ENTRY;
 873
 874         if (unlikely(!inode))
 875                 RETURN(1);
 876
 877         if (d_mountpoint(dentry))
 878                 RETURN(1);
 879
 880         item = sa_prep_data(dir, inode, entry);
 881         if (IS_ERR(item))
 882                 RETURN(PTR_ERR(item));
 883
 884         entry->se_inode = igrab(inode);
 885         rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode),
 886                                 NULL);
 887         if (rc == 1) {
 888                 entry->se_handle = it.it_lock_handle;
 889                 ll_intent_release(&it);
 890                 sa_fini_data(item);
 891                 RETURN(1);
 892         }
 893
 894         rc = md_intent_getattr_async(ll_i2mdexp(dir), item);
 895         if (rc < 0) {
 896                 entry->se_inode = NULL;
 897                 iput(inode);
 898                 sa_fini_data(item);
 899         }
 900
 901         RETURN(rc);
 902 }
 903
 904 /* async stat for file with @name */
 905 static void sa_statahead(struct dentry *parent, const char *name, int len,
 906                          const struct lu_fid *fid)
 907 {
 908         struct inode *dir = parent->d_inode;
 909         struct ll_inode_info *lli = ll_i2info(dir);
 910         struct ll_statahead_info *sai = lli->lli_sai;
 911         struct dentry *dentry = NULL;
 912         struct sa_entry *entry;
 913         int rc;
 914
 915         ENTRY;
 916
 917         entry = sa_alloc(parent, sai, sai->sai_index, name, len, fid);
 918         if (IS_ERR(entry))
 919                 RETURN_EXIT;
 920
 921         dentry = d_lookup(parent, &entry->se_qstr);
 922         if (!dentry) {
 923                 rc = sa_lookup(dir, entry);
 924         } else {
 925                 rc = sa_revalidate(dir, entry, dentry);
 926                 if (rc == 1 && agl_should_run(sai, dentry->d_inode))
 927                         ll_agl_add(sai, dentry->d_inode, entry->se_index);
 928         }
 929
 930         if (dentry)
 931                 dput(dentry);
 932
 933         if (rc != 0)
 934                 sa_make_ready(sai, entry, rc);
 935         else
 936                 sai->sai_sent++;
 937
 938         sai->sai_index++;
 939
 940         EXIT;
 941 }
 942
 943 /* async glimpse (agl) thread main function */
 944 static int ll_agl_thread(void *arg)
 945 {
 946         struct dentry *parent = (struct dentry *)arg;
 947         struct inode *dir = parent->d_inode;
 948         struct ll_inode_info *plli = ll_i2info(dir);
 949         struct ll_inode_info *clli;
 950         /*
 951          * We already own this reference, so it is safe to take it
 952          * without a lock.
 953          */
 954         struct ll_statahead_info *sai = plli->lli_sai;
 955
 956         ENTRY;
 957
 958         CDEBUG(D_READA, "agl thread started: sai %p, parent %pd\n",
 959                sai, parent);
 960
 961         while (({set_current_state(TASK_IDLE);
 962                  !kthread_should_stop(); })) {
 963                 spin_lock(&plli->lli_agl_lock);
 964                 clli = list_first_entry_or_null(&sai->sai_agls,
 965                                                 struct ll_inode_info,
 966                                                 lli_agl_list);
 967                 if (clli) {
 968                         __set_current_state(TASK_RUNNING);
 969                         list_del_init(&clli->lli_agl_list);
 970                         spin_unlock(&plli->lli_agl_lock);
 971                         ll_agl_trigger(&clli->lli_vfs_inode, sai);
 972                         cond_resched();
 973                 } else {
 974                         spin_unlock(&plli->lli_agl_lock);
 975                         schedule();
 976                 }
 977         }
 978         __set_current_state(TASK_RUNNING);
 979         RETURN(0);
 980 }
 981
 982 static void ll_stop_agl(struct ll_statahead_info *sai)
 983 {
 984         struct dentry *parent = sai->sai_dentry;
 985         struct ll_inode_info *plli = ll_i2info(parent->d_inode);
 986         struct ll_inode_info *clli;
 987         struct task_struct *agl_task;
 988
 989         spin_lock(&plli->lli_agl_lock);
 990         agl_task = sai->sai_agl_task;
 991         sai->sai_agl_task = NULL;
 992         spin_unlock(&plli->lli_agl_lock);
 993         if (!agl_task)
 994                 return;
 995
 996         CDEBUG(D_READA, "stop agl thread: sai %p pid %u\n",
 997                sai, (unsigned int)agl_task->pid);
 998         kthread_stop(agl_task);
 999
1000         spin_lock(&plli->lli_agl_lock);
1001         while ((clli = list_first_entry_or_null(&sai->sai_agls,
1002                                                 struct ll_inode_info,
1003                                                 lli_agl_list)) != NULL) {
1004                 list_del_init(&clli->lli_agl_list);
1005                 spin_unlock(&plli->lli_agl_lock);
1006                 clli->lli_agl_index = 0;
1007                 iput(&clli->lli_vfs_inode);
1008                 spin_lock(&plli->lli_agl_lock);
1009         }
1010         spin_unlock(&plli->lli_agl_lock);
1011         CDEBUG(D_READA, "agl thread stopped: sai %p, parent %pd\n",
1012                sai, parent);
1013         ll_sai_put(sai);
1014 }
1015
1016 /* start agl thread */
1017 static void ll_start_agl(struct dentry *parent, struct ll_statahead_info *sai)
1018 {
1019         int node = cfs_cpt_spread_node(cfs_cpt_tab, CFS_CPT_ANY);
1020         struct ll_inode_info *plli;
1021         struct task_struct *task;
1022
1023         ENTRY;
1024
1025         CDEBUG(D_READA, "start agl thread: sai %p, parent %pd\n",
1026                sai, parent);
1027
1028         plli = ll_i2info(parent->d_inode);
1029         task = kthread_create_on_node(ll_agl_thread, parent, node, "ll_agl_%d",
1030                                       plli->lli_opendir_pid);
1031         if (IS_ERR(task)) {
1032                 CERROR("can't start ll_agl thread, rc: %ld\n", PTR_ERR(task));
1033                 RETURN_EXIT;
1034         }
1035         sai->sai_agl_task = task;
1036         atomic_inc(&ll_i2sbi(d_inode(parent))->ll_agl_total);
1037         /* Get an extra reference that the thread holds */
1038         ll_sai_get(d_inode(parent));
1039
1040         wake_up_process(task);
1041
1042         EXIT;
1043 }
1044
1045 /* statahead thread main function */
1046 static int ll_statahead_thread(void *arg)
1047 {
1048         struct dentry *parent = (struct dentry *)arg;
1049         struct inode *dir = parent->d_inode;
1050         struct ll_inode_info *lli = ll_i2info(dir);
1051         struct ll_sb_info *sbi = ll_i2sbi(dir);
1052         struct ll_statahead_info *sai = lli->lli_sai;
1053         int first = 0;
1054         struct md_op_data *op_data;
1055         struct page *page = NULL;
1056         __u64 pos = 0;
1057         int rc = 0;
1058
1059         ENTRY;
1060
1061         CDEBUG(D_READA, "statahead thread starting: sai %p, parent %pd\n",
1062                sai, parent);
1063
1064         OBD_ALLOC_PTR(op_data);
1065         if (!op_data)
1066                 GOTO(out, rc = -ENOMEM);
1067
1068         while (pos != MDS_DIR_END_OFF && sai->sai_task) {
1069                 struct lu_dirpage *dp;
1070                 struct lu_dirent  *ent;
1071
1072                 op_data = ll_prep_md_op_data(op_data, dir, dir, NULL, 0, 0,
1073                                              LUSTRE_OPC_ANY, dir);
1074                 if (IS_ERR(op_data)) {
1075                         rc = PTR_ERR(op_data);
1076                         break;
1077                 }
1078
1079                 page = ll_get_dir_page(dir, op_data, pos, NULL);
1080                 ll_unlock_md_op_lsm(op_data);
1081                 if (IS_ERR(page)) {
1082                         rc = PTR_ERR(page);
1083                         CDEBUG(D_READA,
1084                                "error reading dir "DFID" at %llu /%llu opendir_pid = %u: rc = %d\n",
1085                                PFID(ll_inode2fid(dir)), pos, sai->sai_index,
1086                                lli->lli_opendir_pid, rc);
1087                         break;
1088                 }
1089
1090                 dp = page_address(page);
1091                 for (ent = lu_dirent_start(dp);
1092                      ent != NULL && sai->sai_task &&
1093                      !sa_low_hit(sai);
1094                      ent = lu_dirent_next(ent)) {
1095                         __u64 hash;
1096                         int namelen;
1097                         char *name;
1098                         struct lu_fid fid;
1099                         struct llcrypt_str lltr = LLTR_INIT(NULL, 0);
1100
1101                         hash = le64_to_cpu(ent->lde_hash);
1102                         if (unlikely(hash < pos))
1103                                 /*
1104                                  * Skip until we find target hash value.
1105                                  */
1106                                 continue;
1107
1108                         namelen = le16_to_cpu(ent->lde_namelen);
1109                         if (unlikely(namelen == 0))
1110                                 /*
1111                                  * Skip dummy record.
1112                                  */
1113                                 continue;
1114
1115                         name = ent->lde_name;
1116                         if (name[0] == '.') {
1117                                 if (namelen == 1) {
1118                                         /*
1119                                          * skip "."
1120                                          */
1121                                         continue;
1122                                 } else if (name[1] == '.' && namelen == 2) {
1123                                         /*
1124                                          * skip ".."
1125                                          */
1126                                         continue;
1127                                 } else if (!sai->sai_ls_all) {
1128                                         /*
1129                                          * skip hidden files.
1130                                          */
1131                                         sai->sai_skip_hidden++;
1132                                         continue;
1133                                 }
1134                         }
1135
1136                         /*
1137                          * don't stat-ahead first entry.
1138                          */
1139                         if (unlikely(++first == 1))
1140                                 continue;
1141
1142                         fid_le_to_cpu(&fid, &ent->lde_fid);
1143
1144                         while (({set_current_state(TASK_IDLE);
1145                                  sai->sai_task; })) {
1146                                 spin_lock(&lli->lli_agl_lock);
1147                                 while (sa_sent_full(sai) &&
1148                                        !agl_list_empty(sai)) {
1149                                         struct ll_inode_info *clli;
1150
1151                                         __set_current_state(TASK_RUNNING);
1152                                         clli = agl_first_entry(sai);
1153                                         list_del_init(&clli->lli_agl_list);
1154                                         spin_unlock(&lli->lli_agl_lock);
1155
1156                                         ll_agl_trigger(&clli->lli_vfs_inode,
1157                                                        sai);
1158                                         cond_resched();
1159                                         spin_lock(&lli->lli_agl_lock);
1160                                 }
1161                                 spin_unlock(&lli->lli_agl_lock);
1162
1163                                 if (!sa_sent_full(sai))
1164                                         break;
1165                                 schedule();
1166                         }
1167                         __set_current_state(TASK_RUNNING);
1168
1169                         if (IS_ENCRYPTED(dir)) {
1170                                 struct llcrypt_str de_name =
1171                                         LLTR_INIT(ent->lde_name, namelen);
1172                                 struct lu_fid fid;
1173
1174                                 rc = llcrypt_fname_alloc_buffer(dir, NAME_MAX,
1175                                                                 &lltr);
1176                                 if (rc < 0)
1177                                         continue;
1178
1179                                 fid_le_to_cpu(&fid, &ent->lde_fid);
1180                                 if (ll_fname_disk_to_usr(dir, 0, 0, &de_name,
1181                                                          &lltr, &fid)) {
1182                                         llcrypt_fname_free_buffer(&lltr);
1183                                         continue;
1184                                 }
1185
1186                                 name = lltr.name;
1187                                 namelen = lltr.len;
1188                         }
1189
1190                         sa_statahead(parent, name, namelen, &fid);
1191                         llcrypt_fname_free_buffer(&lltr);
1192                 }
1193
1194                 pos = le64_to_cpu(dp->ldp_hash_end);
1195                 down_read(&lli->lli_lsm_sem);
1196                 ll_release_page(dir, page,
1197                                 le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
1198                 up_read(&lli->lli_lsm_sem);
1199
1200                 if (sa_low_hit(sai)) {
1201                         rc = -EFAULT;
1202                         atomic_inc(&sbi->ll_sa_wrong);
1203                         CDEBUG(D_READA,
1204                                "Statahead for dir "DFID" hit ratio too low: hit/miss %llu/%llu, sent/replied %llu/%llu, stoppingstatahead thread: pid %d\n",
1205                                PFID(&lli->lli_fid), sai->sai_hit,
1206                                sai->sai_miss, sai->sai_sent,
1207                                sai->sai_replied, current->pid);
1208                         break;
1209                 }
1210         }
1211         ll_finish_md_op_data(op_data);
1212
1213         if (rc < 0) {
1214                 spin_lock(&lli->lli_sa_lock);
1215                 sai->sai_task = NULL;
1216                 lli->lli_sa_enabled = 0;
1217                 spin_unlock(&lli->lli_sa_lock);
1218         }
1219
1220         /*
1221          * statahead is finished, but statahead entries need to be cached, wait
1222          * for file release closedir() call to stop me.
1223          */
1224         while (({set_current_state(TASK_IDLE);
1225                  sai->sai_task; })) {
1226                 schedule();
1227         }
1228         __set_current_state(TASK_RUNNING);
1229
1230         EXIT;
1231 out:
1232         ll_stop_agl(sai);
1233
1234         /*
1235          * wait for inflight statahead RPCs to finish, and then we can free sai
1236          * safely because statahead RPC will access sai data
1237          */
1238         while (sai->sai_sent != sai->sai_replied)
1239                 /* in case we're not woken up, timeout wait */
1240                 msleep(125);
1241
1242         CDEBUG(D_READA, "%s: statahead thread stopped: sai %p, parent %pd\n",
1243                sbi->ll_fsname, sai, parent);
1244
1245         spin_lock(&lli->lli_sa_lock);
1246         sai->sai_task = NULL;
1247         spin_unlock(&lli->lli_sa_lock);
1248         wake_up(&sai->sai_waitq);
1249
1250         atomic_add(sai->sai_hit, &sbi->ll_sa_hit_total);
1251         atomic_add(sai->sai_miss, &sbi->ll_sa_miss_total);
1252
1253         ll_sai_put(sai);
1254
1255         return rc;
1256 }
1257
1258 /* authorize opened dir handle @key to statahead */
1259 void ll_authorize_statahead(struct inode *dir, void *key)
1260 {
1261         struct ll_inode_info *lli = ll_i2info(dir);
1262
1263         spin_lock(&lli->lli_sa_lock);
1264         if (!lli->lli_opendir_key && !lli->lli_sai) {
1265                 /*
1266                  * if lli_sai is not NULL, it means previous statahead is not
1267                  * finished yet, we'd better not start a new statahead for now.
1268                  */
1269                 LASSERT(lli->lli_opendir_pid == 0);
1270                 lli->lli_opendir_key = key;
1271                 lli->lli_opendir_pid = current->pid;
1272                 lli->lli_sa_enabled = 1;
1273         }
1274         spin_unlock(&lli->lli_sa_lock);
1275 }
1276
1277 /*
1278  * deauthorize opened dir handle @key to statahead, and notify statahead thread
1279  * to quit if it's running.
1280  */
1281 void ll_deauthorize_statahead(struct inode *dir, void *key)
1282 {
1283         struct ll_inode_info *lli = ll_i2info(dir);
1284         struct ll_statahead_info *sai;
1285
1286         LASSERT(lli->lli_opendir_key == key);
1287         LASSERT(lli->lli_opendir_pid != 0);
1288
1289         CDEBUG(D_READA, "deauthorize statahead for "DFID"\n",
1290                PFID(&lli->lli_fid));
1291
1292         spin_lock(&lli->lli_sa_lock);
1293         lli->lli_opendir_key = NULL;
1294         lli->lli_opendir_pid = 0;
1295         lli->lli_sa_enabled = 0;
1296         sai = lli->lli_sai;
1297         if (sai && sai->sai_task) {
1298                 /*
1299                  * statahead thread may not have quit yet because it needs to
1300                  * cache entries, now it's time to tell it to quit.
1301                  *
1302                  * wake_up_process() provides the necessary barriers
1303                  * to pair with set_current_state().
1304                  */
1305                 struct task_struct *task = sai->sai_task;
1306
1307                 sai->sai_task = NULL;
1308                 wake_up_process(task);
1309         }
1310         spin_unlock(&lli->lli_sa_lock);
1311 }
1312
1313 enum {
1314         /**
1315          * not first dirent, or is "."
1316          */
1317         LS_NOT_FIRST_DE = 0,
1318         /**
1319          * the first non-hidden dirent
1320          */
1321         LS_FIRST_DE,
1322         /**
1323          * the first hidden dirent, that is "."
1324          */
1325         LS_FIRST_DOT_DE
1326 };
1327
1328 /* file is first dirent under @dir */
1329 static int is_first_dirent(struct inode *dir, struct dentry *dentry)
1330 {
1331         struct qstr *target = &dentry->d_name;
1332         struct md_op_data *op_data;
1333         int dot_de;
1334         struct page *page = NULL;
1335         int rc = LS_NOT_FIRST_DE;
1336         __u64 pos = 0;
1337         struct llcrypt_str lltr = LLTR_INIT(NULL, 0);
1338
1339         ENTRY;
1340
1341         op_data = ll_prep_md_op_data(NULL, dir, dir, NULL, 0, 0,
1342                                      LUSTRE_OPC_ANY, dir);
1343         if (IS_ERR(op_data))
1344                 RETURN(PTR_ERR(op_data));
1345
1346         if (IS_ENCRYPTED(dir)) {
1347                 int rc2 = llcrypt_fname_alloc_buffer(dir, NAME_MAX, &lltr);
1348
1349                 if (rc2 < 0)
1350                         RETURN(rc2);
1351         }
1352
1353         /**
1354          *FIXME choose the start offset of the readdir
1355          */
1356
1357         page = ll_get_dir_page(dir, op_data, 0, NULL);
1358
1359         while (1) {
1360                 struct lu_dirpage *dp;
1361                 struct lu_dirent  *ent;
1362
1363                 if (IS_ERR(page)) {
1364                         struct ll_inode_info *lli = ll_i2info(dir);
1365
1366                         rc = PTR_ERR(page);
1367                         CERROR("%s: reading dir "DFID" at %llu opendir_pid = %u : rc = %d\n",
1368                                ll_i2sbi(dir)->ll_fsname,
1369                                PFID(ll_inode2fid(dir)), pos,
1370                                lli->lli_opendir_pid, rc);
1371                         break;
1372                 }
1373
1374                 dp = page_address(page);
1375                 for (ent = lu_dirent_start(dp); ent != NULL;
1376                      ent = lu_dirent_next(ent)) {
1377                         __u64 hash;
1378                         int namelen;
1379                         char *name;
1380
1381                         hash = le64_to_cpu(ent->lde_hash);
1382                         /*
1383                          * The ll_get_dir_page() can return any page containing
1384                          * the given hash which may be not the start hash.
1385                          */
1386                         if (unlikely(hash < pos))
1387                                 continue;
1388
1389                         namelen = le16_to_cpu(ent->lde_namelen);
1390                         if (unlikely(namelen == 0))
1391                                 /*
1392                                  * skip dummy record.
1393                                  */
1394                                 continue;
1395
1396                         name = ent->lde_name;
1397                         if (name[0] == '.') {
1398                                 if (namelen == 1)
1399                                         /*
1400                                          * skip "."
1401                                          */
1402                                         continue;
1403                                 else if (name[1] == '.' && namelen == 2)
1404                                         /*
1405                                          * skip ".."
1406                                          */
1407                                         continue;
1408                                 else
1409                                         dot_de = 1;
1410                         } else {
1411                                 dot_de = 0;
1412                         }
1413
1414                         if (dot_de && target->name[0] != '.') {
1415                                 CDEBUG(D_READA, "%.*s skip hidden file %.*s\n",
1416                                        target->len, target->name,
1417                                        namelen, name);
1418                                 continue;
1419                         }
1420
1421                         if (IS_ENCRYPTED(dir)) {
1422                                 struct llcrypt_str de_name =
1423                                         LLTR_INIT(ent->lde_name, namelen);
1424                                 struct lu_fid fid;
1425
1426                                 fid_le_to_cpu(&fid, &ent->lde_fid);
1427                                 if (ll_fname_disk_to_usr(dir, 0, 0, &de_name,
1428                                                          &lltr, &fid))
1429                                         continue;
1430                                 name = lltr.name;
1431                                 namelen = lltr.len;
1432                         }
1433
1434                         if (target->len != namelen ||
1435                             memcmp(target->name, name, namelen) != 0)
1436                                 rc = LS_NOT_FIRST_DE;
1437                         else if (!dot_de)
1438                                 rc = LS_FIRST_DE;
1439                         else
1440                                 rc = LS_FIRST_DOT_DE;
1441
1442                         ll_release_page(dir, page, false);
1443                         GOTO(out, rc);
1444                 }
1445                 pos = le64_to_cpu(dp->ldp_hash_end);
1446                 if (pos == MDS_DIR_END_OFF) {
1447                         /*
1448                          * End of directory reached.
1449                          */
1450                         ll_release_page(dir, page, false);
1451                         GOTO(out, rc);
1452                 } else {
1453                         /*
1454                          * chain is exhausted
1455                          * Normal case: continue to the next page.
1456                          */
1457                         ll_release_page(dir, page, le32_to_cpu(dp->ldp_flags) &
1458                                               LDF_COLLIDE);
1459                         page = ll_get_dir_page(dir, op_data, pos, NULL);
1460                 }
1461         }
1462         EXIT;
1463 out:
1464         llcrypt_fname_free_buffer(&lltr);
1465         ll_finish_md_op_data(op_data);
1466
1467         return rc;
1468 }
1469
1470 /**
1471  * revalidate @dentryp from statahead cache
1472  *
1473  * \param[in] dir       parent directory
1474  * \param[in] sai       sai structure
1475  * \param[out] dentryp  pointer to dentry which will be revalidated
1476  * \param[in] unplug    unplug statahead window only (normally for negative
1477  *                      dentry)
1478  * \retval              1 on success, dentry is saved in @dentryp
1479  * \retval              0 if revalidation failed (no proper lock on client)
1480  * \retval              negative number upon error
1481  */
1482 static int revalidate_statahead_dentry(struct inode *dir,
1483                                        struct ll_statahead_info *sai,
1484                                        struct dentry **dentryp,
1485                                        bool unplug)
1486 {
1487         struct sa_entry *entry = NULL;
1488         struct ll_inode_info *lli = ll_i2info(dir);
1489         int rc = 0;
1490
1491         ENTRY;
1492
1493         if ((*dentryp)->d_name.name[0] == '.') {
1494                 if (sai->sai_ls_all ||
1495                     sai->sai_miss_hidden >= sai->sai_skip_hidden) {
1496                         /*
1497                          * Hidden dentry is the first one, or statahead
1498                          * thread does not skip so many hidden dentries
1499                          * before "sai_ls_all" enabled as below.
1500                          */
1501                 } else {
1502                         if (!sai->sai_ls_all)
1503                                 /*
1504                                  * It maybe because hidden dentry is not
1505                                  * the first one, "sai_ls_all" was not
1506                                  * set, then "ls -al" missed. Enable
1507                                  * "sai_ls_all" for such case.
1508                                  */
1509                                 sai->sai_ls_all = 1;
1510
1511                         /*
1512                          * Such "getattr" has been skipped before
1513                          * "sai_ls_all" enabled as above.
1514                          */
1515                         sai->sai_miss_hidden++;
1516                         RETURN(-EAGAIN);
1517                 }
1518         }
1519
1520         if (unplug)
1521                 GOTO(out, rc = 1);
1522
1523         entry = sa_get(sai, &(*dentryp)->d_name);
1524         if (!entry)
1525                 GOTO(out, rc = -EAGAIN);
1526
1527         if (!sa_ready(entry)) {
1528                 spin_lock(&lli->lli_sa_lock);
1529                 sai->sai_index_wait = entry->se_index;
1530                 spin_unlock(&lli->lli_sa_lock);
1531                 rc = wait_event_idle_timeout(sai->sai_waitq, sa_ready(entry),
1532                                              cfs_time_seconds(30));
1533                 if (rc == 0) {
1534                         /*
1535                          * entry may not be ready, so it may be used by inflight
1536                          * statahead RPC, don't free it.
1537                          */
1538                         entry = NULL;
1539                         GOTO(out, rc = -EAGAIN);
1540                 }
1541         }
1542
1543         /*
1544          * We need to see the value that was set immediately before we
1545          * were woken up.
1546          */
1547         if (smp_load_acquire(&entry->se_state) == SA_ENTRY_SUCC &&
1548             entry->se_inode) {
1549                 struct inode *inode = entry->se_inode;
1550                 struct lookup_intent it = { .it_op = IT_GETATTR,
1551                                             .it_lock_handle =
1552                                                 entry->se_handle };
1553                 __u64 bits;
1554
1555                 rc = md_revalidate_lock(ll_i2mdexp(dir), &it,
1556                                         ll_inode2fid(inode), &bits);
1557                 if (rc == 1) {
1558                         if (!(*dentryp)->d_inode) {
1559                                 struct dentry *alias;
1560
1561                                 alias = ll_splice_alias(inode, *dentryp);
1562                                 if (IS_ERR(alias)) {
1563                                         ll_intent_release(&it);
1564                                         GOTO(out, rc = PTR_ERR(alias));
1565                                 }
1566                                 *dentryp = alias;
1567                                 /*
1568                                  * statahead prepared this inode, transfer inode
1569                                  * refcount from sa_entry to dentry
1570                                  */
1571                                 entry->se_inode = NULL;
1572                         } else if ((*dentryp)->d_inode != inode) {
1573                                 /* revalidate, but inode is recreated */
1574                                 CDEBUG(D_READA,
1575                                        "%s: stale dentry %pd inode " DFID", statahead inode "DFID "\n",
1576                                        ll_i2sbi(inode)->ll_fsname, *dentryp,
1577                                        PFID(ll_inode2fid((*dentryp)->d_inode)),
1578                                        PFID(ll_inode2fid(inode)));
1579                                 ll_intent_release(&it);
1580                                 GOTO(out, rc = -ESTALE);
1581                         }
1582
1583                         if ((bits & MDS_INODELOCK_LOOKUP) &&
1584                             d_lustre_invalid(*dentryp)) {
1585                                 d_lustre_revalidate(*dentryp);
1586                                 ll_update_dir_depth(dir, (*dentryp)->d_inode);
1587                         }
1588
1589                         ll_intent_release(&it);
1590                 }
1591         }
1592 out:
1593         /*
1594          * statahead cached sa_entry can be used only once, and will be killed
1595          * right after use, so if lookup/revalidate accessed statahead cache,
1596          * set dentry ldd_sa_generation to parent lli_sa_generation, later if we
1597          * stat this file again, we know we've done statahead before, see
1598          * dentry_may_statahead().
1599          */
1600         if (lld_is_init(*dentryp))
1601                 ll_d2d(*dentryp)->lld_sa_generation = lli->lli_sa_generation;
1602         sa_put(sai, entry);
1603         spin_lock(&lli->lli_sa_lock);
1604         if (sai->sai_task)
1605                 wake_up_process(sai->sai_task);
1606         spin_unlock(&lli->lli_sa_lock);
1607
1608         RETURN(rc);
1609 }
1610
1611 /**
1612  * start statahead thread
1613  *
1614  * \param[in] dir       parent directory
1615  * \param[in] dentry    dentry that triggers statahead, normally the first
1616  *                      dirent under @dir
1617  * \param[in] agl       indicate whether AGL is needed
1618  * \retval              -EAGAIN on success, because when this function is
1619  *                      called, it's already in lookup call, so client should
1620  *                      do it itself instead of waiting for statahead thread
1621  *                      to do it asynchronously.
1622  * \retval              negative number upon error
1623  */
1624 static int start_statahead_thread(struct inode *dir, struct dentry *dentry,
1625                                   bool agl)
1626 {
1627         int node = cfs_cpt_spread_node(cfs_cpt_tab, CFS_CPT_ANY);
1628         struct ll_inode_info *lli = ll_i2info(dir);
1629         struct ll_statahead_info *sai = NULL;
1630         struct dentry *parent = dentry->d_parent;
1631         struct task_struct *task;
1632         struct ll_sb_info *sbi = ll_i2sbi(parent->d_inode);
1633         int first = LS_FIRST_DE;
1634         int rc = 0;
1635
1636         ENTRY;
1637
1638         /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */
1639         first = is_first_dirent(dir, dentry);
1640         if (first == LS_NOT_FIRST_DE)
1641                 /* It is not "ls -{a}l" operation, no need statahead for it. */
1642                 GOTO(out, rc = -EFAULT);
1643
1644         if (unlikely(atomic_inc_return(&sbi->ll_sa_running) >
1645                                        sbi->ll_sa_running_max)) {
1646                 CDEBUG(D_READA,
1647                        "Too many concurrent statahead instances, avoid new statahead instance temporarily.\n");
1648                 GOTO(out, rc = -EMFILE);
1649         }
1650
1651         sai = ll_sai_alloc(parent);
1652         if (!sai)
1653                 GOTO(out, rc = -ENOMEM);
1654
1655         sai->sai_ls_all = (first == LS_FIRST_DOT_DE);
1656
1657         /*
1658          * if current lli_opendir_key was deauthorized, or dir re-opened by
1659          * another process, don't start statahead, otherwise the newly spawned
1660          * statahead thread won't be notified to quit.
1661          */
1662         spin_lock(&lli->lli_sa_lock);
1663         if (unlikely(lli->lli_sai || !lli->lli_opendir_key ||
1664                      lli->lli_opendir_pid != current->pid)) {
1665                 spin_unlock(&lli->lli_sa_lock);
1666                 GOTO(out, rc = -EPERM);
1667         }
1668         lli->lli_sai = sai;
1669         spin_unlock(&lli->lli_sa_lock);
1670
1671         CDEBUG(D_READA, "start statahead thread: [pid %d] [parent %pd]\n",
1672                current->pid, parent);
1673
1674         task = kthread_create_on_node(ll_statahead_thread, parent, node,
1675                                       "ll_sa_%u", lli->lli_opendir_pid);
1676         if (IS_ERR(task)) {
1677                 spin_lock(&lli->lli_sa_lock);
1678                 lli->lli_sai = NULL;
1679                 spin_unlock(&lli->lli_sa_lock);
1680                 rc = PTR_ERR(task);
1681                 CERROR("can't start ll_sa thread, rc: %d\n", rc);
1682                 GOTO(out, rc);
1683         }
1684
1685         if (test_bit(LL_SBI_AGL_ENABLED, ll_i2sbi(parent->d_inode)->ll_flags) &&
1686             agl)
1687                 ll_start_agl(parent, sai);
1688
1689         atomic_inc(&ll_i2sbi(parent->d_inode)->ll_sa_total);
1690         sai->sai_task = task;
1691
1692         wake_up_process(task);
1693         /*
1694          * We don't stat-ahead for the first dirent since we are already in
1695          * lookup.
1696          */
1697         RETURN(-EAGAIN);
1698
1699 out:
1700         /*
1701          * once we start statahead thread failed, disable statahead so that
1702          * subsequent stat won't waste time to try it.
1703          */
1704         spin_lock(&lli->lli_sa_lock);
1705         if (lli->lli_opendir_pid == current->pid)
1706                 lli->lli_sa_enabled = 0;
1707         spin_unlock(&lli->lli_sa_lock);
1708
1709         if (sai)
1710                 ll_sai_free(sai);
1711         if (first != LS_NOT_FIRST_DE)
1712                 atomic_dec(&sbi->ll_sa_running);
1713
1714         RETURN(rc);
1715 }
1716
1717 /*
1718  * Check whether statahead for @dir was started.
1719  */
1720 static inline bool ll_statahead_started(struct inode *dir, bool agl)
1721 {
1722         struct ll_inode_info *lli = ll_i2info(dir);
1723         struct ll_statahead_info *sai;
1724
1725         spin_lock(&lli->lli_sa_lock);
1726         sai = lli->lli_sai;
1727         if (sai && (sai->sai_agl_task != NULL) != agl)
1728                 CDEBUG(D_READA,
1729                        "%s: Statahead AGL hint changed from %d to %d\n",
1730                        ll_i2sbi(dir)->ll_fsname,
1731                        sai->sai_agl_task != NULL, agl);
1732         spin_unlock(&lli->lli_sa_lock);
1733
1734         return !!sai;
1735 }
1736
1737 /**
1738  * statahead entry function, this is called when client getattr on a file, it
1739  * will start statahead thread if this is the first dir entry, else revalidate
1740  * dentry from statahead cache.
1741  *
1742  * \param[in]  dir      parent directory
1743  * \param[out] dentryp  dentry to getattr
1744  * \param[in]  agl      whether start the agl thread
1745  *
1746  * \retval              1 on success
1747  * \retval              0 revalidation from statahead cache failed, caller needs
1748  *                      to getattr from server directly
1749  * \retval              negative number on error, caller often ignores this and
1750  *                      then getattr from server
1751  */
1752 int ll_start_statahead(struct inode *dir, struct dentry *dentry, bool agl)
1753 {
1754         if (!ll_statahead_started(dir, agl))
1755                 return start_statahead_thread(dir, dentry, agl);
1756         return 0;
1757 }
1758
1759 /**
1760  * revalidate dentry from statahead cache.
1761  *
1762  * \param[in]  dir      parent directory
1763  * \param[out] dentryp  dentry to getattr
1764  * \param[in]  unplug   unplug statahead window only (normally for negative
1765  *                      dentry)
1766  * \retval              1 on success
1767  * \retval              0 revalidation from statahead cache failed, caller needs
1768  *                      to getattr from server directly
1769  * \retval              negative number on error, caller often ignores this and
1770  *                      then getattr from server
1771  */
1772 int ll_revalidate_statahead(struct inode *dir, struct dentry **dentryp,
1773                             bool unplug)
1774 {
1775         struct ll_statahead_info *sai;
1776         int rc = 0;
1777
1778         sai = ll_sai_get(dir);
1779         if (sai) {
1780                 rc = revalidate_statahead_dentry(dir, sai, dentryp, unplug);
1781                 CDEBUG(D_READA, "revalidate statahead %pd: rc = %d.\n",
1782                        *dentryp, rc);
1783                 ll_sai_put(sai);
1784         }
1785         return rc;
1786 }