LU-4970 lfsck: flush async updating before exit

[fs/lustre-release.git] / lustre / lfsck / lfsck_layout.c

diff --git a/lustre/lfsck/lfsck_layout.c b/lustre/lfsck/lfsck_layout.c
index 2345dc5..87acec9 100644 (file)
--- a/lustre/lfsck/lfsck_layout.c
+++ b/lustre/lfsck/lfsck_layout.c
@@ -39,7 +39,6 @@
 #include <lustre/lustre_idl.h>
 #include <lu_object.h>
 #include <dt_object.h>
-#include <lustre_linkea.h>
 #include <lustre_fid.h>
 #include <lustre_lib.h>
 #include <lustre_net.h>
@@ -1980,6 +1979,7 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
                                        __u32 ea_off)
 {
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
+       struct dt_insert_rec            *dtrec  = &info->lti_dt_rec;
        char                            *name   = info->lti_key;
        struct lu_attr                  *la     = &info->lti_la;
        struct dt_object_format         *dof    = &info->lti_dof;
@@ -1993,6 +1993,9 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
        struct lu_buf                   *pbuf   = NULL;
        struct lu_buf                   *ea_buf = &info->lti_big_buf;
        struct lustre_handle             lh     = { 0 };
+       struct linkea_data               ldata  = { 0 };
+       struct lu_buf                    linkea_buf;
+       const struct lu_name            *pname;
        int                              buflen = ea_buf->lb_len;
        int                              idx    = 0;
        int                              rc     = 0;
@@ -2040,6 +2043,16 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
                        GOTO(put, rc);
        } while (rc == 0);
 
+       rc = linkea_data_new(&ldata,
+                            &lfsck_env_info(env)->lti_linkea_buf);
+       if (rc != 0)
+               GOTO(put, rc);
+
+       pname = lfsck_name_get_const(env, name, strlen(name));
+       rc = linkea_add_buf(&ldata, pname, lfsck_dto2fid(lfsck->li_lpf_obj));
+       if (rc != 0)
+               GOTO(put, rc);
+
        memset(la, 0, sizeof(*la));
        la->la_uid = rec->lor_uid;
        la->la_gid = rec->lor_gid;
@@ -2096,12 +2109,22 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
                GOTO(stop, rc);
 
        /* 4a. Insert the MDT-object to .lustre/lost+found/MDTxxxx/ */
