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 /*
  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         cfs_list_t              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 cfs_spinlock_t sai_generation_lock = CFS_SPIN_LOCK_UNLOCKED;
  67
  68 /**
  69  * Check whether first entry was stated already or not.
  70  * No need to hold lli_sa_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 (!cfs_list_empty(&sai->sai_entries_stated)) {
  80                 entry = cfs_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 cfs_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_index_next + 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(cfs_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         cfs_spin_lock(&sai_generation_lock);
 165         sai->sai_generation = ++sai_generation;
 166         if (unlikely(sai_generation == 0))
 167                 sai->sai_generation = ++sai_generation;
 168         cfs_spin_unlock(&sai_generation_lock);
 169         cfs_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         cfs_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 (cfs_atomic_dec_and_lock(&sai->sai_refcount, &lli->lli_sa_lock)) {
 198                 struct ll_sai_entry *entry, *next;
 199
 200                 if (unlikely(cfs_atomic_read(&sai->sai_refcount) > 0)) {
 201                         /* It is race case, the interpret callback just hold
 202                          * a reference count */
 203                         cfs_spin_unlock(&lli->lli_sa_lock);
 204                         RETURN_EXIT;
 205                 }
 206
 207                 LASSERT(lli->lli_opendir_key == NULL);
 208                 lli->lli_sai = NULL;
 209                 lli->lli_opendir_pid = 0;
 210                 cfs_spin_unlock(&lli->lli_sa_lock);
 211
 212                 LASSERT(sa_is_stopped(sai));
 213
 214                 if (sai->sai_sent > sai->sai_replied)
 215                         CDEBUG(D_READA,"statahead for dir "DFID" does not "
 216                               "finish: [sent:%u] [replied:%u]\n",
 217                               PFID(&lli->lli_fid),
 218                               sai->sai_sent, sai->sai_replied);
 219
 220                 cfs_list_for_each_entry_safe(entry, next,
 221                                              &sai->sai_entries_sent, se_list) {
 222                         cfs_list_del_init(&entry->se_list);
 223                         ll_sai_entry_cleanup(entry, 1);
 224                 }
 225                 cfs_list_for_each_entry_safe(entry, next,
 226                                              &sai->sai_entries_received,
 227                                              se_list) {
 228                         cfs_list_del_init(&entry->se_list);
 229                         ll_sai_entry_cleanup(entry, 1);
 230                 }
 231                 cfs_list_for_each_entry_safe(entry, next,
 232                                              &sai->sai_entries_stated,
 233                                              se_list) {
 234                         cfs_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         cfs_spin_lock(&lli->lli_sa_lock);
 263         cfs_list_add_tail(&entry->se_list, &sai->sai_entries_sent);
 264         cfs_spin_unlock(&lli->lli_sa_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 int 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         int rc = 0;
 278         ENTRY;
 279
 280         cfs_spin_lock(&lli->lli_sa_lock);
 281         sai->sai_index_next++;
 282         if (likely(!cfs_list_empty(&sai->sai_entries_stated))) {
 283                 entry = cfs_list_entry(sai->sai_entries_stated.next,
 284                                        struct ll_sai_entry, se_list);
 285                 if (entry->se_index < sai->sai_index_next) {
 286                         cfs_list_del_init(&entry->se_list);
 287                         rc = entry->se_stat;
 288                         OBD_FREE_PTR(entry);
 289                 }
 290         } else {
 291                 LASSERT(sa_is_stopped(sai));
 292         }
 293         cfs_spin_unlock(&lli->lli_sa_lock);
 294
 295         RETURN(rc);
 296 }
 297
 298 /**
 299  * inside lli_sa_lock.
 300  * \retval NULL : can not find the entry in sai_entries_sent with the index
 301  * \retval entry: find the entry in sai_entries_sent with the index
 302  */
 303 static struct ll_sai_entry *
 304 ll_sai_entry_set(struct ll_statahead_info *sai, unsigned int index, int stat,
 305                  struct ptlrpc_request *req, struct md_enqueue_info *minfo)
 306 {
 307         struct ll_sai_entry *entry;
 308         ENTRY;
 309
 310         if (!cfs_list_empty(&sai->sai_entries_sent)) {
 311                 cfs_list_for_each_entry(entry, &sai->sai_entries_sent,
 312                                         se_list) {
 313                         if (entry->se_index == index) {
 314                                 entry->se_stat = stat;
 315                                 entry->se_req = ptlrpc_request_addref(req);
 316                                 entry->se_minfo = minfo;
 317                                 RETURN(entry);
 318                         } else if (entry->se_index > index) {
 319                                 RETURN(NULL);
 320                         }
 321                 }
 322         }
 323         RETURN(NULL);
 324 }
 325
 326 /**
 327  * inside lli_sa_lock.
 328  * Move entry to sai_entries_received and
 329  * insert it into sai_entries_received tail.
 330  */
 331 static inline void
 332 ll_sai_entry_to_received(struct ll_statahead_info *sai, struct ll_sai_entry *entry)
 333 {
 334         if (!cfs_list_empty(&entry->se_list))
 335                 cfs_list_del_init(&entry->se_list);
 336         cfs_list_add_tail(&entry->se_list, &sai->sai_entries_received);
 337 }
 338
 339 /**
 340  * Move entry to sai_entries_stated and
 341  * sort with the index.
 342  */
 343 static int
 344 ll_sai_entry_to_stated(struct ll_statahead_info *sai, struct ll_sai_entry *entry)
 345 {
 346         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
 347         struct ll_sai_entry  *se;
 348         ENTRY;
 349
 350         ll_sai_entry_cleanup(entry, 0);
 351
 352         cfs_spin_lock(&lli->lli_sa_lock);
 353         if (!cfs_list_empty(&entry->se_list))
 354                 cfs_list_del_init(&entry->se_list);
 355
 356         /* stale entry */
 357         if (unlikely(entry->se_index < sai->sai_index_next)) {
 358                 cfs_spin_unlock(&lli->lli_sa_lock);
 359                 OBD_FREE_PTR(entry);
 360                 RETURN(0);
 361         }
 362
 363         cfs_list_for_each_entry_reverse(se, &sai->sai_entries_stated, se_list) {
 364                 if (se->se_index < entry->se_index) {
 365                         cfs_list_add(&entry->se_list, &se->se_list);
 366                         cfs_spin_unlock(&lli->lli_sa_lock);
 367                         RETURN(1);
 368                 }
 369         }
 370
 371         /*
 372          * I am the first entry.
 373          */
 374         cfs_list_add(&entry->se_list, &sai->sai_entries_stated);
 375         cfs_spin_unlock(&lli->lli_sa_lock);
 376         RETURN(1);
 377 }
 378
 379 /**
 380  * finish lookup/revalidate.
 381  */
 382 static int do_statahead_interpret(struct ll_statahead_info *sai)
 383 {
 384         struct ll_inode_info   *lli = ll_i2info(sai->sai_inode);
 385         struct ll_sai_entry    *entry;
 386         struct ptlrpc_request  *req;
 387         struct md_enqueue_info *minfo;
 388         struct lookup_intent   *it;
 389         struct dentry          *dentry;
 390         int                     rc = 0;
 391         struct mdt_body        *body;
 392         ENTRY;
 393
 394         cfs_spin_lock(&lli->lli_sa_lock);
 395         LASSERT(!sa_received_empty(sai));
 396         entry = cfs_list_entry(sai->sai_entries_received.next,
 397                                struct ll_sai_entry, se_list);
 398         cfs_list_del_init(&entry->se_list);
 399         cfs_spin_unlock(&lli->lli_sa_lock);
 400
 401         if (unlikely(entry->se_index < sai->sai_index_next)) {
 402                 CWARN("Found stale entry: [index %u] [next %u]\n",
 403                       entry->se_index, sai->sai_index_next);
 404                 ll_sai_entry_cleanup(entry, 1);
 405                 RETURN(0);
 406         }
 407
 408         if (entry->se_stat != SA_ENTRY_STATED)
 409                 GOTO(out, rc = entry->se_stat);
 410
 411         req = entry->se_req;
 412         minfo = entry->se_minfo;
 413         it = &minfo->mi_it;
 414         dentry = minfo->mi_dentry;
 415
 416         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 417         if (body == NULL)
 418                 GOTO(out, rc = -EFAULT);
 419
 420         if (dentry->d_inode == NULL) {
 421                 /*
 422                  * lookup.
 423                  */
 424                 struct dentry    *save = dentry;
 425                 struct it_cb_data icbd = {
 426                         .icbd_parent   = minfo->mi_dir,
 427                         .icbd_childp   = &dentry
 428                 };
 429
 430                 LASSERT(fid_is_zero(&minfo->mi_data.op_fid2));
 431
 432                 /* XXX: No fid in reply, this is probaly cross-ref case.
 433                  * SA can't handle it yet. */
 434                 if (body->valid & OBD_MD_MDS)
 435                         GOTO(out, rc = -EAGAIN);
 436
 437                 /* Here dentry->d_inode might be NULL, because the entry may
 438                  * have been removed before we start doing stat ahead. */
 439                 rc = ll_lookup_it_finish(req, it, &icbd);
 440                 if (!rc)
 441                         ll_lookup_finish_locks(it, dentry);
 442
 443                 if (dentry != save) {
 444                         minfo->mi_dentry = dentry;
 445                         dput(save);
 446                 }
 447         } else {
 448                 /*
 449                  * revalidate.
 450                  */
 451                 if (!lu_fid_eq(&minfo->mi_data.op_fid2, &body->fid1)) {
 452                         ll_unhash_aliases(dentry->d_inode);
 453                         GOTO(out, rc = -EAGAIN);
 454                 }
 455
 456                 rc = ll_revalidate_it_finish(req, it, dentry);
 457                 if (rc) {
 458                         ll_unhash_aliases(dentry->d_inode);
 459                         GOTO(out, rc);
 460                 }
 461
 462                 cfs_spin_lock(&ll_lookup_lock);
 463                 spin_lock(&dcache_lock);
 464                 lock_dentry(dentry);
 465                 __d_drop(dentry);
 466                 dentry->d_flags &= ~DCACHE_LUSTRE_INVALID;
 467                 unlock_dentry(dentry);
 468                 d_rehash_cond(dentry, 0);
 469                 spin_unlock(&dcache_lock);
 470                 cfs_spin_unlock(&ll_lookup_lock);
 471
 472                 ll_lookup_finish_locks(it, dentry);
 473         }
 474         EXIT;
 475
 476 out:
 477         /* The "ll_sai_entry_to_stated()" will drop related ldlm ibits lock
 478          * reference count with ll_intent_drop_lock() called in spite of the
 479          * above operations failed or not. Do not worry about calling
 480          * "ll_intent_drop_lock()" more than once. */
 481         if (likely(ll_sai_entry_to_stated(sai, entry)))
 482                 cfs_waitq_signal(&sai->sai_waitq);
 483         return rc;
 484 }
 485
 486 static int ll_statahead_interpret(struct ptlrpc_request *req,
 487                                   struct md_enqueue_info *minfo,
 488                                   int rc)
 489 {
 490         struct lookup_intent     *it = &minfo->mi_it;
 491         struct dentry            *dentry = minfo->mi_dentry;
 492         struct inode             *dir = minfo->mi_dir;
 493         struct ll_inode_info     *lli = ll_i2info(dir);
 494         struct ll_statahead_info *sai;
 495         struct ll_sai_entry      *entry;
 496         ENTRY;
 497
 498         CDEBUG(D_READA, "interpret statahead %.*s rc %d\n",
 499                dentry->d_name.len, dentry->d_name.name, rc);
 500
 501         cfs_spin_lock(&lli->lli_sa_lock);
 502         /* stale entry */
 503         if (unlikely(lli->lli_sai == NULL ||
 504             lli->lli_sai->sai_generation != minfo->mi_generation)) {
 505                 cfs_spin_unlock(&lli->lli_sa_lock);
 506                 ll_intent_release(it);
 507                 dput(dentry);
 508                 iput(dir);
 509                 OBD_FREE_PTR(minfo);
 510                 RETURN(-ESTALE);
 511         } else {
 512                 sai = ll_sai_get(lli->lli_sai);
 513                 entry = ll_sai_entry_set(sai,
 514                                          (unsigned int)(long)minfo->mi_cbdata,
 515                                          rc < 0 ? rc : SA_ENTRY_STATED, req,
 516                                          minfo);
 517                 LASSERT(entry != NULL);
 518                 if (likely(sa_is_running(sai))) {
 519                         ll_sai_entry_to_received(sai, entry);
 520                         sai->sai_replied++;
 521                         cfs_spin_unlock(&lli->lli_sa_lock);
 522                         cfs_waitq_signal(&sai->sai_thread.t_ctl_waitq);
 523                 } else {
 524                         if (!cfs_list_empty(&entry->se_list))
 525                                 cfs_list_del_init(&entry->se_list);
 526                         sai->sai_replied++;
 527                         cfs_spin_unlock(&lli->lli_sa_lock);
 528                         ll_sai_entry_cleanup(entry, 1);
 529                 }
 530                 ll_sai_put(sai);
 531                 RETURN(rc);
 532         }
 533 }
 534
 535 static void sa_args_fini(struct md_enqueue_info *minfo,
 536                          struct ldlm_enqueue_info *einfo)
 537 {
 538         LASSERT(minfo && einfo);
 539         iput(minfo->mi_dir);
 540         capa_put(minfo->mi_data.op_capa1);
 541         capa_put(minfo->mi_data.op_capa2);
 542         OBD_FREE_PTR(minfo);
 543         OBD_FREE_PTR(einfo);
 544 }
 545
 546 /**
 547  * There is race condition between "capa_put" and "ll_statahead_interpret" for
 548  * accessing "op_data.op_capa[1,2]" as following:
 549  * "capa_put" releases "op_data.op_capa[1,2]"'s reference count after calling
 550  * "md_intent_getattr_async". But "ll_statahead_interpret" maybe run first, and
 551  * fill "op_data.op_capa[1,2]" as POISON, then cause "capa_put" access invalid
 552  * "ocapa". So here reserve "op_data.op_capa[1,2]" in "pcapa" before calling
 553  * "md_intent_getattr_async".
 554  */
 555 static int sa_args_init(struct inode *dir, struct dentry *dentry,
 556                         struct md_enqueue_info **pmi,
 557                         struct ldlm_enqueue_info **pei,
 558                         struct obd_capa **pcapa)
 559 {
 560         struct ll_inode_info     *lli = ll_i2info(dir);
 561         struct md_enqueue_info   *minfo;
 562         struct ldlm_enqueue_info *einfo;
 563         struct md_op_data        *op_data;
 564
 565         OBD_ALLOC_PTR(einfo);
 566         if (einfo == NULL)
 567                 return -ENOMEM;
 568
 569         OBD_ALLOC_PTR(minfo);
 570         if (minfo == NULL) {
 571                 OBD_FREE_PTR(einfo);
 572                 return -ENOMEM;
 573         }
 574
 575         op_data = ll_prep_md_op_data(&minfo->mi_data, dir, dentry->d_inode,
 576                                      dentry->d_name.name, dentry->d_name.len,
 577                                      0, LUSTRE_OPC_ANY, NULL);
 578         if (IS_ERR(op_data)) {
 579                 OBD_FREE_PTR(einfo);
 580                 OBD_FREE_PTR(minfo);
 581                 return PTR_ERR(op_data);
 582         }
 583
 584         minfo->mi_it.it_op = IT_GETATTR;
 585         minfo->mi_dentry = dentry;
 586         minfo->mi_dir = igrab(dir);
 587         minfo->mi_cb = ll_statahead_interpret;
 588         minfo->mi_generation = lli->lli_sai->sai_generation;
 589         minfo->mi_cbdata = (void *)(long)lli->lli_sai->sai_index;
 590
 591         einfo->ei_type   = LDLM_IBITS;
 592         einfo->ei_mode   = it_to_lock_mode(&minfo->mi_it);
 593         einfo->ei_cb_bl  = ll_md_blocking_ast;
 594         einfo->ei_cb_cp  = ldlm_completion_ast;
 595         einfo->ei_cb_gl  = NULL;
 596         einfo->ei_cbdata = NULL;
 597
 598         *pmi = minfo;
 599         *pei = einfo;
 600         pcapa[0] = op_data->op_capa1;
 601         pcapa[1] = op_data->op_capa2;
 602
 603         return 0;
 604 }
 605
 606 /**
 607  * similar to ll_lookup_it().
 608  */
 609 static int do_sa_lookup(struct inode *dir, struct dentry *dentry)
 610 {
 611         struct md_enqueue_info   *minfo;
 612         struct ldlm_enqueue_info *einfo;
 613         struct obd_capa          *capas[2];
 614         int                       rc;
 615         ENTRY;
 616
 617         rc = sa_args_init(dir, dentry, &minfo, &einfo, capas);
 618         if (rc)
 619                 RETURN(rc);
 620
 621         rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo);
 622         if (!rc) {
 623                 capa_put(capas[0]);
 624                 capa_put(capas[1]);
 625         } else {
 626                 sa_args_fini(minfo, einfo);
 627         }
 628
 629         RETURN(rc);
 630 }
 631
 632 /**
 633  * similar to ll_revalidate_it().
 634  * \retval      1 -- dentry valid
 635  * \retval      0 -- will send stat-ahead request
 636  * \retval others -- prepare stat-ahead request failed
 637  */
 638 static int do_sa_revalidate(struct inode *dir, struct dentry *dentry)
 639 {
 640         struct inode             *inode = dentry->d_inode;
 641         struct lookup_intent      it = { .it_op = IT_GETATTR };
 642         struct md_enqueue_info   *minfo;
 643         struct ldlm_enqueue_info *einfo;
 644         struct obd_capa          *capas[2];
 645         int rc;
 646         ENTRY;
 647
 648         if (unlikely(inode == NULL))
 649                 RETURN(1);
 650
 651         if (d_mountpoint(dentry))
 652                 RETURN(1);
 653
 654         if (unlikely(dentry == dentry->d_sb->s_root))
 655                 RETURN(1);
 656
 657         rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode));
 658         if (rc == 1) {
 659                 ll_intent_release(&it);
 660                 RETURN(1);
 661         }
 662
 663         rc = sa_args_init(dir, dentry, &minfo, &einfo, capas);
 664         if (rc)
 665                 RETURN(rc);
 666
 667         rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo);
 668         if (!rc) {
 669                 capa_put(capas[0]);
 670                 capa_put(capas[1]);
 671         } else {
 672                 sa_args_fini(minfo, einfo);
 673         }
 674
 675         RETURN(rc);
 676 }
 677
 678 static inline void ll_name2qstr(struct qstr *q, const char *name, int namelen)
 679 {
 680         q->name = name;
 681         q->len  = namelen;
 682         q->hash = full_name_hash(name, namelen);
 683 }
 684
 685 static int ll_statahead_one(struct dentry *parent, const char* entry_name,
 686                             int entry_name_len)
 687 {
 688         struct inode             *dir = parent->d_inode;
 689         struct ll_inode_info     *lli = ll_i2info(dir);
 690         struct ll_statahead_info *sai = lli->lli_sai;
 691         struct qstr               name;
 692         struct dentry            *dentry = NULL;
 693         struct ll_sai_entry      *se;
 694         int                       rc;
 695         ENTRY;
 696
 697         if (parent->d_flags & DCACHE_LUSTRE_INVALID) {
 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                 else
 715                         GOTO(out, rc = -ENOMEM);
 716         } else {
 717                 rc = do_sa_revalidate(dir, dentry);
 718         }
 719
 720         EXIT;
 721
 722 out:
 723         if (rc) {
 724                 if (dentry != NULL)
 725                         dput(dentry);
 726                 se->se_stat = rc < 0 ? rc : SA_ENTRY_STATED;
 727                 CDEBUG(D_READA, "set sai entry %p index %u stat %d rc %d\n",
 728                        se, se->se_index, se->se_stat, rc);
 729                 if (ll_sai_entry_to_stated(sai, se))
 730                         cfs_waitq_signal(&sai->sai_waitq);
 731         } else {
 732                 sai->sai_sent++;
 733         }
 734
 735         sai->sai_index++;
 736         return rc;
 737 }
 738
 739 static int ll_statahead_thread(void *arg)
 740 {
 741         struct dentry            *parent = (struct dentry *)arg;
 742         struct inode             *dir = parent->d_inode;
 743         struct ll_inode_info     *lli = ll_i2info(dir);
 744         struct ll_sb_info        *sbi = ll_i2sbi(dir);
 745         struct ll_statahead_info *sai = ll_sai_get(lli->lli_sai);
 746         struct ptlrpc_thread     *thread = &sai->sai_thread;
 747         struct page              *page;
 748         __u64                     pos = 0;
 749         int                       first = 0;
 750         int                       rc = 0;
 751         struct ll_dir_chain       chain;
 752         ENTRY;
 753
 754         {
 755                 char pname[16];
 756                 snprintf(pname, 15, "ll_sa_%u", lli->lli_opendir_pid);
 757                 cfs_daemonize(pname);
 758         }
 759
 760         atomic_inc(&sbi->ll_sa_total);
 761         cfs_spin_lock(&lli->lli_sa_lock);
 762         thread->t_flags = SVC_RUNNING;
 763         cfs_spin_unlock(&lli->lli_sa_lock);
 764         cfs_waitq_signal(&thread->t_ctl_waitq);
 765         CDEBUG(D_READA, "start doing statahead for %s\n", parent->d_name.name);
 766
 767         sai->sai_pid = cfs_curproc_pid();
 768         lli->lli_sa_pos = 0;
 769         ll_dir_chain_init(&chain);
 770         page = ll_get_dir_page(NULL, dir, pos, 0, &chain);
 771
 772         while (1) {
 773                 struct l_wait_info lwi = { 0 };
 774                 struct lu_dirpage *dp;
 775                 struct lu_dirent  *ent;
 776
 777                 if (IS_ERR(page)) {
 778                         rc = PTR_ERR(page);
 779                         CDEBUG(D_READA, "error reading dir "DFID" at "LPU64
 780                                "/%u: [rc %d] [parent %u]\n",
 781                                PFID(ll_inode2fid(dir)), pos, sai->sai_index,
 782                                rc, lli->lli_opendir_pid);
 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                         __u64 hash;
 790                         int namelen;
 791                         char *name;
 792
 793                         hash = le64_to_cpu(ent->lde_hash);
 794                         if (unlikely(hash < pos))
 795                                 /*
 796                                  * Skip until we find target hash value.
 797                                  */
 798                                 continue;
 799
 800                         namelen = le16_to_cpu(ent->lde_namelen);
 801                         if (unlikely(namelen == 0))
 802                                 /*
 803                                  * Skip dummy record.
 804                                  */
 805                                 continue;
 806
 807                         name = ent->lde_name;
 808                         if (name[0] == '.') {
 809                                 if (namelen == 1) {
 810                                         /*
 811                                          * skip "."
 812                                          */
 813                                         continue;
 814                                 } else if (name[1] == '.' && namelen == 2) {
 815                                         /*
 816                                          * skip ".."
 817                                          */
 818                                         continue;
 819                                 } else if (!sai->sai_ls_all) {
 820                                         /*
 821                                          * skip hidden files.
 822                                          */
 823                                         sai->sai_skip_hidden++;
 824                                         continue;
 825                                 }
 826                         }
 827
 828                         /*
 829                          * don't stat-ahead first entry.
 830                          */
 831                         if (unlikely(!first)) {
 832                                 first++;
 833                                 continue;
 834                         }
 835
 836 keep_de:
 837                         l_wait_event(thread->t_ctl_waitq,
 838                                      !sa_is_running(sai) || sa_not_full(sai) ||
 839                                      !sa_received_empty(sai),
 840                                      &lwi);
 841
 842                         while (!sa_received_empty(sai) && sa_is_running(sai))
 843                                 do_statahead_interpret(sai);
 844
 845                         if (unlikely(!sa_is_running(sai))) {
 846                                 ll_put_page(page);
 847                                 GOTO(out, rc);
 848                         }
 849
 850                         if (!sa_not_full(sai))
 851                                 /*
 852                                  * do not skip the current de.
 853                                  */
 854                                 goto keep_de;
 855
 856                         rc = ll_statahead_one(parent, name, namelen);
 857                         if (rc < 0) {
 858                                 ll_put_page(page);
 859                                 GOTO(out, rc);
 860                         }
 861                 }
 862                 pos = le64_to_cpu(dp->ldp_hash_end);
 863                 ll_put_page(page);
 864                 if (pos == DIR_END_OFF) {
 865                         /*
 866                          * End of directory reached.
 867                          */
 868                         while (1) {
 869                                 l_wait_event(thread->t_ctl_waitq,
 870                                              !sa_is_running(sai) ||
 871                                              !sa_received_empty(sai) ||
 872                                              sai->sai_sent == sai->sai_replied,
 873                                              &lwi);
 874                                 if (!sa_received_empty(sai) &&
 875                                     sa_is_running(sai))
 876                                         do_statahead_interpret(sai);
 877                                 else
 878                                         GOTO(out, rc);
 879                         }
 880                 } else if (1) {
 881                         /*
 882                          * chain is exhausted.
 883                          * Normal case: continue to the next page.
 884                          */
 885                         lli->lli_sa_pos = pos;
 886                         page = ll_get_dir_page(NULL, dir, pos, 1, &chain);
 887                 } else {
 888                         /*
 889                          * go into overflow page.
 890                          */
 891                 }
 892         }
 893         EXIT;
 894
 895 out:
 896         ll_dir_chain_fini(&chain);
 897         cfs_spin_lock(&lli->lli_sa_lock);
 898         thread->t_flags = SVC_STOPPED;
 899         cfs_spin_unlock(&lli->lli_sa_lock);
 900         cfs_waitq_signal(&sai->sai_waitq);
 901         cfs_waitq_signal(&thread->t_ctl_waitq);
 902         ll_sai_put(sai);
 903         dput(parent);
 904         CDEBUG(D_READA, "statahead thread stopped, pid %d\n",
 905                cfs_curproc_pid());
 906         return rc;
 907 }
 908
 909 /**
 910  * called in ll_file_release().
 911  */
 912 void ll_stop_statahead(struct inode *dir, void *key)
 913 {
 914         struct ll_inode_info *lli = ll_i2info(dir);
 915
 916         if (unlikely(key == NULL))
 917                 return;
 918
 919         cfs_spin_lock(&lli->lli_sa_lock);
 920         if (lli->lli_opendir_key != key || lli->lli_opendir_pid == 0) {
 921                 cfs_spin_unlock(&lli->lli_sa_lock);
 922                 return;
 923         }
 924
 925         lli->lli_opendir_key = NULL;
 926
 927         if (lli->lli_sai) {
 928                 struct l_wait_info lwi = { 0 };
 929                 struct ptlrpc_thread *thread = &lli->lli_sai->sai_thread;
 930
 931                 if (!sa_is_stopped(lli->lli_sai)) {
 932                         thread->t_flags = SVC_STOPPING;
 933                         cfs_spin_unlock(&lli->lli_sa_lock);
 934                         cfs_waitq_signal(&thread->t_ctl_waitq);
 935
 936                         CDEBUG(D_READA, "stopping statahead thread, pid %d\n",
 937                                cfs_curproc_pid());
 938                         l_wait_event(thread->t_ctl_waitq,
 939                                      sa_is_stopped(lli->lli_sai),
 940                                      &lwi);
 941                 } else {
 942                         cfs_spin_unlock(&lli->lli_sa_lock);
 943                 }
 944
 945                 /*
 946                  * Put the ref which was held when first statahead_enter.
 947                  * It maybe not the last ref for some statahead requests
 948                  * maybe inflight.
 949                  */
 950                 ll_sai_put(lli->lli_sai);
 951         } else {
 952                 lli->lli_opendir_pid = 0;
 953                 cfs_spin_unlock(&lli->lli_sa_lock);
 954         }
 955 }
 956
 957 enum {
 958         /**
 959          * not first dirent, or is "."
 960          */
 961         LS_NONE_FIRST_DE = 0,
 962         /**
 963          * the first non-hidden dirent
 964          */
 965         LS_FIRST_DE,
 966         /**
 967          * the first hidden dirent, that is "."
 968          */
 969         LS_FIRST_DOT_DE
 970 };
 971
 972 static int is_first_dirent(struct inode *dir, struct dentry *dentry)
 973 {
 974         struct ll_inode_info *lli = ll_i2info(dir);
 975         struct ll_dir_chain chain;
 976         struct qstr        *target = &dentry->d_name;
 977         struct page        *page;
 978         __u64               pos = 0;
 979         int                 dot_de;
 980         int                 rc = LS_NONE_FIRST_DE;
 981         ENTRY;
 982
 983         lli->lli_sa_pos = 0;
 984         ll_dir_chain_init(&chain);
 985         page = ll_get_dir_page(NULL, dir, pos, 0, &chain);
 986
 987         while (1) {
 988                 struct lu_dirpage *dp;
 989                 struct lu_dirent  *ent;
 990
 991                 if (IS_ERR(page)) {
 992                         struct ll_inode_info *lli = ll_i2info(dir);
 993
 994                         rc = PTR_ERR(page);
 995                         CERROR("error reading dir "DFID" at "LPU64": "
 996                                "[rc %d] [parent %u]\n",
 997                                PFID(ll_inode2fid(dir)), pos,
 998                                rc, lli->lli_opendir_pid);
 999                         break;
1000                 }
1001
1002                 dp = page_address(page);
1003                 for (ent = lu_dirent_start(dp); ent != NULL;
1004                      ent = lu_dirent_next(ent)) {
1005                         int namelen;
1006                         char *name;
1007
1008                         namelen = le16_to_cpu(ent->lde_namelen);
1009                         if (unlikely(namelen == 0))
1010                                 /*
1011                                  * skip dummy record.
1012                                  */
1013                                 continue;
1014
1015                         name = ent->lde_name;
1016                         if (name[0] == '.') {
1017                                 if (namelen == 1)
1018                                         /*
1019                                          * skip "."
1020                                          */
1021                                         continue;
1022                                 else if (name[1] == '.' && namelen == 2)
1023                                         /*
1024                                          * skip ".."
1025                                          */
1026                                         continue;
1027                                 else
1028                                         dot_de = 1;
1029                         } else {
1030                                 dot_de = 0;
1031                         }
1032
1033                         if (dot_de && target->name[0] != '.') {
1034                                 CDEBUG(D_READA, "%.*s skip hidden file %.*s\n",
1035                                        target->len, target->name,
1036                                        namelen, name);
1037                                 continue;
1038                         }
1039
1040                         if (target->len != namelen ||
1041                             memcmp(target->name, name, namelen) != 0)
1042                                 rc = LS_NONE_FIRST_DE;
1043                         else if (!dot_de)
1044                                 rc = LS_FIRST_DE;
1045                         else
1046                                 rc = LS_FIRST_DOT_DE;
1047
1048                         ll_put_page(page);
1049                         GOTO(out, rc);
1050                 }
1051                 pos = le64_to_cpu(dp->ldp_hash_end);
1052                 ll_put_page(page);
1053                 if (pos == DIR_END_OFF) {
1054                         /*
1055                          * End of directory reached.
1056                          */
1057                         break;
1058                 } else if (1) {
1059                         /*
1060                          * chain is exhausted
1061                          * Normal case: continue to the next page.
1062                          */
1063                         lli->lli_sa_pos = pos;
1064                         page = ll_get_dir_page(NULL, dir, pos, 1, &chain);
1065                 } else {
1066                         /*
1067                          * go into overflow page.
1068                          */
1069                 }
1070         }
1071         EXIT;
1072
1073 out:
1074         ll_dir_chain_fini(&chain);
1075         return rc;
1076 }
1077
1078 /**
1079  * Start statahead thread if this is the first dir entry.
1080  * Otherwise if a thread is started already, wait it until it is ahead of me.
1081  * \retval 0       -- stat ahead thread process such dentry, for lookup, it miss
1082  * \retval 1       -- stat ahead thread process such dentry, for lookup, it hit
1083  * \retval -EEXIST -- stat ahead thread started, and this is the first dentry
1084  * \retval -EBADFD -- statahead thread exit and not dentry available
1085  * \retval -EAGAIN -- try to stat by caller
1086  * \retval others  -- error
1087  */
1088 int do_statahead_enter(struct inode *dir, struct dentry **dentryp, int lookup)
1089 {
1090         struct ll_inode_info     *lli;
1091         struct ll_statahead_info *sai;
1092         struct dentry            *parent;
1093         struct l_wait_info        lwi = { 0 };
1094         int                       rc = 0;
1095         ENTRY;
1096
1097         LASSERT(dir != NULL);
1098         lli = ll_i2info(dir);
1099         LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
1100         sai = lli->lli_sai;
1101
1102         if (sai) {
1103                 if (unlikely(sa_is_stopped(sai) &&
1104                              cfs_list_empty(&sai->sai_entries_stated)))
1105                         RETURN(-EBADFD);
1106
1107                 if ((*dentryp)->d_name.name[0] == '.') {
1108                         if (likely(sai->sai_ls_all ||
1109                             sai->sai_miss_hidden >= sai->sai_skip_hidden)) {
1110                                 /*
1111                                  * Hidden dentry is the first one, or statahead
1112                                  * thread does not skip so many hidden dentries
1113                                  * before "sai_ls_all" enabled as below.
1114                                  */
1115                         } else {
1116                                 if (!sai->sai_ls_all)
1117                                         /*
1118                                          * It maybe because hidden dentry is not
1119                                          * the first one, "sai_ls_all" was not
1120                                          * set, then "ls -al" missed. Enable
1121                                          * "sai_ls_all" for such case.
1122                                          */
1123                                         sai->sai_ls_all = 1;
1124
1125                                 /*
1126                                  * Such "getattr" has been skipped before
1127                                  * "sai_ls_all" enabled as above.
1128                                  */
1129                                 sai->sai_miss_hidden++;
1130                                 RETURN(-ENOENT);
1131                         }
1132                 }
1133
1134                 if (!ll_sai_entry_stated(sai)) {
1135                         /*
1136                          * thread started already, avoid double-stat.
1137                          */
1138                         lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
1139                         rc = l_wait_event(sai->sai_waitq,
1140                                           ll_sai_entry_stated(sai) ||
1141                                           sa_is_stopped(sai),
1142                                           &lwi);
1143                         if (unlikely(rc == -EINTR))
1144                                 RETURN(rc);
1145                 }
1146
1147                 if (lookup) {
1148                         struct dentry *result;
1149
1150                         result = d_lookup((*dentryp)->d_parent,
1151                                           &(*dentryp)->d_name);
1152                         if (result) {
1153                                 LASSERT(result != *dentryp);
1154                                 /* BUG 16303: do not drop reference count for
1155                                  * "*dentryp", VFS will do that by itself. */
1156                                 *dentryp = result;
1157                                 RETURN(1);
1158                         }
1159                 }
1160                 /*
1161                  * do nothing for revalidate.
1162                  */
1163                 RETURN(0);
1164         }
1165
1166         /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */
1167         rc = is_first_dirent(dir, *dentryp);
1168         if (rc == LS_NONE_FIRST_DE)
1169                 /* It is not "ls -{a}l" operation, no need statahead for it. */
1170                 GOTO(out, rc = -EAGAIN);
1171
1172         sai = ll_sai_alloc();
1173         if (sai == NULL)
1174                 GOTO(out, rc = -ENOMEM);
1175
1176         sai->sai_ls_all = (rc == LS_FIRST_DOT_DE);
1177         sai->sai_inode = igrab(dir);
1178         if (unlikely(sai->sai_inode == NULL)) {
1179                 CWARN("Do not start stat ahead on dying inode "DFID"\n",
1180                       PFID(&lli->lli_fid));
1181                 OBD_FREE_PTR(sai);
1182                 GOTO(out, rc = -ESTALE);
1183         }
1184
1185         /* get parent reference count here, and put it in ll_statahead_thread */
1186         parent = dget((*dentryp)->d_parent);
1187         if (unlikely(sai->sai_inode != parent->d_inode)) {
1188                 struct ll_inode_info *nlli = ll_i2info(parent->d_inode);
1189
1190                 CWARN("Race condition, someone changed %.*s just now: "
1191                       "old parent "DFID", new parent "DFID"\n",
1192                       (*dentryp)->d_name.len, (*dentryp)->d_name.name,
1193                       PFID(&lli->lli_fid), PFID(&nlli->lli_fid));
1194                 dput(parent);
1195                 iput(sai->sai_inode);
1196                 OBD_FREE_PTR(sai);
1197                 RETURN(-EAGAIN);
1198         }
1199
1200         lli->lli_sai = sai;
1201         rc = cfs_kernel_thread(ll_statahead_thread, parent, 0);
1202         if (rc < 0) {
1203                 CERROR("can't start ll_sa thread, rc: %d\n", rc);
1204                 dput(parent);
1205                 lli->lli_opendir_key = NULL;
1206                 sai->sai_thread.t_flags = SVC_STOPPED;
1207                 ll_sai_put(sai);
1208                 LASSERT(lli->lli_sai == NULL);
1209                 RETURN(-EAGAIN);
1210         }
1211
1212         l_wait_event(sai->sai_thread.t_ctl_waitq,
1213                      sa_is_running(sai) || sa_is_stopped(sai),
1214                      &lwi);
1215
1216         /*
1217          * We don't stat-ahead for the first dirent since we are already in
1218          * lookup, and -EEXIST also indicates that this is the first dirent.
1219          */
1220         RETURN(-EEXIST);
1221
1222 out:
1223         cfs_spin_lock(&lli->lli_sa_lock);
1224         lli->lli_opendir_key = NULL;
1225         lli->lli_opendir_pid = 0;
1226         cfs_spin_unlock(&lli->lli_sa_lock);
1227         return rc;
1228 }
1229
1230 /**
1231  * update hit/miss count.
1232  */
1233 void ll_statahead_exit(struct inode *dir, struct dentry *dentry, int result)
1234 {
1235         struct ll_inode_info     *lli;
1236         struct ll_statahead_info *sai;
1237         struct ll_sb_info        *sbi;
1238         struct ll_dentry_data    *ldd = ll_d2d(dentry);
1239         int                       rc;
1240         ENTRY;
1241
1242         LASSERT(dir != NULL);
1243         lli = ll_i2info(dir);
1244         LASSERT(lli->lli_opendir_pid == cfs_curproc_pid());
1245         sai = lli->lli_sai;
1246         LASSERT(sai != NULL);
1247         sbi = ll_i2sbi(dir);
1248
1249         rc = ll_sai_entry_fini(sai);
1250         /* rc == -ENOENT means such dentry was removed just between statahead
1251          * readdir and pre-fetched, count it as hit.
1252          *
1253          * result == -ENOENT has two meanings:
1254          * 1. such dentry was removed just between statahead pre-fetched and
1255          *    main process stat such dentry.
1256          * 2. main process stat non-exist dentry.
1257          * We can not distinguish such two cases, just count them as miss. */
1258         if (result >= 1 || unlikely(rc == -ENOENT)) {
1259                 sai->sai_hit++;
1260                 sai->sai_consecutive_miss = 0;
1261                 sai->sai_max = min(2 * sai->sai_max, sbi->ll_sa_max);
1262         } else {
1263                 sai->sai_miss++;
1264                 sai->sai_consecutive_miss++;
1265                 if (sa_low_hit(sai) && sa_is_running(sai)) {
1266                         atomic_inc(&sbi->ll_sa_wrong);
1267                         CDEBUG(D_READA, "Statahead for dir "DFID" hit ratio "
1268                                "too low: hit/miss %u/%u, sent/replied %u/%u, "
1269                                "stopping statahead thread: pid %d\n",
1270                                PFID(&lli->lli_fid), sai->sai_hit,
1271                                sai->sai_miss, sai->sai_sent,
1272                                sai->sai_replied, cfs_curproc_pid());
1273                         cfs_spin_lock(&lli->lli_sa_lock);
1274                         if (!sa_is_stopped(sai))
1275                                 sai->sai_thread.t_flags = SVC_STOPPING;
1276                         cfs_spin_unlock(&lli->lli_sa_lock);
1277                 }
1278         }
1279
1280         if (!sa_is_stopped(sai))
1281                 cfs_waitq_signal(&sai->sai_thread.t_ctl_waitq);
1282         if (likely(ldd != NULL))
1283                 ldd->lld_sa_generation = sai->sai_generation;
1284
1285         EXIT;
1286 }