From f6b13fb12e35276c72154c52c92bcaf3db71427a Mon Sep 17 00:00:00 2001
From: Roland Fehrenbacher <rf@q-leap.de>
Date: Sat, 15 Feb 2014 15:32:54 +0100
Subject: [PATCH 2/2] QL: Lustre - port patch LU-3270 statahead: statahead
 thread wait for RPCs to finish

Original patch message:

LU-3270 statahead: statahead thread wait for RPCs to finish

Statahead thread should wait for inflight statahead RPCs to
finish before quit, because statahead RPC callback will access
data allocated in context of statahead thread, and this can
also guarantee all referenced inodes are put in client umount.

Several other fixes and cleanups to statahead:
* when only_unplug is set for do_statahead_enter(), it should
  not unplug the entry found, because the inflight statahead
  RPC may not finish yet.
* fix the bug in LU-2272: ll_intent_drop_lock() may sleep, which
  should not be called inside lli_sa_lock.
* removed sa_entry.se_refcount because now statahead threads wait
  for RPCs to finish, it's always safe to access sa_entries, and
  scanner is the only one that removes sa_entry.
* simplify sai refcounting:
  - statahead thread holds one refcount.
  - do_statahead_enter()/ll_deauthorize_statahead() takes refcount
    before use.
* add lli_sa_enabled flag to enable/disable statahead for current
  opened dir.

Signed-off-by: Lai Siyao <lai.siyao@intel.com>
Change-Id: I03788d18a230c697f9a93390306af3f5a2d6ef62
---
 drivers/staging/lustre/lustre/include/obd.h        |    7 +-
 drivers/staging/lustre/lustre/llite/dcache.c       |    2 +-
 drivers/staging/lustre/lustre/llite/file.c         |   23 +-
 .../staging/lustre/lustre/llite/llite_internal.h   |  212 +++--
 drivers/staging/lustre/lustre/llite/llite_lib.c    |    6 +
 drivers/staging/lustre/lustre/llite/statahead.c    |  820 +++++++++-----------
 6 files changed, 483 insertions(+), 587 deletions(-)

