LU-8882 osd: use bydnode methods to access DMU

[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 05eb8a4..b2e89c8 100644 (file)
--- a/lustre/lod/lod_sub_object.c
+++ b/lustre/lod/lod_sub_object.c
@@ -20,7 +20,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2014, Intel Corporation.
+ * Copyright (c) 2015, Intel Corporation.
  */
 /*
  * lustre/lod/lod_sub_object.c
@@ -95,15 +95,19 @@ struct thandle *lod_sub_get_thandle(const struct lu_env *env,
        if (rc < 0)
                RETURN(ERR_PTR(rc));
 
-       if (type == LU_SEQ_RANGE_OST)
+       /* th_complex means we need track all of updates for this
+        * transaction, include changes on OST */
+       if (type == LU_SEQ_RANGE_OST && !th->th_complex)
                RETURN(tth->tt_master_sub_thandle);
 
+       sub_th = thandle_get_sub(env, th, sub_obj);
+       if (IS_ERR(sub_th))
+               RETURN(sub_th);
+
        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);
-
        RETURN(sub_th);
 }
 
@@ -130,11 +134,16 @@ int lod_sub_object_declare_create(const struct lu_env *env,
                                  struct thandle *th)
 {
        struct thandle *sub_th;
+       bool record_update;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                return PTR_ERR(sub_th);
 
+       if (record_update)
+               update_record_size(env, create, th, lu_object_fid(&dt->do_lu),
+                                  attr, hint, dof);
+
        return dt_declare_create(env, dt, attr, hint, dof, sub_th);
 }
 
