LU-15117 ofd: don't take lock for dt_bufs_get()

osd_bufs_get() allocates the pages and can cause new transactions
as part of memory release procedure. this would break Lustre's
"start a transaction, then do locking" rule.

Signed-off-by: Alex Zhuravlev <bzzz@whamcloud.com>
Change-Id: I58a52af8e2fbbc4823aafc133893e1defedf99b7
Reviewed-on: https://review.whamcloud.com/47029
Reviewed-by: John L. Hammond <jhammond@whamcloud.com>
Reviewed-by: Yang Sheng <ys@whamcloud.com>
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/ofd/ofd_io.c b/lustre/ofd/ofd_io.c
index 3f1fd54..6e9a87b 100644 (file)
--- a/lustre/ofd/ofd_io.c
+++ b/lustre/ofd/ofd_io.c
@@ -763,36 +763,13 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
 
        ofd_info(env)->fti_obj = fo;
 
-       ofd_read_lock(env, fo);
        if (!ofd_object_exists(fo)) {
                CERROR("%s: BRW to missing obj "DOSTID"\n",
                       exp->exp_obd->obd_name, POSTID(&obj->ioo_oid));
-               ofd_read_unlock(env, fo);
                ofd_object_put(env, fo);
                GOTO(out, rc = -ENOENT);
        }
 
-       if (ofd->ofd_lfsck_verify_pfid && oa->o_valid & OBD_MD_FLFID) {
-               rc = ofd_verify_ff(env, fo, oa);
-               if (rc != 0) {
-                       ofd_read_unlock(env, fo);
-                       ofd_object_put(env, fo);
-                       GOTO(out, rc);
-               }
-       }
-
-       /* need to verify layout version */
-       if (oa->o_valid & OBD_MD_LAYOUT_VERSION) {
-               rc = ofd_verify_layout_version(env, fo, oa);
-               if (rc) {
-                       ofd_read_unlock(env, fo);
-                       ofd_object_put(env, fo);
-                       GOTO(out, rc);
-               }
-
-               oa->o_valid &= ~OBD_MD_LAYOUT_VERSION;
-       }
-
        if (ptlrpc_connection_is_local(exp->exp_connection))
                dbt |= DT_BUFS_TYPE_LOCAL;
 
@@ -809,7 +786,7 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
                rc = dt_bufs_get(env, ofd_object_child(fo),
                                 rnb + i, lnb + j, maxlnb, dbt);
                if (unlikely(rc < 0))
-                       GOTO(err, rc);
+                       GOTO(err_nolock, rc);
                LASSERT(rc <= PTLRPC_MAX_BRW_PAGES);
                /* correct index for local buffers to continue with */
                for (k = 0; k < rc; k++) {
@@ -826,6 +803,27 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
        }
        LASSERT(*nr_local > 0 && *nr_local <= PTLRPC_MAX_BRW_PAGES);
 
+       ofd_read_lock(env, fo);
+       if (!ofd_object_exists(fo)) {
+               CERROR("%s: BRW to missing obj "DOSTID": rc = -ENOENT\n",
+                      exp->exp_obd->obd_name, POSTID(&obj->ioo_oid));
+               GOTO(err, rc = -ENOENT);
+       }
+
+       if (ofd->ofd_lfsck_verify_pfid && oa->o_valid & OBD_MD_FLFID) {
+               rc = ofd_verify_ff(env, fo, oa);
+               if (rc != 0)
+                       GOTO(err, rc);
+       }
+
+       /* need to verify layout version */
+       if (oa->o_valid & OBD_MD_LAYOUT_VERSION) {
+               rc = ofd_verify_layout_version(env, fo, oa);
+               if (rc)
+                       GOTO(err, rc);
+               oa->o_valid &= ~OBD_MD_LAYOUT_VERSION;
+       }
+
        rc = dt_write_prep(env, ofd_object_child(fo), lnb, *nr_local);
        if (unlikely(rc != 0))
                GOTO(err, rc);
@@ -864,9 +862,11 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
        ofd_info(env)->fti_range_locked = 1;
 
        RETURN(0);
+
 err:
-       dt_bufs_put(env, ofd_object_child(fo), lnb, *nr_local);
        ofd_read_unlock(env, fo);
+err_nolock:
+       dt_bufs_put(env, ofd_object_child(fo), lnb, *nr_local);
        ofd_object_put(env, fo);
        /* tgt_grant_prepare_write() was called, so we must commit */
        tgt_grant_commit(exp, oa->o_grant_used, rc);

diff --git a/lustre/osd-ldiskfs/osd_io.c b/lustre/osd-ldiskfs/osd_io.c
index 9cff3c8..068ef49 100644 (file)
--- a/lustre/osd-ldiskfs/osd_io.c
+++ b/lustre/osd-ldiskfs/osd_io.c
@@ -872,6 +872,9 @@ static int osd_bufs_get(const struct lu_env *env, struct dt_object *dt,
 
        LASSERT(obj->oo_inode);
 
+       if (unlikely(obj->oo_destroyed))
+               RETURN(-ENOENT);
+
        rc = osd_map_remote_to_local(pos, len, &npages, lnb, maxlnb);
        if (rc)
                RETURN(rc);

diff --git a/lustre/osd-zfs/osd_io.c b/lustre/osd-zfs/osd_io.c
index f267367..74644c9 100644 (file)
--- a/lustre/osd-zfs/osd_io.c
+++ b/lustre/osd-zfs/osd_io.c
@@ -589,14 +589,18 @@ static int osd_bufs_get(const struct lu_env *env, struct dt_object *dt,
        struct osd_object *obj  = osd_dt_obj(dt);
        int                rc;
 
-       LASSERT(dt_object_exists(dt));
-       LASSERT(obj->oo_dn);
+       down_read(&obj->oo_guard);
+
+       if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
+               GOTO(out, rc = -ENOENT);
 
        if (rw & DT_BUFS_TYPE_WRITE)
                rc = osd_bufs_get_write(env, obj, offset, len, lnb, maxlnb);
        else
                rc = osd_bufs_get_read(env, obj, offset, len, lnb, maxlnb);
 
+out:
+       up_read(&obj->oo_guard);
        return rc;
 }