LU-3935 scrub: support dryrun mode OI scrub

[fs/lustre-release.git] / lustre / ofd / ofd_io.c

diff --git a/lustre/ofd/ofd_io.c b/lustre/ofd/ofd_io.c
index 8655093..b90bc8a 100644 (file)
--- a/lustre/ofd/ofd_io.c
+++ b/lustre/ofd/ofd_io.c
@@ -27,7 +27,7 @@
  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2012, Intel Corporation.
+ * Copyright (c) 2012, 2013, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -65,18 +65,20 @@ static int ofd_preprw_read(const struct lu_env *env, struct obd_export *exp,
                GOTO(unlock, rc = -ENOENT);
 
        /* parse remote buffers to local buffers and prepare the latter */
+       *nr_local = 0;
        for (i = 0, j = 0; i < niocount; i++) {
                rc = dt_bufs_get(env, ofd_object_child(fo), rnb + i,
                                 lnb + j, 0, ofd_object_capa(env, fo));
-               LASSERT(rc > 0);
+               if (unlikely(rc < 0))
+                       GOTO(buf_put, rc);
                LASSERT(rc <= PTLRPC_MAX_BRW_PAGES);
                /* correct index for local buffers to continue with */
                j += rc;
+               *nr_local += rc;
                LASSERT(j <= PTLRPC_MAX_BRW_PAGES);
                tot_bytes += rnb[i].rnb_len;
        }
 
-       *nr_local = j;
        LASSERT(*nr_local > 0 && *nr_local <= PTLRPC_MAX_BRW_PAGES);
        rc = dt_attr_get(env, ofd_object_child(fo), la,
                         ofd_object_capa(env, fo));
@@ -158,11 +160,13 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
        ofd_grant_prepare_write(env, exp, oa, rnb, obj->ioo_bufcnt);
 
        /* parse remote buffers to local buffers and prepare the latter */
+       *nr_local = 0;
        for (i = 0, j = 0; i < obj->ioo_bufcnt; i++) {
                rc = dt_bufs_get(env, ofd_object_child(fo),
                                 rnb + i, lnb + j, 1,
                                 ofd_object_capa(env, fo));
-               LASSERT(rc > 0);
+               if (unlikely(rc < 0))
+                       GOTO(err, rc);
                LASSERT(rc <= PTLRPC_MAX_BRW_PAGES);
                /* correct index for local buffers to continue with */
                for (k = 0; k < rc; k++) {
@@ -176,26 +180,26 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
                                lnb[j+k].lnb_flags &= ~OBD_BRW_NOQUOTA;
                }
                j += rc;
+               *nr_local += rc;
                LASSERT(j <= PTLRPC_MAX_BRW_PAGES);
                tot_bytes += rnb[i].rnb_len;
        }
-       *nr_local = j;
        LASSERT(*nr_local > 0 && *nr_local <= PTLRPC_MAX_BRW_PAGES);
 
        rc = dt_write_prep(env, ofd_object_child(fo), lnb, *nr_local);
-       if (unlikely(rc != 0)) {
-               dt_bufs_put(env, ofd_object_child(fo), lnb, *nr_local);
-               ofd_read_unlock(env, fo);
-               /* ofd_grant_prepare_write() was called, so we must commit */
-               ofd_grant_commit(env, exp, rc);
-               GOTO(out, rc);
-       }
+       if (unlikely(rc != 0))
+               GOTO(err, rc);
 
        lprocfs_counter_add(ofd_obd(ofd)->obd_stats,
                            LPROC_OFD_WRITE_BYTES, tot_bytes);
        ofd_counter_incr(exp, LPROC_OFD_STATS_WRITE,
                         oti->oti_jobid, tot_bytes);
        RETURN(0);
+err:
+       dt_bufs_put(env, ofd_object_child(fo), lnb, *nr_local);
+       ofd_read_unlock(env, fo);
+       /* ofd_grant_prepare_write() was called, so we must commit */
+       ofd_grant_commit(env, exp, rc);
 out:
        /* let's still process incoming grant information packed in the oa,
         * but without enforcing grant since we won't proceed with the write.
@@ -214,6 +218,13 @@ int ofd_preprw(const struct lu_env* env, int cmd, struct obd_export *exp,
        struct ofd_thread_info  *info;
        int                      rc = 0;
 
+       if (*nr_local > PTLRPC_MAX_BRW_PAGES) {
+               CERROR("%s: bulk has too many pages %d, which exceeds the"
+                      "maximum pages per RPC of %d\n",
+                      exp->exp_obd->obd_name, *nr_local, PTLRPC_MAX_BRW_PAGES);
+               RETURN(-EPROTO);
+       }
+
        rc = lu_env_refill((struct lu_env *)env);
        LASSERT(rc == 0);
        info = ofd_info_init(env, exp);
@@ -243,7 +254,10 @@ int ofd_preprw(const struct lu_env* env, int cmd, struct obd_export *exp,
        LASSERT(objcount == 1);
        LASSERT(obj->ioo_bufcnt > 0);
 
-       ostid_to_fid(&info->fti_fid, &oa->o_oi, 0);
+       rc = ostid_to_fid(&info->fti_fid, &oa->o_oi, 0);
+       if (unlikely(rc != 0))
+               RETURN(rc);
+
        if (cmd == OBD_BRW_WRITE) {
                rc = ofd_auth_capa(exp, &info->fti_fid, ostid_seq(&oa->o_oi),
                                   capa, CAPA_OPC_OSS_WRITE);
@@ -502,7 +516,9 @@ int ofd_commitrw(const struct lu_env *env, int cmd, struct obd_export *exp,
 
        LASSERT(npages > 0);
 
-       ostid_to_fid(&info->fti_fid, &oa->o_oi, 0);
+       rc = ostid_to_fid(&info->fti_fid, &oa->o_oi, 0);
+       if (unlikely(rc != 0))
+               RETURN(rc);
        if (cmd == OBD_BRW_WRITE) {
                /* Don't update timestamps if this write is older than a
                 * setattr which modifies the timestamps. b=10150 */