Whamcloud - gitweb
git://git.whamcloud.com / fs / lustre-release.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
LU-10048 ofd: take local locks within transaction
[fs/lustre-release.git] / lustre / ofd / ofd_objects.c
diff --git a/lustre/ofd/ofd_objects.c b/lustre/ofd/ofd_objects.c
index 53e5b64..e98e72b 100644 (file)
--- a/lustre/ofd/ofd_objects.c
+++ b/lustre/ofd/ofd_objects.c
@@ -316,7 +316,6 @@ int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
                        break;
                }
 
-               ofd_write_lock(env, fo);
                batch[i] = fo;
        }
        info->fti_buf.lb_buf = &tmp;
@@ -394,6 +393,8 @@ int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
                fo = batch[i];
                LASSERT(fo);
 
+               ofd_write_lock(env, fo);
+
                /* Only the new created objects need to be recorded. */
                if (ofd->ofd_osd->dd_record_fid_accessed) {
                        struct lfsck_req_local *lrl = &ofd_info(env)->fti_lrl;
@@ -410,6 +411,7 @@ int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
 
                        rc = dt_create(env, next, &info->fti_attr, NULL,
                                       &info->fti_dof, th);
+                       ofd_write_unlock(env, fo);
                        if (rc < 0) {
                                if (i == 0)
                                        GOTO(trans_stop, rc);
@@ -418,7 +420,10 @@ int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
                                break;
                        }
                        LASSERT(ofd_object_exists(fo));
+               } else {
+                       ofd_write_unlock(env, fo);
                }
+
                ofd_seq_last_oid_set(oseq, id + i);
        }
 
@@ -453,10 +458,9 @@ trans_stop:
 out:
        for (i = 0; i < nr_saved; i++) {
                fo = batch[i];
-               if (fo) {
-                       ofd_write_unlock(env, fo);
-                       ofd_object_put(env, fo);
-               }
+               if (!fo)
+                       continue;
+               ofd_object_put(env, fo);
        }
        OBD_FREE(batch, nr_saved * sizeof(struct ofd_object *));
 