diff --git a/drivers/staging/lustre/lustre/include/obd.h b/drivers/staging/lustre/lustre/include/obd.h
index a612255..35ab976 100644
--- a/drivers/staging/lustre/lustre/include/obd.h
+++ b/drivers/staging/lustre/lustre/include/obd.h
@@ -1101,10 +1101,9 @@ struct md_enqueue_info {
 	struct md_op_data       mi_data;
 	struct lookup_intent    mi_it;
 	struct lustre_handle    mi_lockh;
-	struct inode	   *mi_dir;
-	md_enqueue_cb_t	 mi_cb;
-	__u64		   mi_cbdata;
-	unsigned int	    mi_generation;
+	struct inode	       *mi_dir;
+	md_enqueue_cb_t	        mi_cb;
+	void		       *mi_cbdata;
 };
 
 struct obd_ops {
diff --git a/drivers/staging/lustre/lustre/llite/dcache.c b/drivers/staging/lustre/lustre/llite/dcache.c
index 01e709c..a928a1d 100644
--- a/drivers/staging/lustre/lustre/llite/dcache.c
+++ b/drivers/staging/lustre/lustre/llite/dcache.c
@@ -633,7 +633,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 bc534db..4974884 100644
--- a/drivers/staging/lustre/lustre/llite/file.c
+++ b/drivers/staging/lustre/lustre/llite/file.c
@@ -321,9 +321,8 @@ int ll_file_release(struct inode *inode, struct file *file)
 	/* The last ref on @file, maybe not the 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. */
-	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 (inode->i_sb->s_root == file->f_dentry) {
 		LUSTRE_FPRIVATE(file) = NULL;
@@ -504,7 +503,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=%lu/%u(%p), flags %o\n", inode->i_ino,
 	       inode->i_generation, inode, file->f_flags);
@@ -517,16 +516,8 @@ int ll_file_open(struct inode *inode, struct file *file)
 		GOTO(out_openerr, rc = -ENOMEM);
 
 	fd->fd_file = file;
-	if (S_ISDIR(inode->i_mode)) {
-		spin_lock(&lli->lli_sa_lock);
-		if (lli->lli_opendir_key == NULL && lli->lli_sai == NULL &&
-		    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 (inode->i_sb->s_root == file->f_dentry) {
 		LUSTRE_FPRIVATE(file) = fd;
@@ -665,8 +656,8 @@ out_och_free:
 		mutex_unlock(&lli->lli_och_mutex);
 
 out_openerr:
-		if (opendir_set != 0)
-			ll_stop_statahead(inode, lli->lli_opendir_key);
+		if (lli->lli_opendir_key == fd)
+			ll_deauthorize_statahead(inode, fd);
 		if (fd != NULL)
 			ll_file_data_put(fd);
 	} else {
diff --git a/drivers/staging/lustre/lustre/llite/llite_internal.h b/drivers/staging/lustre/lustre/llite/llite_internal.h
index 47e443d..1612516 100644
--- a/drivers/staging/lustre/lustre/llite/llite_internal.h
+++ b/drivers/staging/lustre/lustre/llite/llite_internal.h
@@ -179,36 +179,32 @@ 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
 			 * struct, "opendir_key" is the token when dir close for
 			 * case of parent exit before child -- it is me should
 			 * cleanup the dir readahead. */
-			void			   *d_opendir_key;
-			struct ll_statahead_info       *d_sai;
-			struct posix_acl	       *d_def_acl;
+			void			   *lli_opendir_key;
+			struct ll_statahead_info       *lli_sai;
+			unsigned int			lli_sa_generation;
+			/* statahead is enabled for current opened-dir */
+			unsigned int			lli_sa_enabled:1;
+			struct posix_acl	       *lli_def_acl;
 			/* protect statahead stuff. */
-			spinlock_t			d_sa_lock;
+			spinlock_t			lli_sa_lock;
 			/* "opendir_pid" is the token when lookup/revalid
 			 * -- I am the owner of dir statahead. */
-			pid_t			   d_opendir_pid;
-		} 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_def_acl	     u.d.d_def_acl
-#define lli_sa_lock	     u.d.d_sa_lock
-#define lli_opendir_pid	 u.d.d_opendir_pid
+			pid_t		 		lli_opendir_pid;
+		};
 
 		/* for non-directory */
 		struct {
-			struct semaphore		f_size_sem;
-			void				*f_size_sem_owner;
-			char				*f_symlink_name;
-			__u64				f_maxbytes;
+			struct semaphore		lli_size_sem;
+			void			       *lli_size_sem_owner;
+			char			       *lli_symlink_name;
+			__u64				lli_maxbytes;
 			/*
 			 * struct rw_semaphore {
 			 *    signed long	count;     // align d.d_def_acl
@@ -216,22 +212,22 @@ struct ll_inode_info {
 			 *    struct list_head wait_list;
 			 * }
 			 */
-			struct rw_semaphore		f_trunc_sem;
-			struct mutex			f_write_mutex;
+			struct rw_semaphore		lli_trunc_sem;
+			struct mutex			lli_write_mutex;
 
-			struct rw_semaphore		f_glimpse_sem;
-			cfs_time_t			f_glimpse_time;
-			struct list_head			f_agl_list;
-			__u64				f_agl_index;
+			struct rw_semaphore		lli_glimpse_sem;
+			cfs_time_t			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;
 
 			/* volatile file criteria is based on file name, this
 			 * flag is used to keep the test result, so the strcmp
 			 * is done only once
 			 */
-			bool				f_volatile;
+			bool				lli_volatile;
 			/*
 			 * whenever a process try to read/write the file, the
 			 * jobid of the process will be saved here, and it'll
@@ -240,24 +236,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[JOBSTATS_JOBID_SIZE];
-		} f;
-
-#define lli_size_sem	    u.f.f_size_sem
-#define lli_size_sem_owner      u.f.f_size_sem_owner
-#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_mutex	 u.f.f_write_mutex
-#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
-#define lli_volatile		u.f.f_volatile
-
-	} u;
+			char			lli_jobid[JOBSTATS_JOBID_SIZE];
+		};
+	};
 
 	/* XXX: For following frequent used members, although they maybe special
 	 *      used for non-directory object, it is some time-wasting to check
@@ -268,8 +249,8 @@ struct ll_inode_info {
 	 *
 	 *      In the future, if more members are added only for directory,
 	 *      some of the following members can be moved into u.f.
-	 */
-	bool			    lli_has_smd;
+	*/
+	bool				lli_has_smd;
 	struct cl_object	       *lli_clob;
 
 	/* mutex to request for layout lock exclusively. */
@@ -494,17 +475,19 @@ struct ll_sb_info {
 	int		       ll_rw_stats_on;
 
 	/* metadata stat-ahead */
-	unsigned int	      ll_sa_max;     /* max statahead RPCs */
+	unsigned int	          ll_sa_max;     /* max statahead RPCs */
 	atomic_t		  ll_sa_total;   /* statahead thread started
 						  * count */
+	atomic_t                  ll_sa_running; /* running statahead thread
+						  * count */
 	atomic_t		  ll_sa_wrong;   /* statahead thread stopped for
 						  * low hit ratio */
 	atomic_t		  ll_agl_total;  /* AGL thread started count */
 
-	dev_t		     ll_sdev_orig; /* save s_dev before assign for
-						 * clustred nfs */
+	dev_t		          ll_sdev_orig;  /* save s_dev before assign for
+						  * clustred nfs */
 	struct rmtacl_ctl_table   ll_rct;
-	struct eacl_table	 ll_et;
+	struct eacl_table	  ll_et;
 	__kernel_fsid_t		  ll_fsid;
 };
 
@@ -1203,29 +1186,28 @@ static inline obd_valid rce_ops2valid(int ops)
 
 /* 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
-						 * reply */
-	__u64		   sai_index;      /* index of statahead entry */
-	__u64		   sai_index_wait; /* index of entry which is the
-						 * caller is waiting for */
-	__u64		   sai_hit;	/* hit count */
-	__u64		   sai_miss;       /* miss count:
-						 * for "ls -al" case, it includes
-						 * hidden dentry miss;
-						 * for "ls -l" case, it does not
-						 * include hidden dentry miss.
-						 * "sai_miss_hidden" is used for
-						 * the later case.
-						 */
+					     * refcount */
+	unsigned int	    sai_max;	    /* max ahead of lookup */
+	__u64		    sai_sent;       /* stat requests sent count */
+	__u64		    sai_replied;    /* stat requests which received
+					     * reply */
+	__u64		    sai_index;      /* index of statahead entry */
+	__u64		    sai_index_wait; /* index of entry which is the
+					     * caller is waiting for */
+	__u64		    sai_hit;        /* hit count */
+	__u64		    sai_miss;       /* miss count:
+					     * for "ls -al" case, it includes
+					     * hidden dentry miss;
+					     * for "ls -l" case, it does not
+					     * include hidden dentry miss.
+					     * "sai_miss_hidden" is used for
+					     * the later case.
+					     */
 	unsigned int	    sai_consecutive_miss; /* consecutive miss */
 	unsigned int	    sai_miss_hidden;/* "ls -al", but first dentry
-						 * is not a hidden one */
+					     * is not a hidden one */
 	unsigned int	    sai_skip_hidden;/* skipped hidden dentry count */
 	unsigned int	    sai_ls_all:1,   /* "ls -al", do stat-ahead for
 						 * hidden entries */
@@ -1234,18 +1216,19 @@ struct ll_statahead_info {
 	wait_queue_head_t	     sai_waitq;      /* stat-ahead wait queue */
 	struct ptlrpc_thread    sai_thread;     /* stat-ahead thread */
 	struct ptlrpc_thread    sai_agl_thread; /* AGL thread */
-	struct list_head	      sai_entries;    /* entry list */
-	struct list_head	      sai_entries_received; /* entries returned */
-	struct list_head	      sai_entries_stated;   /* entries stated */
-	struct list_head	      sai_entries_agl; /* AGL entries to be sent */
-	struct list_head	      sai_cache[LL_SA_CACHE_SIZE];
+	struct list_head	sai_entries;    /* entry list */
+	struct list_head	sai_entries_received; /* entries returned */
+	struct list_head	sai_entries_stated;   /* entries stated */
+	struct list_head	sai_entries_agl; /* AGL entries to be sent */
+	struct list_head	sai_cache[LL_SA_CACHE_SIZE];
 	spinlock_t		sai_cache_lock[LL_SA_CACHE_SIZE];
 	atomic_t		sai_cache_count; /* entry count in cache */
 };
 
 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);
 
 static inline int ll_glimpse_size(struct inode *inode)
 {
@@ -1263,66 +1246,63 @@ 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;
 
-	if (sai != NULL && ldd != NULL)
-		ldd->lld_sa_generation = sai->sai_generation;
+	LASSERT(ldd != NULL);
+	ldd->lld_sa_generation = lli->lli_sa_generation;
 }
 
-static inline int
-ll_need_statahead(struct inode *dir, struct dentry *dentryp)
+/*
+ * 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. The
+ * solution is as following: Assign "lld_sa_generation" with "lli_sa_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->lli_sa_generation"
+ */
+static inline bool
+dentry_was_stated(struct ll_inode_info *lli, struct dentry *dentry)
+{
+	struct ll_dentry_data *ldd = ll_d2d(dentry);
+
+	return (ldd != NULL &&
+		ldd->lld_sa_generation == lli->lli_sa_generation);
+}
+
+static inline bool
+dentry_need_statahead(struct inode *dir, struct dentry *dentry)
 {
-	struct ll_inode_info  *lli;
-	struct ll_dentry_data *ldd;
+	struct ll_inode_info  *lli = ll_i2info(dir);
 
 	if (ll_i2sbi(dir)->ll_sa_max == 0)
-		return -EAGAIN;
+		return false;
+
+	if (!lli->lli_sa_enabled)
+		return false;
 
-	lli = ll_i2info(dir);
-	/* not the same process, don't statahead */
 	if (lli->lli_opendir_pid != current_pid())
-		return -EAGAIN;
+		return false;
 
-	/* statahead has been stopped */
-	if (lli->lli_opendir_key == NULL)
-		return -EAGAIN;
+	if (lli->lli_sai != NULL && dentry_was_stated(lli, dentry))
+		return false;
 
-	ldd = ll_d2d(dentryp);
-	/*
-	 * 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.
-	 * 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"
-	 */
-	if (ldd && lli->lli_sai &&
-	    ldd->lld_sa_generation == lli->lli_sai->sai_generation)
-		return -EAGAIN;
-
-	return 1;
+	return true;
 }
 
 static inline int
 ll_statahead_enter(struct inode *dir, struct dentry **dentryp, int only_unplug)
 {
-	int ret;
-
-	ret = ll_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 b868c2b..94d9a0d 100644
--- a/drivers/staging/lustre/lustre/llite/llite_lib.c
+++ b/drivers/staging/lustre/lustre/llite/llite_lib.c
@@ -135,6 +135,7 @@ static struct ll_sb_info *ll_init_sbi(void)
 	/* metadata statahead is enabled by default */
 	sbi->ll_sa_max = LL_SA_RPC_DEF;
 	atomic_set(&sbi->ll_sa_total, 0);
+	atomic_set(&sbi->ll_sa_running, 0);
 	atomic_set(&sbi->ll_sa_wrong, 0);
 	atomic_set(&sbi->ll_agl_total, 0);
 	sbi->ll_flags |= LL_SBI_AGL_ENABLED;
@@ -717,6 +718,10 @@ void ll_kill_super(struct super_block *sb)
 		sb->s_dev = sbi->ll_sdev_orig;
 		sbi->ll_umounting = 1;
 	}
+
+	/* wait running statahead threads quit */
+	while (atomic_read(&sbi->ll_sa_running) > 0)
+		schedule_timeout_and_set_state(TASK_UNINTERRUPTIBLE, HZ >> 3);
 }
 
 char *ll_read_opt(const char *opt, char *data)
@@ -926,6 +931,7 @@ void ll_lli_init(struct ll_inode_info *lli)
 	if (S_ISDIR(lli->lli_vfs_inode.i_mode)) {
 		mutex_init(&lli->lli_readdir_mutex);
 		lli->lli_opendir_key = NULL;
+		lli->lli_sa_enabled = 0;
 		lli->lli_sai = NULL;
 		lli->lli_def_acl = NULL;
 		spin_lock_init(&lli->lli_sa_lock);
diff --git a/drivers/staging/lustre/lustre/llite/statahead.c b/drivers/staging/lustre/lustre/llite/statahead.c
index 8eaa38e9..3c0aed9 100644
--- a/drivers/staging/lustre/lustre/llite/statahead.c
+++ b/drivers/staging/lustre/lustre/llite/statahead.c
@@ -64,8 +64,6 @@ struct ll_sa_entry {
 	struct list_head	      se_list;
 	/* link into sai hash table locally */
 	struct list_head	      se_hash;
-	/* entry reference count */
-	atomic_t	    se_refcount;
 	/* entry index in the sai */
 	__u64		   se_index;
 	/* low layer ldlm lock handle */
@@ -212,29 +210,11 @@ ll_sa_entry_alloc(struct ll_statahead_info *sai, __u64 index,
 	entry->se_index = index;
 
 	/*
-	 * Statahead entry reference rules:
-	 *
-	 * 1) When statahead entry is initialized, its reference is set as 2.
-	 *    One reference is used by the directory scanner. When the scanner
-	 *    searches the statahead cache for the given name, it can perform
-	 *    lockless hash lookup (only the scanner can remove entry from hash
-	 *    list), and once found, it needn't to call "atomic_inc()" for the
-	 *    entry reference. So the performance is improved. After using the
-	 *    statahead entry, the scanner will call "atomic_dec()" to drop the
-	 *    reference held when initialization. If it is the last reference,
-	 *    the statahead entry will be freed.
-	 *
-	 * 2) All other threads, including statahead thread and ptlrpcd thread,
-	 *    when they process the statahead entry, the reference for target
-	 *    should be held to guarantee the entry will not be released by the
-	 *    directory scanner. After processing the entry, these threads will
-	 *    drop the entry reference. If it is the last reference, the entry
-	 *    will be freed.
-	 *
-	 *    The second reference when initializes the statahead entry is used
-	 *    by the statahead thread, following the rule 2).
+	 * NB sa_entry is not refcounted, because only directory scanner can
+	 * remove entry, and it only removes stated entries, that means
+	 * after ll_post_statahead() the entries cannot be touched except the
+	 * scanner.
 	 */
-	atomic_set(&entry->se_refcount, 2);
 	entry->se_stat = SA_ENTRY_INIT;
 	entry->se_size = entry_size;
 	dname = (char *)entry + sizeof(struct ll_sa_entry);
@@ -244,7 +224,8 @@ ll_sa_entry_alloc(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);
 	list_add_tail(&entry->se_link, &sai->sai_entries);
 	INIT_LIST_HEAD(&entry->se_list);
@@ -278,30 +259,6 @@ ll_sa_entry_get_byname(struct ll_statahead_info *sai, const struct qstr *qstr)
 	return NULL;
 }
 
-/*
- * Used by the async getattr request callback to find entry with index.
- *
- * Inside lli_sa_lock to prevent others to change the list during the search.
- * It needs to increase entry refcount before returning to guarantee that the
- * entry cannot be freed by others.
- */
-static struct ll_sa_entry *
-ll_sa_entry_get_byindex(struct ll_statahead_info *sai, __u64 index)
-{
-	struct ll_sa_entry *entry;
-
-	list_for_each_entry(entry, &sai->sai_entries, se_link) {
-		if (entry->se_index == index) {
-			LASSERT(atomic_read(&entry->se_refcount) > 0);
-			atomic_inc(&entry->se_refcount);
-			return entry;
-		}
-		if (entry->se_index > index)
-			break;
-	}
-	return NULL;
-}
-
 static void ll_sa_entry_cleanup(struct ll_statahead_info *sai,
 				 struct ll_sa_entry *entry)
 {
@@ -321,31 +278,29 @@ static void ll_sa_entry_cleanup(struct ll_statahead_info *sai,
 	}
 }
 
-static void ll_sa_entry_put(struct ll_statahead_info *sai,
+static void ll_sa_entry_free(struct ll_statahead_info *sai,
 			     struct ll_sa_entry *entry)
 {
-	if (atomic_dec_and_test(&entry->se_refcount)) {
-		CDEBUG(D_READA, "free sa entry %.*s(%p) index "LPU64"\n",
-		       entry->se_qstr.len, entry->se_qstr.name, entry,
-		       entry->se_index);
+	CDEBUG(D_READA, "free sa entry %.*s(%p) index "LPU64"\n",
+	       entry->se_qstr.len, entry->se_qstr.name, entry,
+	       entry->se_index);
 
-		LASSERT(list_empty(&entry->se_link));
-		LASSERT(list_empty(&entry->se_list));
-		LASSERT(ll_sa_entry_unhashed(entry));
+	LASSERT(list_empty(&entry->se_link));
+	LASSERT(list_empty(&entry->se_list));
+	LASSERT(ll_sa_entry_unhashed(entry));
 
-		ll_sa_entry_cleanup(sai, entry);
-		if (entry->se_inode)
-			iput(entry->se_inode);
+	ll_sa_entry_cleanup(sai, entry);
+	if (entry->se_inode)
+		iput(entry->se_inode);
 
-		OBD_FREE(entry, entry->se_size);
-		atomic_dec(&sai->sai_cache_count);
-	}
+	OBD_FREE(entry, entry->se_size);
+	atomic_dec(&sai->sai_cache_count);
 }
 
 static inline void
 do_sa_entry_fini(struct ll_statahead_info *sai, struct ll_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(!ll_sa_entry_unhashed(entry));
 	LASSERT(!list_empty(&entry->se_link));
@@ -359,7 +314,7 @@ do_sa_entry_fini(struct ll_statahead_info *sai, struct ll_sa_entry *entry)
 		list_del_init(&entry->se_list);
 	spin_unlock(&lli->lli_sa_lock);
 
-	ll_sa_entry_put(sai, entry);
+	ll_sa_entry_free(sai, entry);
 }
 
 /*
@@ -377,7 +332,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);
 	}
 }
 
@@ -414,7 +372,7 @@ static int
 ll_sa_entry_to_stated(struct ll_statahead_info *sai,
 		      struct ll_sa_entry *entry, se_stat_t stat)
 {
-	struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
+	struct ll_inode_info *lli = ll_i2info(sai->sai_dentry->d_inode);
 	int		   ret = 1;
 
 	ll_sa_entry_cleanup(sai, entry);
@@ -436,7 +394,7 @@ 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);
@@ -460,23 +418,19 @@ static void ll_agl_add(struct ll_statahead_info *sai,
 		wake_up(&sai->sai_agl_thread.t_ctl_waitq);
 }
 
-static struct ll_statahead_info *ll_sai_alloc(void)
+static struct ll_statahead_info *ll_sai_alloc(struct dentry *dentry)
 {
 	struct ll_statahead_info *sai;
+	struct ll_inode_info     *lli = ll_i2info(dentry->d_inode);
 	int		       i;
 
 	OBD_ALLOC_PTR(sai);
 	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);
@@ -494,44 +448,46 @@ 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 == 0))
+		lli->lli_sa_generation = ++sai_generation;
+	spin_unlock(&sai_generation_lock);
+
+	lli->lli_sai = sai;
+	atomic_inc(&ll_i2sbi(dentry->d_inode)->ll_sa_running);
+
 	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 != NULL)
+		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_dentry->d_inode);
 
 	if (atomic_dec_and_lock(&sai->sai_refcount, &lli->lli_sa_lock)) {
 		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 == NULL);
-		LASSERT(thread_is_stopped(&sai->sai_thread));
-		LASSERT(thread_is_stopped(&sai->sai_agl_thread));
+		struct ll_sb_info *sbi = ll_i2sbi(sai->sai_dentry->d_inode);
 
 		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:"LPU64"] [replied:"LPU64"]\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)
@@ -543,9 +499,11 @@ static void ll_sai_put(struct ll_statahead_info *sai)
 
 		LASSERT(atomic_read(&sai->sai_cache_count) == 0);
 		LASSERT(agl_list_empty(sai));
+		LASSERT(atomic_read(&sai->sai_refcount) == 0);
 
-		iput(inode);
+		dput(sai->sai_dentry);
 		OBD_FREE_PTR(sai);
+		atomic_dec(&sbi->ll_sa_running);
 	}
 }
 
@@ -611,15 +569,16 @@ static void ll_agl_trigger(struct inode *inode, struct ll_statahead_info *sai)
 
 static void ll_post_statahead(struct ll_statahead_info *sai)
 {
-	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;
+	struct inode		*dir = sai->sai_dentry->d_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;
+	__u64			index;
+	int			rc = 0;
 
 	spin_lock(&lli->lli_sa_lock);
 	if (unlikely(sa_received_empty(sai))) {
@@ -627,7 +586,6 @@ static void ll_post_statahead(struct ll_statahead_info *sai)
 		return;
 	}
 	entry = sa_first_received_entry(sai);
-	atomic_inc(&entry->se_refcount);
 	list_del_init(&entry->se_list);
 	spin_unlock(&lli->lli_sa_lock);
 
@@ -686,79 +644,68 @@ out:
 	 * 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. */
+	index = entry->se_index;
 	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)
+	if (rc == 0 && index == sai->sai_index_wait)
 		wake_up(&sai->sai_waitq);
-	ll_sa_entry_put(sai, entry);
 }
 
 static int ll_statahead_interpret(struct ptlrpc_request *req,
 				  struct md_enqueue_info *minfo, int rc)
 {
-	struct lookup_intent     *it  = &minfo->mi_it;
-	struct inode	     *dir = minfo->mi_dir;
-	struct ll_inode_info     *lli = ll_i2info(dir);
-	struct ll_statahead_info *sai = NULL;
-	struct ll_sa_entry       *entry;
-	int		       wakeup;
+	struct lookup_intent	  *it  = &minfo->mi_it;
+	struct inode		  *dir = minfo->mi_dir;
+	struct ll_inode_info	  *lli = ll_i2info(dir);
+	struct ll_statahead_info  *sai = ll_sai_get(dir);
+	struct ll_sa_entry	  *entry =
+					 (struct ll_sa_entry *)minfo->mi_cbdata;
+	bool			   wakeup;
+	__u64			   handle = 0;
+
+	LASSERT(sai != NULL);
+	LASSERT(!thread_is_stopped(&sai->sai_thread));
 
 	if (it_disposition(it, DISP_LOOKUP_NEG))
 		rc = -ENOENT;
 
-	spin_lock(&lli->lli_sa_lock);
-	/* stale entry */
-	if (unlikely(lli->lli_sai == NULL ||
-		     lli->lli_sai->sai_generation != minfo->mi_generation)) {
-		spin_unlock(&lli->lli_sa_lock);
-		GOTO(out, rc = -ESTALE);
+	if (rc == 0) {
+		/*
+		 * 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.
+		 *
+		 * ll_intent_drop_lock() may sleep, call it outside of
+		 * lli_sa_lock.
+		 */
+		handle = it->d.lustre.it_lock_handle;
+		ll_intent_drop_lock(it);
 	} 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);
-			GOTO(out, rc = -EBADFD);
-		}
-
-		entry = ll_sa_entry_get_byindex(sai, minfo->mi_cbdata);
-		if (entry == NULL) {
-			sai->sai_replied++;
-			spin_unlock(&lli->lli_sa_lock);
-			GOTO(out, rc = -EIDRM);
-		}
-
-		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 = it->d.lustre.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);
-
-		ll_sa_entry_put(sai, entry);
-		if (wakeup)
-			wake_up(&sai->sai_thread.t_ctl_waitq);
-	}
-
-out:
-	if (rc != 0) {
 		ll_intent_release(it);
 		iput(dir);
 		OBD_FREE_PTR(minfo);
 	}
