staging: lustre: change sai_thread to sai_task.

Rather than allocating a ptlrpc_thread for the
stat-ahead thread, just use the task_struct provided
by kthreads directly.

As nothing ever waits for the sai_task, it must call do_exit()
directly rather than simply return from the function.
Also it cannot use kthread_should_stop() to know when to stop.

There is one caller which can ask it to stop so we need a simple
signaling mechanism.  I've chosen to set ->sai_task to NULL
when the thread should finish up.  The thread notices this and
cleans up and exits.
lli_sa_lock is used to avoid races between waking up the process
and the process exiting.

Signed-off-by: NeilBrown <neilb@suse.com>
Signed-off-by: Greg Kroah-Hartman <gregkh@linuxfoundation.org>

diff --git a/drivers/staging/lustre/lustre/llite/llite_internal.h b/drivers/staging/lustre/lustre/llite/llite_internal.h
index 0c2d717..d46bcf7 100644 (file)
--- a/drivers/staging/lustre/lustre/llite/llite_internal.h
+++ b/drivers/staging/lustre/lustre/llite/llite_internal.h
@@ -1070,7 +1070,7 @@ struct ll_statahead_info {
                                sai_agl_valid:1,/* AGL is valid for the dir */
                                sai_in_readpage:1;/* statahead in readdir() */
        wait_queue_head_t       sai_waitq;      /* stat-ahead wait queue */
-       struct ptlrpc_thread    sai_thread;     /* stat-ahead thread */
+       struct task_struct     *sai_task;       /* stat-ahead thread */
        struct task_struct     *sai_agl_task;   /* AGL thread */
        struct list_head        sai_interim_entries; /* entries which got async
                                                      * stat reply, but not

diff --git a/drivers/staging/lustre/lustre/llite/statahead.c b/drivers/staging/lustre/lustre/llite/statahead.c
index 39241b9..155ce3c 100644 (file)
--- a/drivers/staging/lustre/lustre/llite/statahead.c
+++ b/drivers/staging/lustre/lustre/llite/statahead.c
@@ -267,7 +267,7 @@ sa_kill(struct ll_statahead_info *sai, struct sa_entry *entry)
 
 /* called by scanner after use, sa_entry will be killed */
 static void
-sa_put(struct ll_statahead_info *sai, struct sa_entry *entry)
+sa_put(struct ll_statahead_info *sai, struct sa_entry *entry, struct ll_inode_info *lli)
 {
        struct sa_entry *tmp, *next;
 
@@ -295,7 +295,11 @@ sa_put(struct ll_statahead_info *sai, struct sa_entry *entry)
                sa_kill(sai, tmp);
        }
 
-       wake_up(&sai->sai_thread.t_ctl_waitq);
+       spin_lock(&lli->lli_sa_lock);
+       if (sai->sai_task)
+               wake_up_process(sai->sai_task);
+       spin_unlock(&lli->lli_sa_lock);
+
 }
 
 /*
@@ -403,7 +407,6 @@ static struct ll_statahead_info *ll_sai_alloc(struct dentry *dentry)
        sai->sai_max = LL_SA_RPC_MIN;
        sai->sai_index = 1;
        init_waitqueue_head(&sai->sai_waitq);
-       init_waitqueue_head(&sai->sai_thread.t_ctl_waitq);
 
        INIT_LIST_HEAD(&sai->sai_interim_entries);
        INIT_LIST_HEAD(&sai->sai_entries);
@@ -465,7 +468,7 @@ static void ll_sai_put(struct ll_statahead_info *sai)
                lli->lli_sai = NULL;
                spin_unlock(&lli->lli_sa_lock);
 
-               LASSERT(thread_is_stopped(&sai->sai_thread));
+               LASSERT(sai->sai_task == NULL);
                LASSERT(sai->sai_agl_task == NULL);
                LASSERT(sai->sai_sent == sai->sai_replied);
                LASSERT(!sa_has_callback(sai));
@@ -646,7 +649,6 @@ static int ll_statahead_interpret(struct ptlrpc_request *req,
        struct ll_inode_info     *lli = ll_i2info(dir);
        struct ll_statahead_info *sai = lli->lli_sai;
        struct sa_entry *entry = (struct sa_entry *)minfo->mi_cbdata;
-       wait_queue_head_t *waitq = NULL;
        __u64 handle = 0;
 
        if (it_disposition(it, DISP_LOOKUP_NEG))
@@ -657,7 +659,6 @@ static int ll_statahead_interpret(struct ptlrpc_request *req,
         * sai should be always valid, no need to refcount
         */
        LASSERT(sai);