@@ -657,9 +661,8 @@ int ofd_attr_set(const struct lu_env *env, struct ofd_object *fo,
 
        ENTRY;
 
-       ofd_write_lock(env, fo);
        if (!ofd_object_exists(fo))
-               GOTO(unlock, rc = -ENOENT);
+               GOTO(out, rc = -ENOENT);
 
        if (la->la_valid & (LA_ATIME | LA_MTIME | LA_CTIME))
                tgt_fmd_update(info->fti_exp, &fo->ofo_header.loh_fid,
@@ -668,50 +671,50 @@ int ofd_attr_set(const struct lu_env *env, struct ofd_object *fo,
        /* VBR: version recovery check */
        rc = ofd_version_get_check(info, fo);
        if (rc)
-               GOTO(unlock, rc);
+               GOTO(out, rc);
 
        rc = ofd_attr_handle_id(env, fo, la, 1 /* is_setattr */);
        if (rc != 0)
-               GOTO(unlock, rc);
-
-       fl = ofd_object_ff_update(env, fo, oa, ff);
-       if (fl < 0)
-               GOTO(unlock, rc = fl);
+               GOTO(out, rc);
 
        th = ofd_trans_create(env, ofd);
        if (IS_ERR(th))
-               GOTO(unlock, rc = PTR_ERR(th));
+               GOTO(out, rc = PTR_ERR(th));
 
        rc = dt_declare_attr_set(env, ofd_object_child(fo), la, th);
        if (rc)
                GOTO(stop, rc);
 
-       if (fl) {
-               if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
-                       ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
-               else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
-                       le32_add_cpu(&ff->ff_parent.f_oid, -1);
-
-               info->fti_buf.lb_buf = ff;
-               info->fti_buf.lb_len = sizeof(*ff);
-               rc = dt_declare_xattr_set(env, ofd_object_child(fo),
-                                         &info->fti_buf, XATTR_NAME_FID, fl,
-                                         th);
-               if (rc)
-                       GOTO(stop, rc);
-       }
+       info->fti_buf.lb_buf = ff;
+       info->fti_buf.lb_len = sizeof(*ff);
+       rc = dt_declare_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
+                                 XATTR_NAME_FID, 0, th);
+       if (rc)
+               GOTO(stop, rc);
 
        rc = ofd_trans_start(env, ofd, la->la_valid & LA_SIZE ? fo : NULL, th);
        if (rc)
                GOTO(stop, rc);
 
+       ofd_write_lock(env, fo);
+       if (!ofd_object_exists(fo))
+               GOTO(unlock, rc = -ENOENT);
+
        rc = dt_attr_set(env, ofd_object_child(fo), la, th);
        if (rc)
-               GOTO(stop, rc);
+               GOTO(unlock, rc);
+
+       fl = ofd_object_ff_update(env, fo, oa, ff);
+       if (fl < 0)
+               GOTO(unlock, rc = fl);
 
        if (fl) {
-               if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
-                       GOTO(stop, rc);
+               if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
+                       ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
+               else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
+                       le32_add_cpu(&ff->ff_parent.f_oid, -1);
+               else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
+                       GOTO(unlock, rc);
 
                info->fti_buf.lb_buf = ff;
                info->fti_buf.lb_len = sizeof(*ff);
@@ -721,8 +724,10 @@ int ofd_attr_set(const struct lu_env *env, struct ofd_object *fo,
                        filter_fid_le_to_cpu(&fo->ofo_ff, ff, sizeof(*ff));
        }
 
-       GOTO(stop, rc);
+       GOTO(unlock, rc);
 
+unlock:
+       ofd_write_unlock(env, fo);
 stop:
        rc2 = ofd_trans_stop(env, ofd, th, rc);
        if (rc2)
@@ -730,10 +735,7 @@ stop:
                       ofd_name(ofd), rc2);
        if (!rc)
                rc = rc2;
-
-unlock:
-       ofd_write_unlock(env, fo);
-
+out:
        return rc;
 }
 
@@ -771,45 +773,27 @@ int ofd_object_punch(const struct lu_env *env, struct ofd_object *fo,
        /* we support truncate, not punch yet */
        LASSERT(end == OBD_OBJECT_EOF);
 
-       ofd_write_lock(env, fo);
-       if (la->la_valid & (LA_ATIME | LA_MTIME | LA_CTIME))
-               tgt_fmd_update(info->fti_exp, &fo->ofo_header.loh_fid,
-                              info->fti_xid);
-
        if (!ofd_object_exists(fo))
-               GOTO(unlock, rc = -ENOENT);
+               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)
-                       GOTO(unlock, 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(unlock, rc);
-
-               oa->o_valid &= ~OBD_MD_LAYOUT_VERSION;
+                       GOTO(out, rc);
        }
 
        /* VBR: version recovery check */
        rc = ofd_version_get_check(info, fo);
        if (rc)
-               GOTO(unlock, rc);
+               GOTO(out, rc);
 
        rc = ofd_attr_handle_id(env, fo, la, 0 /* !is_setattr */);
        if (rc != 0)
-               GOTO(unlock, rc);
-
-       fl = ofd_object_ff_update(env, fo, oa, ff);
-       if (fl < 0)
-               GOTO(unlock, rc = fl);
+               GOTO(out, rc);
 
        th = ofd_trans_create(env, ofd);
        if (IS_ERR(th))
-               GOTO(unlock, rc = PTR_ERR(th));
+               GOTO(out, rc = PTR_ERR(th));
 
        rc = dt_declare_attr_set(env, dob, la, th);
        if (rc)
@@ -819,45 +803,67 @@ int ofd_object_punch(const struct lu_env *env, struct ofd_object *fo,
        if (rc)
                GOTO(stop, rc);
 
-       if (fl) {
-               if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
-                       ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
-               else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
-                       le32_add_cpu(&ff->ff_parent.f_oid, -1);
-
-               info->fti_buf.lb_buf = ff;
-               info->fti_buf.lb_len = sizeof(*ff);
-               rc = dt_declare_xattr_set(env, ofd_object_child(fo),
-                                         &info->fti_buf, XATTR_NAME_FID, fl,
-                                         th);
-               if (rc)
-                       GOTO(stop, rc);
-       }
+       info->fti_buf.lb_buf = ff;
+       info->fti_buf.lb_len = sizeof(*ff);
+       rc = dt_declare_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
+                                 XATTR_NAME_FID, 0, th);
+       if (rc)
+               GOTO(stop, rc);
 
        rc = ofd_trans_start(env, ofd, fo, th);
        if (rc)
                GOTO(stop, rc);
 
+       ofd_write_lock(env, fo);
+
+       if (la->la_valid & (LA_ATIME | LA_MTIME | LA_CTIME))
+               tgt_fmd_update(info->fti_exp, &fo->ofo_header.loh_fid,
+                              info->fti_xid);
+
+       if (!ofd_object_exists(fo))
+               GOTO(unlock, rc = -ENOENT);
+
+       /* need to verify layout version */
+       if (oa->o_valid & OBD_MD_LAYOUT_VERSION) {
+               rc = ofd_verify_layout_version(env, fo, oa);
+               if (rc)
+                       GOTO(unlock, rc);
+
+               oa->o_valid &= ~OBD_MD_LAYOUT_VERSION;
+       }
+
        rc = dt_punch(env, dob, start, OBD_OBJECT_EOF, th);
        if (rc)
-               GOTO(stop, rc);
+               GOTO(unlock, rc);
+
+       fl = ofd_object_ff_update(env, fo, oa, ff);
+       if (fl < 0)
+               GOTO(unlock, rc = fl);
 
        rc = dt_attr_set(env, dob, la, th);
        if (rc)
-               GOTO(stop, rc);
+               GOTO(unlock, rc);
 
        if (fl) {
-               if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
-                       GOTO(stop, rc);
+               if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
+                       ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
+               else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
+                       le32_add_cpu(&ff->ff_parent.f_oid, -1);
+               else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
+                       GOTO(unlock, rc);
 
+               info->fti_buf.lb_buf = ff;
+               info->fti_buf.lb_len = sizeof(*ff);
                rc = dt_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
                                  XATTR_NAME_FID, fl, th);
                if (!rc)
                        filter_fid_le_to_cpu(&fo->ofo_ff, ff, sizeof(*ff));
        }
 