-	if (sai != NULL)
-		ll_sai_put(sai);
+
+	spin_lock(&lli->lli_sa_lock);
+	if (rc == 0) {
+		entry->se_handle = handle;
+		entry->se_minfo = minfo;
+		entry->se_req = ptlrpc_request_addref(req);
+		wakeup = sa_received_empty(sai);
+		list_add_tail(&entry->se_list,
+				  &sai->sai_entries_received);
+	} else {
+		wakeup = (entry->se_index == sai->sai_index_wait);
+		do_sa_entry_to_stated(sai, entry, SA_ENTRY_INVA);
+	}
+	sai->sai_replied++;
+	spin_unlock(&lli->lli_sa_lock);
+
+	if (wakeup)
+		wake_up(&sai->sai_thread.t_ctl_waitq);
+	ll_sai_put(sai);
+
 	return rc;
 }
 
@@ -788,7 +735,6 @@ static int sa_args_init(struct inode *dir, struct inode *child,
 			struct obd_capa **pcapa)
 {
 	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;
@@ -814,8 +760,7 @@ 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;
+	minfo->mi_cbdata = entry;
 
 	einfo->ei_type   = LDLM_IBITS;
 	einfo->ei_mode   = it_to_lock_mode(&minfo->mi_it);
@@ -937,29 +882,30 @@ static void ll_statahead_one(struct dentry *parent, const char* entry_name,
 		dput(dentry);
 
 	if (rc) {
+		__u64 index = entry->se_index;
+
 		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)
+		if (rc1 == 0 && index == sai->sai_index_wait)
 			wake_up(&sai->sai_waitq);
 	} else {
 		sai->sai_sent++;
 	}
 
 	sai->sai_index++;
-	/* drop one refcount on entry by ll_sa_entry_alloc */
-	ll_sa_entry_put(sai, entry);
 }
 
 static int ll_agl_thread(void *arg)
 {
-	struct dentry	    *parent = (struct dentry *)arg;
-	struct inode	     *dir    = parent->d_inode;
-	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 dentry		 *parent = (struct dentry *)arg;
+	struct inode		 *dir    = parent->d_inode;
+	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    = plli->lli_sai;
 	struct ptlrpc_thread     *thread = &sai->sai_agl_thread;
-	struct l_wait_info	lwi    = { 0 };
+	struct l_wait_info	  lwi    = { 0 };
+
 
 	CDEBUG(D_READA, "agl thread started: [pid %d] [parent %.*s]\n",
 	       current_pid(), parent->d_name.len, parent->d_name.name);
@@ -1006,9 +952,8 @@ static int ll_agl_thread(void *arg)
 	thread_set_flags(thread, SVC_STOPPED);
 	spin_unlock(&plli->lli_agl_lock);
 	wake_up(&thread->t_ctl_waitq);
-	ll_sai_put(sai);
-	CDEBUG(D_READA, "agl thread stopped: [pid %d] [parent %.*s]\n",
-	       current_pid(), parent->d_name.len, parent->d_name.name);
+	CDEBUG(D_READA, "agl thread for "DFID" stopped.\n",
+	       PFID(&plli->lli_fid));
 	return 0;
 }
 
