LU-2983 osd: osd-zfs to handle errors in IO path

- handle an error returned by dmu_buf_hold_array_by_bonus():
  release already pinned buffers, return an error to the caller.
- OFD to handle an error returned by dt_bufs_get()

Signed-off-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Change-Id: I1fe46364967dbc527d0d80f3729673c00ab7154c
Reviewed-on: http://review.whamcloud.com/5784
Tested-by: Hudson
Reviewed-by: Brian Behlendorf <behlendorf1@llnl.gov>
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>

diff --git a/lustre/ofd/ofd_io.c b/lustre/ofd/ofd_io.c
index 598a551..74bd9b5 100644 (file)
--- a/lustre/ofd/ofd_io.c
+++ b/lustre/ofd/ofd_io.c
@@ -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.

diff --git a/lustre/osd-zfs/osd_io.c b/lustre/osd-zfs/osd_io.c
index b081857..5ec8a33 100644 (file)
--- a/lustre/osd-zfs/osd_io.c
+++ b/lustre/osd-zfs/osd_io.c
@@ -271,9 +271,10 @@ static int osd_bufs_get_read(const struct lu_env *env, struct osd_object *obj,
         */
        while (len > 0) {
                rc = -dmu_buf_hold_array_by_bonus(obj->oo_db, off, len, TRUE,
-                                               osd_zerocopy_tag, &numbufs,
-                                               &dbp);
-               LASSERT(rc == 0);
+                                                 osd_zerocopy_tag, &numbufs,
+                                                 &dbp);
+               if (unlikely(rc))
+                       GOTO(err, rc);
 
                for (i = 0; i < numbufs; i++) {
                        int bufoff, tocpy, thispage;
@@ -323,6 +324,11 @@ static int osd_bufs_get_read(const struct lu_env *env, struct osd_object *obj,
        }
 
        RETURN(npages);
+
+err:
+       LASSERT(rc < 0);
+       osd_bufs_put(env, &obj->oo_dt, lnb - npages, npages);
+       RETURN(rc);
 }
 
 static int osd_bufs_get_write(const struct lu_env *env, struct osd_object *obj,