-       GOTO(stop, rc);
+       GOTO(unlock, rc);
 
+unlock:
+       ofd_write_unlock(env, fo);
 stop:
        rc2 = ofd_trans_stop(env, ofd, th, rc);
        if (rc2 != 0)
@@ -865,9 +871,7 @@ stop:
                       ofd_name(ofd), rc2);
        if (!rc)
                rc = rc2;
-unlock:
-       ofd_write_unlock(env, fo);
-
+out:
        return rc;
 }
 
@@ -895,13 +899,12 @@ int ofd_destroy(const struct lu_env *env, struct ofd_object *fo,
 
        ENTRY;
 
-       ofd_write_lock(env, fo);
        if (!ofd_object_exists(fo))
-               GOTO(unlock, rc = -ENOENT);
+               GOTO(out, rc = -ENOENT);
 
        th = ofd_trans_create(env, ofd);
        if (IS_ERR(th))
-               GOTO(unlock, rc = PTR_ERR(th));
+               GOTO(out, rc = PTR_ERR(th));
 
        rc = dt_declare_ref_del(env, ofd_object_child(fo), th);
        if (rc < 0)
@@ -918,10 +921,16 @@ int ofd_destroy(const struct lu_env *env, struct ofd_object *fo,
        if (rc)
                GOTO(stop, rc);
 
+       ofd_write_lock(env, fo);
+       if (!ofd_object_exists(fo))
+               GOTO(stop, rc = -ENOENT);
+
        tgt_fmd_drop(ofd_info(env)->fti_exp, &fo->ofo_header.loh_fid);
 
        dt_ref_del(env, ofd_object_child(fo), th);
        dt_destroy(env, ofd_object_child(fo), th);
+       ofd_write_unlock(env, fo);
+
 stop:
        rc2 = ofd_trans_stop(env, ofd, th, rc);
        if (rc2)
@@ -929,8 +938,7 @@ stop:
                       ofd_name(ofd), rc2);
        if (!rc)
                rc = rc2;
-unlock:
-       ofd_write_unlock(env, fo);
+out:
        RETURN(rc);
 }
 
Lustre primary development and releases