@@ -1038,28 +983,29 @@ static void ll_start_agl(struct dentry *parent, struct ll_statahead_info *sai)
 
 static int ll_statahead_thread(void *arg)
 {
-	struct dentry	    *parent = (struct dentry *)arg;
-	struct inode	     *dir    = parent->d_inode;
-	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_thread;
-	struct ptlrpc_thread *agl_thread = &sai->sai_agl_thread;
-	struct page	      *page;
-	__u64		     pos    = 0;
-	int		       first  = 0;
-	int		       rc     = 0;
-	struct ll_dir_chain       chain;
-	struct l_wait_info	lwi    = { 0 };
-
-	CDEBUG(D_READA, "statahead thread started: [pid %d] [parent %.*s]\n",
-	       current_pid(), parent->d_name.len, parent->d_name.name);
+	struct dentry		 *parent = (struct dentry *)arg;
+	struct inode		 *dir    = parent->d_inode;
+	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(dir);
+	struct ptlrpc_thread	 *thread = &sai->sai_thread;
+	struct ptlrpc_thread	 *agl_thread = &sai->sai_agl_thread;
+	struct page		 *page;
+	__u64			  pos    = 0;
+	int			  first  = 0;
+	int			  rc     = 0;
+	struct ll_dir_chain	  chain;
+	struct l_wait_info	  lwi    = { 0 };
+
+	CDEBUG(D_READA, "statahead thread started: parent %.*s "DFID".\n",
+		parent->d_name.len, parent->d_name.name, PFID(&plli->lli_fid));
 
 	if (sbi->ll_flags & LL_SBI_AGL_ENABLED)
 		ll_start_agl(parent, sai);
 
 	atomic_inc(&sbi->ll_sa_total);
+
 	spin_lock(&plli->lli_sa_lock);
 	thread_set_flags(thread, SVC_RUNNING);
 	spin_unlock(&plli->lli_sa_lock);
@@ -1075,9 +1021,8 @@ static int ll_statahead_thread(void *arg)
 		if (IS_ERR(page)) {
 			rc = PTR_ERR(page);
 			CDEBUG(D_READA, "error reading dir "DFID" at "LPU64
-			       "/"LPU64": [rc %d] [parent %u]\n",
-			       PFID(ll_inode2fid(dir)), pos, sai->sai_index,
-			       rc, plli->lli_opendir_pid);
+			       "/"LPU64": %d\n", PFID(&plli->lli_fid),
+			       pos, sai->sai_index, rc);
 			GOTO(out, rc);
 		}
 
