lustre/llite/statahead.c

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