From: Lai Siyao <lai.siyao@xxxxxxxxx> small fixes: * when 'unplug' is set for ll_statahead(), sa_put() shouldn't kill the entry found, because its inflight RPC may not finish yet. * remove 'sai_generation', add 'lli_sa_generation' because the former one is not safe to access without lock. * revalidate_statahead_dentry() may fail to wait for statahead entry to become ready, in this case it should not release this entry, because it may be used by inflight statahead RPC. cleanups: * rename ll_statahead_enter() to ll_statahead(). * move dentry 'lld_sa_generation' update to ll_statahead() to simplify code and logic. * other small cleanups. Signed-off-by: Lai Siyao <lai.siyao@xxxxxxxxx> Intel-bug-id: https://jira.hpdd.intel.com/browse/LU-3270 Reviewed-on: http://review.whamcloud.com/9667 Intel-bug-id: https://jira.hpdd.intel.com/browse/LU-6222 Reviewed-on: http://review.whamcloud.com/13708 Reviewed-by: Fan Yong <fan.yong@xxxxxxxxx> Reviewed-by: Bobi Jam <bobijam@xxxxxxxxxxx> Reviewed-by: James Simmons <uja.ornl@xxxxxxxxx> Reviewed-by: Oleg Drokin <oleg.drokin@xxxxxxxxx> Signed-off-by: James Simmons <jsimmons@xxxxxxxxxxxxx> --- drivers/staging/lustre/lustre/llite/dcache.c | 5 +- .../staging/lustre/lustre/llite/llite_internal.h | 137 +++----- drivers/staging/lustre/lustre/llite/namei.c | 11 +- drivers/staging/lustre/lustre/llite/statahead.c | 353 +++++++++++--------- drivers/staging/lustre/lustre/mdc/mdc_request.c | 2 +- 5 files changed, 250 insertions(+), 258 deletions(-) diff --git a/drivers/staging/lustre/lustre/llite/dcache.c b/drivers/staging/lustre/lustre/llite/dcache.c index 8500080..0e45d8f 100644 --- a/drivers/staging/lustre/lustre/llite/dcache.c +++ b/drivers/staging/lustre/lustre/llite/dcache.c @@ -278,14 +278,13 @@ static int ll_revalidate_dentry(struct dentry *dentry, if (lookup_flags & (LOOKUP_PARENT | LOOKUP_OPEN | LOOKUP_CREATE)) return 1; - if (!dentry_need_statahead(dir, dentry)) + if (!dentry_may_statahead(dir, dentry)) return 1; if (lookup_flags & LOOKUP_RCU) return -ECHILD; - do_statahead_enter(dir, &dentry, !d_inode(dentry)); - ll_statahead_mark(dir, dentry); + ll_statahead(dir, &dentry, !d_inode(dentry)); return 1; } diff --git a/drivers/staging/lustre/lustre/llite/llite_internal.h b/drivers/staging/lustre/lustre/llite/llite_internal.h index a68bea1..bdfdff5 100644 --- a/drivers/staging/lustre/lustre/llite/llite_internal.h +++ b/drivers/staging/lustre/lustre/llite/llite_internal.h @@ -161,7 +161,7 @@ struct ll_inode_info { /* for directory */ struct { /* serialize normal readdir and statahead-readdir. */ - struct mutex d_readdir_mutex; + struct mutex lli_readdir_mutex; /* metadata statahead */ /* since parent-child threads can share the same @file @@ -169,44 +169,35 @@ struct ll_inode_info { * case of parent exit before child -- it is me should * cleanup the dir readahead. */ - void *d_opendir_key; - struct ll_statahead_info *d_sai; + void *lli_opendir_key; + struct ll_statahead_info *lli_sai; /* protect statahead stuff. */ - spinlock_t d_sa_lock; + spinlock_t lli_sa_lock; /* "opendir_pid" is the token when lookup/revalidate * -- I am the owner of dir statahead. */ - pid_t d_opendir_pid; + pid_t lli_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; + unsigned int lli_sa_enabled:1; + /* generation for statahead */ + unsigned int lli_sa_generation; /* directory stripe information */ - struct lmv_stripe_md *d_lsm_md; + struct lmv_stripe_md *lli_lsm_md; /* striped directory size */ - loff_t d_stripe_size; - /* striped directory nlink */ - __u64 d_stripe_nlink; - } d; - -#define lli_readdir_mutex u.d.d_readdir_mutex -#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 -#define lli_stripe_dir_nlink u.d.d_stripe_nlink + loff_t lli_stripe_dir_size; + u64 lli_stripe_dir_nlink; + }; /* for non-directory */ struct { - struct mutex f_size_mutex; - char *f_symlink_name; - __u64 f_maxbytes; + struct mutex lli_size_mutex; + char *lli_symlink_name; + __u64 lli_maxbytes; /* * struct rw_semaphore { * signed long count; // align d.d_def_acl @@ -214,16 +205,16 @@ struct ll_inode_info { * struct list_head wait_list; * } */ - struct rw_semaphore f_trunc_sem; - struct range_lock_tree f_write_tree; + struct rw_semaphore lli_trunc_sem; + struct range_lock_tree lli_write_tree; - struct rw_semaphore f_glimpse_sem; - unsigned long f_glimpse_time; - struct list_head f_agl_list; - __u64 f_agl_index; + struct rw_semaphore lli_glimpse_sem; + unsigned long lli_glimpse_time; + struct list_head lli_agl_list; + __u64 lli_agl_index; /* for writepage() only to communicate to fsync */ - int f_async_rc; + int lli_async_rc; /* * whenever a process try to read/write the file, the @@ -233,22 +224,9 @@ struct ll_inode_info { * so the read/write statistics for jobid will not be * accurate if the file is shared by different jobs. */ - char f_jobid[LUSTRE_JOBID_SIZE]; - } f; - -#define lli_size_mutex u.f.f_size_mutex -#define lli_symlink_name u.f.f_symlink_name -#define lli_maxbytes u.f.f_maxbytes -#define lli_trunc_sem u.f.f_trunc_sem -#define lli_write_tree u.f.f_write_tree -#define lli_glimpse_sem u.f.f_glimpse_sem -#define lli_glimpse_time u.f.f_glimpse_time -#define lli_agl_list u.f.f_agl_list -#define lli_agl_index u.f.f_agl_index -#define lli_async_rc u.f.f_async_rc -#define lli_jobid u.f.f_jobid - - } u; + char lli_jobid[LUSTRE_JOBID_SIZE]; + }; + }; /* XXX: For following frequent used members, although they maybe special * used for non-directory object, it is some time-wasting to check @@ -1095,11 +1073,10 @@ void ll_ra_stats_inc(struct inode *inode, enum ra_stat which); /* per inode struct, for dir only */ struct ll_statahead_info { - struct inode *sai_inode; + struct dentry *sai_dentry; atomic_t sai_refcount; /* when access this struct, hold * refcount */ - unsigned int sai_generation; /* generation for statahead */ unsigned int sai_max; /* max ahead of lookup */ __u64 sai_sent; /* stat requests sent count */ __u64 sai_replied; /* stat requests which received @@ -1142,8 +1119,7 @@ struct ll_statahead_info { atomic_t sai_cache_count; /* entry count in cache */ }; -int do_statahead_enter(struct inode *dir, struct dentry **dentry, - int only_unplug); +int ll_statahead(struct inode *dir, struct dentry **dentry, bool unplug); void ll_authorize_statahead(struct inode *dir, void *key); void ll_deauthorize_statahead(struct inode *dir, void *key); @@ -1175,24 +1151,12 @@ static inline int ll_glimpse_size(struct inode *inode) return rc; } -static inline void -ll_statahead_mark(struct inode *dir, struct dentry *dentry) -{ - struct ll_inode_info *lli = ll_i2info(dir); - struct ll_statahead_info *sai = lli->lli_sai; - struct ll_dentry_data *ldd = ll_d2d(dentry); - - /* not the same process, don't mark */ - if (lli->lli_opendir_pid != current_pid()) - return; - - LASSERT(ldd); - if (sai) - ldd->lld_sa_generation = sai->sai_generation; -} - +/* + * dentry may statahead when statahead is enabled and current process has opened + * parent directory, and this dentry hasn't accessed statahead cache before + */ static inline bool -dentry_need_statahead(struct inode *dir, struct dentry *dentry) +dentry_may_statahead(struct inode *dir, struct dentry *dentry) { struct ll_inode_info *lli; struct ll_dentry_data *ldd; @@ -1215,38 +1179,27 @@ dentry_need_statahead(struct inode *dir, struct dentry *dentry) if (lli->lli_opendir_pid != current_pid()) return false; - 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. - * Under patchless client mode, the operation intent is not accurate, - * which maybe misguide the statahead thread. For example: - * The "revalidate" call for "getattr" and "getxattr" of a dentry maybe - * have the same operation intent -- "IT_GETATTR". - * In fact, one dentry should has only one chance to interact with the - * statahead thread, otherwise the statahead windows will be confused. + * When stating a dentry, kernel may trigger 'revalidate' or 'lookup' + * multiple times, eg. for 'getattr', 'getxattr' and etc. + * For patchless client, lookup intent is not accurate, which may + * misguide statahead. For example: + * The 'revalidate' call for 'getattr' and 'getxattr' of a dentry will + * have the same intent -- IT_GETATTR, while one dentry should access + * statahead cache once, otherwise statahead windows is messed up. * The solution is as following: - * Assign "lld_sa_generation" with "sai_generation" when a dentry - * "IT_GETATTR" for the first time, and the subsequent "IT_GETATTR" - * will bypass interacting with statahead thread for checking: - * "lld_sa_generation == lli_sai->sai_generation" + * Assign 'lld_sa_generation' with 'lli_sa_generation' when a dentry + * IT_GETATTR for the first time, and subsequent IT_GETATTR will + * bypass interacting with statahead cache by checking + * 'lld_sa_generation == lli->lli_sa_generation'. */ - if (ldd && lli->lli_sai && - ldd->lld_sa_generation == lli->lli_sai->sai_generation) + ldd = ll_d2d(dentry); + if (ldd && ldd->lld_sa_generation == lli->lli_sa_generation) return false; return true; } -static inline int -ll_statahead_enter(struct inode *dir, struct dentry **dentryp, int only_unplug) -{ - if (!dentry_need_statahead(dir, *dentryp)) - return -EAGAIN; - - return do_statahead_enter(dir, dentryp, only_unplug); -} - /* llite ioctl register support routine */ enum llioc_iter { LLIOC_CONT = 0, diff --git a/drivers/staging/lustre/lustre/llite/namei.c b/drivers/staging/lustre/lustre/llite/namei.c index 85f8ce7..494140a 100644 --- a/drivers/staging/lustre/lustre/llite/namei.c +++ b/drivers/staging/lustre/lustre/llite/namei.c @@ -522,8 +522,8 @@ static struct dentry *ll_lookup_it(struct inode *parent, struct dentry *dentry, if (!it || it->it_op == IT_GETXATTR) it = &lookup_it; - if (it->it_op == IT_GETATTR) { - rc = ll_statahead_enter(parent, &dentry, 0); + if (it->it_op == IT_GETATTR && dentry_may_statahead(parent, dentry)) { + rc = ll_statahead(parent, &dentry, 0); if (rc == 1) { if (dentry == save) retval = NULL; @@ -574,11 +574,8 @@ static struct dentry *ll_lookup_it(struct inode *parent, struct dentry *dentry, retval = NULL; else retval = dentry; - out: - if (req) - ptlrpc_req_finished(req); - if (it->it_op == IT_GETATTR && (!retval || retval == dentry)) - ll_statahead_mark(parent, dentry); +out: + ptlrpc_req_finished(req); return retval; } diff --git a/drivers/staging/lustre/lustre/llite/statahead.c b/drivers/staging/lustre/lustre/llite/statahead.c index 323a175..9e76a68 100644 --- a/drivers/staging/lustre/lustre/llite/statahead.c +++ b/drivers/staging/lustre/lustre/llite/statahead.c @@ -54,12 +54,12 @@ enum se_stat { /* * sa_entry is not refcounted: statahead thread allocates it and do async stat, * and in async stat callback ll_statahead_interpret() will add it into - * sai_cb_entries, later statahead thread will call sa_handle_callback() to + * sai_interim_entries, later statahead thread will call sa_handle_callback() to * instantiate entry and move it into sai_entries, and then only scanner process * can access and free it. */ struct sa_entry { - /* link into sai_cb_entries or sai_entries */ + /* link into sai_interim_entries or sai_entries */ struct list_head se_list; /* link into sai hash table locally */ struct list_head se_hash; @@ -84,23 +84,20 @@ struct sa_entry { static unsigned int sai_generation; static DEFINE_SPINLOCK(sai_generation_lock); -/* - * The entry only can be released by the caller, it is necessary to hold lock. - */ +/* sa_entry is ready to use */ static inline int sa_ready(struct sa_entry *entry) { smp_rmb(); return (entry->se_state != SA_ENTRY_INIT); } +/* hash value to put in sai_cache */ static inline int sa_hash(int val) { return val & LL_SA_CACHE_MASK; } -/* - * Insert entry to hash SA table. - */ +/* hash entry into sai_cache */ static inline void sa_rehash(struct ll_statahead_info *sai, struct sa_entry *entry) { @@ -130,11 +127,13 @@ static inline int agl_should_run(struct ll_statahead_info *sai, return (inode && S_ISREG(inode->i_mode) && sai->sai_agl_valid); } +/* statahead window is full */ static inline int sa_sent_full(struct ll_statahead_info *sai) { return atomic_read(&sai->sai_cache_count) >= sai->sai_max; } +/* got async stat replies */ static inline int sa_has_callback(struct ll_statahead_info *sai) { return !list_empty(&sai->sai_interim_entries); @@ -158,7 +157,7 @@ static inline int sa_low_hit(struct ll_statahead_info *sai) } /* - * If the given index is behind of statahead window more than + * if the given index is behind of statahead window more than * SA_OMITTED_ENTRY_MAX, then it is old. */ static inline int is_omitted_entry(struct ll_statahead_info *sai, __u64 index) @@ -167,9 +166,7 @@ static inline int is_omitted_entry(struct ll_statahead_info *sai, __u64 index) sai->sai_index); } -/* - * Insert it into sai_entries tail when init. - */ +/* allocate sa_entry and hash it to allow scanner process to find it */ static struct sa_entry * sa_alloc(struct dentry *parent, struct ll_statahead_info *sai, __u64 index, const char *name, int len) @@ -198,7 +195,7 @@ sa_alloc(struct dentry *parent, struct ll_statahead_info *sai, __u64 index, entry->se_qstr.len = len; entry->se_qstr.name = dname; - lli = ll_i2info(sai->sai_inode); + lli = ll_i2info(sai->sai_dentry->d_inode); spin_lock(&lli->lli_sa_lock); INIT_LIST_HEAD(&entry->se_list); sa_rehash(sai, entry); @@ -246,7 +243,7 @@ sa_get(struct ll_statahead_info *sai, const struct qstr *qstr) static inline void sa_kill(struct ll_statahead_info *sai, struct sa_entry *entry) { - struct ll_inode_info *lli = ll_i2info(sai->sai_inode); + struct ll_inode_info *lli = ll_i2info(sai->sai_dentry->d_inode); LASSERT(!list_empty(&entry->se_hash)); LASSERT(!list_empty(&entry->se_list)); @@ -271,7 +268,7 @@ sa_put(struct ll_statahead_info *sai, struct sa_entry *entry) struct sa_entry *tmp, *next; if (entry && entry->se_state == SA_ENTRY_SUCC) { - struct ll_sb_info *sbi = ll_i2sbi(sai->sai_inode); + struct ll_sb_info *sbi = ll_i2sbi(sai->sai_dentry->d_inode); sai->sai_hit++; sai->sai_consecutive_miss = 0; @@ -293,6 +290,7 @@ sa_put(struct ll_statahead_info *sai, struct sa_entry *entry) break; sa_kill(sai, tmp); } + wake_up(&sai->sai_thread.t_ctl_waitq); } @@ -329,7 +327,7 @@ __sa_make_ready(struct ll_statahead_info *sai, struct sa_entry *entry, int ret) static void sa_make_ready(struct ll_statahead_info *sai, struct sa_entry *entry, int ret) { - struct ll_inode_info *lli = ll_i2info(sai->sai_inode); + struct ll_inode_info *lli = ll_i2info(sai->sai_dentry->d_inode); struct md_enqueue_info *minfo = entry->se_minfo; struct ptlrpc_request *req = entry->se_req; bool wakeup; @@ -355,14 +353,12 @@ sa_make_ready(struct ll_statahead_info *sai, struct sa_entry *entry, int ret) wake_up(&sai->sai_waitq); } -/* - * Insert inode into the list of sai_agls. - */ +/* Insert inode into the list of sai_agls. */ static void ll_agl_add(struct ll_statahead_info *sai, struct inode *inode, int index) { struct ll_inode_info *child = ll_i2info(inode); - struct ll_inode_info *parent = ll_i2info(sai->sai_inode); + struct ll_inode_info *parent = ll_i2info(sai->sai_dentry->d_inode); int added = 0; spin_lock(&child->lli_agl_lock); @@ -387,8 +383,9 @@ static void ll_agl_add(struct ll_statahead_info *sai, } /* allocate sai */ -static struct ll_statahead_info *ll_sai_alloc(void) +static struct ll_statahead_info *ll_sai_alloc(struct dentry *dentry) { + struct ll_inode_info *lli = ll_i2info(dentry->d_inode); struct ll_statahead_info *sai; int i; @@ -396,14 +393,9 @@ static struct ll_statahead_info *ll_sai_alloc(void) if (!sai) return NULL; + sai->sai_dentry = dget(dentry); atomic_set(&sai->sai_refcount, 1); - spin_lock(&sai_generation_lock); - sai->sai_generation = ++sai_generation; - if (unlikely(sai_generation == 0)) - sai->sai_generation = ++sai_generation; - spin_unlock(&sai_generation_lock); - sai->sai_max = LL_SA_RPC_MIN; sai->sai_index = 1; init_waitqueue_head(&sai->sai_waitq); @@ -420,9 +412,27 @@ static struct ll_statahead_info *ll_sai_alloc(void) } atomic_set(&sai->sai_cache_count, 0); + spin_lock(&sai_generation_lock); + lli->lli_sa_generation = ++sai_generation; + if (unlikely(!sai_generation)) + lli->lli_sa_generation = ++sai_generation; + spin_unlock(&sai_generation_lock); + return sai; } +/* free sai */ +static inline void ll_sai_free(struct ll_statahead_info *sai) +{ + LASSERT(sai->sai_dentry); + dput(sai->sai_dentry); + kfree(sai); +} + +/* + * take refcount of sai if sai for @dir exists, which means statahead is on for + * this directory. + */ static inline struct ll_statahead_info *ll_sai_get(struct inode *dir) { struct ll_inode_info *lli = ll_i2info(dir); @@ -437,12 +447,16 @@ static inline struct ll_statahead_info *ll_sai_get(struct inode *dir) return sai; } +/* + * put sai refcount after use, if refcount reaches zero, free sai and sa_entries + * attached to it. + */ static void ll_sai_put(struct ll_statahead_info *sai) { - struct ll_inode_info *lli = ll_i2info(sai->sai_inode); + struct ll_inode_info *lli = ll_i2info(sai->sai_dentry->d_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_sb_info *sbi = ll_i2sbi(sai->sai_dentry->d_inode); struct sa_entry *entry, *next; lli->lli_sai = NULL; @@ -460,8 +474,7 @@ static void ll_sai_put(struct ll_statahead_info *sai) LASSERT(atomic_read(&sai->sai_cache_count) == 0); LASSERT(list_empty(&sai->sai_agls)); - iput(sai->sai_inode); - kfree(sai); + ll_sai_free(sai); atomic_dec(&sbi->ll_sa_running); } } @@ -533,7 +546,7 @@ static void ll_agl_trigger(struct inode *inode, struct ll_statahead_info *sai) static void sa_instantiate(struct ll_statahead_info *sai, struct sa_entry *entry) { - struct inode *dir = sai->sai_inode; + struct inode *dir = sai->sai_dentry->d_inode; struct inode *child; struct md_enqueue_info *minfo; struct lookup_intent *it; @@ -609,12 +622,12 @@ out: sa_make_ready(sai, entry, rc); } -/* once there are async stat replies, instantiate sa_entry */ +/* once there are async stat replies, instantiate sa_entry from replies */ static void sa_handle_callback(struct ll_statahead_info *sai) { struct ll_inode_info *lli; - lli = ll_i2info(sai->sai_inode); + lli = ll_i2info(sai->sai_dentry->d_inode); while (sa_has_callback(sai)) { struct sa_entry *entry; @@ -631,21 +644,6 @@ static void sa_handle_callback(struct ll_statahead_info *sai) sa_instantiate(sai, entry); } - - spin_lock(&lli->lli_agl_lock); - while (!agl_list_empty(sai)) { - struct ll_inode_info *clli; - - clli = list_entry(sai->sai_agls.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); } /* @@ -718,6 +716,7 @@ static int ll_statahead_interpret(struct ptlrpc_request *req, return rc; } +/* finish async stat RPC arguments */ static void sa_fini_data(struct md_enqueue_info *minfo, struct ldlm_enqueue_info *einfo) { @@ -775,6 +774,7 @@ static int sa_prep_data(struct inode *dir, struct inode *child, return 0; } +/* async stat for file not found in dcache */ static int sa_lookup(struct inode *dir, struct sa_entry *entry) { struct md_enqueue_info *minfo; @@ -786,17 +786,18 @@ static int sa_lookup(struct inode *dir, struct sa_entry *entry) return rc; rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo); - if (rc < 0) + if (rc) sa_fini_data(minfo, einfo); return rc; } /** - * similar to ll_revalidate_it(). - * \retval 1 -- dentry valid - * \retval 0 -- will send stat-ahead request - * \retval others -- prepare stat-ahead request failed + * async stat for file found in dcache, similar to .revalidate + * + * \retval 1 dentry valid, no RPC sent + * \retval 0 dentry invalid, will send async stat RPC + * \retval negative number upon error */ static int sa_revalidate(struct inode *dir, struct sa_entry *entry, struct dentry *dentry) @@ -831,7 +832,7 @@ static int sa_revalidate(struct inode *dir, struct sa_entry *entry, } rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo, einfo); - if (rc < 0) { + if (rc) { entry->se_inode = NULL; iput(inode); sa_fini_data(minfo, einfo); @@ -840,6 +841,7 @@ static int sa_revalidate(struct inode *dir, struct sa_entry *entry, return rc; } +/* async stat for file with @name */ static void sa_statahead(struct dentry *parent, const char *name, int len) { struct inode *dir = d_inode(parent); @@ -873,6 +875,7 @@ static void sa_statahead(struct dentry *parent, const char *name, int len) sai->sai_index++; } +/* async glimpse (agl) thread main function */ static int ll_agl_thread(void *arg) { struct dentry *parent = arg; @@ -946,6 +949,7 @@ static int ll_agl_thread(void *arg) return 0; } +/* start agl thread */ static void ll_start_agl(struct dentry *parent, struct ll_statahead_info *sai) { struct ptlrpc_thread *thread = &sai->sai_agl_thread; @@ -970,6 +974,7 @@ static void ll_start_agl(struct dentry *parent, struct ll_statahead_info *sai) &lwi); } +/* statahead thread main function */ static int ll_statahead_thread(void *arg) { struct dentry *parent = arg; @@ -977,7 +982,7 @@ static int ll_statahead_thread(void *arg) struct ll_inode_info *lli = ll_i2info(dir); struct ll_sb_info *sbi = ll_i2sbi(dir); struct ll_statahead_info *sai; - struct ptlrpc_thread *thread; + struct ptlrpc_thread *sa_thread; struct ptlrpc_thread *agl_thread; struct page *page = NULL; __u64 pos = 0; @@ -987,9 +992,9 @@ static int ll_statahead_thread(void *arg) struct l_wait_info lwi = { 0 }; sai = ll_sai_get(dir); - thread = &sai->sai_thread; + sa_thread = &sai->sai_thread; agl_thread = &sai->sai_agl_thread; - thread->t_pid = current_pid(); + sa_thread->t_pid = current_pid(); CDEBUG(D_READA, "statahead thread starting: sai %p, parent %pd\n", sai, parent); @@ -1007,16 +1012,16 @@ static int ll_statahead_thread(void *arg) atomic_inc(&sbi->ll_sa_total); spin_lock(&lli->lli_sa_lock); - if (thread_is_init(thread)) + if (thread_is_init(sa_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); + thread_set_flags(sa_thread, SVC_RUNNING); spin_unlock(&lli->lli_sa_lock); - wake_up(&thread->t_ctl_waitq); + wake_up(&sa_thread->t_ctl_waitq); - while (pos != MDS_DIR_END_OFF && thread_is_running(thread)) { + while (pos != MDS_DIR_END_OFF && thread_is_running(sa_thread)) { struct lu_dirpage *dp; struct lu_dirent *ent; @@ -1033,7 +1038,7 @@ static int ll_statahead_thread(void *arg) dp = page_address(page); for (ent = lu_dirent_start(dp); - ent && thread_is_running(thread) && !sa_low_hit(sai); + ent && thread_is_running(sa_thread) && !sa_low_hit(sai); ent = lu_dirent_next(ent)) { __u64 hash; int namelen; @@ -1082,15 +1087,32 @@ static int ll_statahead_thread(void *arg) /* wait for spare statahead window */ do { - l_wait_event(thread->t_ctl_waitq, + l_wait_event(sa_thread->t_ctl_waitq, !sa_sent_full(sai) || sa_has_callback(sai) || !list_empty(&sai->sai_agls) || - !thread_is_running(thread), + !thread_is_running(sa_thread), &lwi); sa_handle_callback(sai); + + spin_lock(&lli->lli_agl_lock); + while (sa_sent_full(sai) && + !agl_list_empty(sai)) { + struct ll_inode_info *clli; + + clli = list_entry(sai->sai_agls.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); } while (sa_sent_full(sai) && - thread_is_running(thread)); + thread_is_running(sa_thread)); sa_statahead(parent, name, namelen); } @@ -1113,7 +1135,7 @@ static int ll_statahead_thread(void *arg) if (rc < 0) { spin_lock(&lli->lli_sa_lock); - thread_set_flags(thread, SVC_STOPPING); + thread_set_flags(sa_thread, SVC_STOPPING); lli->lli_sa_enabled = 0; spin_unlock(&lli->lli_sa_lock); } @@ -1122,11 +1144,11 @@ static int ll_statahead_thread(void *arg) * 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, + while (thread_is_running(sa_thread)) { + l_wait_event(sa_thread->t_ctl_waitq, sa_has_callback(sai) || !agl_list_empty(sai) || - !thread_is_running(thread), + !thread_is_running(sa_thread), &lwi); sa_handle_callback(sai); @@ -1156,7 +1178,7 @@ out: /* in case we're not woken up, timeout wait */ lwi = LWI_TIMEOUT(msecs_to_jiffies(MSEC_PER_SEC >> 3), NULL, NULL); - l_wait_event(thread->t_ctl_waitq, + l_wait_event(sa_thread->t_ctl_waitq, sai->sai_sent == sai->sai_replied, &lwi); } @@ -1164,19 +1186,20 @@ out: sa_handle_callback(sai); spin_lock(&lli->lli_sa_lock); - thread_set_flags(thread, SVC_STOPPED); + thread_set_flags(sa_thread, SVC_STOPPED); spin_unlock(&lli->lli_sa_lock); - wake_up(&sai->sai_waitq); - wake_up(&thread->t_ctl_waitq); - ll_sai_put(sai); CDEBUG(D_READA, "statahead thread stopped: sai %p, parent %pd\n", sai, parent); - dput(parent); + + wake_up(&sai->sai_waitq); + wake_up(&sa_thread->t_ctl_waitq); + ll_sai_put(sai); + return rc; } -/* authorize opened dir handle @key to statahead later */ +/* authorize opened dir handle @key to statahead */ void ll_authorize_statahead(struct inode *dir, void *key) { struct ll_inode_info *lli = ll_i2info(dir); @@ -1230,7 +1253,7 @@ enum { /** * not first dirent, or is "." */ - LS_NONE_FIRST_DE = 0, + LS_NOT_FIRST_DE = 0, /** * the first non-hidden dirent */ @@ -1241,6 +1264,7 @@ enum { LS_FIRST_DOT_DE }; +/* file is first dirent under @dir */ static int is_first_dirent(struct inode *dir, struct dentry *dentry) { const struct qstr *target = &dentry->d_name; @@ -1248,7 +1272,7 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry) struct page *page; __u64 pos = 0; int dot_de; - int rc = LS_NONE_FIRST_DE; + int rc = LS_NOT_FIRST_DE; op_data = ll_prep_md_op_data(NULL, dir, dir, NULL, 0, 0, LUSTRE_OPC_ANY, dir); @@ -1324,7 +1348,7 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry) if (target->len != namelen || memcmp(target->name, name, namelen) != 0) - rc = LS_NONE_FIRST_DE; + rc = LS_NOT_FIRST_DE; else if (!dot_de) rc = LS_FIRST_DE; else @@ -1356,13 +1380,27 @@ out: return rc; } +/** + * revalidate @dentryp from statahead cache + * + * \param[in] dir parent directory + * \param[in] sai sai structure + * \param[out] dentryp pointer to dentry which will be revalidated + * \param[in] unplug unplug statahead window only (normally for negative + * dentry) + * \retval 1 on success, dentry is saved in @dentryp + * \retval 0 if revalidation failed (no proper lock on client) + * \retval negative number upon error + */ static int revalidate_statahead_dentry(struct inode *dir, struct ll_statahead_info *sai, struct dentry **dentryp, - int only_unplug) + bool unplug) { struct sa_entry *entry = NULL; struct l_wait_info lwi = { 0 }; + struct ll_dentry_data *ldd; + struct ll_inode_info *lli; int rc = 0; if ((*dentryp)->d_name.name[0] == '.') { @@ -1392,10 +1430,15 @@ static int revalidate_statahead_dentry(struct inode *dir, } } + if (unplug) { + rc = 1; + goto out_unplug; + } + entry = sa_get(sai, &(*dentryp)->d_name); - if (!entry || only_unplug) { - sa_put(sai, entry); - return entry ? 1 : -EAGAIN; + if (!entry) { + rc = -EAGAIN; + goto out_unplug; } /* if statahead is busy in readdir, help it do post-work */ @@ -1406,13 +1449,15 @@ static int revalidate_statahead_dentry(struct inode *dir, 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, - sa_ready(entry) || - thread_is_stopped(&sai->sai_thread), - &lwi); + rc = l_wait_event(sai->sai_waitq, sa_ready(entry), &lwi); if (rc < 0) { - sa_put(sai, entry); - return -EAGAIN; + /* + * entry may not be ready, so it may be used by inflight + * statahead RPC, don't free it. + */ + entry = NULL; + rc = -EAGAIN; + goto out_unplug; } } @@ -1430,10 +1475,15 @@ static int revalidate_statahead_dentry(struct inode *dir, alias = ll_splice_alias(inode, *dentryp); if (IS_ERR(alias)) { - sa_put(sai, entry); - return PTR_ERR(alias); + rc = PTR_ERR(alias); + goto out_unplug; } *dentryp = alias; + /** + * statahead prepared this inode, transfer inode + * refcount from sa_entry to dentry + */ + entry->se_inode = NULL; } else if ((*dentryp)->d_inode != inode) { /* revalidate, but inode is recreated */ CDEBUG(D_READA, @@ -1445,10 +1495,7 @@ static int revalidate_statahead_dentry(struct inode *dir, PFID(ll_inode2fid(inode))); rc = -ESTALE; goto out_unplug; - } else { - iput(inode); } - entry->se_inode = NULL; if ((bits & MDS_INODELOCK_LOOKUP) && d_lustre_invalid(*dentryp)) @@ -1457,10 +1504,34 @@ static int revalidate_statahead_dentry(struct inode *dir, } } out_unplug: + /* + * statahead cached sa_entry can be used only once, and will be killed + * right after use, so if lookup/revalidate accessed statahead cache, + * set dentry ldd_sa_generation to parent lli_sa_generation, later if we + * stat this file again, we know we've done statahead before, see + * dentry_may_statahead(). + */ + ldd = ll_d2d(*dentryp); + lli = ll_i2info(dir); + /* ldd can be NULL if llite lookup failed. */ + if (ldd) + ldd->lld_sa_generation = lli->lli_sa_generation; sa_put(sai, entry); return rc; } +/** + * start statahead thread + * + * \param[in] dir parent directory + * \param[in] dentry dentry that triggers statahead, normally the first + * dirent under @dir + * \retval -EAGAIN on success, because when this function is + * called, it's already in lookup call, so client should + * do it itself instead of waiting for statahead thread + * to do it asynchronously. + * \retval negative number upon error + */ static int start_statahead_thread(struct inode *dir, struct dentry *dentry) { struct ll_inode_info *lli = ll_i2info(dir); @@ -1468,60 +1539,34 @@ static int start_statahead_thread(struct inode *dir, struct dentry *dentry) struct l_wait_info lwi = { 0 }; struct ptlrpc_thread *thread; struct task_struct *task; - struct dentry *parent; + struct dentry *parent = dentry->d_parent; int rc; /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */ rc = is_first_dirent(dir, dentry); - if (rc == LS_NONE_FIRST_DE) { + if (rc == LS_NOT_FIRST_DE) { /* It is not "ls -{a}l" operation, no need statahead for it. */ - rc = -EAGAIN; + rc = -EFAULT; goto out; } - sai = ll_sai_alloc(); + sai = ll_sai_alloc(parent); if (!sai) { rc = -ENOMEM; goto out; } sai->sai_ls_all = (rc == LS_FIRST_DOT_DE); - sai->sai_inode = igrab(dir); - if (unlikely(!sai->sai_inode)) { - CWARN("Do not start stat ahead on dying inode "DFID"\n", - PFID(&lli->lli_fid)); - rc = -ESTALE; - goto out; - } - - /* get parent reference count here, and put it in ll_statahead_thread */ - 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", - dentry, PFID(&lli->lli_fid), PFID(&nlli->lli_fid)); - dput(parent); - iput(sai->sai_inode); - rc = -EAGAIN; - goto out; - } - - CDEBUG(D_READA, "start statahead thread: sai %p, parent %pd\n", - sai, parent); - /* - * if another process started statahead thread, or deauthorized current - * lli_opendir_key, don't start statahead. + * if current lli_opendir_key was deauthorized, or dir re-opened by + * another process, don't start statahead, otherwise the newly spawned + * statahead thread won't be notified to quit. */ spin_lock(&lli->lli_sa_lock); if (unlikely(lli->lli_sai || lli->lli_opendir_key || lli->lli_opendir_pid != current->pid)) { spin_unlock(&lli->lli_sa_lock); - - dput(parent); - iput(sai->sai_inode); - rc = -EAGAIN; + rc = -EPERM; goto out; } lli->lli_sai = sai; @@ -1529,22 +1574,16 @@ static int start_statahead_thread(struct inode *dir, struct dentry *dentry) atomic_inc(&ll_i2sbi(parent->d_inode)->ll_sa_running); + CDEBUG(D_READA, "start statahead thread: [pid %d] [parent %pd]\n", + current_pid(), parent); + task = kthread_run(ll_statahead_thread, parent, "ll_sa_%u", lli->lli_opendir_pid); thread = &sai->sai_thread; if (IS_ERR(task)) { rc = PTR_ERR(task); - CERROR("cannot start ll_sa thread: rc = %d\n", rc); - dput(parent); - - spin_lock(&lli->lli_sa_lock); - thread_set_flags(thread, SVC_STOPPED); - thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED); - spin_unlock(&lli->lli_sa_lock); - - ll_sai_put(sai); - LASSERT(!lli->lli_sai); - return -EAGAIN; + CERROR("can't start ll_sa thread, rc : %d\n", rc); + goto out; } l_wait_event(thread->t_ctl_waitq, @@ -1559,29 +1598,35 @@ static int start_statahead_thread(struct inode *dir, struct dentry *dentry) return -EAGAIN; out: - kfree(sai); /* * once we start statahead thread failed, disable statahead so - * subsequent won't waste time to try it. + * that subsequent stat won't waste time to try it. */ spin_lock(&lli->lli_sa_lock); lli->lli_sa_enabled = 0; + lli->lli_sai = NULL; spin_unlock(&lli->lli_sa_lock); - + if (sai) + ll_sai_free(sai); 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. + * statahead entry function, this is called when client getattr on a file, it + * will start statahead thread if this is the first dir entry, else revalidate + * dentry from statahead cache. + * + * \param[in] dir parent directory + * \param[out] dentryp dentry to getattr + * \param[in] unplug unplug statahead window only (normally for negative + * dentry) + * \retval 1 on success + * \retval 0 revalidation from statahead cache failed, caller needs + * to getattr from server directly + * \retval negative number on error, caller often ignores this and + * then getattr from server */ -int do_statahead_enter(struct inode *dir, struct dentry **dentryp, - int only_unplug) +int ll_statahead(struct inode *dir, struct dentry **dentryp, bool unplug) { struct ll_statahead_info *sai; @@ -1589,13 +1634,11 @@ int do_statahead_enter(struct inode *dir, struct dentry **dentryp, if (sai) { int rc; - rc = revalidate_statahead_dentry(dir, sai, dentryp, - only_unplug); + rc = revalidate_statahead_dentry(dir, sai, dentryp, unplug); CDEBUG(D_READA, "revalidate statahead %pd: %d.\n", *dentryp, rc); ll_sai_put(sai); return rc; } - return start_statahead_thread(dir, *dentryp); } diff --git a/drivers/staging/lustre/lustre/mdc/mdc_request.c b/drivers/staging/lustre/lustre/mdc/mdc_request.c index 166f0c4..125d882 100644 --- a/drivers/staging/lustre/lustre/mdc/mdc_request.c +++ b/drivers/staging/lustre/lustre/mdc/mdc_request.c @@ -1367,7 +1367,7 @@ static int mdc_read_page(struct obd_export *exp, struct md_op_data *op_data, page = mdc_page_locate(mapping, &rp_param.rp_off, &start, &end, rp_param.rp_hash64); if (IS_ERR(page)) { - CERROR("%s: dir page locate: "DFID" at %llu: rc %ld\n", + CDEBUG(D_INFO, "%s: dir page locate: " DFID " at %llu: rc %ld\n", exp->exp_obd->obd_name, PFID(&op_data->op_fid1), rp_param.rp_off, PTR_ERR(page)); rc = PTR_ERR(page); -- 1.7.1 _______________________________________________ devel mailing list devel@xxxxxxxxxxxxxxxxxxxxxx http://driverdev.linuxdriverproject.org/mailman/listinfo/driverdev-devel