@@ -1141,6 +1086,21 @@ interpret_it:
 			while (!sa_received_empty(sai))
 				ll_post_statahead(sai);
 
+			if (sa_low_hit(sai)) {
+				spin_lock(&plli->lli_sa_lock);
+				thread_set_flags(thread, SVC_STOPPING);
+				spin_unlock(&plli->lli_sa_lock);
+
+				atomic_inc(&sbi->ll_sa_wrong);
+				CDEBUG(D_READA, "Statahead for dir "DFID" hit "
+					"ratio too low: hit/miss "LPU64"/"LPU64
+					", sent/replied "LPU64"/"LPU64", "
+					"stopping statahead thread.\n",
+					PFID(&plli->lli_fid), sai->sai_hit,
+					sai->sai_miss, sai->sai_sent,
+					sai->sai_replied);
+			}
+
 			if (unlikely(!thread_is_running(thread))) {
 				ll_release_page(page, 0);
 				GOTO(out, rc = 0);
@@ -1186,35 +1146,16 @@ do_it:
 			 * End of directory reached.
 			 */
 			ll_release_page(page, 0);
-			while (1) {
+			/* wait for ll_file_release stop statahead */
+			while (thread_is_running(thread)) {
 				l_wait_event(thread->t_ctl_waitq,
 					     !sa_received_empty(sai) ||
-					     sai->sai_sent == sai->sai_replied||
 					     !thread_is_running(thread),
 					     &lwi);
 
-				while (!sa_received_empty(sai))
+				if (!sa_received_empty(sai))
 					ll_post_statahead(sai);
-
-				if (unlikely(!thread_is_running(thread)))
-					GOTO(out, rc = 0);
-
-				if (sai->sai_sent == sai->sai_replied &&
-				    sa_received_empty(sai))
-					break;
 			}
-
-			spin_lock(&plli->lli_agl_lock);
-			while (!agl_list_empty(sai) &&
-			       thread_is_running(thread)) {
-				clli = agl_first_entry(sai);
-				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);
-
 			GOTO(out, rc = 0);
 		} else if (1) {
 			/*
@@ -1236,14 +1177,15 @@ do_it:
 	}
 
 out:
+	ll_dir_chain_fini(&chain);
 	if (sai->sai_agl_valid) {
 		spin_lock(&plli->lli_agl_lock);
 		thread_set_flags(agl_thread, SVC_STOPPING);
 		spin_unlock(&plli->lli_agl_lock);
 		wake_up(&agl_thread->t_ctl_waitq);
 
-		CDEBUG(D_READA, "stop agl thread: [pid %d]\n",
-		       current_pid());
+		CDEBUG(D_READA, "stop agl thread for "DFID".\n",
+		       PFID(&plli->lli_fid));
 		l_wait_event(agl_thread->t_ctl_waitq,
 			     thread_is_stopped(agl_thread),
 			     &lwi);
@@ -1251,74 +1193,90 @@ out:
 		/* Set agl_thread flags anyway. */
 		thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED);
 	}
