LU-8753 llog: remove lgh_write_offset

Remove lgh_write_offset for remote llog object,
since it is not easy to be refreshed when remote
transaction fails, which might cause llog file
corruption. Instead let's rely on the osp object
cache to invalidate and update the llog object
size.

Signed-off-by: Di Wang <di.wang@intel.com>
Change-Id: I38aa9ce2c5ac6485bf0cf6be023b8edde986b5d3
Reviewed-on: https://review.whamcloud.com/24008
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Fan Yong <fan.yong@intel.com>
Reviewed-by: Lai Siyao <lai.siyao@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index 668b73e..237da21 100644 (file)
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -273,9 +273,6 @@ struct llog_handle {
        int                      lgh_last_idx;
        int                      lgh_cur_idx; /* used during llog_process */
        __u64                    lgh_cur_offset; /* used during llog_process */
-       /* used during llog_osd_write_rec */
-       __u64                    lgh_write_offset;
-       int                      lgh_max_size;
        struct llog_ctxt        *lgh_ctxt;
        union {
                struct plain_handle_data         phd;
@@ -286,6 +283,7 @@ struct llog_handle {
        struct llog_operations  *lgh_logops;
        atomic_t                 lgh_refcount;
 
+       int                     lgh_max_size;
        __u32                   lgh_stale:1;
 };
 

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index b5aa8a3..80e4f12 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -316,7 +316,6 @@ int llog_read_header(const struct lu_env *env, struct llog_handle *handle,
 
                /* lrh_len should be initialized in llog_init_handle */
                handle->lgh_last_idx = 0; /* header is record with index 0 */
-               handle->lgh_write_offset = 0;
                llh->llh_count = 1;         /* for the header record */
                llh->llh_hdr.lrh_type = LLOG_HDR_MAGIC;
                LASSERT(handle->lgh_ctxt->loc_chunk_size >=

diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c
index 32d74d8..551a949 100644 (file)
--- a/lustre/obdclass/llog_osd.c
+++ b/lustre/obdclass/llog_osd.c
@@ -286,7 +286,6 @@ static int llog_osd_read_header(const struct lu_env *env,
 
        handle->lgh_hdr->llh_flags |= (flags & LLOG_F_EXT_MASK);
        handle->lgh_last_idx = LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_index;
-       handle->lgh_write_offset = lgi->lgi_attr.la_size;
 
        RETURN(0);
 }
@@ -390,7 +389,6 @@ static int llog_osd_write_rec(const struct lu_env *env,
        __u32                   chunk_size;
        size_t                   left;
        __u32                   orig_last_idx;
-       __u64                   orig_write_offset;
        ENTRY;
 
        llh = loghandle->lgh_hdr;
@@ -556,7 +554,6 @@ static int llog_osd_write_rec(const struct lu_env *env,
 
        LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
        orig_last_idx = loghandle->lgh_last_idx;
-       orig_write_offset = loghandle->lgh_write_offset;
        lgi->lgi_off = lgi->lgi_attr.la_size;
 
        if (loghandle->lgh_max_size > 0 &&
@@ -580,9 +577,6 @@ static int llog_osd_write_rec(const struct lu_env *env,
                if (rc)
                        RETURN(rc);
 
-               if (dt_object_remote(o))
-                       loghandle->lgh_write_offset = lgi->lgi_off;
-
                loghandle->lgh_last_idx++; /* for pad rec */
        }
        /* if it's the last idx in log file, then return -ENOSPC
@@ -684,9 +678,6 @@ out_unlock:
         * records. This also allows to handle Catalog wrap around case */
        if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
                lgi->lgi_off = llh->llh_hdr.lrh_len + (index - 1) * reclen;
-       } else if (dt_object_remote(o)) {
-               lgi->lgi_off = max_t(__u64, loghandle->lgh_write_offset,
-                                    lgi->lgi_off);
        } else {
                rc = dt_attr_get(env, o, &lgi->lgi_attr);
                if (rc)
@@ -703,9 +694,6 @@ out_unlock:
        if (rc < 0)
                GOTO(out, rc);
 
-       if (dt_object_remote(o))
-               loghandle->lgh_write_offset = lgi->lgi_off;
-
        CDEBUG(D_HA, "added record "DFID": idx: %u, %u off%llu\n",
               PFID(lu_object_fid(&o->do_lu)), index, rec->lrh_len,
               lgi->lgi_off);
@@ -732,7 +720,6 @@ out:
        /* restore llog last_idx */
        if (dt_object_remote(o)) {
                loghandle->lgh_last_idx = orig_last_idx;
-               loghandle->lgh_write_offset = orig_write_offset;
        } else if (--loghandle->lgh_last_idx == 0 &&
            (llh->llh_flags & LLOG_F_IS_CAT) && llh->llh_cat_idx != 0) {
                /* catalog had just wrap-around case */
@@ -936,11 +923,13 @@ static int llog_osd_next_block(const struct lu_env *env,
 
                if (last_rec->lrh_index != tail->lrt_index) {
                        CERROR("%s: invalid llog tail at log id "DOSTID"/%u "
-                              "offset %llu last_rec idx %u tail idx %u\n",
+                              "offset %llu last_rec idx %u tail idx %u"
+                              "lrt len %u read_size %d\n",
                               o->do_lu.lo_dev->ld_obd->obd_name,
                               POSTID(&loghandle->lgh_id.lgl_oi),
                               loghandle->lgh_id.lgl_ogen, *cur_offset,
-                              last_rec->lrh_index, tail->lrt_index);
+                              last_rec->lrh_index, tail->lrt_index,
+                              tail->lrt_len, rc);
                        GOTO(out, rc = -EINVAL);
                }
 

diff --git a/lustre/osp/osp_object.c b/lustre/osp/osp_object.c
index 84dda71..1fcbac7 100644 (file)
--- a/lustre/osp/osp_object.c
+++ b/lustre/osp/osp_object.c
@@ -1243,6 +1243,8 @@ int osp_invalidate(const struct lu_env *env, struct dt_object *dt)
        struct osp_object *obj = dt2osp_obj(dt);
        ENTRY;
 
+       CDEBUG(D_HA, "Invalidate osp_object "DFID"\n",
+              PFID(lu_object_fid(&dt->do_lu)));
        osp_obj_invalidate_cache(obj);
 
        spin_lock(&obj->opo_lock);