lustre/llite/statahead.c

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