[PATCH 070/124] staging: lustre: statahead: small fixes and cleanup

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

 



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




[Index of Archives]     [Linux Driver Backports]     [DMA Engine]     [Linux GPIO]     [Linux SPI]     [Video for Linux]     [Linux USB Devel]     [Linux Coverity]     [Linux Audio Users]     [Linux Kernel]     [Linux SCSI]     [Yosemite Backpacking]
  Powered by Linux