LU-2056 mdd: create transaction before locking object

[fs/lustre-release.git] / lustre / mdd / mdd_object.c

diff --git a/lustre/mdd/mdd_object.c b/lustre/mdd/mdd_object.c
index 1937db4..c3f6958 100644 (file)
--- a/lustre/mdd/mdd_object.c
+++ b/lustre/mdd/mdd_object.c
@@ -1790,7 +1790,9 @@ static int mdd_close(const struct lu_env *env, struct md_object *obj,
         ENTRY;
 
         if (ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_KEEP_ORPHAN) {
-                mdd_obj->mod_count--;
+               mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
+               mdd_obj->mod_count--;
+               mdd_write_unlock(env, mdd_obj);
 
                 if (mdd_obj->mod_flags & ORPHAN_OBJ && !mdd_obj->mod_count)
                         CDEBUG(D_HA, "Object "DFID" is retained in orphan "
@@ -1798,10 +1800,14 @@ static int mdd_close(const struct lu_env *env, struct md_object *obj,
                 RETURN(0);
         }
 
-        /* check without any lock */
-        if (mdd_obj->mod_count == 1 &&
-            (mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0) {
- again:
+       /* mdd_finish_unlink() will always set orphan object as DEAD_OBJ, but
+        * it might fail to add the object to orphan list (w/o ORPHAN_OBJ). */
+       /* check without any lock */
+       is_orphan = mdd_obj->mod_count == 1 &&
+                   (mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0;
+
+again:
+       if (is_orphan) {
                 handle = mdd_trans_create(env, mdo2mdd(obj));
                 if (IS_ERR(handle))
                         RETURN(PTR_ERR(handle));
@@ -1820,71 +1826,62 @@ static int mdd_close(const struct lu_env *env, struct md_object *obj,
         }
 
         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
-        if (handle == NULL && mdd_obj->mod_count == 1 &&
-            (mdd_obj->mod_flags & ORPHAN_OBJ) != 0) {
-                mdd_write_unlock(env, mdd_obj);
-                goto again;
-        }
-
-        /* release open count */
-        mdd_obj->mod_count --;
-
-        if (mdd_obj->mod_count == 0 && mdd_obj->mod_flags & ORPHAN_OBJ) {
-                /* remove link to object from orphan index */
-                LASSERT(handle != NULL);
-                rc = __mdd_orphan_del(env, mdd_obj, handle);
-                if (rc == 0) {
-                        CDEBUG(D_HA, "Object "DFID" is deleted from orphan "
-                               "list, OSS objects to be destroyed.\n",
-                               PFID(mdd_object_fid(mdd_obj)));
-                        is_orphan = 1;
-                } else {
-                        CERROR("Object "DFID" can not be deleted from orphan "
-                                "list, maybe cause OST objects can not be "
-                                "destroyed (err: %d).\n",
-                                PFID(mdd_object_fid(mdd_obj)), rc);
-                        /* If object was not deleted from orphan list, do not
-                         * destroy OSS objects, which will be done when next
-                         * recovery. */
-                        GOTO(out, rc);
-                }
-        }
-
        rc = mdd_la_get(env, mdd_obj, &ma->ma_attr,
                        mdd_object_capa(env, mdd_obj));
-        /* Object maybe not in orphan list originally, it is rare case for
-         * mdd_finish_unlink() failure. */
-        if (rc == 0 && (ma->ma_attr.la_nlink == 0 || is_orphan)) {
-               if (handle == NULL) {
-                       handle = mdd_trans_create(env, mdo2mdd(obj));
-                       if (IS_ERR(handle))
-                               GOTO(out, rc = PTR_ERR(handle));
-
-                       rc = mdo_declare_destroy(env, mdd_obj, handle);
-                       if (rc)
-                               GOTO(out, rc);
+       if (rc != 0) {
+               CERROR("Failed to get lu_attr of "DFID": %d\n",
+                      PFID(mdd_object_fid(mdd_obj)), rc);
+               GOTO(out, rc);
+       }
 
-                       rc = mdd_declare_changelog_store(env, mdd,
-                                       NULL, handle);
-                       if (rc)
-                               GOTO(stop, rc);
+       /* check again with lock */
+       is_orphan = (mdd_obj->mod_count == 1) &&
+                   ((mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0 ||
+                    ma->ma_attr.la_nlink == 0);
 
-                       rc = mdd_trans_start(env, mdo2mdd(obj), handle);
-                       if (rc)
-                               GOTO(out, rc);
+       if (is_orphan && handle == NULL) {
+               mdd_write_unlock(env, mdd_obj);
+               goto again;
+       }
+
+       mdd_obj->mod_count--; /*release open count */
+
+       if (!is_orphan)
+               GOTO(out, rc = 0);
+
+       /* Orphan object */
+       /* NB: Object maybe not in orphan list originally, it is rare case for
+        * mdd_finish_unlink() failure, in that case, the object doesn't have
+        * ORPHAN_OBJ flag */
+       if ((mdd_obj->mod_flags & ORPHAN_OBJ) != 0) {
+               /* remove link to object from orphan index */
+               LASSERT(handle != NULL);
+               rc = __mdd_orphan_del(env, mdd_obj, handle);
+               if (rc != 0) {
+                       CERROR("%s: unable to delete "DFID" from orphan list: "
+                              "rc = %d\n", lu_dev_name(mdd2lu_dev(mdd)),
+                              PFID(mdd_object_fid(mdd_obj)), rc);
+                       /* If object was not deleted from orphan list, do not
+                        * destroy OSS objects, which will be done when next
+                        * recovery. */
+                       GOTO(out, rc);
                }
 
-               rc = mdo_destroy(env, mdd_obj, handle);
+               CDEBUG(D_HA, "Object "DFID" is deleted from orphan "
+                      "list, OSS objects to be destroyed.\n",
+                      PFID(mdd_object_fid(mdd_obj)));
+       }
 
-                if (rc != 0)
-                        CERROR("Error when prepare to delete Object "DFID" , "
-                               "which will cause OST objects can not be "
-                               "destroyed.\n",  PFID(mdd_object_fid(mdd_obj)));
-        }
-        EXIT;
+       rc = mdo_destroy(env, mdd_obj, handle);
 
-out:
+       if (rc != 0) {
+               CERROR("%s: unable to delete "DFID" from orphan list: "
+                      "rc = %d\n", lu_dev_name(mdd2lu_dev(mdd)),
+                      PFID(mdd_object_fid(mdd_obj)), rc);
+       }
+       EXIT;
 
+out:
         mdd_write_unlock(env, mdd_obj);
 
         if (rc == 0 &&