@@ -199,13 +208,17 @@ int lod_sub_object_declare_ref_add(const struct lu_env *env,
                                   struct thandle *th)
 {
        struct thandle  *sub_th;
+       bool            record_update;
        int             rc;
        ENTRY;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                RETURN(PTR_ERR(sub_th));
 
+       if (record_update)
+               update_record_size(env, ref_add, th, lu_object_fid(&dt->do_lu));
+
        rc = dt_declare_ref_add(env, dt, sub_th);
 
        RETURN(rc);
@@ -265,13 +278,17 @@ int lod_sub_object_declare_ref_del(const struct lu_env *env,
                                   struct thandle *th)
 {
        struct thandle  *sub_th;
+       bool            record_update;
        int             rc;
        ENTRY;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                RETURN(PTR_ERR(sub_th));
 
+       if (record_update)
+               update_record_size(env, ref_del, th, lu_object_fid(&dt->do_lu));
+
        rc = dt_declare_ref_del(env, dt, sub_th);
 
        RETURN(rc);
@@ -331,13 +348,18 @@ int lod_sub_object_declare_destroy(const struct lu_env *env,
                                   struct thandle *th)
 {
        struct thandle  *sub_th;
+       bool            record_update;
        int             rc;
        ENTRY;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                RETURN(PTR_ERR(sub_th));
 
+       if (record_update)
+               update_record_size(env, object_destroy, th,
+                                  lu_object_fid(&dt->do_lu));
+
        rc = dt_declare_destroy(env, dt, sub_th);
 
        RETURN(rc);
@@ -401,11 +423,16 @@ int lod_sub_object_declare_insert(const struct lu_env *env,
                                  struct thandle *th)
 {
        struct thandle *sub_th;
+       bool            record_update;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                return PTR_ERR(sub_th);
 
+       if (record_update)
+               update_record_size(env, index_insert, th,
+                                  lu_object_fid(&dt->do_lu), rec, key);
+
        return dt_declare_insert(env, dt, rec, key, sub_th);
 }
 
@@ -467,11 +494,16 @@ int lod_sub_object_declare_delete(const struct lu_env *env,
                                  struct thandle *th)
 {
        struct thandle *sub_th;
+       bool            record_update;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                return PTR_ERR(sub_th);
 
+       if (record_update)
+               update_record_size(env, index_delete, th,
+                                  lu_object_fid(&dt->do_lu), key);
+
        return dt_declare_delete(env, dt, key, sub_th);
 }
 
@@ -533,13 +565,19 @@ int lod_sub_object_declare_xattr_set(const struct lu_env *env,
                                     struct thandle *th)
 {
        struct thandle  *sub_th;
+       bool            record_update;
        int             rc;
        ENTRY;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                RETURN(PTR_ERR(sub_th));
 
+       if (record_update)
+               update_record_size(env, xattr_set, th,
+                                  lu_object_fid(&dt->do_lu),
+                                  buf, name, fl);
+
        rc = dt_declare_xattr_set(env, dt, buf, name, fl, sub_th);
 
        RETURN(rc);
@@ -606,13 +644,18 @@ int lod_sub_object_declare_attr_set(const struct lu_env *env,
                                    struct thandle *th)
 {
        struct thandle  *sub_th;
+       bool            record_update;
        int             rc;
        ENTRY;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                RETURN(PTR_ERR(sub_th));
 
+       if (record_update)
+               update_record_size(env, attr_set, th,
+                                  lu_object_fid(&dt->do_lu), attr);
+
        rc = dt_declare_attr_set(env, dt, attr, sub_th);
 
        RETURN(rc);
@@ -677,13 +720,19 @@ int lod_sub_object_declare_xattr_del(const struct lu_env *env,
                                     struct thandle *th)
 {
        struct thandle  *sub_th;
+       bool            record_update;
        int             rc;
        ENTRY;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                RETURN(PTR_ERR(sub_th));
 
+       if (record_update)
+               update_record_size(env, xattr_del, th,
+                                  lu_object_fid(&dt->do_lu),
+                                  name);
+
        rc = dt_declare_xattr_del(env, dt, name, sub_th);
 
        RETURN(rc);
@@ -749,13 +798,19 @@ int lod_sub_object_declare_write(const struct lu_env *env,
                                 struct thandle *th)
 {
        struct thandle  *sub_th;
+       bool            record_update;
        int             rc;
        ENTRY;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                RETURN(PTR_ERR(sub_th));
 
+       if (record_update)
+               update_record_size(env, write, th,
+                                  lu_object_fid(&dt->do_lu),
+                                  buf, pos);
+
        rc = dt_declare_write(env, dt, buf, pos, sub_th);
 
        RETURN(rc);
@@ -821,13 +876,19 @@ int lod_sub_object_declare_punch(const struct lu_env *env,
                                 struct thandle *th)
 {
        struct thandle  *sub_th;
+       bool            record_update;
        int             rc;
        ENTRY;
 
-       sub_th = lod_sub_get_thandle(env, th, dt, NULL);
+       sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
        if (IS_ERR(sub_th))
                RETURN(PTR_ERR(sub_th));
 
+       if (record_update)
+               update_record_size(env, punch, th,
+                                  lu_object_fid(&dt->do_lu),
+                                  start, end);
+
        rc = dt_declare_punch(env, dt, start, end, sub_th);
 
        RETURN(rc);
@@ -903,7 +964,7 @@ int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
        ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
        LASSERT(ctxt != NULL);
        ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
-
+       ctxt->loc_chunk_size = LLOG_MIN_CHUNK_SIZE * 4;
        if (likely(logid_id(&cid->lci_logid) != 0)) {
                rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
                               LLOG_OPEN_EXISTS);
@@ -924,9 +985,8 @@ int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
        }
 
        LASSERT(lgh != NULL);
-       ctxt->loc_handle = lgh;
 
-       rc = llog_cat_init_and_process(env, lgh);
+       rc = llog_init_handle(env, lgh, LLOG_F_IS_CAT, NULL);
        if (rc != 0)
                GOTO(out_close, rc);
 
@@ -936,16 +996,15 @@ int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
                        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),
+       ctxt->loc_handle = lgh;
+
+       CDEBUG(D_INFO, "%s: init llog for index %d - catid "DFID":%x\n",
+              obd->obd_name, index, PFID(&cid->lci_logid.lgl_oi.oi_fid),
               cid->lci_logid.lgl_ogen);
 out_close:
-       if (rc != 0) {
-               llog_cat_close(env, ctxt->loc_handle);
-               ctxt->loc_handle = NULL;
-       }
+       if (rc != 0)
+               llog_cat_close(env, lgh);
 out_put:
        llog_ctxt_put(ctxt);
        RETURN(rc);
 }
-