LU-2179 osc: truncate partial page correctly

If a partial page is being truncated, the corresponding osc extent
should be held until the truncate finished.

Debug patch for osc_extent_wait() and don't wait for completion
of RPC it's not even sent in truncate.

Signed-off-by: Jinshan Xiong <jinshan.xiong@intel.com>
Change-Id: I96a5ec1fdbb3133c735ebdfdd0330a45a2a8ab1a
Reviewed-on: http://review.whamcloud.com/4317
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Niu Yawei <niu@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/osc/osc_cache.c b/lustre/osc/osc_cache.c
index 23dc755..061fdea 100644 (file)
--- a/lustre/osc/osc_cache.c
+++ b/lustre/osc/osc_cache.c
@@ -845,6 +845,17 @@ int osc_extent_finish(const struct lu_env *env, struct osc_extent *ext,
        RETURN(0);
 }
 
        RETURN(0);
 }
 

+static int extent_wait_cb(struct osc_extent *ext, int state)
+{
+       int ret;
+
+       osc_object_lock(ext->oe_obj);
+       ret = ext->oe_state == state;
+       osc_object_unlock(ext->oe_obj);
+
+       return ret;
+}
+

 /**
  * Wait for the extent's state to become @state.
  */
 /**
  * Wait for the extent's state to become @state.
  */


@@ -852,7 +863,8 @@ static int osc_extent_wait(const struct lu_env *env, struct osc_extent *ext,
                           int state)
 {
        struct osc_object *obj = ext->oe_obj;
                           int state)
 {
        struct osc_object *obj = ext->oe_obj;

-       struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
+       struct l_wait_info lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(600), NULL,
+                                                 LWI_ON_SIGNAL_NOOP, NULL);

        int rc = 0;
        ENTRY;
 
        int rc = 0;
        ENTRY;
 


