From: Lai Siyao <lai.si...@intel.com> Statahead thread should wait for inflight stat RPCs to finish in case statahead RPC callback may access data allocated in statahead thread context.
ll_sa_entry_fini() should keep old entry if stat RPC is not finished yet. Simplify sai refcounting: * newly allocated sai will hold one refcount, and it will put it after starting statahead thread. * statahead thread holds one refcount. * agl thread holds one refcount. * stat process calls do_statahead_enter() which will try to get sai, and if it's valid, it will revalidate from statahead cache, and put refcount after use. Signed-off-by: Lai Siyao <lai.si...@intel.com> Intel-bug-id: https://jira.hpdd.intel.com/browse/LU-3270 Reviewed-on: http://review.whamcloud.com/9663 Reviewed-by: Fan Yong <fan.y...@intel.com> Reviewed-by: James Simmons <uja.o...@gmail.com> Reviewed-by: Oleg Drokin <oleg.dro...@intel.com> Signed-off-by: James Simmons <jsimm...@infradead.org> --- drivers/staging/lustre/lustre/include/obd.h | 1 - drivers/staging/lustre/lustre/llite/dcache.c | 2 +- drivers/staging/lustre/lustre/llite/file.c | 31 +- .../staging/lustre/lustre/llite/llite_internal.h | 49 +- drivers/staging/lustre/lustre/llite/llite_lib.c | 8 + drivers/staging/lustre/lustre/llite/statahead.c | 849 +++++++++----------- 6 files changed, 434 insertions(+), 506 deletions(-) diff --git a/drivers/staging/lustre/lustre/include/obd.h b/drivers/staging/lustre/lustre/include/obd.h index 838a428..89633f7 100644 --- a/drivers/staging/lustre/lustre/include/obd.h +++ b/drivers/staging/lustre/lustre/include/obd.h @@ -806,7 +806,6 @@ struct md_enqueue_info { int (*mi_cb)(struct ptlrpc_request *req, struct md_enqueue_info *minfo, int rc); __u64 mi_cbdata; - unsigned int mi_generation; }; struct obd_ops { diff --git a/drivers/staging/lustre/lustre/llite/dcache.c b/drivers/staging/lustre/lustre/llite/dcache.c index f4b6f38..8c00cc6 100644 --- a/drivers/staging/lustre/lustre/llite/dcache.c +++ b/drivers/staging/lustre/lustre/llite/dcache.c @@ -279,7 +279,7 @@ static int ll_revalidate_dentry(struct dentry *dentry, if (lookup_flags & (LOOKUP_PARENT | LOOKUP_OPEN | LOOKUP_CREATE)) return 1; - if (d_need_statahead(dir, dentry) <= 0) + if (!dentry_need_statahead(dir, dentry)) return 1; if (lookup_flags & LOOKUP_RCU) diff --git a/drivers/staging/lustre/lustre/llite/file.c b/drivers/staging/lustre/lustre/llite/file.c index e9791e3..273b563 100644 --- a/drivers/staging/lustre/lustre/llite/file.c +++ b/drivers/staging/lustre/lustre/llite/file.c @@ -351,13 +351,11 @@ int ll_file_release(struct inode *inode, struct file *file) fd = LUSTRE_FPRIVATE(file); LASSERT(fd); - /* The last ref on @file, maybe not be the owner pid of statahead. - * Different processes can open the same dir, "ll_opendir_key" means: - * it is me that should stop the statahead thread. + /* The last ref on @file, maybe not be the owner pid of statahead, + * because parent and child process can share the same file handle. */ - if (S_ISDIR(inode->i_mode) && lli->lli_opendir_key == fd && - lli->lli_opendir_pid != 0) - ll_stop_statahead(inode, lli->lli_opendir_key); + if (S_ISDIR(inode->i_mode) && lli->lli_opendir_key == fd) + ll_deauthorize_statahead(inode, fd); if (is_root_inode(inode)) { LUSTRE_FPRIVATE(file) = NULL; @@ -530,7 +528,7 @@ int ll_file_open(struct inode *inode, struct file *file) struct obd_client_handle **och_p = NULL; __u64 *och_usecount = NULL; struct ll_file_data *fd; - int rc = 0, opendir_set = 0; + int rc = 0; CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p), flags %o\n", PFID(ll_inode2fid(inode)), inode, file->f_flags); @@ -545,16 +543,8 @@ int ll_file_open(struct inode *inode, struct file *file) } fd->fd_file = file; - if (S_ISDIR(inode->i_mode)) { - spin_lock(&lli->lli_sa_lock); - if (!lli->lli_opendir_key && !lli->lli_sai && - lli->lli_opendir_pid == 0) { - lli->lli_opendir_key = fd; - lli->lli_opendir_pid = current_pid(); - opendir_set = 1; - } - spin_unlock(&lli->lli_sa_lock); - } + if (S_ISDIR(inode->i_mode)) + ll_authorize_statahead(inode, fd); if (is_root_inode(inode)) { LUSTRE_FPRIVATE(file) = fd; @@ -713,9 +703,10 @@ out_och_free: mutex_unlock(&lli->lli_och_mutex); out_openerr: - if (opendir_set != 0) - ll_stop_statahead(inode, lli->lli_opendir_key); - ll_file_data_put(fd); + if (lli->lli_opendir_key == fd) + ll_deauthorize_statahead(inode, fd); + if (fd) + ll_file_data_put(fd); } else { ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_OPEN, 1); } diff --git a/drivers/staging/lustre/lustre/llite/llite_internal.h b/drivers/staging/lustre/lustre/llite/llite_internal.h index cbd5bc5..f903f2a 100644 --- a/drivers/staging/lustre/lustre/llite/llite_internal.h +++ b/drivers/staging/lustre/lustre/llite/llite_internal.h @@ -172,6 +172,13 @@ struct ll_inode_info { * -- I am the owner of dir statahead. */ pid_t d_opendir_pid; + /* stat will try to access statahead entries or start + * statahead if this flag is set, and this flag will be + * set upon dir open, and cleared when dir is closed, + * statahead hit ratio is too low, or start statahead + * thread failed. + */ + unsigned int d_sa_enabled:1; /* directory stripe information */ struct lmv_stripe_md *d_lsm_md; /* striped directory size */ @@ -184,6 +191,7 @@ struct ll_inode_info { #define lli_opendir_key u.d.d_opendir_key #define lli_sai u.d.d_sai #define lli_sa_lock u.d.d_sa_lock +#define lli_sa_enabled u.d.d_sa_enabled #define lli_opendir_pid u.d.d_opendir_pid #define lli_lsm_md u.d.d_lsm_md #define lli_stripe_dir_size u.d.d_stripe_size @@ -495,6 +503,9 @@ struct ll_sb_info { atomic_t ll_sa_wrong; /* statahead thread stopped for * low hit ratio */ + atomic_t ll_sa_running; /* running statahead thread + * count + */ atomic_t ll_agl_total; /* AGL thread started count */ dev_t ll_sdev_orig; /* save s_dev before assign for @@ -1040,7 +1051,8 @@ struct ll_statahead_info { int do_statahead_enter(struct inode *dir, struct dentry **dentry, int only_unplug); -void ll_stop_statahead(struct inode *dir, void *key); +void ll_authorize_statahead(struct inode *dir, void *key); +void ll_deauthorize_statahead(struct inode *dir, void *key); blkcnt_t dirty_cnt(struct inode *inode); @@ -1086,25 +1098,31 @@ ll_statahead_mark(struct inode *dir, struct dentry *dentry) ldd->lld_sa_generation = sai->sai_generation; } -static inline int -d_need_statahead(struct inode *dir, struct dentry *dentryp) +static inline bool +dentry_need_statahead(struct inode *dir, struct dentry *dentry) { struct ll_inode_info *lli; struct ll_dentry_data *ldd; if (ll_i2sbi(dir)->ll_sa_max == 0) - return -EAGAIN; + return false; lli = ll_i2info(dir); + + /* + * statahead is not allowed for this dir, there may be three causes: + * 1. dir is not opened. + * 2. statahead hit ratio is too low. + * 3. previous stat started statahead thread failed. + */ + if (!lli->lli_sa_enabled) + return false; + /* not the same process, don't statahead */ if (lli->lli_opendir_pid != current_pid()) - return -EAGAIN; - - /* statahead has been stopped */ - if (!lli->lli_opendir_key) - return -EAGAIN; + return false; - ldd = ll_d2d(dentryp); + ldd = ll_d2d(dentry); /* * When stats a dentry, the system trigger more than once "revalidate" * or "lookup", for "getattr", for "getxattr", and maybe for others. @@ -1122,19 +1140,16 @@ d_need_statahead(struct inode *dir, struct dentry *dentryp) */ if (ldd && lli->lli_sai && ldd->lld_sa_generation == lli->lli_sai->sai_generation) - return -EAGAIN; + return false; - return 1; + return true; } static inline int ll_statahead_enter(struct inode *dir, struct dentry **dentryp, int only_unplug) { - int ret; - - ret = d_need_statahead(dir, *dentryp); - if (ret <= 0) - return ret; + if (!dentry_need_statahead(dir, *dentryp)) + return -EAGAIN; return do_statahead_enter(dir, dentryp, only_unplug); } diff --git a/drivers/staging/lustre/lustre/llite/llite_lib.c b/drivers/staging/lustre/lustre/llite/llite_lib.c index 99aba6b..93fd69b 100644 --- a/drivers/staging/lustre/lustre/llite/llite_lib.c +++ b/drivers/staging/lustre/lustre/llite/llite_lib.c @@ -116,6 +116,7 @@ static struct ll_sb_info *ll_init_sbi(struct super_block *sb) sbi->ll_sa_max = LL_SA_RPC_DEF; atomic_set(&sbi->ll_sa_total, 0); atomic_set(&sbi->ll_sa_wrong, 0); + atomic_set(&sbi->ll_sa_running, 0); atomic_set(&sbi->ll_agl_total, 0); sbi->ll_flags |= LL_SBI_AGL_ENABLED; @@ -630,6 +631,12 @@ void ll_kill_super(struct super_block *sb) if (sbi) { sb->s_dev = sbi->ll_sdev_orig; sbi->ll_umounting = 1; + + /* wait running statahead threads to quit */ + while (atomic_read(&sbi->ll_sa_running) > 0) { + set_current_state(TASK_UNINTERRUPTIBLE); + schedule_timeout(msecs_to_jiffies(MSEC_PER_SEC >> 3)); + } } } @@ -795,6 +802,7 @@ void ll_lli_init(struct ll_inode_info *lli) lli->lli_sai = NULL; spin_lock_init(&lli->lli_sa_lock); lli->lli_opendir_pid = 0; + lli->lli_sa_enabled = 0; } else { mutex_init(&lli->lli_size_mutex); lli->lli_symlink_name = NULL; diff --git a/drivers/staging/lustre/lustre/llite/statahead.c b/drivers/staging/lustre/lustre/llite/statahead.c index 016463b..6577a66 100644 --- a/drivers/staging/lustre/lustre/llite/statahead.c +++ b/drivers/staging/lustre/lustre/llite/statahead.c @@ -281,25 +281,6 @@ ll_sa_entry_get_byindex(struct ll_statahead_info *sai, __u64 index) return NULL; } -static void ll_sa_entry_cleanup(struct ll_statahead_info *sai, - struct ll_sa_entry *entry) -{ - struct md_enqueue_info *minfo = entry->se_minfo; - struct ptlrpc_request *req = entry->se_req; - - if (minfo) { - entry->se_minfo = NULL; - ll_intent_release(&minfo->mi_it); - iput(minfo->mi_dir); - kfree(minfo); - } - - if (req) { - entry->se_req = NULL; - ptlrpc_req_finished(req); - } -} - static void ll_sa_entry_put(struct ll_statahead_info *sai, struct ll_sa_entry *entry) { @@ -312,7 +293,6 @@ static void ll_sa_entry_put(struct ll_statahead_info *sai, LASSERT(list_empty(&entry->se_list)); LASSERT(list_empty(&entry->se_hash)); - ll_sa_entry_cleanup(sai, entry); iput(entry->se_inode); kfree(entry); @@ -355,7 +335,10 @@ ll_sa_entry_fini(struct ll_statahead_info *sai, struct ll_sa_entry *entry) list_for_each_entry_safe(pos, next, &sai->sai_entries, se_link) { if (!is_omitted_entry(sai, pos->se_index)) break; - do_sa_entry_fini(sai, pos); + /* keep those whose statahead RPC not finished */ + if (pos->se_stat == SA_ENTRY_SUCC || + pos->se_stat == SA_ENTRY_INVA) + do_sa_entry_fini(sai, pos); } } @@ -363,12 +346,14 @@ ll_sa_entry_fini(struct ll_statahead_info *sai, struct ll_sa_entry *entry) * Inside lli_sa_lock. */ static void -do_sa_entry_to_stated(struct ll_statahead_info *sai, - struct ll_sa_entry *entry, enum se_stat stat) +__sa_entry_post_stat(struct ll_statahead_info *sai, struct ll_sa_entry *entry, + enum se_stat stat) { struct ll_sa_entry *se; struct list_head *pos = &sai->sai_entries_stated; + LASSERT(entry->se_stat == SA_ENTRY_INIT); + if (!list_empty(&entry->se_list)) list_del_init(&entry->se_list); @@ -388,23 +373,30 @@ do_sa_entry_to_stated(struct ll_statahead_info *sai, * \retval 1 -- entry to be destroyed. * \retval 0 -- entry is inserted into stated list. */ -static int -ll_sa_entry_to_stated(struct ll_statahead_info *sai, - struct ll_sa_entry *entry, enum se_stat stat) +static void +sa_entry_post_stat(struct ll_statahead_info *sai, struct ll_sa_entry *entry, + enum se_stat stat) { struct ll_inode_info *lli = ll_i2info(sai->sai_inode); - int ret = 1; + struct md_enqueue_info *minfo = entry->se_minfo; + struct ptlrpc_request *req = entry->se_req; + + /* release resources used in RPC */ + if (minfo) { + entry->se_minfo = NULL; + ll_intent_release(&minfo->mi_it); + iput(minfo->mi_dir); + kfree(minfo); + } - ll_sa_entry_cleanup(sai, entry); + if (req) { + entry->se_req = NULL; + ptlrpc_req_finished(req); + } spin_lock(&lli->lli_sa_lock); - if (likely(entry->se_stat != SA_ENTRY_DEST)) { - do_sa_entry_to_stated(sai, entry, stat); - ret = 0; - } + __sa_entry_post_stat(sai, entry, stat); spin_unlock(&lli->lli_sa_lock); - - return ret; } /* @@ -475,56 +467,46 @@ static struct ll_statahead_info *ll_sai_alloc(void) return sai; } -static inline struct ll_statahead_info * -ll_sai_get(struct ll_statahead_info *sai) +static inline struct ll_statahead_info *ll_sai_get(struct inode *dir) { - atomic_inc(&sai->sai_refcount); + struct ll_inode_info *lli = ll_i2info(dir); + struct ll_statahead_info *sai = NULL; + + spin_lock(&lli->lli_sa_lock); + sai = lli->lli_sai; + if (sai) + atomic_inc(&sai->sai_refcount); + spin_unlock(&lli->lli_sa_lock); + return sai; } static void ll_sai_put(struct ll_statahead_info *sai) { - struct inode *inode = sai->sai_inode; - struct ll_inode_info *lli = ll_i2info(inode); + struct ll_inode_info *lli = ll_i2info(sai->sai_inode); if (atomic_dec_and_lock(&sai->sai_refcount, &lli->lli_sa_lock)) { + struct ll_sb_info *sbi = ll_i2sbi(sai->sai_inode); struct ll_sa_entry *entry, *next; - if (unlikely(atomic_read(&sai->sai_refcount) > 0)) { - /* It is race case, the interpret callback just hold - * a reference count - */ - spin_unlock(&lli->lli_sa_lock); - return; - } - - LASSERT(!lli->lli_opendir_key); - LASSERT(thread_is_stopped(&sai->sai_thread)); - LASSERT(thread_is_stopped(&sai->sai_agl_thread)); - lli->lli_sai = NULL; - lli->lli_opendir_pid = 0; spin_unlock(&lli->lli_sa_lock); - if (sai->sai_sent > sai->sai_replied) - CDEBUG(D_READA, "statahead for dir "DFID - " does not finish: [sent:%llu] [replied:%llu]\n", - PFID(&lli->lli_fid), - sai->sai_sent, sai->sai_replied); + LASSERT(thread_is_stopped(&sai->sai_thread)); + LASSERT(thread_is_stopped(&sai->sai_agl_thread)); + LASSERT(sai->sai_sent == sai->sai_replied); list_for_each_entry_safe(entry, next, &sai->sai_entries, se_link) do_sa_entry_fini(sai, entry); - LASSERT(list_empty(&sai->sai_entries)); - LASSERT(list_empty(&sai->sai_entries_received)); - LASSERT(list_empty(&sai->sai_entries_stated)); - LASSERT(atomic_read(&sai->sai_cache_count) == 0); LASSERT(list_empty(&sai->sai_entries_agl)); + LASSERT(atomic_read(&sai->sai_refcount) == 0); - iput(inode); + iput(sai->sai_inode); kfree(sai); + atomic_dec(&sbi->ll_sa_running); } } @@ -588,29 +570,18 @@ static void ll_agl_trigger(struct inode *inode, struct ll_statahead_info *sai) iput(inode); } -static void ll_post_statahead(struct ll_statahead_info *sai) +/* prepare inode for received statahead entry, and add it into agl list */ +static void sa_post_one(struct ll_statahead_info *sai, + struct ll_sa_entry *entry) { struct inode *dir = sai->sai_inode; struct inode *child; - struct ll_inode_info *lli = ll_i2info(dir); - struct ll_sa_entry *entry; struct md_enqueue_info *minfo; struct lookup_intent *it; struct ptlrpc_request *req; struct mdt_body *body; int rc = 0; - spin_lock(&lli->lli_sa_lock); - if (unlikely(list_empty(&sai->sai_entries_received))) { - spin_unlock(&lli->lli_sa_lock); - return; - } - entry = list_entry(sai->sai_entries_received.next, - struct ll_sa_entry, se_list); - atomic_inc(&entry->se_refcount); - list_del_init(&entry->se_list); - spin_unlock(&lli->lli_sa_lock); - LASSERT(entry->se_handle != 0); minfo = entry->se_minfo; @@ -670,18 +641,56 @@ static void ll_post_statahead(struct ll_statahead_info *sai) ll_agl_add(sai, child, entry->se_index); out: - /* The "ll_sa_entry_to_stated()" will drop related ldlm ibits lock + /* The "sa_entry_post_stat()" will drop related ldlm ibits lock * reference count by calling "ll_intent_drop_lock()" in spite of the * above operations failed or not. Do not worry about calling * "ll_intent_drop_lock()" more than once. */ - rc = ll_sa_entry_to_stated(sai, entry, - rc < 0 ? SA_ENTRY_INVA : SA_ENTRY_SUCC); - if (rc == 0 && entry->se_index == sai->sai_index_wait) + sa_entry_post_stat(sai, entry, rc < 0 ? SA_ENTRY_INVA : SA_ENTRY_SUCC); + if (entry->se_index == sai->sai_index_wait) wake_up(&sai->sai_waitq); ll_sa_entry_put(sai, entry); } +static void ll_post_statahead(struct ll_statahead_info *sai) +{ + struct ll_inode_info *lli; + + lli = ll_i2info(sai->sai_inode); + + while (!sa_received_empty(sai)) { + struct ll_sa_entry *entry; + + spin_lock(&lli->lli_sa_lock); + if (unlikely(sa_received_empty(sai))) { + spin_unlock(&lli->lli_sa_lock); + break; + } + entry = list_entry(sai->sai_entries_received.next, + struct ll_sa_entry, se_list); + atomic_inc(&entry->se_refcount); + list_del_init(&entry->se_list); + spin_unlock(&lli->lli_sa_lock); + + sa_post_one(sai, entry); + } + + spin_lock(&lli->lli_agl_lock); + while (!agl_list_empty(sai)) { + struct ll_inode_info *clli; + + clli = list_entry(sai->sai_entries_agl.next, + struct ll_inode_info, lli_agl_list); + list_del_init(&clli->lli_agl_list); + spin_unlock(&lli->lli_agl_lock); + + ll_agl_trigger(&clli->lli_vfs_inode, sai); + + spin_lock(&lli->lli_agl_lock); + } + spin_unlock(&lli->lli_agl_lock); +} + static int ll_statahead_interpret(struct ptlrpc_request *req, struct md_enqueue_info *minfo, int rc) { @@ -690,72 +699,43 @@ static int ll_statahead_interpret(struct ptlrpc_request *req, struct ll_inode_info *lli = ll_i2info(dir); struct ll_statahead_info *sai = NULL; struct ll_sa_entry *entry; - __u64 handle = 0; int wakeup; if (it_disposition(it, DISP_LOOKUP_NEG)) rc = -ENOENT; - if (rc == 0) { - /* release ibits lock ASAP to avoid deadlock when statahead - * thread enqueues lock on parent in readdir and another - * process enqueues lock on child with parent lock held, eg. - * unlink. - */ - handle = it->it_lock_handle; - ll_intent_drop_lock(it); - } + sai = ll_sai_get(dir); + LASSERT(sai); + LASSERT(!thread_is_stopped(&sai->sai_thread)); spin_lock(&lli->lli_sa_lock); - /* stale entry */ - if (unlikely(!lli->lli_sai || - lli->lli_sai->sai_generation != minfo->mi_generation)) { - spin_unlock(&lli->lli_sa_lock); - rc = -ESTALE; - goto out; + entry = ll_sa_entry_get_byindex(sai, minfo->mi_cbdata); + LASSERT(entry); + if (rc) { + __sa_entry_post_stat(sai, entry, SA_ENTRY_INVA); + wakeup = (entry->se_index == sai->sai_index_wait); } else { - sai = ll_sai_get(lli->lli_sai); - if (unlikely(!thread_is_running(&sai->sai_thread))) { - sai->sai_replied++; - spin_unlock(&lli->lli_sa_lock); - rc = -EBADFD; - goto out; - } - - entry = ll_sa_entry_get_byindex(sai, minfo->mi_cbdata); - if (!entry) { - sai->sai_replied++; - spin_unlock(&lli->lli_sa_lock); - rc = -EIDRM; - goto out; - } - - if (rc != 0) { - do_sa_entry_to_stated(sai, entry, SA_ENTRY_INVA); - wakeup = (entry->se_index == sai->sai_index_wait); - } else { - entry->se_minfo = minfo; - entry->se_req = ptlrpc_request_addref(req); - /* Release the async ibits lock ASAP to avoid deadlock - * when statahead thread tries to enqueue lock on parent - * for readpage and other tries to enqueue lock on child - * with parent's lock held, for example: unlink. - */ - entry->se_handle = handle; - wakeup = list_empty(&sai->sai_entries_received); - list_add_tail(&entry->se_list, - &sai->sai_entries_received); - } - sai->sai_replied++; - spin_unlock(&lli->lli_sa_lock); - - ll_sa_entry_put(sai, entry); - if (wakeup) - wake_up(&sai->sai_thread.t_ctl_waitq); + entry->se_minfo = minfo; + entry->se_req = ptlrpc_request_addref(req); + /* + * Release the async ibits lock ASAP to avoid deadlock + * when statahead thread tries to enqueue lock on parent + * for readpage and other tries to enqueue lock on child + * with parent's lock held, for example: unlink. + */ + entry->se_handle = it->it_lock_handle; + ll_intent_drop_lock(it); + wakeup = sa_received_empty(sai); + list_add_tail(&entry->se_list, &sai->sai_entries_received); } + sai->sai_replied++; + spin_unlock(&lli->lli_sa_lock); -out: - if (rc != 0) { + ll_sa_entry_put(sai, entry); + if (wakeup) + wake_up(&sai->sai_thread.t_ctl_waitq); + + if (rc) { ll_intent_release(it); iput(dir); kfree(minfo); @@ -782,7 +762,6 @@ static int sa_args_init(struct inode *dir, struct inode *child, struct ldlm_enqueue_info **pei) { const struct qstr *qstr = &entry->se_qstr; - struct ll_inode_info *lli = ll_i2info(dir); struct md_enqueue_info *minfo; struct ldlm_enqueue_info *einfo; struct md_op_data *op_data; @@ -808,7 +787,6 @@ static int sa_args_init(struct inode *dir, struct inode *child, minfo->mi_it.it_op = IT_GETATTR; minfo->mi_dir = igrab(dir); minfo->mi_cb = ll_statahead_interpret; - minfo->mi_generation = lli->lli_sai->sai_generation; minfo->mi_cbdata = entry->se_index; einfo->ei_type = LDLM_IBITS; @@ -889,8 +867,8 @@ static int do_sa_revalidate(struct inode *dir, struct ll_sa_entry *entry, return rc; } -static void ll_statahead_one(struct dentry *parent, const char *entry_name, - int entry_name_len) +static void ll_statahead_one(struct dentry *parent, const char *name, + const int name_len) { struct inode *dir = d_inode(parent); struct ll_inode_info *lli = ll_i2info(dir); @@ -898,10 +876,9 @@ static void ll_statahead_one(struct dentry *parent, const char *entry_name, struct dentry *dentry = NULL; struct ll_sa_entry *entry; int rc; - int rc1; - entry = ll_sa_entry_alloc(parent, sai, sai->sai_index, entry_name, - entry_name_len); + entry = ll_sa_entry_alloc(parent, sai, sai->sai_index, name, + name_len); if (IS_ERR(entry)) return; @@ -912,15 +889,15 @@ static void ll_statahead_one(struct dentry *parent, const char *entry_name, rc = do_sa_revalidate(dir, entry, dentry); if (rc == 1 && agl_should_run(sai, d_inode(dentry))) ll_agl_add(sai, d_inode(dentry), entry->se_index); + } + if (dentry) dput(dentry); - } if (rc) { - rc1 = ll_sa_entry_to_stated(sai, entry, - rc < 0 ? SA_ENTRY_INVA : - SA_ENTRY_SUCC); - if (rc1 == 0 && entry->se_index == sai->sai_index_wait) + sa_entry_post_stat(sai, entry, + rc < 0 ? SA_ENTRY_INVA : SA_ENTRY_SUCC); + if (entry->se_index == sai->sai_index_wait) wake_up(&sai->sai_waitq); } else { sai->sai_sent++; @@ -938,10 +915,12 @@ static int ll_agl_thread(void *arg) struct ll_inode_info *plli = ll_i2info(dir); struct ll_inode_info *clli; struct ll_sb_info *sbi = ll_i2sbi(dir); - struct ll_statahead_info *sai = ll_sai_get(plli->lli_sai); - struct ptlrpc_thread *thread = &sai->sai_agl_thread; + struct ll_statahead_info *sai; + struct ptlrpc_thread *thread; struct l_wait_info lwi = { 0 }; + sai = ll_sai_get(dir); + thread = &sai->sai_agl_thread; thread->t_pid = current_pid(); CDEBUG(D_READA, "agl thread started: sai %p, parent %pd\n", sai, parent); @@ -1030,12 +1009,11 @@ static int ll_statahead_thread(void *arg) { struct dentry *parent = arg; struct inode *dir = d_inode(parent); - struct ll_inode_info *plli = ll_i2info(dir); - struct ll_inode_info *clli; + struct ll_inode_info *lli = ll_i2info(dir); struct ll_sb_info *sbi = ll_i2sbi(dir); - struct ll_statahead_info *sai = ll_sai_get(plli->lli_sai); - struct ptlrpc_thread *thread = &sai->sai_thread; - struct ptlrpc_thread *agl_thread = &sai->sai_agl_thread; + struct ll_statahead_info *sai; + struct ptlrpc_thread *thread; + struct ptlrpc_thread *agl_thread; struct page *page = NULL; __u64 pos = 0; int first = 0; @@ -1044,6 +1022,9 @@ static int ll_statahead_thread(void *arg) struct ll_dir_chain chain; struct l_wait_info lwi = { 0 }; + sai = ll_sai_get(dir); + thread = &sai->sai_thread; + agl_thread = &sai->sai_agl_thread; thread->t_pid = current_pid(); CDEBUG(D_READA, "statahead thread starting: sai %p, parent %pd\n", sai, parent); @@ -1052,7 +1033,7 @@ static int ll_statahead_thread(void *arg) LUSTRE_OPC_ANY, dir); if (IS_ERR(op_data)) { rc = PTR_ERR(op_data); - goto out_put; + goto out; } op_data->op_max_pages = ll_i2sbi(dir)->ll_md_brw_pages; @@ -1061,33 +1042,35 @@ static int ll_statahead_thread(void *arg) ll_start_agl(parent, sai); atomic_inc(&sbi->ll_sa_total); - spin_lock(&plli->lli_sa_lock); + spin_lock(&lli->lli_sa_lock); if (thread_is_init(thread)) /* If someone else has changed the thread state * (e.g. already changed to SVC_STOPPING), we can't just * blindly overwrite that setting. */ thread_set_flags(thread, SVC_RUNNING); - spin_unlock(&plli->lli_sa_lock); + spin_unlock(&lli->lli_sa_lock); wake_up(&thread->t_ctl_waitq); ll_dir_chain_init(&chain); - page = ll_get_dir_page(dir, op_data, pos, &chain); - - while (1) { + while (pos != MDS_DIR_END_OFF && thread_is_running(thread)) { struct lu_dirpage *dp; struct lu_dirent *ent; + sai->sai_in_readpage = 1; + page = ll_get_dir_page(dir, op_data, pos, &chain); + sai->sai_in_readpage = 0; if (IS_ERR(page)) { rc = PTR_ERR(page); CDEBUG(D_READA, "error reading dir "DFID" at %llu/%llu: opendir_pid = %u: rc = %d\n", PFID(ll_inode2fid(dir)), pos, sai->sai_index, - plli->lli_opendir_pid, rc); - goto out; + lli->lli_opendir_pid, rc); + break; } dp = page_address(page); - for (ent = lu_dirent_start(dp); ent; + for (ent = lu_dirent_start(dp); + ent && thread_is_running(thread) && !sa_low_hit(sai); ent = lu_dirent_next(ent)) { __u64 hash; int namelen; @@ -1134,120 +1117,63 @@ static int ll_statahead_thread(void *arg) if (unlikely(++first == 1)) continue; -keep_it: - l_wait_event(thread->t_ctl_waitq, - !sa_sent_full(sai) || - !list_empty(&sai->sai_entries_received) || - !list_empty(&sai->sai_entries_agl) || - !thread_is_running(thread), - &lwi); - -interpret_it: - while (!list_empty(&sai->sai_entries_received)) + /* wait for spare statahead window */ + do { + l_wait_event(thread->t_ctl_waitq, + !sa_sent_full(sai) || + !list_empty(&sai->sai_entries_received) || + !list_empty(&sai->sai_entries_agl) || + !thread_is_running(thread), + &lwi); ll_post_statahead(sai); + } while (sa_sent_full(sai) && + thread_is_running(thread)); - if (unlikely(!thread_is_running(thread))) { - ll_release_page(dir, page, false); - rc = 0; - goto out; - } - - /* If no window for metadata statahead, but there are - * some AGL entries to be triggered, then try to help - * to process the AGL entries. - */ - if (sa_sent_full(sai)) { - spin_lock(&plli->lli_agl_lock); - while (!list_empty(&sai->sai_entries_agl)) { - clli = list_entry(sai->sai_entries_agl.next, - struct ll_inode_info, lli_agl_list); - list_del_init(&clli->lli_agl_list); - spin_unlock(&plli->lli_agl_lock); - ll_agl_trigger(&clli->lli_vfs_inode, - sai); - - if (!list_empty(&sai->sai_entries_received)) - goto interpret_it; - - if (unlikely(!thread_is_running(thread))) { - ll_release_page(dir, page, false); - rc = 0; - goto out; - } - - if (!sa_sent_full(sai)) - goto do_it; - - spin_lock(&plli->lli_agl_lock); - } - spin_unlock(&plli->lli_agl_lock); - - goto keep_it; - } -do_it: ll_statahead_one(parent, name, namelen); } pos = le64_to_cpu(dp->ldp_hash_end); - if (pos == MDS_DIR_END_OFF) { - /* - * End of directory reached. - */ - ll_release_page(dir, page, false); - while (1) { - l_wait_event(thread->t_ctl_waitq, - !list_empty(&sai->sai_entries_received) || - sai->sai_sent == sai->sai_replied || - !thread_is_running(thread), - &lwi); + ll_release_page(dir, page, + le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE); - while (!list_empty(&sai->sai_entries_received)) - ll_post_statahead(sai); - - if (unlikely(!thread_is_running(thread))) { - rc = 0; - goto out; - } + if (sa_low_hit(sai)) { + rc = -EFAULT; + atomic_inc(&sbi->ll_sa_wrong); + CDEBUG(D_READA, "Statahead for dir "DFID" hit ratio too low: hit/miss %llu/%llu, sent/replied %llu/%llu, stopping statahead thread: pid %d\n", + PFID(&lli->lli_fid), sai->sai_hit, + sai->sai_miss, sai->sai_sent, + sai->sai_replied, current_pid()); + break; + } + } + ll_dir_chain_fini(&chain); + ll_finish_md_op_data(op_data); - if (sai->sai_sent == sai->sai_replied && - list_empty(&sai->sai_entries_received)) - break; - } + if (rc < 0) { + spin_lock(&lli->lli_sa_lock); + thread_set_flags(thread, SVC_STOPPING); + lli->lli_sa_enabled = 0; + spin_unlock(&lli->lli_sa_lock); + } - spin_lock(&plli->lli_agl_lock); - while (!list_empty(&sai->sai_entries_agl) && - thread_is_running(thread)) { - clli = list_entry(sai->sai_entries_agl.next, - struct ll_inode_info, lli_agl_list); - list_del_init(&clli->lli_agl_list); - spin_unlock(&plli->lli_agl_lock); - ll_agl_trigger(&clli->lli_vfs_inode, sai); - spin_lock(&plli->lli_agl_lock); - } - spin_unlock(&plli->lli_agl_lock); + /* + * statahead is finished, but statahead entries need to be cached, wait + * for file release to stop me. + */ + while (thread_is_running(thread)) { + l_wait_event(thread->t_ctl_waitq, + !sa_received_empty(sai) || + !agl_list_empty(sai) || + !thread_is_running(thread), + &lwi); - rc = 0; - goto out; - } else { - /* - * chain is exhausted. - * Normal case: continue to the next page. - */ - ll_release_page(dir, page, - le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE); - sai->sai_in_readpage = 1; - page = ll_get_dir_page(dir, op_data, pos, &chain); - sai->sai_in_readpage = 0; - } + ll_post_statahead(sai); } out: - ll_dir_chain_fini(&chain); - ll_finish_md_op_data(op_data); -out_put: if (sai->sai_agl_valid) { - spin_lock(&plli->lli_agl_lock); + spin_lock(&lli->lli_agl_lock); thread_set_flags(agl_thread, SVC_STOPPING); - spin_unlock(&plli->lli_agl_lock); + spin_unlock(&lli->lli_agl_lock); wake_up(&agl_thread->t_ctl_waitq); CDEBUG(D_READA, "stop agl thread: sai %p pid %u\n", @@ -1257,21 +1183,27 @@ out_put: &lwi); } else { /* Set agl_thread flags anyway. */ - thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED); + thread_set_flags(agl_thread, SVC_STOPPED); } - spin_lock(&plli->lli_sa_lock); - if (!list_empty(&sai->sai_entries_received)) { - thread_set_flags(thread, SVC_STOPPING); - spin_unlock(&plli->lli_sa_lock); - - /* To release the resources held by received entries. */ - while (!list_empty(&sai->sai_entries_received)) - ll_post_statahead(sai); - spin_lock(&plli->lli_sa_lock); + /* + * wait for inflight statahead RPCs to finish, and then we can free sai + * safely because statahead RPC will access sai data + */ + while (sai->sai_sent != sai->sai_replied) { + /* in case we're not woken up, timeout wait */ + lwi = LWI_TIMEOUT(HZ >> 3, NULL, NULL); + l_wait_event(thread->t_ctl_waitq, + sai->sai_sent == sai->sai_replied, &lwi); } + + /* release resources held by received entries. */ + ll_post_statahead(sai); + + spin_lock(&lli->lli_sa_lock); thread_set_flags(thread, SVC_STOPPED); - spin_unlock(&plli->lli_sa_lock); + spin_unlock(&lli->lli_sa_lock); + wake_up(&sai->sai_waitq); wake_up(&thread->t_ctl_waitq); ll_sai_put(sai); @@ -1281,52 +1213,54 @@ out_put: return rc; } -/** - * called in ll_file_release(). - */ -void ll_stop_statahead(struct inode *dir, void *key) +/* authorize opened dir handle @key to statahead later */ +void ll_authorize_statahead(struct inode *dir, void *key) { struct ll_inode_info *lli = ll_i2info(dir); - if (unlikely(!key)) - return; - spin_lock(&lli->lli_sa_lock); - if (lli->lli_opendir_key != key || lli->lli_opendir_pid == 0) { - spin_unlock(&lli->lli_sa_lock); - return; + if (!lli->lli_opendir_key && !lli->lli_sai) { + /* + * if lli_sai is not NULL, it means previous statahead is not + * finished yet, we'd better not start a new statahead for now. + */ + LASSERT(!lli->lli_opendir_pid); + lli->lli_opendir_key = key; + lli->lli_opendir_pid = current_pid(); + lli->lli_sa_enabled = 1; } + spin_unlock(&lli->lli_sa_lock); +} - lli->lli_opendir_key = NULL; +/* + * deauthorize opened dir handle @key to statahead, but statahead thread may + * still be running, notify it to quit. + */ +void ll_deauthorize_statahead(struct inode *dir, void *key) +{ + struct ll_inode_info *lli = ll_i2info(dir); + struct ll_statahead_info *sai; - if (lli->lli_sai) { - struct l_wait_info lwi = { 0 }; - struct ptlrpc_thread *thread = &lli->lli_sai->sai_thread; + LASSERT(lli->lli_opendir_key == key); + LASSERT(lli->lli_opendir_pid); - if (!thread_is_stopped(thread)) { - thread_set_flags(thread, SVC_STOPPING); - spin_unlock(&lli->lli_sa_lock); - wake_up(&thread->t_ctl_waitq); - - CDEBUG(D_READA, "stop statahead thread: sai %p pid %u\n", - lli->lli_sai, (unsigned int)thread->t_pid); - l_wait_event(thread->t_ctl_waitq, - thread_is_stopped(thread), - &lwi); - } else { - spin_unlock(&lli->lli_sa_lock); - } + CDEBUG(D_READA, "deauthorize statahead for "DFID"\n", + PFID(&lli->lli_fid)); + spin_lock(&lli->lli_sa_lock); + lli->lli_opendir_key = NULL; + lli->lli_opendir_pid = 0; + lli->lli_sa_enabled = 0; + sai = lli->lli_sai; + if (sai && thread_is_running(&sai->sai_thread)) { /* - * Put the ref which was held when first statahead_enter. - * It maybe not the last ref for some statahead requests - * maybe inflight. + * statahead thread may not quit yet because it needs to cache + * stated entries, now it's time to tell it to quit. */ - ll_sai_put(lli->lli_sai); - } else { - lli->lli_opendir_pid = 0; - spin_unlock(&lli->lli_sa_lock); + thread_set_flags(&sai->sai_thread, SVC_STOPPING); + wake_up(&sai->sai_thread.t_ctl_waitq); } + spin_unlock(&lli->lli_sa_lock); } enum { @@ -1465,175 +1399,137 @@ out: static void ll_sai_unplug(struct ll_statahead_info *sai, struct ll_sa_entry *entry) { - struct ptlrpc_thread *thread = &sai->sai_thread; - struct ll_sb_info *sbi = ll_i2sbi(sai->sai_inode); - int hit; + if (entry && entry->se_stat == SA_ENTRY_SUCC) { + struct ll_sb_info *sbi = ll_i2sbi(sai->sai_inode); - if (entry && entry->se_stat == SA_ENTRY_SUCC) - hit = 1; - else - hit = 0; - - ll_sa_entry_fini(sai, entry); - if (hit) { sai->sai_hit++; sai->sai_consecutive_miss = 0; sai->sai_max = min(2 * sai->sai_max, sbi->ll_sa_max); } else { - struct ll_inode_info *lli = ll_i2info(sai->sai_inode); - sai->sai_miss++; sai->sai_consecutive_miss++; - if (sa_low_hit(sai) && thread_is_running(thread)) { - atomic_inc(&sbi->ll_sa_wrong); - CDEBUG(D_READA, "Statahead for dir " DFID " hit ratio too low: hit/miss %llu/%llu, sent/replied %llu/%llu, stopping statahead thread\n", - PFID(&lli->lli_fid), sai->sai_hit, - sai->sai_miss, sai->sai_sent, - sai->sai_replied); - spin_lock(&lli->lli_sa_lock); - if (!thread_is_stopped(thread)) - thread_set_flags(thread, SVC_STOPPING); - spin_unlock(&lli->lli_sa_lock); - } } - - if (!thread_is_stopped(thread)) - wake_up(&thread->t_ctl_waitq); + ll_sa_entry_fini(sai, entry); + wake_up(&sai->sai_thread.t_ctl_waitq); } -/** - * Start statahead thread if this is the first dir entry. - * Otherwise if a thread is started already, wait it until it is ahead of me. - * \retval 1 -- find entry with lock in cache, the caller needs to do - * nothing. - * \retval 0 -- find entry in cache, but without lock, the caller needs - * refresh from MDS. - * \retval others -- the caller need to process as non-statahead. - */ -int do_statahead_enter(struct inode *dir, struct dentry **dentryp, - int only_unplug) +static int revalidate_statahead_dentry(struct inode *dir, + struct ll_statahead_info *sai, + struct dentry **dentryp, + int only_unplug) { - struct ll_inode_info *lli = ll_i2info(dir); - struct ll_statahead_info *sai = lli->lli_sai; - struct dentry *parent; - struct ll_sa_entry *entry; - struct ptlrpc_thread *thread; - struct l_wait_info lwi = { 0 }; - struct task_struct *task; - int rc = 0; - struct ll_inode_info *plli; - - LASSERT(lli->lli_opendir_pid == current_pid()); - - if (sai) { - thread = &sai->sai_thread; - if (unlikely(thread_is_stopped(thread) && - list_empty(&sai->sai_entries_stated))) { - /* to release resource */ - ll_stop_statahead(dir, lli->lli_opendir_key); - return -EAGAIN; - } + struct ll_sa_entry *entry = NULL; + struct l_wait_info lwi = { 0 }; + int rc = 0; - if ((*dentryp)->d_name.name[0] == '.') { - if (sai->sai_ls_all || - sai->sai_miss_hidden >= sai->sai_skip_hidden) { + if ((*dentryp)->d_name.name[0] == '.') { + if (sai->sai_ls_all || + sai->sai_miss_hidden >= sai->sai_skip_hidden) { + /* + * Hidden dentry is the first one, or statahead + * thread does not skip so many hidden dentries + * before "sai_ls_all" enabled as below. + */ + } else { + if (!sai->sai_ls_all) /* - * Hidden dentry is the first one, or statahead - * thread does not skip so many hidden dentries - * before "sai_ls_all" enabled as below. + * It maybe because hidden dentry is not + * the first one, "sai_ls_all" was not + * set, then "ls -al" missed. Enable + * "sai_ls_all" for such case. */ - } else { - if (!sai->sai_ls_all) - /* - * It maybe because hidden dentry is not - * the first one, "sai_ls_all" was not - * set, then "ls -al" missed. Enable - * "sai_ls_all" for such case. - */ - sai->sai_ls_all = 1; + sai->sai_ls_all = 1; - /* - * Such "getattr" has been skipped before - * "sai_ls_all" enabled as above. - */ - sai->sai_miss_hidden++; - return -EAGAIN; - } + /* + * Such "getattr" has been skipped before + * "sai_ls_all" enabled as above. + */ + sai->sai_miss_hidden++; + return -EAGAIN; } + } - entry = ll_sa_entry_get_byname(sai, &(*dentryp)->d_name); - if (!entry || only_unplug) { + entry = ll_sa_entry_get_byname(sai, &(*dentryp)->d_name); + if (!entry || only_unplug) { + ll_sai_unplug(sai, entry); + return entry ? 1 : -EAGAIN; + } + + /* if statahead is busy in readdir, help it do post-work */ + if (!ll_sa_entry_stated(entry) && sai->sai_in_readpage) + ll_post_statahead(sai); + + if (!ll_sa_entry_stated(entry)) { + sai->sai_index_wait = entry->se_index; + lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(30), NULL, + LWI_ON_SIGNAL_NOOP, NULL); + rc = l_wait_event(sai->sai_waitq, + ll_sa_entry_stated(entry) || + thread_is_stopped(&sai->sai_thread), + &lwi); + if (rc < 0) { ll_sai_unplug(sai, entry); - return entry ? 1 : -EAGAIN; + return -EAGAIN; } + } - /* if statahead is busy in readdir, help it do post-work */ - while (!ll_sa_entry_stated(entry) && sai->sai_in_readpage && - !sa_received_empty(sai)) - ll_post_statahead(sai); - - if (!ll_sa_entry_stated(entry)) { - sai->sai_index_wait = entry->se_index; - lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(30), NULL, - LWI_ON_SIGNAL_NOOP, NULL); - rc = l_wait_event(sai->sai_waitq, - ll_sa_entry_stated(entry) || - thread_is_stopped(thread), - &lwi); - if (rc < 0) { - ll_sai_unplug(sai, entry); - return -EAGAIN; - } - } + if (entry->se_stat == SA_ENTRY_SUCC && entry->se_inode) { + struct inode *inode = entry->se_inode; + struct lookup_intent it = { .it_op = IT_GETATTR, + .it_lock_handle = entry->se_handle }; + __u64 bits; + + rc = md_revalidate_lock(ll_i2mdexp(dir), &it, + ll_inode2fid(inode), &bits); + if (rc == 1) { + if (!(*dentryp)->d_inode) { + struct dentry *alias; - if (entry->se_stat == SA_ENTRY_SUCC && entry->se_inode) { - struct inode *inode = entry->se_inode; - struct lookup_intent it = { .it_op = IT_GETATTR, - .it_lock_handle = - entry->se_handle }; - __u64 bits; - - rc = md_revalidate_lock(ll_i2mdexp(dir), &it, - ll_inode2fid(inode), &bits); - if (rc == 1) { - if (!d_inode(*dentryp)) { - struct dentry *alias; - - alias = ll_splice_alias(inode, - *dentryp); - if (IS_ERR(alias)) { - ll_sai_unplug(sai, entry); - return PTR_ERR(alias); - } - *dentryp = alias; - } else if (d_inode(*dentryp) != inode) { - /* revalidate, but inode is recreated */ - CDEBUG(D_READA, "%s: stale dentry %pd inode "DFID", statahead inode "DFID"\n", - ll_get_fsname(d_inode(*dentryp)->i_sb, NULL, 0), - *dentryp, - PFID(ll_inode2fid(d_inode(*dentryp))), - PFID(ll_inode2fid(inode))); - ll_intent_release(&it); + alias = ll_splice_alias(inode, *dentryp); + if (IS_ERR(alias)) { ll_sai_unplug(sai, entry); - return -ESTALE; - } else { - iput(inode); + return PTR_ERR(alias); } - entry->se_inode = NULL; - - if ((bits & MDS_INODELOCK_LOOKUP) && - d_lustre_invalid(*dentryp)) - d_lustre_revalidate(*dentryp); - ll_intent_release(&it); + *dentryp = alias; + } else if ((*dentryp)->d_inode != inode) { + /* revalidate, but inode is recreated */ + CDEBUG(D_READA, + "%s: stale dentry %pd inode "DFID", statahead inode "DFID"\n", + ll_get_fsname((*dentryp)->d_inode->i_sb, + NULL, 0), + *dentryp, + PFID(ll_inode2fid((*dentryp)->d_inode)), + PFID(ll_inode2fid(inode))); + rc = -ESTALE; + goto out_unplug; + } else { + iput(inode); } - } + entry->se_inode = NULL; - ll_sai_unplug(sai, entry); - return rc; + if ((bits & MDS_INODELOCK_LOOKUP) && + d_lustre_invalid(*dentryp)) + d_lustre_revalidate(*dentryp); + ll_intent_release(&it); + } } +out_unplug: + ll_sai_unplug(sai, entry); + return rc; +} + +static int start_statahead_thread(struct inode *dir, struct dentry *dentry) +{ + struct ll_inode_info *lli = ll_i2info(dir); + struct ll_statahead_info *sai = NULL; + struct l_wait_info lwi = { 0 }; + struct ptlrpc_thread *thread; + struct task_struct *task; + struct dentry *parent; + int rc; /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */ - rc = is_first_dirent(dir, *dentryp); + rc = is_first_dirent(dir, dentry); if (rc == LS_NONE_FIRST_DE) { /* It is not "ls -{a}l" operation, no need statahead for it. */ rc = -EAGAIN; @@ -1656,13 +1552,12 @@ int do_statahead_enter(struct inode *dir, struct dentry **dentryp, } /* get parent reference count here, and put it in ll_statahead_thread */ - parent = dget((*dentryp)->d_parent); + parent = dget(dentry->d_parent); if (unlikely(sai->sai_inode != d_inode(parent))) { struct ll_inode_info *nlli = ll_i2info(d_inode(parent)); CWARN("Race condition, someone changed %pd just now: old parent "DFID", new parent "DFID"\n", - *dentryp, - PFID(&lli->lli_fid), PFID(&nlli->lli_fid)); + dentry, PFID(&lli->lli_fid), PFID(&nlli->lli_fid)); dput(parent); iput(sai->sai_inode); rc = -EAGAIN; @@ -1672,30 +1567,18 @@ int do_statahead_enter(struct inode *dir, struct dentry **dentryp, CDEBUG(D_READA, "start statahead thread: sai %p, parent %pd\n", sai, parent); - /* The sai buffer already has one reference taken at allocation time, - * but as soon as we expose the sai by attaching it to the lli that - * default reference can be dropped by another thread calling - * ll_stop_statahead. We need to take a local reference to protect - * the sai buffer while we intend to access it. - */ - ll_sai_get(sai); lli->lli_sai = sai; - plli = ll_i2info(d_inode(parent)); task = kthread_run(ll_statahead_thread, parent, "ll_sa_%u", - plli->lli_opendir_pid); + lli->lli_opendir_pid); thread = &sai->sai_thread; if (IS_ERR(task)) { rc = PTR_ERR(task); - CERROR("can't start ll_sa thread, rc: %d\n", rc); + CERROR("cannot start ll_sa thread: rc = %d\n", rc); dput(parent); lli->lli_opendir_key = NULL; thread_set_flags(thread, SVC_STOPPED); thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED); - /* Drop both our own local reference and the default - * reference from allocation time. - */ - ll_sai_put(sai); ll_sai_put(sai); LASSERT(!lli->lli_sai); return -EAGAIN; @@ -1704,6 +1587,7 @@ int do_statahead_enter(struct inode *dir, struct dentry **dentryp, l_wait_event(thread->t_ctl_waitq, thread_is_running(thread) || thread_is_stopped(thread), &lwi); + atomic_inc(&ll_i2sbi(d_inode(parent))->ll_sa_running); ll_sai_put(sai); /* @@ -1717,6 +1601,37 @@ out: spin_lock(&lli->lli_sa_lock); lli->lli_opendir_key = NULL; lli->lli_opendir_pid = 0; + lli->lli_sa_enabled = 0; spin_unlock(&lli->lli_sa_lock); + return rc; } + +/** + * Start statahead thread if this is the first dir entry. + * Otherwise if a thread is started already, wait it until it is ahead of me. + * \retval 1 -- find entry with lock in cache, the caller needs to do + * nothing. + * \retval 0 -- find entry in cache, but without lock, the caller needs + * refresh from MDS. + * \retval others -- the caller need to process as non-statahead. + */ +int do_statahead_enter(struct inode *dir, struct dentry **dentryp, + int only_unplug) +{ + struct ll_statahead_info *sai; + + sai = ll_sai_get(dir); + if (sai) { + int rc; + + rc = revalidate_statahead_dentry(dir, sai, dentryp, + only_unplug); + CDEBUG(D_READA, "revalidate statahead %pd: %d.\n", + *dentryp, rc); + ll_sai_put(sai); + return rc; + } + + return start_statahead_thread(dir, *dentryp); +} -- 1.7.1 _______________________________________________ devel mailing list de...@linuxdriverproject.org http://driverdev.linuxdriverproject.org/mailman/listinfo/driverdev-devel
