LU-3536 lod: cancel update log after all committed

[fs/lustre-release.git] / lustre / lod / lod_sub_object.c

diff --git a/lustre/lod/lod_sub_object.c b/lustre/lod/lod_sub_object.c
index a703944..05eb8a4 100644 (file)
--- a/lustre/lod/lod_sub_object.c
+++ b/lustre/lod/lod_sub_object.c
@@ -50,6 +50,7 @@
 #include <lustre_param.h>
 #include <md_object.h>
 #include <lustre_linkea.h>
+#include <lustre_log.h>
 
 #include "lod_internal.h"
 
@@ -73,7 +74,7 @@ struct thandle *lod_sub_get_thandle(const struct lu_env *env,
                RETURN(th);
 
        tth = container_of(th, struct top_thandle, tt_super);
-       LASSERT(tth->tt_magic == TOP_THANDLE_MAGIC);
+
        /* local object must be mdt object, Note: during ost object
         * creation, FID is not assigned until osp_object_create(),
         * so if the FID of sub_obj is zero, it means OST object. */
@@ -81,8 +82,9 @@ struct thandle *lod_sub_get_thandle(const struct lu_env *env,
            fid_is_zero(lu_object_fid(&sub_obj->do_lu))) {
                /* local MDT object */
                if (fid_is_sane(lu_object_fid(&sub_obj->do_lu)) &&
-                   tth->tt_update_records != NULL &&
-                   record_update != NULL)
+                   tth->tt_multiple_thandle != NULL &&
+                   record_update != NULL &&
+                   th->th_result == 0)
                        *record_update = true;
 
                RETURN(tth->tt_master_sub_thandle);
@@ -96,7 +98,8 @@ struct thandle *lod_sub_get_thandle(const struct lu_env *env,
        if (type == LU_SEQ_RANGE_OST)
                RETURN(tth->tt_master_sub_thandle);
 
-       if (tth->tt_update_records != NULL && record_update != NULL)
+       if (tth->tt_multiple_thandle != NULL && record_update != NULL &&
+           th->th_result == 0)
                *record_update = true;
 
        sub_th = thandle_get_sub(env, th, sub_obj);
@@ -869,3 +872,80 @@ int lod_sub_object_punch(const struct lu_env *env, struct dt_object *dt,
 
        RETURN(rc);
 }
+
+int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
+                     struct dt_device *dt, int index)
+{
+       struct lod_thread_info  *lti = lod_env_info(env);
+       struct llog_ctxt        *ctxt;
+       struct llog_handle      *lgh;
+       struct llog_catid       *cid = &lti->lti_cid;
+       struct lu_fid           *fid = &lti->lti_fid;
+       struct obd_device       *obd;
+       int                     rc;
+       bool                    need_put = false;
+       ENTRY;
+
+       lu_update_log_fid(fid, index);
+
+       rc = lodname2mdt_index(lod2obd(lod)->obd_name, (__u32 *)&index);
+       if (rc < 0)
+               RETURN(rc);
+
+       rc = llog_osd_get_cat_list(env, dt, index, 1, cid, fid);
+       if (rc != 0) {
+               CERROR("%s: can't get id from catalogs: rc = %d\n",
+                      lod2obd(lod)->obd_name, rc);
+               RETURN(rc);
+       }
+
+       obd = dt->dd_lu_dev.ld_obd;
+       ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
+       LASSERT(ctxt != NULL);
+       ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
+
+       if (likely(logid_id(&cid->lci_logid) != 0)) {
+               rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
+                              LLOG_OPEN_EXISTS);
+
+               /* re-create llog if it is missing */
+               if (rc == -ENOENT)
+                       logid_set_id(&cid->lci_logid, 0);
+               else if (rc < 0)
+                       GOTO(out_put, rc);
+       }
+
+       if (unlikely(logid_id(&cid->lci_logid) == 0)) {
+               rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
+               if (rc < 0)
+                       GOTO(out_put, rc);
+               cid->lci_logid = lgh->lgh_id;
+               need_put = true;
+       }
+
+       LASSERT(lgh != NULL);
+       ctxt->loc_handle = lgh;
+
+       rc = llog_cat_init_and_process(env, lgh);
+       if (rc != 0)
+               GOTO(out_close, rc);
+
+       if (need_put) {
+               rc = llog_osd_put_cat_list(env, dt, index, 1, cid, fid);
+               if (rc != 0)
+                       GOTO(out_close, rc);
+       }
+
+       CDEBUG(D_INFO, "%s: Init llog for %d - catid "DOSTID":%x\n",
+              obd->obd_name, index, POSTID(&cid->lci_logid.lgl_oi),
+              cid->lci_logid.lgl_ogen);
+out_close:
+       if (rc != 0) {
+               llog_cat_close(env, ctxt->loc_handle);
+               ctxt->loc_handle = NULL;
+       }
+out_put:
+       llog_ctxt_put(ctxt);
+       RETURN(rc);
+}
+