-	ll_dir_chain_fini(&chain);
-	spin_lock(&plli->lli_sa_lock);
-	if (!sa_received_empty(sai)) {
-		thread_set_flags(thread, SVC_STOPPING);
-		spin_unlock(&plli->lli_sa_lock);
-
-		/* To release the resources held by received entries. */
-		while (!sa_received_empty(sai))
-			ll_post_statahead(sai);
 
-		spin_lock(&plli->lli_sa_lock);
+	/* wait for inflight statahead RPCs to finish. */
+	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);
 	}
+	/* To release the resources held by received entries. */
+	while (!sa_received_empty(sai))
+		ll_post_statahead(sai);
+	spin_lock(&plli->lli_sa_lock);
 	thread_set_flags(thread, SVC_STOPPED);
 	spin_unlock(&plli->lli_sa_lock);
+
+	CDEBUG(D_READA, "statahead thread for "DFID" stopped.\n",
+	       PFID(&plli->lli_fid));
+
 	wake_up(&sai->sai_waitq);
 	wake_up(&thread->t_ctl_waitq);
 	ll_sai_put(sai);
-	dput(parent);
-	CDEBUG(D_READA, "statahead thread stopped: [pid %d] [parent %.*s]\n",
-	       current_pid(), parent->d_name.len, parent->d_name.name);
+
 	return rc;
 }
 