-       LASSERT(!thread_is_stopped(&sai->sai_thread));
        LASSERT(entry);
 
        CDEBUG(D_READA, "sa_entry %.*s rc %d\n",
@@ -681,8 +682,9 @@ static int ll_statahead_interpret(struct ptlrpc_request *req,
        spin_lock(&lli->lli_sa_lock);
        if (rc) {
                if (__sa_make_ready(sai, entry, rc))
-                       waitq = &sai->sai_waitq;
+                       wake_up(&sai->sai_waitq);
        } else {
+               int first = 0;
                entry->se_minfo = minfo;
                entry->se_req = ptlrpc_request_addref(req);
                /*
@@ -693,14 +695,15 @@ static int ll_statahead_interpret(struct ptlrpc_request *req,
                 */
                entry->se_handle = handle;
                if (!sa_has_callback(sai))
-                       waitq = &sai->sai_thread.t_ctl_waitq;
+                       first = 1;
 
                list_add_tail(&entry->se_list, &sai->sai_interim_entries);
+
+               if (first && sai->sai_task)
+                       wake_up_process(sai->sai_task);
        }
        sai->sai_replied++;
 
-       if (waitq)
-               wake_up(waitq);
        spin_unlock(&lli->lli_sa_lock);
 
        return rc;
@@ -942,17 +945,13 @@ static int ll_statahead_thread(void *arg)
        struct inode         *dir    = d_inode(parent);
        struct ll_inode_info     *lli   = ll_i2info(dir);
        struct ll_sb_info       *sbi    = ll_i2sbi(dir);
-       struct ll_statahead_info *sai;
-       struct ptlrpc_thread *sa_thread;
+       struct ll_statahead_info *sai = lli->lli_sai;
        struct page           *page = NULL;
        __u64                pos    = 0;
        int                    first  = 0;
        int                    rc     = 0;
        struct md_op_data *op_data;
 
-       sai = ll_sai_get(dir);
-       sa_thread = &sai->sai_thread;
-       sa_thread->t_pid = current_pid();
        CDEBUG(D_READA, "statahead thread starting: sai %p, parent %pd\n",
               sai, parent);
 
@@ -965,21 +964,7 @@ static int ll_statahead_thread(void *arg)
 
        op_data->op_max_pages = ll_i2sbi(dir)->ll_md_brw_pages;
 
-       if (sbi->ll_flags & LL_SBI_AGL_ENABLED)
-               ll_start_agl(parent, sai);
-
-       atomic_inc(&sbi->ll_sa_total);
-       spin_lock(&lli->lli_sa_lock);
-       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(sa_thread, SVC_RUNNING);
-       spin_unlock(&lli->lli_sa_lock);
-       wake_up(&sa_thread->t_ctl_waitq);
-
-       while (pos != MDS_DIR_END_OFF && thread_is_running(sa_thread)) {
+       while (pos != MDS_DIR_END_OFF && sai->sai_task) {
                struct lu_dirpage *dp;
                struct lu_dirent  *ent;
 
@@ -996,7 +981,7 @@ static int ll_statahead_thread(void *arg)
 
                dp = page_address(page);
                for (ent = lu_dirent_start(dp);
-                    ent && thread_is_running(sa_thread) && !sa_low_hit(sai);
+                    ent && sai->sai_task && !sa_low_hit(sai);
                     ent = lu_dirent_next(ent)) {
                        struct lu_fid fid;
                        __u64 hash;
@@ -1046,13 +1031,7 @@ static int ll_statahead_thread(void *arg)
 
                        fid_le_to_cpu(&fid, &ent->lde_fid);
 
-                       /* wait for spare statahead window */
                        do {
-                               wait_event_idle(sa_thread->t_ctl_waitq,
-                                               !sa_sent_full(sai) ||
-                                               sa_has_callback(sai) ||
-                                               !list_empty(&sai->sai_agls) ||
-                                               !thread_is_running(sa_thread));
                                sa_handle_callback(sai);
 
                                spin_lock(&lli->lli_agl_lock);
@@ -1072,8 +1051,16 @@ static int ll_statahead_thread(void *arg)
                                        spin_lock(&lli->lli_agl_lock);
                                }
                                spin_unlock(&lli->lli_agl_lock);
-                       } while (sa_sent_full(sai) &&
-                                thread_is_running(sa_thread));
+
+                               set_current_state(TASK_IDLE);
+                               if (sa_sent_full(sai) &&
+                                   !sa_has_callback(sai) &&
+                                   agl_list_empty(sai) &&
+                                   sai->sai_task)
+                                       /* wait for spare statahead window */
+                                       schedule();
+                               __set_current_state(TASK_RUNNING);
+                       } while (sa_sent_full(sai) && sai->sai_task);
 
                        sa_statahead(parent, name, namelen, &fid);
                }
@@ -1096,7 +1083,7 @@ static int ll_statahead_thread(void *arg)
 
        if (rc < 0) {
                spin_lock(&lli->lli_sa_lock);
-               thread_set_flags(sa_thread, SVC_STOPPING);
+               sai->sai_task = NULL;
                lli->lli_sa_enabled = 0;
                spin_unlock(&lli->lli_sa_lock);
        }
@@ -1105,12 +1092,14 @@ 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(sa_thread)) {
-               wait_event_idle(sa_thread->t_ctl_waitq,
-                               sa_has_callback(sai) ||
-                               !thread_is_running(sa_thread));
-
+       while (sai->sai_task) {
                sa_handle_callback(sai);
+
+               set_current_state(TASK_IDLE);
+               if (!sa_has_callback(sai) &&
+                   sai->sai_task)
+                       schedule();
+               __set_current_state(TASK_RUNNING);
        }
 out:
        if (sai->sai_agl_task) {
@@ -1126,26 +1115,23 @@ out:
         */
        while (sai->sai_sent != sai->sai_replied) {
                /* in case we're not woken up, timeout wait */
-               wait_event_idle_timeout(sa_thread->t_ctl_waitq,
-                                       sai->sai_sent == sai->sai_replied,
-                                       HZ>>3);
+               schedule_timeout_idle(HZ>>3);
        }
 
        /* release resources held by statahead RPCs */
        sa_handle_callback(sai);
 
-       spin_lock(&lli->lli_sa_lock);
-       thread_set_flags(sa_thread, SVC_STOPPED);
-       spin_unlock(&lli->lli_sa_lock);
-
        CDEBUG(D_READA, "statahead thread stopped: sai %p, parent %pd\n",
               sai, parent);
 
+       spin_lock(&lli->lli_sa_lock);
+       sai->sai_task = NULL;
+       spin_unlock(&lli->lli_sa_lock);
+
        wake_up(&sai->sai_waitq);
-       wake_up(&sa_thread->t_ctl_waitq);
        ll_sai_put(sai);
 
-       return rc;
+       do_exit(rc);
 }
 
 /* authorize opened dir handle @key to statahead */
@@ -1187,13 +1173,13 @@ void ll_deauthorize_statahead(struct inode *dir, void *key)
        lli->lli_opendir_pid = 0;
        lli->lli_sa_enabled = 0;
        sai = lli->lli_sai;
-       if (sai && thread_is_running(&sai->sai_thread)) {
+       if (sai && sai->sai_task) {
                /*
                 * statahead thread may not quit yet because it needs to cache
                 * entries, now it's time to tell it to quit.
                 */
-               thread_set_flags(&sai->sai_thread, SVC_STOPPING);
-               wake_up(&sai->sai_thread.t_ctl_waitq);
+               wake_up_process(sai->sai_task);
+               sai->sai_task = NULL;
        }
        spin_unlock(&lli->lli_sa_lock);
 }
@@ -1463,7 +1449,7 @@ out_unplug:
         */
        ldd = ll_d2d(*dentryp);
        ldd->lld_sa_generation = lli->lli_sa_generation;
-       sa_put(sai, entry);
+       sa_put(sai, entry, lli);
        return rc;
 }
 
@@ -1483,7 +1469,6 @@ static int start_statahead_thread(struct inode *dir, struct dentry *dentry)
 {
        struct ll_inode_info *lli = ll_i2info(dir);
        struct ll_statahead_info *sai = NULL;
-       struct ptlrpc_thread *thread;
        struct task_struct *task;
        struct dentry *parent = dentry->d_parent;
        int rc;
@@ -1523,18 +1508,21 @@ static int start_statahead_thread(struct inode *dir, struct dentry *dentry)
        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;
+       task = kthread_create(ll_statahead_thread, parent, "ll_sa_%u",
+                             lli->lli_opendir_pid);
        if (IS_ERR(task)) {
                rc = PTR_ERR(task);
                CERROR("can't start ll_sa thread, rc : %d\n", rc);
                goto out;
        }
 
-       wait_event_idle(thread->t_ctl_waitq,
-                       thread_is_running(thread) || thread_is_stopped(thread));
-       ll_sai_put(sai);
+       if (ll_i2sbi(parent->d_inode)->ll_flags & LL_SBI_AGL_ENABLED)
+               ll_start_agl(parent, sai);
+
+       atomic_inc(&ll_i2sbi(parent->d_inode)->ll_sa_total);
+       sai->sai_task = task;
+
+       wake_up_process(task);
 
        /*
         * We don't stat-ahead for the first dirent since we are already in