LU-14139 llite: simplify callback handling for async getattr

In this patch, it prepares the inode and set lock data directly in
the callback interpret of the intent async getattr RPC request (in
ptlrpcd context), simplifies the old impementation that defer this
work in the statahead thread.

According to the benchmark result, the workload "ls -l" to a large
directory on a client without any caching (server and client),
containing 1M files (47001 bytes) shows the results with measured
elapsed time:
- w/o patch:	180 seconds;
- w patch:	181 seconds;

There is no any obvious performance regession.

Change-Id: Ifcfad3eb26d831bec3beea0c3d7045f31d35fa6a
Signed-off-by: Qian Yingjin <qian@ddn.com>
Reviewed-on: https://review.whamcloud.com/40712
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: James Simmons <jsimmons@infradead.org>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index e4daa4c..10f728c 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -949,20 +949,27 @@ struct md_callback {
                               void *data, int flag);
 };
 
                               void *data, int flag);
 };
 

-struct md_enqueue_info;
-/* metadata stat-ahead */
-typedef int (* md_enqueue_cb_t)(struct ptlrpc_request *req,
-                                struct md_enqueue_info *minfo,
-                                int rc);
-
-struct md_enqueue_info {
-       struct md_op_data               mi_data;
-       struct lookup_intent            mi_it;
-       struct lustre_handle            mi_lockh;
-       struct inode                   *mi_dir;
-       struct ldlm_enqueue_info        mi_einfo;
-       md_enqueue_cb_t                 mi_cb;
-       void                           *mi_cbdata;
+struct md_op_item;
+typedef int (*md_op_item_cb_t)(struct req_capsule *pill,
+                              struct md_op_item *item,
+                              int rc);
+
+enum md_opcode {
+       MD_OP_NONE      = 0,
+       MD_OP_GETATTR   = 1,
+       MD_OP_MAX,
+};
+
+struct md_op_item {
+       enum md_opcode                   mop_opc;
+       struct md_op_data                mop_data;
+       struct lookup_intent             mop_it;
+       struct lustre_handle             mop_lockh;
+       struct ldlm_enqueue_info         mop_einfo;
+       md_op_item_cb_t                  mop_cb;
+       void                            *mop_cbdata;
+       struct inode                    *mop_dir;
+       __u64                            mop_lock_flags;

 };
 
 struct obd_ops {
 };
 
 struct obd_ops {


@@ -1160,7 +1167,7 @@ struct md_ops {
                          u64, const char *, size_t, struct ptlrpc_request **);
 
        int (*m_intent_getattr_async)(struct obd_export *,
                          u64, const char *, size_t, struct ptlrpc_request **);
 
        int (*m_intent_getattr_async)(struct obd_export *,

-                                     struct md_enqueue_info *);
+                                     struct md_op_item *);

 
         int (*m_revalidate_lock)(struct obd_export *, struct lookup_intent *,
                                  struct lu_fid *, __u64 *bits);
 
         int (*m_revalidate_lock)(struct obd_export *, struct lookup_intent *,
                                  struct lu_fid *, __u64 *bits);

diff --git a/lustre/include/obd_class.h b/lustre/include/obd_class.h
index 093682f..66932d5 100644 (file)
--- a/lustre/include/obd_class.h
+++ b/lustre/include/obd_class.h
@@ -1731,7 +1731,7 @@ static inline int md_init_ea_size(struct obd_export *exp, __u32 ea_size,
 }
 
 static inline int md_intent_getattr_async(struct obd_export *exp,
 }
 
 static inline int md_intent_getattr_async(struct obd_export *exp,

-                                         struct md_enqueue_info *minfo)
+                                         struct md_op_item *item)

 {
        int rc;
 
 {
        int rc;
 


@@ -1742,7 +1742,7 @@ static inline int md_intent_getattr_async(struct obd_export *exp,
        lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
                             LPROC_MD_INTENT_GETATTR_ASYNC);
 
        lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
                             LPROC_MD_INTENT_GETATTR_ASYNC);
 

-       return MDP(exp->exp_obd, intent_getattr_async)(exp, minfo);
+       return MDP(exp->exp_obd, intent_getattr_async)(exp, item);

 }
 
 static inline int md_revalidate_lock(struct obd_export *exp,
 }
 
 static inline int md_revalidate_lock(struct obd_export *exp,

diff --git a/lustre/llite/llite_internal.h b/lustre/llite/llite_internal.h
index 6a86b6d..9c37101 100644 (file)
--- a/lustre/llite/llite_internal.h
+++ b/lustre/llite/llite_internal.h
@@ -1473,15 +1473,11 @@ struct ll_statahead_info {
                                                 * is not a hidden one */
        unsigned int            sai_skip_hidden;/* skipped hidden dentry count
                                                 */
                                                 * 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
+       unsigned int            sai_ls_all:1;   /* "ls -al", do stat-ahead for

                                                 * hidden entries */
                                                 * hidden entries */

-                               sai_in_readpage:1;/* statahead is in readdir()*/

        wait_queue_head_t       sai_waitq;      /* stat-ahead wait queue */
        struct task_struct      *sai_task;      /* stat-ahead thread */
        struct task_struct      *sai_agl_task;  /* AGL thread */
        wait_queue_head_t       sai_waitq;      /* stat-ahead wait queue */
        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
-                                                     * instantiated */

        struct list_head        sai_entries;    /* completed entries */
        struct list_head        sai_agls;       /* AGLs to be sent */
        struct list_head        sai_cache[LL_SA_CACHE_SIZE];
        struct list_head        sai_entries;    /* completed entries */
        struct list_head        sai_agls;       /* AGLs to be sent */
        struct list_head        sai_cache[LL_SA_CACHE_SIZE];

diff --git a/lustre/llite/statahead.c b/lustre/llite/statahead.c
index 775c488..1db80ae 100644 (file)
--- a/lustre/llite/statahead.c
+++ b/lustre/llite/statahead.c
@@ -54,13 +54,12 @@ typedef enum {
 
 /*
  * sa_entry is not refcounted: statahead thread allocates it and do async 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_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.
+ * and in async stat callback ll_statahead_interpret() will prepare the inode
+ * and set lock data in the ptlrpcd context. Then the scanner process will be
+ * woken up if this entry is the waiting one, can access and free it.

  */
 struct sa_entry {
  */
 struct sa_entry {

-       /* link into sai_interim_entries or sai_entries */
+       /* link into sai_entries */

        struct list_head        se_list;
        /* link into sai hash table locally */
        struct list_head        se_hash;
        struct list_head        se_list;
        /* link into sai hash table locally */
        struct list_head        se_hash;


@@ -72,10 +71,6 @@ struct sa_entry {
        se_state_t              se_state;
        /* entry size, contains name */
        int                     se_size;
        se_state_t              se_state;
        /* entry size, contains name */
        int                     se_size;

-       /* pointer to async getattr enqueue info */
-       struct md_enqueue_info *se_minfo;
-       /* pointer to the async getattr request */
-       struct ptlrpc_request  *se_req;

        /* pointer to the target inode */
        struct inode           *se_inode;
        /* entry name */
        /* pointer to the target inode */
        struct inode           *se_inode;
        /* entry name */


@@ -147,12 +142,6 @@ static inline int sa_sent_full(struct ll_statahead_info *sai)
        return atomic_read(&sai->sai_cache_count) >= sai->sai_max;
 }
 
        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);
-}
-

 static inline int agl_list_empty(struct ll_statahead_info *sai)
 {
        return list_empty(&sai->sai_agls);
 static inline int agl_list_empty(struct ll_statahead_info *sai)
 {
        return list_empty(&sai->sai_agls);


@@ -341,55 +330,55 @@ __sa_make_ready(struct ll_statahead_info *sai, struct sa_entry *entry, int ret)
 }
 
 /* finish async stat RPC arguments */
 }
 
 /* finish async stat RPC arguments */

-static void sa_fini_data(struct md_enqueue_info *minfo)
+static void sa_fini_data(struct md_op_item *item)

 {
 {

-       ll_unlock_md_op_lsm(&minfo->mi_data);
-       iput(minfo->mi_dir);
-       OBD_FREE_PTR(minfo);
+       ll_unlock_md_op_lsm(&item->mop_data);
+       iput(item->mop_dir);
+       OBD_FREE_PTR(item);

 }
 
 }
 

-static int ll_statahead_interpret(struct ptlrpc_request *req,
-                                 struct md_enqueue_info *minfo, int rc);
+static int ll_statahead_interpret(struct req_capsule *pill,
+                                 struct md_op_item *item, int rc);

 
 /*
  * prepare arguments for async stat RPC.
  */
 
 /*
  * prepare arguments for async stat RPC.
  */

-static struct md_enqueue_info *
+static struct md_op_item *

 sa_prep_data(struct inode *dir, struct inode *child, struct sa_entry *entry)
 {
 sa_prep_data(struct inode *dir, struct inode *child, struct sa_entry *entry)
 {

-       struct md_enqueue_info   *minfo;
+       struct md_op_item *item;

        struct ldlm_enqueue_info *einfo;
        struct ldlm_enqueue_info *einfo;

-       struct md_op_data        *op_data;
+       struct md_op_data *op_data;

 
 

-       OBD_ALLOC_PTR(minfo);
-       if (!minfo)
+       OBD_ALLOC_PTR(item);
+       if (!item)

                return ERR_PTR(-ENOMEM);
 
                return ERR_PTR(-ENOMEM);
 

-       op_data = ll_prep_md_op_data(&minfo->mi_data, dir, child,
+       op_data = ll_prep_md_op_data(&item->mop_data, dir, child,

                                     entry->se_qstr.name, entry->se_qstr.len, 0,
                                     LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data)) {
                                     entry->se_qstr.name, entry->se_qstr.len, 0,
                                     LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data)) {

-               OBD_FREE_PTR(minfo);
-               return (struct md_enqueue_info *)op_data;
+               OBD_FREE_PTR(item);
+               return (struct md_op_item *)op_data;

        }
 
        if (!child)
                op_data->op_fid2 = entry->se_fid;
 
        }
 
        if (!child)
                op_data->op_fid2 = entry->se_fid;
 

-       minfo->mi_it.it_op = IT_GETATTR;
-       minfo->mi_dir = igrab(dir);
-       minfo->mi_cb = ll_statahead_interpret;
-       minfo->mi_cbdata = entry;
-
-       einfo = &minfo->mi_einfo;
-       einfo->ei_type   = LDLM_IBITS;
-       einfo->ei_mode   = it_to_lock_mode(&minfo->mi_it);
-       einfo->ei_cb_bl  = ll_md_blocking_ast;
-       einfo->ei_cb_cp  = ldlm_completion_ast;
-       einfo->ei_cb_gl  = NULL;
+       item->mop_it.it_op = IT_GETATTR;
+       item->mop_dir = igrab(dir);
+       item->mop_cb = ll_statahead_interpret;
+       item->mop_cbdata = entry;
+
+       einfo = &item->mop_einfo;
+       einfo->ei_type = LDLM_IBITS;
+       einfo->ei_mode = it_to_lock_mode(&item->mop_it);
+       einfo->ei_cb_bl = ll_md_blocking_ast;
+       einfo->ei_cb_cp = ldlm_completion_ast;
+       einfo->ei_cb_gl = NULL;

        einfo->ei_cbdata = NULL;
 
        einfo->ei_cbdata = NULL;
 

-       return minfo;
+       return item;

 }
 
 /*
 }
 
 /*


@@ -400,22 +389,8 @@ 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_dentry->d_inode);
 sa_make_ready(struct ll_statahead_info *sai, struct sa_entry *entry, int ret)
 {
        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;
 
        bool wakeup;
 

-       /* release resources used in RPC */
-       if (minfo) {
-               entry->se_minfo = NULL;
-               ll_intent_release(&minfo->mi_it);
-               sa_fini_data(minfo);
-       }
-
-       if (req) {
-               entry->se_req = NULL;
-               ptlrpc_req_finished(req);
-       }
-

        spin_lock(&lli->lli_sa_lock);
        wakeup = __sa_make_ready(sai, entry, ret);
        spin_unlock(&lli->lli_sa_lock);
        spin_lock(&lli->lli_sa_lock);
        wakeup = __sa_make_ready(sai, entry, ret);
        spin_unlock(&lli->lli_sa_lock);


@@ -472,7 +447,6 @@ static struct ll_statahead_info *ll_sai_alloc(struct dentry *dentry)
        sai->sai_index = 1;
        init_waitqueue_head(&sai->sai_waitq);
 
        sai->sai_index = 1;
        init_waitqueue_head(&sai->sai_waitq);
 

-       INIT_LIST_HEAD(&sai->sai_interim_entries);

        INIT_LIST_HEAD(&sai->sai_entries);
        INIT_LIST_HEAD(&sai->sai_agls);
 
        INIT_LIST_HEAD(&sai->sai_entries);
        INIT_LIST_HEAD(&sai->sai_agls);
 


@@ -535,7 +509,6 @@ static void ll_sai_put(struct ll_statahead_info *sai)
                LASSERT(!sai->sai_task);
                LASSERT(!sai->sai_agl_task);
                LASSERT(sai->sai_sent == sai->sai_replied);
                LASSERT(!sai->sai_task);
                LASSERT(!sai->sai_agl_task);
                LASSERT(sai->sai_sent == sai->sai_replied);

-               LASSERT(!sa_has_callback(sai));

 
                list_for_each_entry_safe(entry, next, &sai->sai_entries,
                                         se_list)
 
                list_for_each_entry_safe(entry, next, &sai->sai_entries,
                                         se_list)


@@ -631,34 +604,71 @@ static void ll_agl_trigger(struct inode *inode, struct ll_statahead_info *sai)
 }
 
 /*
 }
 
 /*

- * prepare inode for sa entry, add it into agl list, now sa_entry is ready
- * to be used by scanner process.
+ * Callback for async stat RPC, this is called in ptlrpcd context. It prepares
+ * the inode and set lock data directly in the ptlrpcd context. It will wake up
+ * the directory listing process if the dentry is the waiting one.

  */
  */

-static void sa_instantiate(struct ll_statahead_info *sai,
-                          struct sa_entry *entry)
+static int ll_statahead_interpret(struct req_capsule *pill,
+                                 struct md_op_item *item, int rc)

 {
 {

-       struct inode *dir = sai->sai_dentry->d_inode;
-       struct inode *child;
-       struct md_enqueue_info *minfo;
-       struct lookup_intent *it;
-       struct ptlrpc_request *req;
+       struct lookup_intent *it = &item->mop_it;
+       struct inode *dir = item->mop_dir;
+       struct ll_inode_info *lli = ll_i2info(dir);
+       struct ll_statahead_info *sai = lli->lli_sai;
+       struct sa_entry *entry = (struct sa_entry *)item->mop_cbdata;

        struct mdt_body *body;
        struct mdt_body *body;

-       int rc = 0;
+       struct inode *child;
+       __u64 handle = 0;

 
        ENTRY;
 
 
        ENTRY;
 

-       LASSERT(entry->se_handle != 0);
+       if (it_disposition(it, DISP_LOOKUP_NEG))
+               rc = -ENOENT;
+
+       /*
+        * because statahead thread will wait for all inflight RPC to finish,
+        * sai should be always valid, no need to refcount
+        */
+       LASSERT(sai != NULL);
+       LASSERT(entry != NULL);
+
+       CDEBUG(D_READA, "sa_entry %.*s rc %d\n",
+              entry->se_qstr.len, entry->se_qstr.name, rc);
+
+       if (rc != 0) {
+               ll_intent_release(it);
+               sa_fini_data(item);
+       } else {
+               /*
+                * release ibits lock ASAP to avoid deadlock when statahead
+                * thread enqueues lock on parent in readdir and another
+                * process enqueues lock on child with parent lock held, eg.
+                * unlink.
+                */
+               handle = it->it_lock_handle;
+               ll_intent_drop_lock(it);
+               ll_unlock_md_op_lsm(&item->mop_data);
+       }
+
+       if (rc != 0) {
+               spin_lock(&lli->lli_sa_lock);
+               if (__sa_make_ready(sai, entry, rc))
+                       wake_up(&sai->sai_waitq);
+
+               sai->sai_replied++;
+               spin_unlock(&lli->lli_sa_lock);
+
+               RETURN(rc);
+       }

 
 

-       minfo = entry->se_minfo;
-       it = &minfo->mi_it;
-       req = entry->se_req;
-       body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
+       entry->se_handle = handle;
+       body = req_capsule_server_get(pill, &RMF_MDT_BODY);

        if (!body)
                GOTO(out, rc = -EFAULT);
 
        child = entry->se_inode;
        /* revalidate; unlinked and re-created with the same name */
        if (!body)
                GOTO(out, rc = -EFAULT);
 
        child = entry->se_inode;
        /* revalidate; unlinked and re-created with the same name */

-       if (unlikely(!lu_fid_eq(&minfo->mi_data.op_fid2, &body->mbo_fid1))) {
+       if (unlikely(!lu_fid_eq(&item->mop_data.op_fid2, &body->mbo_fid1))) {

                if (child) {
                        entry->se_inode = NULL;
                        iput(child);
                if (child) {
                        entry->se_inode = NULL;
                        iput(child);


@@ -672,7 +682,7 @@ static void sa_instantiate(struct ll_statahead_info *sai,
        if (rc != 1)
                GOTO(out, rc = -EAGAIN);
 
        if (rc != 1)
                GOTO(out, rc = -EAGAIN);
 

-       rc = ll_prep_inode(&child, &req->rq_pill, dir->i_sb, it);
+       rc = ll_prep_inode(&child, pill, dir->i_sb, it);

        if (rc)
                GOTO(out, rc);
 
        if (rc)
                GOTO(out, rc);
 


@@ -686,109 +696,18 @@ static void sa_instantiate(struct ll_statahead_info *sai,
        if (agl_should_run(sai, child))
                ll_agl_add(sai, child, entry->se_index);
 
        if (agl_should_run(sai, child))
                ll_agl_add(sai, child, entry->se_index);
 

-       EXIT;
-

 out:
        /*
 out:
        /*

-        * sa_make_ready() will drop ldlm ibits lock refcount by calling
+        * First it will drop ldlm ibits lock refcount by calling

         * ll_intent_drop_lock() in spite of failures. Do not worry about
         * calling ll_intent_drop_lock() more than once.
         */
         * ll_intent_drop_lock() in spite of failures. Do not worry about
         * calling ll_intent_drop_lock() more than once.
         */

+       ll_intent_release(&item->mop_it);
+       sa_fini_data(item);

        sa_make_ready(sai, entry, rc);
        sa_make_ready(sai, entry, rc);

-}
-
-/* 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_dentry->d_inode);
-
-       spin_lock(&lli->lli_sa_lock);
-       while (sa_has_callback(sai)) {
-               struct sa_entry *entry;
-
-               entry = list_entry(sai->sai_interim_entries.next,
-                                  struct sa_entry, se_list);
-               list_del_init(&entry->se_list);
-               spin_unlock(&lli->lli_sa_lock);
-
-               sa_instantiate(sai, entry);
-               spin_lock(&lli->lli_sa_lock);
-       }
-       spin_unlock(&lli->lli_sa_lock);
-}
-
-/*
- * callback for async stat RPC, because this is called in ptlrpcd context, we
- * only put sa_entry in sai_interim_entries, and wake up statahead thread to
- * really prepare inode and instantiate sa_entry later.
- */
-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 = lli->lli_sai;
-       struct sa_entry *entry = (struct sa_entry *)minfo->mi_cbdata;
-       __u64 handle = 0;
-
-       ENTRY;
-
-       if (it_disposition(it, DISP_LOOKUP_NEG))
-               rc = -ENOENT;
-
-       /*
-        * because statahead thread will wait for all inflight RPC to finish,
-        * sai should be always valid, no need to refcount
-        */
-       LASSERT(sai != NULL);
-       LASSERT(entry != NULL);
-
-       CDEBUG(D_READA, "sa_entry %.*s rc %d\n",
-              entry->se_qstr.len, entry->se_qstr.name, rc);
-
-       if (rc != 0) {
-               ll_intent_release(it);
-               sa_fini_data(minfo);
-       } else {
-               /*
-                * release ibits lock ASAP to avoid deadlock when statahead
-                * thread enqueues lock on parent in readdir and another
-                * process enqueues lock on child with parent lock held, eg.
-                * unlink.
-                */
-               handle = it->it_lock_handle;
-               ll_intent_drop_lock(it);
-               ll_unlock_md_op_lsm(&minfo->mi_data);
-       }

 
        spin_lock(&lli->lli_sa_lock);
 
        spin_lock(&lli->lli_sa_lock);

-       if (rc != 0) {
-               if (__sa_make_ready(sai, entry, rc))
-                       wake_up(&sai->sai_waitq);
-       } else {
-               int first = 0;
-
-               entry->se_minfo = minfo;
-               entry->se_req = ptlrpc_request_addref(req);
-               /*
-                * Release the async ibits lock ASAP to avoid deadlock
-                * when statahead thread tries to enqueue lock on parent
-                * for readpage and other tries to enqueue lock on child
-                * with parent's lock held, for example: unlink.
-                */
-               entry->se_handle = handle;
-               if (!sa_has_callback(sai))
-                       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++;
        sai->sai_replied++;

-

        spin_unlock(&lli->lli_sa_lock);
 
        RETURN(rc);
        spin_unlock(&lli->lli_sa_lock);
 
        RETURN(rc);


@@ -797,18 +716,18 @@ static int ll_statahead_interpret(struct ptlrpc_request *req,
 /* async stat for file not found in dcache */
 static int sa_lookup(struct inode *dir, struct sa_entry *entry)
 {
 /* async stat for file not found in dcache */
 static int sa_lookup(struct inode *dir, struct sa_entry *entry)
 {

-       struct md_enqueue_info   *minfo;
-       int                       rc;
+       struct md_op_item *item;
+       int rc;

 
        ENTRY;
 
 
        ENTRY;
 

-       minfo = sa_prep_data(dir, NULL, entry);
-       if (IS_ERR(minfo))
-               RETURN(PTR_ERR(minfo));
+       item = sa_prep_data(dir, NULL, entry);
+       if (IS_ERR(item))
+               RETURN(PTR_ERR(item));

 
 

-       rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo);
+       rc = md_intent_getattr_async(ll_i2mdexp(dir), item);

        if (rc < 0)
        if (rc < 0)

-               sa_fini_data(minfo);
+               sa_fini_data(item);

 
        RETURN(rc);
 }
 
        RETURN(rc);
 }


@@ -826,7 +745,7 @@ static int sa_revalidate(struct inode *dir, struct sa_entry *entry,
        struct inode *inode = dentry->d_inode;
        struct lookup_intent it = { .it_op = IT_GETATTR,
                                    .it_lock_handle = 0 };
        struct inode *inode = dentry->d_inode;
        struct lookup_intent it = { .it_op = IT_GETATTR,
                                    .it_lock_handle = 0 };

-       struct md_enqueue_info *minfo;
+       struct md_op_item *item;

        int rc;
 
        ENTRY;
        int rc;
 
        ENTRY;


@@ -837,9 +756,9 @@ static int sa_revalidate(struct inode *dir, struct sa_entry *entry,
        if (d_mountpoint(dentry))
                RETURN(1);
 
        if (d_mountpoint(dentry))
                RETURN(1);
 

-       minfo = sa_prep_data(dir, inode, entry);
-       if (IS_ERR(minfo))
-               RETURN(PTR_ERR(minfo));
+       item = sa_prep_data(dir, inode, entry);
+       if (IS_ERR(item))
+               RETURN(PTR_ERR(item));

 
        entry->se_inode = igrab(inode);
        rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode),
 
        entry->se_inode = igrab(inode);
        rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode),


@@ -847,15 +766,15 @@ static int sa_revalidate(struct inode *dir, struct sa_entry *entry,
        if (rc == 1) {
                entry->se_handle = it.it_lock_handle;
                ll_intent_release(&it);
        if (rc == 1) {
                entry->se_handle = it.it_lock_handle;
                ll_intent_release(&it);

-               sa_fini_data(minfo);
+               sa_fini_data(item);

                RETURN(1);
        }
 
                RETURN(1);
        }
 

-       rc = md_intent_getattr_async(ll_i2mdexp(dir), minfo);
+       rc = md_intent_getattr_async(ll_i2mdexp(dir), item);

        if (rc < 0) {
                entry->se_inode = NULL;
                iput(inode);
        if (rc < 0) {
                entry->se_inode = NULL;
                iput(inode);

-               sa_fini_data(minfo);
+               sa_fini_data(item);

        }
 
        RETURN(rc);
        }
 
        RETURN(rc);


@@ -1033,10 +952,8 @@ static int ll_statahead_thread(void *arg)
                        break;
                }
 
                        break;
                }
 

-               sai->sai_in_readpage = 1;

                page = ll_get_dir_page(dir, op_data, pos);
                ll_unlock_md_op_lsm(op_data);
                page = ll_get_dir_page(dir, op_data, pos);
                ll_unlock_md_op_lsm(op_data);

-               sai->sai_in_readpage = 0;

                if (IS_ERR(page)) {
                        rc = PTR_ERR(page);
                        CDEBUG(D_READA,
                if (IS_ERR(page)) {
                        rc = PTR_ERR(page);
                        CDEBUG(D_READA,


@@ -1101,11 +1018,6 @@ static int ll_statahead_thread(void *arg)
 
                        while (({set_current_state(TASK_IDLE);
                                 sai->sai_task; })) {
 
                        while (({set_current_state(TASK_IDLE);
                                 sai->sai_task; })) {

-                               if (sa_has_callback(sai)) {
-                                       __set_current_state(TASK_RUNNING);
-                                       sa_handle_callback(sai);
-                               }
-

                                spin_lock(&lli->lli_agl_lock);
                                while (sa_sent_full(sai) &&
                                       !agl_list_empty(sai)) {
                                spin_lock(&lli->lli_agl_lock);
                                while (sa_sent_full(sai) &&
                                       !agl_list_empty(sai)) {


@@ -1158,16 +1070,11 @@ static int ll_statahead_thread(void *arg)
 
        /*
         * statahead is finished, but statahead entries need to be cached, wait
 
        /*
         * statahead is finished, but statahead entries need to be cached, wait

-        * for file release to stop me.
+        * for file release closedir() call to stop me.

         */
        while (({set_current_state(TASK_IDLE);
                 sai->sai_task; })) {
         */
        while (({set_current_state(TASK_IDLE);
                 sai->sai_task; })) {

-               if (sa_has_callback(sai)) {
-                       __set_current_state(TASK_RUNNING);
-                       sa_handle_callback(sai);
-               } else {
-                       schedule();
-               }
+               schedule();

        }
        __set_current_state(TASK_RUNNING);
 
        }
        __set_current_state(TASK_RUNNING);
 


@@ -1183,9 +1090,6 @@ out:
                /* in case we're not woken up, timeout wait */
                msleep(125);
 
                /* in case we're not woken up, timeout wait */
                msleep(125);
 

-       /* release resources held by statahead RPCs */
-       sa_handle_callback(sai);
-

        CDEBUG(D_READA, "%s: statahead thread stopped: sai %p, parent %pd\n",
               sbi->ll_fsname, sai, parent);
 
        CDEBUG(D_READA, "%s: statahead thread stopped: sai %p, parent %pd\n",
               sbi->ll_fsname, sai, parent);
 


@@ -1446,10 +1350,6 @@ static int revalidate_statahead_dentry(struct inode *dir,
        if (!entry)
                GOTO(out, rc = -EAGAIN);
 
        if (!entry)
                GOTO(out, rc = -EAGAIN);
 

-       /* if statahead is busy in readdir, help it do post-work */
-       if (!sa_ready(entry) && sai->sai_in_readpage)
-               sa_handle_callback(sai);
-

        if (!sa_ready(entry)) {
                spin_lock(&lli->lli_sa_lock);
                sai->sai_index_wait = entry->se_index;
        if (!sa_ready(entry)) {
                spin_lock(&lli->lli_sa_lock);
                sai->sai_index_wait = entry->se_index;

diff --git a/lustre/lmv/lmv_obd.c b/lustre/lmv/lmv_obd.c
index ed44459..9560c16 100644 (file)
--- a/lustre/lmv/lmv_obd.c
+++ b/lustre/lmv/lmv_obd.c
@@ -3493,9 +3493,9 @@ static int lmv_clear_open_replay_data(struct obd_export *exp,
 }
 
 static int lmv_intent_getattr_async(struct obd_export *exp,
 }
 
 static int lmv_intent_getattr_async(struct obd_export *exp,

-                                   struct md_enqueue_info *minfo)
+                                   struct md_op_item *item)

 {
 {

-       struct md_op_data *op_data = &minfo->mi_data;
+       struct md_op_data *op_data = &item->mop_data;

        struct obd_device *obd = exp->exp_obd;
        struct lmv_obd *lmv = &obd->u.lmv;
        struct lmv_tgt_desc *ptgt;
        struct obd_device *obd = exp->exp_obd;
        struct lmv_obd *lmv = &obd->u.lmv;
        struct lmv_tgt_desc *ptgt;


@@ -3522,7 +3522,7 @@ static int lmv_intent_getattr_async(struct obd_export *exp,
        if (ctgt != ptgt)
                RETURN(-EREMOTE);
 
        if (ctgt != ptgt)
                RETURN(-EREMOTE);
 

-       rc = md_intent_getattr_async(ptgt->ltd_exp, minfo);
+       rc = md_intent_getattr_async(ptgt->ltd_exp, item);

 
        RETURN(rc);
 }
 
        RETURN(rc);
 }

diff --git a/lustre/mdc/mdc_internal.h b/lustre/mdc/mdc_internal.h
index ab94861..3e734e0 100644 (file)
--- a/lustre/mdc/mdc_internal.h
+++ b/lustre/mdc/mdc_internal.h
@@ -129,8 +129,7 @@ int mdc_cancel_unused(struct obd_export *exp, const struct lu_fid *fid,
 int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
                         struct lu_fid *fid, __u64 *bits);
 
 int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
                         struct lu_fid *fid, __u64 *bits);
 

-int mdc_intent_getattr_async(struct obd_export *exp,
-                            struct md_enqueue_info *minfo);
+int mdc_intent_getattr_async(struct obd_export *exp, struct md_op_item *item);

 
 enum ldlm_mode mdc_lock_match(struct obd_export *exp, __u64 flags,
                              const struct lu_fid *fid, enum ldlm_type type,
 
 enum ldlm_mode mdc_lock_match(struct obd_export *exp, __u64 flags,
                              const struct lu_fid *fid, enum ldlm_type type,

diff --git a/lustre/mdc/mdc_locks.c b/lustre/mdc/mdc_locks.c
index ac4574c..7873412 100644 (file)
--- a/lustre/mdc/mdc_locks.c
+++ b/lustre/mdc/mdc_locks.c
@@ -47,8 +47,8 @@
 #include "mdc_internal.h"
 
 struct mdc_getattr_args {
 #include "mdc_internal.h"
 
 struct mdc_getattr_args {

-       struct obd_export               *ga_exp;
-       struct md_enqueue_info          *ga_minfo;
+       struct obd_export       *ga_exp;
+       struct md_op_item       *ga_item;

 };
 
 int it_open_error(int phase, struct lookup_intent *it)
 };
 
 int it_open_error(int phase, struct lookup_intent *it)


@@ -1368,10 +1368,10 @@ static int mdc_intent_getattr_async_interpret(const struct lu_env *env,
 {
        struct mdc_getattr_args *ga = args;
        struct obd_export *exp = ga->ga_exp;
 {
        struct mdc_getattr_args *ga = args;
        struct obd_export *exp = ga->ga_exp;

-       struct md_enqueue_info *minfo = ga->ga_minfo;
-       struct ldlm_enqueue_info *einfo = &minfo->mi_einfo;
-       struct lookup_intent *it = &minfo->mi_it;
-       struct lustre_handle *lockh = &minfo->mi_lockh;
+       struct md_op_item *item = ga->ga_item;
+       struct ldlm_enqueue_info *einfo = &item->mop_einfo;
+       struct lookup_intent *it = &item->mop_it;
+       struct lustre_handle *lockh = &item->mop_lockh;

        struct ldlm_reply *lockrep;
        __u64 flags = LDLM_FL_HAS_INTENT;
 
        struct ldlm_reply *lockrep;
        __u64 flags = LDLM_FL_HAS_INTENT;
 


@@ -1398,19 +1398,19 @@ static int mdc_intent_getattr_async_interpret(const struct lu_env *env,
        if (rc)
                GOTO(out, rc);
 
        if (rc)
                GOTO(out, rc);
 

-       rc = mdc_finish_intent_lock(exp, req, &minfo->mi_data, it, lockh);
+       rc = mdc_finish_intent_lock(exp, req, &item->mop_data, it, lockh);

        EXIT;
 
 out:
        EXIT;
 
 out:

-       minfo->mi_cb(req, minfo, rc);
+       item->mop_cb(&req->rq_pill, item, rc);

        return 0;
 }
 
 int mdc_intent_getattr_async(struct obd_export *exp,
        return 0;
 }
 
 int mdc_intent_getattr_async(struct obd_export *exp,

-                            struct md_enqueue_info *minfo)
+                            struct md_op_item *item)

 {
 {

-       struct md_op_data *op_data = &minfo->mi_data;
-       struct lookup_intent *it = &minfo->mi_it;
+       struct md_op_data *op_data = &item->mop_data;
+       struct lookup_intent *it = &item->mop_it;

        struct ptlrpc_request *req;
        struct mdc_getattr_args *ga;
        struct ldlm_res_id res_id;
        struct ptlrpc_request *req;
        struct mdc_getattr_args *ga;
        struct ldlm_res_id res_id;


@@ -1440,11 +1440,11 @@ int mdc_intent_getattr_async(struct obd_export *exp,
         * to avoid possible races. It is safe to have glimpse handler
         * for non-DOM locks and costs nothing.
         */
         * to avoid possible races. It is safe to have glimpse handler
         * for non-DOM locks and costs nothing.
         */

-       if (minfo->mi_einfo.ei_cb_gl == NULL)
-               minfo->mi_einfo.ei_cb_gl = mdc_ldlm_glimpse_ast;
+       if (item->mop_einfo.ei_cb_gl == NULL)
+               item->mop_einfo.ei_cb_gl = mdc_ldlm_glimpse_ast;

 
 

-       rc = ldlm_cli_enqueue(exp, &req, &minfo->mi_einfo, &res_id, &policy,
-                             &flags, NULL, 0, LVB_T_NONE, &minfo->mi_lockh, 1);
+       rc = ldlm_cli_enqueue(exp, &req, &item->mop_einfo, &res_id, &policy,
+                             &flags, NULL, 0, LVB_T_NONE, &item->mop_lockh, 1);

        if (rc < 0) {
                ptlrpc_req_finished(req);
                RETURN(rc);
        if (rc < 0) {
                ptlrpc_req_finished(req);
                RETURN(rc);


@@ -1452,7 +1452,7 @@ int mdc_intent_getattr_async(struct obd_export *exp,
 
        ga = ptlrpc_req_async_args(ga, req);
        ga->ga_exp = exp;
 
        ga = ptlrpc_req_async_args(ga, req);
        ga->ga_exp = exp;

-       ga->ga_minfo = minfo;
+       ga->ga_item = item;

 
        req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
        ptlrpcd_add_req(req);
 
        req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
        ptlrpcd_add_req(req);