-/**
- * called in ll_file_release().
+/*
+ * disable statahead for current opened file to avoid retry statahead for
+ * subsequent stat.
  */
-void ll_stop_statahead(struct inode *dir, void *key)
+static inline void ll_disable_statahead(struct ll_inode_info *lli)
 {
-	struct ll_inode_info *lli = ll_i2info(dir);
+	lli->lli_sa_enabled = 0;
+}
 
-	if (unlikely(key == NULL))
-		return;
+static inline void ll_enable_statahead(struct ll_inode_info *lli)
+{
+	lli->lli_sa_enabled = 1;
+}
+
+void ll_authorize_statahead(struct inode *dir, void *key)
+{
+	struct ll_inode_info *lli = ll_i2info(dir);
 
 	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 == NULL && lli->lli_sai == NULL) {
+		/*
+		 * 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 == 0);
+		lli->lli_opendir_key = key;
+		lli->lli_opendir_pid = current_pid();
+
+		ll_enable_statahead(lli);
 	}
+	spin_unlock(&lli->lli_sa_lock);
+}
 
-	lli->lli_opendir_key = NULL;
+void ll_deauthorize_statahead(struct inode *dir, void *key)
+{
+	struct ll_inode_info *lli = ll_i2info(dir);
+	struct ll_statahead_info *sai = ll_sai_get(dir);
 
-	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 != 0);
 
-		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, "deauthorize statahead for "DFID"\n",
+	       PFID(&lli->lli_fid));
 
-			CDEBUG(D_READA, "stop statahead thread: [pid %d]\n",
-			       current_pid());
-			l_wait_event(thread->t_ctl_waitq,
-				     thread_is_stopped(thread),
-				     &lwi);
-		} else {
-			spin_unlock(&lli->lli_sa_lock);
-		}
+	spin_lock(&lli->lli_sa_lock);
+	lli->lli_opendir_key = NULL;
+	lli->lli_opendir_pid = 0;
+	spin_unlock(&lli->lli_sa_lock);
 
+	if (sai) {
 		/*
-		 * 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);
+		if (thread_is_running(&sai->sai_thread)) {
+			thread_set_flags(&sai->sai_thread, SVC_STOPPING);
+			wake_up(&sai->sai_thread.t_ctl_waitq);
+		}
+		ll_sai_put(sai);
 	}
 }
 
@@ -1452,228 +1410,166 @@ 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 != NULL && entry->se_stat == SA_ENTRY_SUCC)
-		hit = 1;
-	else
-		hit = 0;
+	struct ll_sb_info *sbi = ll_i2sbi(sai->sai_dentry->d_inode);
 
-	ll_sa_entry_fini(sai, entry);
-	if (hit) {
+	if (entry != NULL && entry->se_stat == SA_ENTRY_SUCC) {
 		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 "LPU64"/"LPU64
-			       ", sent/replied "LPU64"/"LPU64", stopping "
-			       "statahead thread: pid %d\n",
-			       PFID(&lli->lli_fid), sai->sai_hit,
-			       sai->sai_miss, sai->sai_sent,
-			       sai->sai_replied, current_pid());
-			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 };
-	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 == NULL || only_unplug) {
-			ll_sai_unplug(sai, entry);
-			return entry ? 1 : -EAGAIN;
-		}
+	if (only_unplug)
+		GOTO(out, rc = 1);
 
-		/* 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;
-			}
-		}
+	entry = ll_sa_entry_get_byname(sai, &(*dentryp)->d_name);
+	if (entry == NULL)
+		GOTO(out, rc = -EAGAIN);
 
-		if (entry->se_stat == SA_ENTRY_SUCC &&
-		    entry->se_inode != NULL) {
-			struct inode *inode = entry->se_inode;
-			struct lookup_intent it = { .it_op = IT_GETATTR,
-						    .d.lustre.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 == NULL) {
-					*dentryp = ll_splice_alias(inode,
-								   *dentryp);
-				} else if ((*dentryp)->d_inode != inode) {
-					/* revalidate, but inode is recreated */
-					CDEBUG(D_READA,
-					      "stale dentry %.*s inode %lu/%u, "
-					      "statahead inode %lu/%u\n",
-					      (*dentryp)->d_name.len,
-					      (*dentryp)->d_name.name,
-					      (*dentryp)->d_inode->i_ino,
-					      (*dentryp)->d_inode->i_generation,
-					      inode->i_ino,
-					      inode->i_generation);
-					ll_sai_unplug(sai, entry);
-					return -ESTALE;
-				} else {
-					iput(inode);
-				}
-				entry->se_inode = NULL;
+	/* 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(&sai->sai_thread),
+	                          &lwi);
+	        if (rc < 0)
+	                GOTO(out, entry = NULL);
+	}
 
-				if ((bits & MDS_INODELOCK_LOOKUP) &&
-				    d_lustre_invalid(*dentryp))
-					d_lustre_revalidate(*dentryp);
-				ll_intent_release(&it);
+	if (entry->se_stat == SA_ENTRY_SUCC &&
+	    entry->se_inode != NULL) {
+		struct inode *inode = entry->se_inode;
+		struct lookup_intent it = { .it_op = IT_GETATTR,
+					    .d.lustre.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 == NULL) {
+				struct dentry *alias;
+
+				alias = ll_splice_alias(inode,
+							*dentryp);
+				if (IS_ERR(alias))
+					GOTO(out, rc = PTR_ERR(alias));
+				*dentryp = alias;
+				/* transfer inode refcount to dentry */
+				entry->se_inode = NULL;
+			} else if ((*dentryp)->d_inode != inode) {
+				/* revalidate, but inode is recreated */
+				CDEBUG(D_READA, "stale dentry %.*s inode "DFID
+					"!= statahead inode "DFID"\n",
+					(*dentryp)->d_name.len,
+					(*dentryp)->d_name.name,
+					PFID(ll_inode2fid((*dentryp)->d_inode)),
+					PFID(ll_inode2fid(inode)));
+				GOTO(out, rc = -ESTALE);
 			}