+       dtrec->rec_fid = pfid;
+       dtrec->rec_type = S_IFREG;
        rc = dt_declare_insert(env, lfsck->li_lpf_obj,
-                              (const struct dt_rec *)pfid,
+                              (const struct dt_rec *)dtrec,
                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 
+       /* 5a. insert linkEA for parent. */
+       linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
+       linkea_buf.lb_len = ldata.ld_leh->leh_len;
+       rc = dt_declare_xattr_set(env, pobj, &linkea_buf,
+                                 XATTR_NAME_LINK, 0, th);
+       if (rc != 0)
+               GOTO(stop, rc);
+
        rc = dt_trans_start(env, next, th);
        if (rc != 0)
                GOTO(stop, rc);
@@ -2127,9 +2150,14 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
                GOTO(stop, rc);
 
        /* 4b. Insert the MDT-object to .lustre/lost+found/MDTxxxx/ */
-       rc = dt_insert(env, lfsck->li_lpf_obj,
-                      (const struct dt_rec *)pfid,
+       rc = dt_insert(env, lfsck->li_lpf_obj, (const struct dt_rec *)dtrec,
                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
+       if (rc != 0)
+               GOTO(stop, rc);
+
+       /* 5b. insert linkEA for parent. */
+       rc = dt_xattr_set(env, pobj, &linkea_buf,
+                         XATTR_NAME_LINK, 0, th, BYPASS_CAPA);
 
        GOTO(stop, rc);
 
@@ -2444,6 +2472,7 @@ static int lfsck_layout_recreate_lovea(const struct lu_env *env,
        struct lfsck_bookmark    *bk            = &lfsck->li_bookmark_ram;
        struct thandle            *handle       = NULL;
        size_t                    buflen        = buf->lb_len;
+       size_t                    lovea_size;
        struct lov_mds_md_v1     *lmm;
        struct lov_ost_data_v1   *objs;
        struct lustre_handle      lh            = { 0 };
@@ -2482,8 +2511,9 @@ again:
        if (rc < 0)
                GOTO(unlock_layout, rc);
 
-       if (buf->lb_len < rc) {
-               lu_buf_realloc(buf, rc);
+       lovea_size = rc;
+       if (buf->lb_len < lovea_size) {
+               lu_buf_realloc(buf, lovea_size);
                buflen = buf->lb_len;
                if (buf->lb_buf == NULL)
                        GOTO(unlock_layout, rc = -ENOMEM);
@@ -2513,11 +2543,12 @@ again:
                LASSERT(rc != 0);
                goto again;
        } else if (rc == -ENODATA || rc == 0) {
-               rc = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
+               lovea_size = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
                /* If the declared is not big enough, re-try. */
-               if (buf->lb_len < rc)
+               if (buf->lb_len < lovea_size) {
+                       rc = lovea_size;
                        goto again;
-
+               }
                fl = LU_XATTR_CREATE;
        } else if (rc < 0) {
                GOTO(unlock_parent, rc);
@@ -2525,15 +2556,16 @@ again:
                goto again;
        } else {
                fl = LU_XATTR_REPLACE;
+               lovea_size = rc;
        }
 
        if (fl == LU_XATTR_CREATE) {
                if (bk->lb_param & LPF_DRYRUN)
                        GOTO(unlock_parent, rc = 1);
 
-               LASSERT(buf->lb_len >= rc);
+               LASSERT(buf->lb_len >= lovea_size);
 
-               buf->lb_len = rc;
+               buf->lb_len = lovea_size;
                rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
                                               buf, fl, ost_idx, ea_off, false);
 
@@ -2548,9 +2580,9 @@ again:
                if (bk->lb_param & LPF_DRYRUN)
                        GOTO(unlock_parent, rc = 1);
 
-               LASSERT(buf->lb_len >= rc);
+               LASSERT(buf->lb_len >= lovea_size);
 
-               buf->lb_len = rc;
+               buf->lb_len = lovea_size;
                memset(lmm, 0, buf->lb_len);
                rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
                                               buf, fl, ost_idx, ea_off, true);
@@ -2584,12 +2616,13 @@ again:
                if (bk->lb_param & LPF_DRYRUN)
                        GOTO(unlock_parent, rc = 1);
 
-               rc = lov_mds_md_size(ea_off + 1, magic);
+               lovea_size = lov_mds_md_size(ea_off + 1, magic);
                /* If the declared is not big enough, re-try. */
-               if (buf->lb_len < rc)
+               if (buf->lb_len < lovea_size) {
+                       rc = lovea_size;
                        goto again;
-
-               buf->lb_len = rc;
+               }
+               buf->lb_len = lovea_size;
                rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
                                               buf, fl, ost_idx, ea_off, false);
 
@@ -2598,7 +2631,7 @@ again:
 
        LASSERTF(rc > 0, "invalid rc = %d\n", rc);
 
-       buf->lb_len = rc;
+       buf->lb_len = lovea_size;
        for (i = 0; i < count; i++, objs++) {
                /* The MDT-object was created via lfsck_layout_recover_create()
                 * by others before, and we fill the dummy layout EA. */
@@ -3774,6 +3807,9 @@ cleanup2:
                rc = rc1;
        }
 
+       /* flush all async updating before exit. */
+       dt_sync(env, lfsck->li_next);
+
        /* Under force exit case, some requests may be just freed without
         * verification, those objects should be re-handled when next run.
         * So not update the on-disk tracing file under such case. */