@@ -874,7 +886,16 @@ static int osc_extent_wait(const struct lu_env *env, struct osc_extent *ext,
                osc_extent_release(env, ext);
 
        /* wait for the extent until its state becomes @state */
                osc_extent_release(env, ext);
 
        /* wait for the extent until its state becomes @state */

-       rc = l_wait_event(ext->oe_waitq, ext->oe_state == state, &lwi);
+       rc = l_wait_event(ext->oe_waitq, extent_wait_cb(ext, state), &lwi);
+       if (rc == -ETIMEDOUT) {
+               OSC_EXTENT_DUMP(D_ERROR, ext,
+                       "%s: wait ext to %d timedout, recovery in progress?\n",
+                       osc_export(obj)->exp_obd->obd_name, state);
+
+               lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
+               rc = l_wait_event(ext->oe_waitq, extent_wait_cb(ext, state),
+                                 &lwi);
+       }

        if (rc == 0 && ext->oe_rc < 0)
                rc = ext->oe_rc;
        RETURN(rc);
        if (rc == 0 && ext->oe_rc < 0)
                rc = ext->oe_rc;
        RETURN(rc);


@@ -884,7 +905,8 @@ static int osc_extent_wait(const struct lu_env *env, struct osc_extent *ext,
  * Discard pages with index greater than @size. If @ext is overlapped with
  * @size, then partial truncate happens.
  */
  * Discard pages with index greater than @size. If @ext is overlapped with
  * @size, then partial truncate happens.
  */

-static int osc_extent_truncate(struct osc_extent *ext, pgoff_t trunc_index)
+static int osc_extent_truncate(struct osc_extent *ext, pgoff_t trunc_index,
+                               bool partial)

 {
        struct cl_env_nest     nest;
        struct lu_env         *env;
 {
        struct cl_env_nest     nest;
        struct lu_env         *env;


@@ -925,7 +947,8 @@ static int osc_extent_truncate(struct osc_extent *ext, pgoff_t trunc_index)
 
                /* only discard the pages with their index greater than
                 * trunc_index, and ... */
 
                /* only discard the pages with their index greater than
                 * trunc_index, and ... */

-               if (sub->cp_index < trunc_index) {
+               if (sub->cp_index < trunc_index ||
+                   (sub->cp_index == trunc_index && partial)) {

                        /* accounting how many pages remaining in the chunk
                         * so that we can calculate grants correctly. */
                        if (sub->cp_index >> ppc_bits == trunc_chunk)
                        /* accounting how many pages remaining in the chunk
                         * so that we can calculate grants correctly. */
                        if (sub->cp_index >> ppc_bits == trunc_chunk)


@@ -953,8 +976,9 @@ static int osc_extent_truncate(struct osc_extent *ext, pgoff_t trunc_index)
                --ext->oe_nr_pages;
                ++nr_pages;
        }
                --ext->oe_nr_pages;
                ++nr_pages;
        }

-       EASSERTF(ergo(ext->oe_start >= trunc_index, ext->oe_nr_pages == 0),
-                ext, "trunc_index %lu\n", trunc_index);
+       EASSERTF(ergo(ext->oe_start >= trunc_index + !!partial,
+                     ext->oe_nr_pages == 0),
+               ext, "trunc_index %lu, partial %d\n", trunc_index, partial);

 
        osc_object_lock(obj);
        if (ext->oe_nr_pages == 0) {
 
        osc_object_lock(obj);
        if (ext->oe_nr_pages == 0) {


@@ -2560,10 +2584,12 @@ int osc_cache_truncate_start(const struct lu_env *env, struct osc_io *oio,
        pgoff_t index;
        CFS_LIST_HEAD(list);
        int result = 0;
        pgoff_t index;
        CFS_LIST_HEAD(list);
        int result = 0;

+       bool partial;

        ENTRY;
 
        /* pages with index greater or equal to index will be truncated. */
        ENTRY;
 
        /* pages with index greater or equal to index will be truncated. */

-       index = cl_index(osc2cl(obj), size + CFS_PAGE_SIZE - 1);
+       index = cl_index(osc2cl(obj), size);
+       partial = size > cl_offset(osc2cl(obj), index);

 
 again:
        osc_object_lock(obj);
 
 again:
        osc_object_lock(obj);


@@ -2621,7 +2647,7 @@ again:
                if (ext->oe_state != OES_TRUNC)
                        osc_extent_wait(env, ext, OES_TRUNC);
 
                if (ext->oe_state != OES_TRUNC)
                        osc_extent_wait(env, ext, OES_TRUNC);
 

-               rc = osc_extent_truncate(ext, index);
+               rc = osc_extent_truncate(ext, index, partial);

                if (rc < 0) {
                        if (result == 0)
                                result = rc;
                if (rc < 0) {
                        if (result == 0)
                                result = rc;


@@ -2634,10 +2660,11 @@ again:
                        /* this must be an overlapped extent which means only
                         * part of pages in this extent have been truncated.
                         */
                        /* this must be an overlapped extent which means only
                         * part of pages in this extent have been truncated.
                         */

-                       EASSERTF(ext->oe_start < index, ext,
-                                "trunc index = %lu.\n", index);
+                       EASSERTF(ext->oe_start <= index, ext,
+                                "trunc index = %lu/%d.\n", index, partial);

                        /* fix index to skip this partially truncated extent */
                        index = ext->oe_end + 1;
                        /* fix index to skip this partially truncated extent */
                        index = ext->oe_end + 1;

+                       partial = false;

 
                        /* we need to hold this extent in OES_TRUNC state so
                         * that no writeback will happen. This is to avoid
 
                        /* we need to hold this extent in OES_TRUNC state so
                         * that no writeback will happen. This is to avoid

diff --git a/lustre/osc/osc_cl_internal.h b/lustre/osc/osc_cl_internal.h
index 79dcc8b..ac26f4f 100644 (file)
--- a/lustre/osc/osc_cl_internal.h
+++ b/lustre/osc/osc_cl_internal.h
@@ -83,6 +83,7 @@ struct osc_io {
        struct obd_info    oi_info;
        struct obdo        oi_oa;
        struct osc_async_cbargs {
        struct obd_info    oi_info;
        struct obdo        oi_oa;
        struct osc_async_cbargs {

+               bool              opc_rpc_sent;

                int               opc_rc;
                cfs_completion_t  opc_sync;
        } oi_cbarg;
                int               opc_rc;
                cfs_completion_t  opc_sync;
        } oi_cbarg;

diff --git a/lustre/osc/osc_io.c b/lustre/osc/osc_io.c
index b66019b..f75aa76 100644 (file)
--- a/lustre/osc/osc_io.c
+++ b/lustre/osc/osc_io.c
@@ -482,6 +482,7 @@ static int osc_io_setattr_start(const struct lu_env *env,
                                                         &oinfo, NULL,
                                                        osc_async_upcall,
                                                         cbargs, PTLRPCD_SET);
                                                         &oinfo, NULL,
                                                        osc_async_upcall,
                                                         cbargs, PTLRPCD_SET);

+               cbargs->opc_rpc_sent = result == 0;

         }
         return result;
 }
         }
         return result;
 }


@@ -493,11 +494,12 @@ static void osc_io_setattr_end(const struct lu_env *env,
        struct osc_io    *oio = cl2osc_io(env, slice);
        struct cl_object *obj = slice->cis_obj;
        struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
        struct osc_io    *oio = cl2osc_io(env, slice);
        struct cl_object *obj = slice->cis_obj;
        struct osc_async_cbargs *cbargs = &oio->oi_cbarg;

-        int result;
-
-        cfs_wait_for_completion(&cbargs->opc_sync);
+        int result = 0;

 
 

-        result = io->ci_result = cbargs->opc_rc;
+       if (cbargs->opc_rpc_sent) {
+               cfs_wait_for_completion(&cbargs->opc_sync);
+               result = io->ci_result = cbargs->opc_rc;
+       }

         if (result == 0) {
                 if (oio->oi_lockless) {
                         /* lockless truncate */
         if (result == 0) {
                 if (oio->oi_lockless) {
                         /* lockless truncate */