lustre/llite/statahead.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #include <linux/fs.h>
  38 #include <linux/sched.h>
  39 #include <linux/mm.h>
  40 #include <linux/smp_lock.h>
  41 #include <linux/highmem.h>
  42 #include <linux/pagemap.h>
  43
  44 #define DEBUG_SUBSYSTEM S_LLITE
  45
  46 #include <obd_support.h>
  47 #include <lustre_lite.h>
  48 #include <lustre_dlm.h>
  49 #include <linux/lustre_version.h>
  50 #include "llite_internal.h"
  51
  52 struct ll_sai_entry {
  53         struct list_head        se_list;
  54         unsigned int            se_index;
  55         int                     se_stat;
  56         struct ptlrpc_request  *se_req;
  57         struct md_enqueue_info *se_minfo;
  58 };
  59
  60 enum {
  61         SA_ENTRY_UNSTATED = 0,
  62         SA_ENTRY_STATED
  63 };
  64
  65 static unsigned int sai_generation = 0;
  66 static spinlock_t sai_generation_lock = SPIN_LOCK_UNLOCKED;
  67
  68 /**
  69  * Check whether first entry was stated already or not.
  70  * No need to hold lli_lock, for:
  71  * (1) it is me that remove entry from the list
  72  * (2) the statahead thread only add new entry to the list
  73  */
  74 static int ll_sai_entry_stated(struct ll_statahead_info *sai)
  75 {
  76         struct ll_sai_entry  *entry;
  77         int                   rc = 0;
  78
  79         if (!list_empty(&sai->sai_entries_stated)) {
  80                 entry = list_entry(sai->sai_entries_stated.next,
  81                                    struct ll_sai_entry, se_list);
  82                 if (entry->se_index == sai->sai_index_next)
  83                         rc = 1;
  84         }
  85         return rc;
  86 }
  87
  88 static inline int sa_received_empty(struct ll_statahead_info *sai)
  89 {
  90         return list_empty(&sai->sai_entries_received);
  91 }
  92
  93 static inline int sa_not_full(struct ll_statahead_info *sai)
  94 {
  95         return (sai->sai_index < sai->sai_hit + sai->sai_miss + sai->sai_max);
  96 }
  97
  98 static inline int sa_is_running(struct ll_statahead_info *sai)
  99 {
 100         return !!(sai->sai_thread.t_flags & SVC_RUNNING);
 101 }
 102
 103 static inline int sa_is_stopping(struct ll_statahead_info *sai)
 104 {
 105         return !!(sai->sai_thread.t_flags & SVC_STOPPING);
 106 }
 107
 108 static inline int sa_is_stopped(struct ll_statahead_info *sai)
 109 {
 110         return !!(sai->sai_thread.t_flags & SVC_STOPPED);
 111 }
 112
 113 /**
 114  * (1) hit ratio less than 80%
 115  * or
 116  * (2) consecutive miss more than 8
 117  */
 118 static inline int sa_low_hit(struct ll_statahead_info *sai)
 119 {
 120         return ((sai->sai_hit > 7 && sai->sai_hit < 4 * sai->sai_miss) ||
 121                 (sai->sai_consecutive_miss > 8));
 122 }
 123
 124 /**
 125  * process the deleted entry's member and free the entry.
 126  * (1) release intent
 127  * (2) free md_enqueue_info
 128  * (3) drop dentry's ref count
 129  * (4) release request's ref count
 130  */
 131 static void ll_sai_entry_cleanup(struct ll_sai_entry *entry, int free)
 132 {
 133         struct md_enqueue_info *minfo = entry->se_minfo;
 134         struct ptlrpc_request  *req = entry->se_req;
 135         ENTRY;
 136
 137         if (minfo) {
 138                 entry->se_minfo = NULL;
 139                 ll_intent_release(&minfo->mi_it);
 140                 dput(minfo->mi_dentry);
 141                 iput(minfo->mi_dir);
 142                 OBD_FREE_PTR(minfo);
 143         }
 144         if (req) {
 145                 entry->se_req = NULL;
 146                 ptlrpc_req_finished(req);
 147         }
 148         if (free) {
 149                 LASSERT(list_empty(&entry->se_list));
 150                 OBD_FREE_PTR(entry);
 151         }
 152
 153         EXIT;
 154 }
 155
 156 static struct ll_statahead_info *ll_sai_alloc(void)
 157 {
 158         struct ll_statahead_info *sai;
 159
 160         OBD_ALLOC_PTR(sai);
 161         if (!sai)
 162                 return NULL;
 163
 164         spin_lock(&sai_generation_lock);
 165         sai->sai_generation = ++sai_generation;
 166         if (unlikely(sai_generation == 0))
 167                 sai->sai_generation = ++sai_generation;
 168         spin_unlock(&sai_generation_lock);
 169         atomic_set(&sai->sai_refcount, 1);
 170         sai->sai_max = LL_SA_RPC_MIN;
 171         cfs_waitq_init(&sai->sai_waitq);
 172         cfs_waitq_init(&sai->sai_thread.t_ctl_waitq);
 173         CFS_INIT_LIST_HEAD(&sai->sai_entries_sent);
 174         CFS_INIT_LIST_HEAD(&sai->sai_entries_received);
 175         CFS_INIT_LIST_HEAD(&sai->sai_entries_stated);
 176         return sai;
 177 }
 178
 179 static inline
 180 struct ll_statahead_info *ll_sai_get(struct ll_statahead_info *sai)
 181 {
 182         LASSERT(sai);
 183         atomic_inc(&sai->sai_refcount);
 184         return sai;
 185 }
 186
 187 static void ll_sai_put(struct ll_statahead_info *sai)
 188 {
 189         struct inode         *inode = sai->sai_inode;
 190         struct ll_inode_info *lli;
 191         ENTRY;
 192
 193         LASSERT(inode != NULL);
 194         lli = ll_i2info(inode);
 195         LASSERT(lli->lli_sai == sai);
 196
 197         if (atomic_dec_and_test(&sai->sai_refcount)) {
 198                 struct ll_sai_entry *entry, *next;
 199
 200                 spin_lock(&lli->lli_lock);
 201                 if (unlikely(atomic_read(&sai->sai_refcount) > 0)) {
 202                         /* It is race case, the interpret callback just hold
 203                          * a reference count */
 204                         spin_unlock(&lli->lli_lock);
 205                         EXIT;
 206                         return;
 207                 }
 208
 209                 LASSERT(lli->lli_opendir_key == NULL);
 210                 lli->lli_sai = NULL;
 211                 lli->lli_opendir_pid = 0;
 212                 spin_unlock(&lli->lli_lock);
 213
 214                 LASSERT(sa_is_stopped(sai));
 215
 216                 if (sai->sai_sent > sai->sai_replied)
 217                         CDEBUG(D_READA,"statahead for dir "DFID" does not "
 218                               "finish: [sent:%u] [replied:%u]\n",
 219                               PFID(&lli->lli_fid),
 220                               sai->sai_sent, sai->sai_replied);
 221
 222                 list_for_each_entry_safe(entry, next, &sai->sai_entries_sent,
 223                                          se_list) {
 224                         list_del_init(&entry->se_list);
 225                         ll_sai_entry_cleanup(entry, 1);
 226                 }
 227                 list_for_each_entry_safe(entry, next, &sai->sai_entries_received,
 228                                          se_list) {
 229                         list_del_init(&entry->se_list);
 230                         ll_sai_entry_cleanup(entry, 1);
 231                 }
 232                 list_for_each_entry_safe(entry, next, &sai->sai_entries_stated,
 233                                          se_list) {
 234                         list_del_init(&entry->se_list);
 235                         ll_sai_entry_cleanup(entry, 1);
 236                 }
 237                 iput(inode);
 238                 OBD_FREE_PTR(sai);
 239         }
 240         EXIT;
 241 }
 242
 243 /**
 244  * insert it into sai_entries_sent tail when init.
 245  */
 246 static struct ll_sai_entry *
 247 ll_sai_entry_init(struct ll_statahead_info *sai, unsigned int index)
 248 {
 249         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 250         struct ll_sai_entry  *entry;
 251         ENTRY;
 252
 253         OBD_ALLOC_PTR(entry);
 254         if (entry == NULL)
 255                 RETURN(ERR_PTR(-ENOMEM));
 256
 257         CDEBUG(D_READA, "alloc sai entry %p index %u\n",
 258                entry, index);
 259         entry->se_index = index;
 260         entry->se_stat  = SA_ENTRY_UNSTATED;
 261
 262         spin_lock(&lli->lli_lock);
 263         list_add_tail(&entry->se_list, &sai->sai_entries_sent);
 264         spin_unlock(&lli->lli_lock);
 265
 266         RETURN(entry);
 267 }
 268
 269 /**
 270  * delete it from sai_entries_stated head when fini, it need not
 271  * to process entry's member.
 272  */
 273 static void ll_sai_entry_fini(struct ll_statahead_info *sai)
 274 {
 275         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 276         struct ll_sai_entry  *entry;
 277         ENTRY;
 278
 279         spin_lock(&lli->lli_lock);
 280         sai->sai_index_next++;
 281         if (likely(!list_empty(&sai->sai_entries_stated))) {
 282                 entry = list_entry(sai->sai_entries_stated.next,
 283                                    struct ll_sai_entry, se_list);
 284                 if (entry->se_index < sai->sai_index_next) {
 285                         list_del(&entry->se_list);
 286                         OBD_FREE_PTR(entry);
 287                 }
 288         } else
 289                 LASSERT(sa_is_stopped(sai));
 290         spin_unlock(&lli->lli_lock);
 291
 292         EXIT;
 293 }
 294
 295 /**
 296  * inside lli_lock.
 297  * \retval NULL : can not find the entry in sai_entries_sent with the index
 298  * \retval entry: find the entry in sai_entries_sent with the index
 299  */
 300 static struct ll_sai_entry *
 301 ll_sai_entry_set(struct ll_statahead_info *sai, unsigned int index, int stat,
 302                  struct ptlrpc_request *req, struct md_enqueue_info *minfo)
 303 {
 304         struct ll_sai_entry *entry;
 305         ENTRY;
 306
 307         if (!list_empty(&sai->sai_entries_sent)) {
 308                 list_for_each_entry(entry, &sai->sai_entries_sent, se_list) {
 309                         if (entry->se_index == index) {
 310                                 entry->se_stat = stat;
 311                                 entry->se_req = ptlrpc_request_addref(req);
 312                                 entry->se_minfo = minfo;
 313                                 RETURN(entry);
 314                         } else if (entry->se_index > index)
 315                                 RETURN(NULL);
 316                 }
 317         }
 318         RETURN(NULL);
 319 }
 320
 321 /**
 322  * inside lli_lock.
 323  * Move entry to sai_entries_received and
 324  * insert it into sai_entries_received tail.
 325  */
 326 static inline void
 327 ll_sai_entry_to_received(struct ll_statahead_info *sai, struct ll_sai_entry *entry)
 328 {
 329         if (!list_empty(&entry->se_list))
 330                 list_del_init(&entry->se_list);
 331         list_add_tail(&entry->se_list, &sai->sai_entries_received);
 332 }
 333
 334 /**
 335  * Move entry to sai_entries_stated and
 336  * sort with the index.
 337  */
 338 static int
 339 ll_sai_entry_to_stated(struct ll_statahead_info *sai, struct ll_sai_entry *entry)
 340 {
 341         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 342         struct ll_sai_entry  *se;
 343         ENTRY;
 344
 345         ll_sai_entry_cleanup(entry, 0);
 346
 347         spin_lock(&lli->lli_lock);
 348         if (!list_empty(&entry->se_list))
 349                 list_del_init(&entry->se_list);
 350
 351         if (unlikely(entry->se_index < sai->sai_index_next)) {
 352                 spin_unlock(&lli->lli_lock);
 353                 OBD_FREE_PTR(entry);
 354                 RETURN(0);
 355         }
 356
 357         list_for_each_entry_reverse(se, &sai->sai_entries_stated, se_list) {
 358                 if (se->se_index < entry->se_index) {
 359                         list_add(&entry->se_list, &se->se_list);
 360                         spin_unlock(&lli->lli_lock);
 361                         RETURN(1);
 362                 }
 363         }
 364
 365         /*
 366          * I am the first entry.
 367          */
 368         list_add(&entry->se_list, &sai->sai_entries_stated);
 369         spin_unlock(&lli->lli_lock);
 370         RETURN(1);
 371 }
 372
 373 /**
 374  * finish lookup/revalidate.
 375  */
 376 static int do_statahead_interpret(struct ll_statahead_info *sai)
 377 {
 378         struct ll_inode_info   *lli = ll_i2info(sai->sai_inode);
 379         struct ll_sai_entry    *entry;
 380         struct ptlrpc_request  *req;
 381         struct md_enqueue_info *minfo;
 382         struct lookup_intent   *it;
 383         struct dentry          *dentry;
 384         int                     rc = 0;
 385         struct mdt_body        *body;
 386         ENTRY;
 387
 388         spin_lock(&lli->lli_lock);
 389         LASSERT(!sa_received_empty(sai));
 390         entry = list_entry(sai->sai_entries_received.next, struct ll_sai_entry,
 391                            se_list);
 392         list_del_init(&entry->se_list);
 393         spin_unlock(&lli->lli_lock);
 394
 395         if (unlikely(entry->se_index < sai->sai_index_next)) {
 396                 CWARN("Found stale entry: [index %u] [next %u]\n",
 397                       entry->se_index, sai->sai_index_next);
 398                 ll_sai_entry_cleanup(entry, 1);
 399                 RETURN(0);
 400         }
 401
 402         if (entry->se_stat != SA_ENTRY_STATED)
 403                 GOTO(out, rc = entry->se_stat);
 404
 405         req = entry->se_req;
 406         minfo = entry->se_minfo;
 407         it = &minfo->mi_it;
 408         dentry = minfo->mi_dentry;
 409
 410         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 411         if (body == NULL)
 412                 GOTO(out, rc = -EFAULT);
 413
 414         if (dentry->d_inode == NULL) {
 415                 /*
 416                  * lookup.
 417                  */
 418                 struct dentry    *save = dentry;
 419                 struct it_cb_data icbd = {
 420                         .icbd_parent   = minfo->mi_dir,
 421                         .icbd_childp   = &dentry
 422                 };
 423
 424                 LASSERT(fid_is_zero(&minfo->mi_data.op_fid2));
 425
 426                 /*
 427                  * XXX: No fid in reply, this is probaly cross-ref case.
 428                  * SA can't handle it yet.
 429                  */
 430                 if (body->valid & OBD_MD_MDS)
 431                         GOTO(out, rc = -EAGAIN);
 432
 433                 rc = ll_lookup_it_finish(req, it, &icbd);
 434                 if (!rc)
 435                         /*
 436                          * Here dentry->d_inode might be NULL,
 437                          * because the entry may have been removed before
 438                          * we start doing stat ahead.
 439                          */
 440                         ll_lookup_finish_locks(it, dentry);
 441
 442                 if (dentry != save) {
 443                         minfo->mi_dentry = dentry;
 444                         dput(save);
 445                 }
 446         } else {
 447                 /*
 448                  * revalidate.
 449                  */
 450                 if (!lu_fid_eq(&minfo->mi_data.op_fid2, &body->fid1)) {
 451                         ll_unhash_aliases(dentry->d_inode);
 452                         GOTO(out, rc = -EAGAIN);
 453                 }
 454
 455                 rc = ll_revalidate_it_finish(req, it, dentry);
 456                 if (rc) {
 457                         ll_unhash_aliases(dentry->d_inode);
 458                         GOTO(out, rc);
 459                 }
 460
 461                 spin_lock(&ll_lookup_lock);
 462                 spin_lock(&dcache_lock);
 463                 lock_dentry(dentry);
 464                 __d_drop(dentry);
 465 #ifdef DCACHE_LUSTRE_INVALID
 466                 dentry->d_flags &= ~DCACHE_LUSTRE_INVALID;
 467 #endif
 468                 unlock_dentry(dentry);
 469                 d_rehash_cond(dentry, 0);
 470                 spin_unlock(&dcache_lock);
 471                 spin_unlock(&ll_lookup_lock);
 472
 473                 ll_lookup_finish_locks(it, dentry);
 474         }
 475         EXIT;
 476
 477 out:
 478         if (likely(ll_sai_entry_to_stated(sai, entry)))
 479                 cfs_waitq_signal(&sai->sai_waitq);
 480         return rc;
 481 }
 482
 483 static int ll_statahead_interpret(struct ptlrpc_request *req,
 484                                   struct md_enqueue_info *minfo,
 485                                   int rc)
 486 {
 487         struct lookup_intent     *it = &minfo->mi_it;
 488         struct dentry            *dentry = minfo->mi_dentry;
 489         struct inode             *dir = minfo->mi_dir;
 490         struct ll_inode_info     *lli = ll_i2info(dir);
 491         struct ll_statahead_info *sai;
 492         struct ll_sai_entry      *entry;
 493         ENTRY;
 494
 495         CDEBUG(D_READA, "interpret statahead %.*s rc %d\n",
 496                dentry->d_name.len, dentry->d_name.name, rc);
 497
 498         spin_lock(&lli->lli_lock);
 499         if (unlikely(lli->lli_sai == NULL ||
 500             lli->lli_sai->sai_generation != minfo->mi_generation)) {
 501                 spin_unlock(&lli->lli_lock);
 502                 ll_intent_release(it);
 503                 dput(dentry);
 504                 iput(dir);
 505                 OBD_FREE_PTR(minfo);
 506                 RETURN(-ESTALE);
 507         } else {
 508                 sai = ll_sai_get(lli->lli_sai);
 509                 entry = ll_sai_entry_set(sai,
 510                                          (unsigned int)(long)minfo->mi_cbdata,
 511                                          rc ? SA_ENTRY_UNSTATED :
 512                                          SA_ENTRY_STATED, req, minfo);
 513                 LASSERT(entry != NULL);
 514                 if (likely(sa_is_running(sai))) {
 515                         ll_sai_entry_to_received(sai, entry);
 516                         sai->sai_replied++;
 517                         spin_unlock(&lli->lli_lock);
 518                         cfs_waitq_signal(&sai->sai_thread.t_ctl_waitq);
 519                 } else {
 520                         if (!list_empty(&entry->se_list))
 521                                 list_del_init(&entry->se_list);
 522                         sai->sai_replied++;
 523                         spin_unlock(&lli->lli_lock);
 524                         ll_sai_entry_cleanup(entry, 1);
 525                 }
 526                 ll_sai_put(sai);
 527                 RETURN(rc);
 528         }
 529 }
 530
 531 static void sa_args_fini(struct md_enqueue_info *minfo,
 532                          struct ldlm_enqueue_info *einfo)
 533 {
 534         LASSERT(minfo && einfo);
 535         iput(minfo->mi_dir);
 536         capa_put(minfo->mi_data.op_capa1);
 537         capa_put(minfo->mi_data.op_capa2);
 538         OBD_FREE_PTR(minfo);
 539         OBD_FREE_PTR(einfo);
 540 }
 541
 542 /**
 543  * There is race condition between "capa_put" and "ll_statahead_interpret" for
 544  * accessing "op_data.op_capa[1,2]" as following:
 545  * "capa_put" releases "op_data.op_capa[1,2]"'s reference count after calling
 546  * "md_intent_getattr_async". But "ll_statahead_interpret" maybe run first, and
 547  * fill "op_data.op_capa[1,2]" as POISON, then cause "capa_put" access invalid
 548  * "ocapa". So here reserve "op_data.op_capa[1,2]" in "pcapa" before calling
 549  * "md_intent_getattr_async".
 550  */
 551 static int sa_args_init(struct inode *dir, struct dentry *dentry,
 552                         struct md_enqueue_info **pmi,
 553                         struct ldlm_enqueue_info **pei,
 554                         struct obd_capa **pcapa)
 555 {
 556         struct ll_inode_info     *lli = ll_i2info(dir);
 557         struct md_enqueue_info   *minfo;
 558         struct ldlm_enqueue_info *einfo;
 559         struct md_op_data        *op_data;
 560
 561         OBD_ALLOC_PTR(einfo);
 562         if (einfo == NULL)
 563                 return -ENOMEM;
 564
 565         OBD_ALLOC_PTR(minfo);
 566         if (minfo == NULL) {
 567                 OBD_FREE_PTR(einfo);
 568                 return -ENOMEM;
 569         }
 570
 571         op_data = ll_prep_md_op_data(&minfo->mi_data, dir, dentry->d_inode,
 572                                      dentry->d_name.name, dentry->d_name.len,
 573                                      0, LUSTRE_OPC_ANY, NULL);
 574         if (IS_ERR(op_data)) {
 575                 OBD_FREE_PTR(einfo);
 576                 OBD_FREE_PTR(minfo);
 577                 return PTR_ERR(op_data);
 578         }
 579
 580         minfo->mi_it.it_op = IT_GETATTR;
 581         minfo->mi_dentry = dentry;
 582         minfo->mi_dir = igrab(dir);
 583         minfo->mi_cb = ll_statahead_interpret;
 584         minfo->mi_generation = lli->lli_sai->sai_generation;
 585         minfo->mi_cbdata = (void *)(long)lli->lli_sai->sai_index;
 586
 587         einfo->ei_type   = LDLM_IBITS;
 588         einfo->ei_mode   = it_to_lock_mode(&minfo->mi_it);
 589         einfo->ei_cb_bl  = ll_md_blocking_ast;
 590         einfo->ei_cb_cp  = ldlm_completion_ast;
 591         einfo->ei_cb_gl  = NULL;
 592         einfo->ei_cbdata = NULL;
 593
 594         *pmi = minfo;
 595         *pei = einfo;
 596         pcapa[0] = op_data->op_capa1;
 597         pcapa[1] = op_data->op_capa2;
 598
 599         return 0;
 600 }
 601
 602 /**
 603  * similar to ll_lookup_it().
 604  */
 605 static int do_sa_lookup(struct inode *dir, struct dentry *dentry)
 606 {
 607         struct md_enqueue_info   *minfo;
 608         struct ldlm_enqueue_info *einfo;
 609         struct obd_capa          *capas[2];
 610         int                       rc;
 611         ENTRY;
 612
 613         rc = sa_args_init(dir, dentry, &minfo, &einfo, capas);
 614         if (rc)
 615                 RETURN(rc);
 616
 617         rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo);
 618         if (!rc) {
 619                 capa_put(capas[0]);
 620                 capa_put(capas[1]);
 621         } else {
 622                 sa_args_fini(minfo, einfo);
 623         }
 624
 625         RETURN(rc);
 626 }
 627
 628 /**
 629  * similar to ll_revalidate_it().
 630  * \retval      1 -- dentry valid
 631  * \retval      0 -- will send stat-ahead request
 632  * \retval others -- prepare stat-ahead request failed
 633  */
 634 static int do_sa_revalidate(struct inode *dir, struct dentry *dentry)
 635 {
 636         struct inode             *inode = dentry->d_inode;
 637         struct lookup_intent      it = { .it_op = IT_GETATTR };
 638         struct md_enqueue_info   *minfo;
 639         struct ldlm_enqueue_info *einfo;
 640         struct obd_capa          *capas[2];
 641         int rc;
 642         ENTRY;
 643
 644         if (inode == NULL)
 645                 RETURN(1);
 646
 647         if (d_mountpoint(dentry))
 648                 RETURN(1);
 649
 650         if (dentry == dentry->d_sb->s_root)
 651                 RETURN(1);
 652
 653         rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode));
 654         if (rc == 1) {
 655                 ll_intent_release(&it);
 656                 RETURN(1);
 657         }
 658
 659         rc = sa_args_init(dir, dentry, &minfo, &einfo, capas);
 660         if (rc)
 661                 RETURN(rc);
 662
 663         rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo);
 664         if (!rc) {
 665                 capa_put(capas[0]);
 666                 capa_put(capas[1]);
 667         } else {
 668                 sa_args_fini(minfo, einfo);
 669         }
 670
 671         RETURN(rc);
 672 }
 673
 674 static inline void ll_name2qstr(struct qstr *q, const char *name, int namelen)
 675 {
 676         q->name = name;
 677         q->len  = namelen;
 678         q->hash = full_name_hash(name, namelen);
 679 }
 680
 681 static int ll_statahead_one(struct dentry *parent, const char* entry_name,
 682                             int entry_name_len)
 683 {
 684         struct inode             *dir = parent->d_inode;
 685         struct ll_inode_info     *lli = ll_i2info(dir);
 686         struct ll_statahead_info *sai = lli->lli_sai;
 687         struct qstr               name;
 688         struct dentry            *dentry;
 689         struct ll_sai_entry      *se;
 690         int                       rc;
 691         ENTRY;
 692
 693 #ifdef DCACHE_LUSTRE_INVALID
 694         if (parent->d_flags & DCACHE_LUSTRE_INVALID) {
 695 #else
 696         if (d_unhashed(parent)) {
 697 #endif
 698                 CDEBUG(D_READA, "parent dentry@%p %.*s is "
 699                        "invalid, skip statahead\n",
 700                        parent, parent->d_name.len, parent->d_name.name);
 701                 RETURN(-EINVAL);
 702         }
 703
 704         se = ll_sai_entry_init(sai, sai->sai_index);
 705         if (IS_ERR(se))
 706                 RETURN(PTR_ERR(se));
 707
 708         ll_name2qstr(&name, entry_name, entry_name_len);
 709         dentry = d_lookup(parent, &name);
 710         if (!dentry) {
 711                 dentry = d_alloc(parent, &name);
 712                 if (dentry) {
 713                         rc = do_sa_lookup(dir, dentry);
 714                         if (rc)
 715                                 dput(dentry);
 716                 } else {
 717                         GOTO(out, rc = -ENOMEM);
 718                 }
 719         } else {
 720                 rc = do_sa_revalidate(dir, dentry);
 721                 if (rc)
 722                         dput(dentry);
 723         }
 724
 725         EXIT;
 726
 727 out:
 728         if (rc) {
 729                 CDEBUG(D_READA, "set sai entry %p index %u stat %d rc %d\n",
 730                        se, se->se_index, se->se_stat, rc);
 731                 se->se_stat = rc;
 732                 if (ll_sai_entry_to_stated(sai, se))
 733                         cfs_waitq_signal(&sai->sai_waitq);
 734         } else {
 735                 sai->sai_sent++;
 736         }
 737
 738         sai->sai_index++;
 739         return rc;
 740 }
 741
 742 static int ll_statahead_thread(void *arg)
 743 {
 744         struct dentry            *parent = (struct dentry *)arg;
 745         struct inode             *dir = parent->d_inode;
 746         struct ll_inode_info     *lli = ll_i2info(dir);
 747         struct ll_sb_info        *sbi = ll_i2sbi(dir);
 748         struct ll_statahead_info *sai = ll_sai_get(lli->lli_sai);
 749         struct ptlrpc_thread     *thread = &sai->sai_thread;
 750         struct page              *page;
 751         __u64                     pos = 0;
 752         int                       first = 0;
 753         int                       rc = 0;
 754         struct ll_dir_chain       chain;
 755         ENTRY;
 756
 757         {
 758                 char pname[16];
 759                 snprintf(pname, 15, "ll_sa_%u", lli->lli_opendir_pid);
 760                 cfs_daemonize(pname);
 761         }
 762
 763         sbi->ll_sa_total++;
 764         spin_lock(&lli->lli_lock);
 765         thread->t_flags = SVC_RUNNING;
 766         spin_unlock(&lli->lli_lock);
 767         cfs_waitq_signal(&thread->t_ctl_waitq);
 768         CDEBUG(D_READA, "start doing statahead for %s\n", parent->d_name.name);
 769
 770         ll_dir_chain_init(&chain);
 771         page = ll_get_dir_page(dir, pos, 0, &chain);
 772
 773         while (1) {
 774                 struct l_wait_info lwi = { 0 };
 775                 struct lu_dirpage *dp;
 776                 struct lu_dirent  *ent;
 777
 778                 if (IS_ERR(page)) {
 779                         rc = PTR_ERR(page);
 780                         CERROR("error reading dir "DFID" at "LPU64"/%u: rc %d\n",
 781                                PFID(ll_inode2fid(dir)), pos,
 782                                sai->sai_index, rc);
 783                         break;
 784                 }
 785
 786                 dp = page_address(page);
 787                 for (ent = lu_dirent_start(dp); ent != NULL;
 788                      ent = lu_dirent_next(ent)) {
 789                         char *name = ent->lde_name;
 790                         int namelen = le16_to_cpu(ent->lde_namelen);
 791
 792                         if (namelen == 0)
 793                                 /*
 794                                  * Skip dummy record.
 795                                  */
 796                                 continue;
 797
 798                         if (name[0] == '.') {
 799                                 if (namelen == 1) {
 800                                         /*
 801                                          * skip "."
 802                                          */
 803                                         continue;
 804                                 } else if (name[1] == '.' && namelen == 2) {
 805                                         /*
 806                                          * skip ".."
 807                                          */
 808                                         continue;
 809                                 } else if (!sai->sai_ls_all) {
 810                                         /*
 811                                          * skip hidden files.
 812                                          */
 813                                         sai->sai_skip_hidden++;
 814                                         continue;
 815                                 }
 816                         }
 817
 818                         /*
 819                          * don't stat-ahead first entry.
 820                          */
 821                         if (unlikely(!first)) {
 822                                 first++;
 823                                 continue;
 824                         }
 825
 826 keep_de:
 827                         l_wait_event(thread->t_ctl_waitq,
 828                                      !sa_is_running(sai) || sa_not_full(sai) ||
 829                                      !sa_received_empty(sai),
 830                                      &lwi);
 831
 832                         while (!sa_received_empty(sai) && sa_is_running(sai))
 833                                 do_statahead_interpret(sai);
 834
 835                         if (unlikely(!sa_is_running(sai))) {
 836                                 ll_put_page(page);
 837                                 GOTO(out, rc);
 838                         }
 839
 840                         if (!sa_not_full(sai))
 841                                 /*
 842                                  * do not skip the current de.
 843                                  */
 844                                 goto keep_de;
 845
 846                         rc = ll_statahead_one(parent, name, namelen);
 847                         if (rc < 0) {
 848                                 ll_put_page(page);
 849                                 GOTO(out, rc);
 850                         }
 851                 }
 852                 pos = le64_to_cpu(dp->ldp_hash_end);
 853                 ll_put_page(page);
 854                 if (pos == DIR_END_OFF) {
 855                         /*
 856                          * End of directory reached.
 857                          */
 858                         while (1) {
 859                                 l_wait_event(thread->t_ctl_waitq,
 860                                              !sa_is_running(sai) ||
 861                                              !sa_received_empty(sai) ||
 862                                              sai->sai_sent == sai->sai_replied,
 863                                              &lwi);
 864                                 if (!sa_received_empty(sai) &&
 865                                     sa_is_running(sai))
 866                                         do_statahead_interpret(sai);
 867                                 else
 868                                         GOTO(out, rc);
 869                         }
 870                 } else if (1) {
 871                         /*
 872                          * chain is exhausted.
 873                          * Normal case: continue to the next page.
 874                          */
 875                         page = ll_get_dir_page(dir, pos, 1, &chain);
 876                 } else {
 877                         /*
 878                          * go into overflow page.
 879                          */
 880                 }
 881         }
 882         EXIT;
 883
 884 out:
 885         ll_dir_chain_fini(&chain);
 886         spin_lock(&lli->lli_lock);
 887         thread->t_flags = SVC_STOPPED;
 888         spin_unlock(&lli->lli_lock);
 889         cfs_waitq_signal(&sai->sai_waitq);
 890         cfs_waitq_signal(&thread->t_ctl_waitq);
 891         ll_sai_put(sai);
 892         dput(parent);
 893         CDEBUG(D_READA, "statahead thread stopped, pid %d\n",
 894                cfs_curproc_pid());
 895         return rc;
 896 }
 897
 898 /**
 899  * called in ll_file_release().
 900  */
 901 void ll_stop_statahead(struct inode *inode, void *key)
 902 {
 903         struct ll_inode_info *lli = ll_i2info(inode);
 904
 905         if (unlikely(key == NULL))
 906                 return;
 907
 908         spin_lock(&lli->lli_lock);
 909         if (lli->lli_opendir_key != key || lli->lli_opendir_pid == 0) {
 910                 spin_unlock(&lli->lli_lock);
 911                 return;
 912         }
 913
 914         lli->lli_opendir_key = NULL;
 915
 916         if (lli->lli_sai) {
 917                 struct l_wait_info lwi = { 0 };
 918                 struct ptlrpc_thread *thread = &lli->lli_sai->sai_thread;
 919
 920                 if (!sa_is_stopped(lli->lli_sai)) {
 921                         thread->t_flags = SVC_STOPPING;
 922                         spin_unlock(&lli->lli_lock);
 923                         cfs_waitq_signal(&thread->t_ctl_waitq);
 924
 925                         CDEBUG(D_READA, "stopping statahead thread, pid %d\n",
 926                                cfs_curproc_pid());
 927                         l_wait_event(thread->t_ctl_waitq,
 928                                      sa_is_stopped(lli->lli_sai),
 929                                      &lwi);
 930                 } else {
 931                         spin_unlock(&lli->lli_lock);
 932                 }
 933
 934                 /*
 935                  * Put the ref which was held when first statahead_enter.
 936                  * It maybe not the last ref for some statahead requests
 937                  * maybe inflight.
 938                  */
 939                 ll_sai_put(lli->lli_sai);
 940         } else {
 941                 lli->lli_opendir_pid = 0;
 942                 spin_unlock(&lli->lli_lock);
 943         }
 944 }
 945
 946 enum {
 947         /**
 948          * not first dirent, or is "."
 949          */
 950         LS_NONE_FIRST_DE = 0,
 951         /**
 952          * the first non-hidden dirent
 953          */
 954         LS_FIRST_DE,
 955         /**
 956          * the first hidden dirent, that is ".xxx
 957          */
 958         LS_FIRST_DOT_DE
 959 };
 960
 961 static int is_first_dirent(struct inode *dir, struct dentry *dentry)
 962 {
 963         struct ll_dir_chain chain;
 964         struct qstr        *target = &dentry->d_name;
 965         struct page        *page;
 966         __u64               pos = 0;
 967         int                 dot_de;
 968         int                 rc = LS_NONE_FIRST_DE;
 969         ENTRY;
 970
 971         ll_dir_chain_init(&chain);
 972         page = ll_get_dir_page(dir, pos, 0, &chain);
 973
 974         while (1) {
 975                 struct lu_dirpage *dp;
 976                 struct lu_dirent  *ent;
 977
 978                 if (IS_ERR(page)) {
 979                         rc = PTR_ERR(page);
 980                         CERROR("error reading dir "DFID" at "LPU64": rc %d\n",
 981                                PFID(ll_inode2fid(dir)), pos, rc);
 982                         break;
 983                 }
 984
 985                 dp = page_address(page);
 986                 for (ent = lu_dirent_start(dp); ent != NULL;
 987                      ent = lu_dirent_next(ent)) {
 988                         char *name = ent->lde_name;
 989                         int namelen = le16_to_cpu(ent->lde_namelen);
 990
 991                         if (namelen == 0)
 992                                 /*
 993                                  * skip dummy record.
 994                                  */
 995                                 continue;
 996
 997                         if (name[0] == '.') {
 998                                 if (namelen == 1)
 999                                         /*
1000                                          * skip "."
1001                                          */
1002                                         continue;
1003                                 else if (name[1] == '.' && namelen == 2)
1004                                         /*
1005                                          * skip ".."
1006                                          */
1007                                         continue;
1008                                 else
1009                                         dot_de = 1;
1010                         } else {
1011                                 dot_de = 0;
1012                         }
1013
1014                         if (dot_de && target->name[0] != '.') {
1015                                 CDEBUG(D_READA, "%.*s skip hidden file %.*s\n",
1016                                        target->len, target->name,
1017                                        namelen, name);
1018                                 continue;
1019                         }
1020
1021                         if (target->len == namelen &&
1022                             memcmp(target->name, name, namelen) == 0)
1023                                 rc = LS_FIRST_DE + dot_de;
1024                         else
1025                                 rc = LS_NONE_FIRST_DE;
1026                         ll_put_page(page);
1027                         GOTO(out, rc);
1028                 }
1029                 pos = le64_to_cpu(dp->ldp_hash_end);
1030                 ll_put_page(page);
1031                 if (pos == DIR_END_OFF) {
1032                         /*
1033                          * End of directory reached.
1034                          */
1035                         break;
1036                 } else if (1) {
1037                         /*
1038                          * chain is exhausted
1039                          * Normal case: continue to the next page.
1040                          */
1041                         page = ll_get_dir_page(dir, pos, 1, &chain);
1042                 } else {
1043                         /*
1044                          * go into overflow page.
1045                          */
1046                 }
1047         }
1048         EXIT;
1049
1050 out:
1051         ll_dir_chain_fini(&chain);
1052         return rc;
1053 }
1054
1055 /**
1056  * Start statahead thread if this is the first dir entry.
1057  * Otherwise if a thread is started already, wait it until it is ahead of me.
1058  * \retval 0       -- stat ahead thread process such dentry, for lookup, it miss
1059  * \retval 1       -- stat ahead thread process such dentry, for lookup, it hit
1060  * \retval -EEXIST -- stat ahead thread started, and this is the first dentry
1061  * \retval -EBADFD -- statahead thread exit and not dentry available
1062  * \retval -EAGAIN -- try to stat by caller
1063  * \retval others  -- error
1064  */
1065 int do_statahead_enter(struct inode *dir, struct dentry **dentryp, int lookup)
1066 {
1067         struct ll_inode_info     *lli;
1068         struct ll_statahead_info *sai;
1069         struct dentry            *parent;
1070         struct l_wait_info        lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
1071         int                       rc = 0;
1072         ENTRY;
1073
1074         LASSERT(dir != NULL);
1075         lli = ll_i2info(dir);
1076         LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
1077         sai = lli->lli_sai;
1078
1079         if (sai) {
1080                 struct ll_sb_info *sbi;
1081
1082                 if (unlikely(sa_is_stopped(sai) &&
1083                              list_empty(&sai->sai_entries_stated)))
1084                         RETURN(-EBADFD);
1085
1086                 if ((*dentryp)->d_name.name[0] == '.') {
1087                         if (likely(sai->sai_ls_all ||
1088                             sai->sai_miss_hidden >= sai->sai_skip_hidden)) {
1089                                 /*
1090                                  * Hidden dentry is the first one, or statahead
1091                                  * thread does not skip so many hidden dentries
1092                                  * before "sai_ls_all" enabled as below.
1093                                  */
1094                         } else {
1095                                 if (!sai->sai_ls_all)
1096                                         /*
1097                                          * It maybe because hidden dentry is not
1098                                          * the first one, "sai_ls_all" was not
1099                                          * set, then "ls -al" missed. Enable
1100                                          * "sai_ls_all" for such case.
1101                                          */
1102                                         sai->sai_ls_all = 1;
1103
1104                                 /*
1105                                  * Such "getattr" has been skipped before
1106                                  * "sai_ls_all" enabled as above.
1107                                  */
1108                                 sai->sai_miss_hidden++;
1109                                 RETURN(-ENOENT);
1110                         }
1111                 }
1112
1113                 sbi = ll_i2sbi(dir);
1114                 if (ll_sai_entry_stated(sai)) {
1115                         sbi->ll_sa_cached++;
1116                 } else {
1117                         sbi->ll_sa_blocked++;
1118                         /*
1119                          * thread started already, avoid double-stat.
1120                          */
1121                         rc = l_wait_event(sai->sai_waitq,
1122                                           ll_sai_entry_stated(sai) ||
1123                                           sa_is_stopped(sai),
1124                                           &lwi);
1125                 }
1126
1127                 if (lookup) {
1128                         struct dentry *result;
1129
1130                         result = d_lookup((*dentryp)->d_parent,
1131                                           &(*dentryp)->d_name);
1132                         if (result) {
1133                                 LASSERT(result != *dentryp);
1134                                 /* BUG 16303: do not drop reference count for
1135                                  * "*dentryp", VFS will do that by itself. */
1136                                 *dentryp = result;
1137                                 RETURN(1);
1138                         }
1139                 }
1140                 /*
1141                  * do nothing for revalidate.
1142                  */
1143                 RETURN(rc);
1144         }
1145
1146         /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */
1147         rc = is_first_dirent(dir, *dentryp);
1148         if (rc == LS_NONE_FIRST_DE)
1149                 /* It is not "ls -{a}l" operation, no need statahead for it. */
1150                 GOTO(out, rc = -EAGAIN);
1151
1152         sai = ll_sai_alloc();
1153         if (sai == NULL)
1154                 GOTO(out, rc = -ENOMEM);
1155
1156         sai->sai_ls_all = (rc == LS_FIRST_DOT_DE);
1157         sai->sai_inode = igrab(dir);
1158         if (unlikely(sai->sai_inode == NULL)) {
1159                 CWARN("Do not start stat ahead on dying inode "DFID" .\n",
1160                       PFID(&lli->lli_fid));
1161                 OBD_FREE_PTR(sai);
1162                 GOTO(out, rc = -ESTALE);
1163         }
1164
1165         /* get parent reference count here, and put it in ll_statahead_thread */
1166         parent = dget((*dentryp)->d_parent);
1167         if (unlikely(sai->sai_inode != parent->d_inode)) {
1168                 struct ll_inode_info *nlli = ll_i2info(parent->d_inode);
1169
1170                 CWARN("Race condition, someone changed %.*s just now: "
1171                       "old parent "DFID", new parent "DFID" .\n",
1172                       (*dentryp)->d_name.len, (*dentryp)->d_name.name,
1173                       PFID(&lli->lli_fid), PFID(&nlli->lli_fid));
1174                 dput(parent);
1175                 iput(sai->sai_inode);
1176                 OBD_FREE_PTR(sai);
1177                 RETURN(-EAGAIN);
1178         }
1179
1180         lli->lli_sai = sai;
1181         rc = cfs_kernel_thread(ll_statahead_thread, parent, 0);
1182         if (rc < 0) {
1183                 CERROR("can't start ll_sa thread, rc: %d\n", rc);
1184                 dput(parent);
1185                 lli->lli_opendir_key = NULL;
1186                 sai->sai_thread.t_flags = SVC_STOPPED;
1187                 ll_sai_put(sai);
1188                 LASSERT(lli->lli_sai == NULL);
1189                 RETURN(-EAGAIN);
1190         }
1191
1192         l_wait_event(sai->sai_thread.t_ctl_waitq,
1193                      sa_is_running(sai) || sa_is_stopped(sai),
1194                      &lwi);
1195
1196         /*
1197          * We don't stat-ahead for the first dirent since we are already in
1198          * lookup, and -EEXIST also indicates that this is the first dirent.
1199          */
1200         RETURN(-EEXIST);
1201
1202 out:
1203         spin_lock(&lli->lli_lock);
1204         lli->lli_opendir_key = NULL;
1205         lli->lli_opendir_pid = 0;
1206         spin_unlock(&lli->lli_lock);
1207         return rc;
1208 }
1209
1210 /**
1211  * update hit/miss count.
1212  */
1213 void ll_statahead_exit(struct inode *dir, struct dentry *dentry, int result)
1214 {
1215         struct ll_inode_info     *lli;
1216         struct ll_statahead_info *sai;
1217         struct ll_sb_info        *sbi;
1218         struct ll_dentry_data    *ldd = ll_d2d(dentry);
1219         ENTRY;
1220
1221         LASSERT(dir != NULL);
1222         lli = ll_i2info(dir);
1223         LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
1224         sai = lli->lli_sai;
1225         LASSERT(sai != NULL);
1226         sbi = ll_i2sbi(dir);
1227
1228         if (result >= 1) {
1229                 sbi->ll_sa_hit++;
1230                 sai->sai_hit++;
1231                 sai->sai_consecutive_miss = 0;
1232                 sai->sai_max = min(2 * sai->sai_max, sbi->ll_sa_max);
1233         } else {
1234                 sbi->ll_sa_miss++;
1235                 sai->sai_miss++;
1236                 sai->sai_consecutive_miss++;
1237                 if (sa_low_hit(sai) && sa_is_running(sai)) {
1238                         sbi->ll_sa_wrong++;
1239                         CDEBUG(D_READA, "Statahead for dir "DFID" hit ratio "
1240                                "too low: hit/miss %u/%u, sent/replied %u/%u, "
1241                                "stopping statahead thread: pid %d\n",
1242                                PFID(&lli->lli_fid), sai->sai_hit,
1243                                sai->sai_miss, sai->sai_sent,
1244                                sai->sai_replied, cfs_curproc_pid());
1245                         spin_lock(&lli->lli_lock);
1246                         if (!sa_is_stopped(sai))
1247                                 sai->sai_thread.t_flags = SVC_STOPPING;
1248                         spin_unlock(&lli->lli_lock);
1249                 }
1250         }
1251
1252         if (!sa_is_stopped(sai))
1253                 cfs_waitq_signal(&sai->sai_thread.t_ctl_waitq);
1254         ll_sai_entry_fini(sai);
1255         if (likely(ldd != NULL))
1256                 ldd->lld_sa_generation = sai->sai_generation;
1257
1258         EXIT;
1259 }