LU-2970 lov: wait for active IOs before deleting the object

lov_layout_operations::llo_delete will be called when the file's
layout is going to change or file is being taken out of cache so we
must make sure there is no active IOs on this object.

For layout change case, it's lov_layout_change()'s responsibility to
make sure there is no active IOs before entering into llo_delete.

Signed-off-by: Jinshan Xiong <jinshan.xiong@intel.com>
Change-Id: Id56a66548f9f05555e6d112355d2a71da5ab4705
Reviewed-on: http://review.whamcloud.com/5812
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Bobi Jam <bobijam@gmail.com>
Reviewed-by: Niu Yawei <yawei.niu@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/llite/rw.c b/lustre/llite/rw.c
index c8b0626..f8eb719 100644 (file)
--- a/lustre/llite/rw.c
+++ b/lustre/llite/rw.c
@@ -1305,7 +1305,6 @@ int ll_readpage(struct file *file, struct page *vmpage)
                         result = cl_io_read_page(env, io, page);
                 } else {
                         /* Page from a non-object file. */
-                       LASSERT(!ll_i2info(vmpage->mapping->host)->lli_has_smd);
                         unlock_page(vmpage);
                         result = 0;
                 }

diff --git a/lustre/lov/lov_object.c b/lustre/lov/lov_object.c
index 060e629..eecfbcb 100644 (file)
--- a/lustre/lov/lov_object.c
+++ b/lustre/lov/lov_object.c
@@ -78,6 +78,8 @@ struct lov_layout_operations {
                             struct cl_attr *attr);
 };
 
+static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov);
+
 /*****************************************************************************
  *
  * Lov object layout operations.
@@ -257,8 +259,8 @@ static int lov_delete_empty(const struct lu_env *env, struct lov_object *lov,
                            union lov_layout_state *state)
 {
        LASSERT(lov->lo_type == LLT_EMPTY);
-       if (cfs_atomic_read(&lov->lo_active_ios) > 0)
-               RETURN(-EBUSY);
+
+       lov_layout_wait(env, lov);
 
        cl_object_prune(env, &lov->lo_cl);
        return 0;
@@ -321,9 +323,8 @@ static int lov_delete_raid0(const struct lu_env *env, struct lov_object *lov,
        ENTRY;
 
        dump_lsm(D_INODE, lsm);
-       if (cfs_atomic_read(&lov->lo_active_ios) > 0)
-               RETURN(-EBUSY);
 
+       lov_layout_wait(env, lov);
         if (r0->lo_sub != NULL) {
                 for (i = 0; i < r0->lo_nr; ++i) {
                         struct lovsub_object *los = r0->lo_sub[i];
@@ -579,19 +580,13 @@ static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov)
        struct l_wait_info lwi = { 0 };
        ENTRY;
 
-       if (!lov->lo_layout_invalid)
-               RETURN(0);
-
        while (cfs_atomic_read(&lov->lo_active_ios) > 0) {
-               lov_conf_unlock(lov);
-
                CDEBUG(D_INODE, "file:"DFID" wait for active IO, now: %d.\n",
                        PFID(lu_object_fid(lov2lu(lov))),
                        cfs_atomic_read(&lov->lo_active_ios));
 
                l_wait_event(lov->lo_waitq,
                             cfs_atomic_read(&lov->lo_active_ios) == 0, &lwi);
-               lov_conf_lock(lov);
        }
        RETURN(0);
 }
@@ -702,7 +697,12 @@ static int lov_conf_set(const struct lu_env *env, struct cl_object *obj,
        }
 
        if (conf->coc_opc == OBJECT_CONF_WAIT) {
-               result = lov_layout_wait(env, lov);
+               if (lov->lo_layout_invalid &&
+                   cfs_atomic_read(&lov->lo_active_ios) > 0) {
+                       lov_conf_unlock(lov);
+                       result = lov_layout_wait(env, lov);
+                       lov_conf_lock(lov);
+               }
                GOTO(out, result);
        }
 
@@ -719,7 +719,7 @@ static int lov_conf_set(const struct lu_env *env, struct cl_object *obj,
        }
 
        /* will change layout - check if there still exists active IO. */
-       if (cfs_atomic_read(&lov->lo_active_ios) > 1) {
+       if (cfs_atomic_read(&lov->lo_active_ios) > 0) {
                lov->lo_layout_invalid = true;
                GOTO(out, result = -EBUSY);
        }