-		}
 
-		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:
+	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;
+	struct ptlrpc_thread *thread;
+	struct l_wait_info lwi = { 0 };
+	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. */
 		GOTO(out, rc = -EAGAIN);
 
-	sai = ll_sai_alloc();
+	sai = ll_sai_alloc(dentry->d_parent);
 	if (sai == NULL)
 		GOTO(out, rc = -ENOMEM);
 
 	sai->sai_ls_all = (rc == LS_FIRST_DOT_DE);
-	sai->sai_inode = igrab(dir);
-	if (unlikely(sai->sai_inode == NULL)) {
-		CWARN("Do not start stat ahead on dying inode "DFID"\n",
-		      PFID(&lli->lli_fid));
-		GOTO(out, rc = -ESTALE);
-	}
-
-	/* get parent reference count here, and put it in ll_statahead_thread */
-	parent = dget((*dentryp)->d_parent);
-	if (unlikely(sai->sai_inode != parent->d_inode)) {
-		struct ll_inode_info *nlli = ll_i2info(parent->d_inode);
-
-		CWARN("Race condition, someone changed %.*s just now: "
-		      "old parent "DFID", new parent "DFID"\n",
-		      (*dentryp)->d_name.len, (*dentryp)->d_name.name,
-		      PFID(&lli->lli_fid), PFID(&nlli->lli_fid));
-		dput(parent);
-		iput(sai->sai_inode);
-		GOTO(out, rc = -EAGAIN);
-	}
 
-	CDEBUG(D_READA, "start statahead thread: [pid %d] [parent %.*s]\n",
-	       current_pid(), parent->d_name.len, parent->d_name.name);
+	CDEBUG(D_READA, "start statahead thread for dir "DFID"\n",
+	       PFID(&lli->lli_fid));
 
-	lli->lli_sai = sai;
-
-	plli = ll_i2info(parent->d_inode);
-	rc = PTR_ERR(kthread_run(ll_statahead_thread, parent,
-				 "ll_sa_%u", plli->lli_opendir_pid));
+	rc = PTR_ERR(kthread_run(ll_statahead_thread, dentry->d_parent,
+				 "ll_sa_%u", lli->lli_opendir_pid));
 	thread = &sai->sai_thread;
 	if (IS_ERR_VALUE(rc)) {
 		CERROR("can't start ll_sa thread, rc: %d\n", rc);
-		dput(parent);
-		lli->lli_opendir_key = NULL;
+		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 == NULL);
-		return -EAGAIN;
+		GOTO(out, rc = -EAGAIN);
 	}
 
 	l_wait_event(thread->t_ctl_waitq,
 		     thread_is_running(thread) || thread_is_stopped(thread),
 		     &lwi);
+	ll_sai_put(sai);
 
 	/*
 	 * We don't stat-ahead for the first dirent since we are already in
@@ -1682,11 +1578,35 @@ int do_statahead_enter(struct inode *dir, struct dentry **dentryp,
 	return -EAGAIN;
 
 out:
-	if (sai != NULL)
-		OBD_FREE_PTR(sai);
-	spin_lock(&lli->lli_sa_lock);
-	lli->lli_opendir_key = NULL;
-	lli->lli_opendir_pid = 0;
-	spin_unlock(&lli->lli_sa_lock);
+	ll_disable_statahead(lli);
 	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 = ll_sai_get(dir);
+
+	if (sai != NULL) {
+		int rc;
+
+		rc = revalidate_statahead_dentry(dir, sai, dentryp,
+						 only_unplug);
+		CDEBUG(D_READA, "revalidate statahead %.*s: %d.\n",
+			(*dentryp)->d_name.len, (*dentryp)->d_name.name, rc);
+		ll_sai_put(sai);
+		return rc;
+	}
+
+	return start_statahead_thread(dir, *dentryp);
+}
+
-- 
1.7.9.5

