LU-10198 llog: keep llog handle alive until last reference

[fs/lustre-release.git] / lustre / obdclass / llog_osd.c

diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c
index a374221..54f3dd4 100644 (file)
--- a/lustre/obdclass/llog_osd.c
+++ b/lustre/obdclass/llog_osd.c
@@ -44,6 +44,8 @@
 
 #define DEBUG_SUBSYSTEM S_LOG
 
+#include <linux/delay.h>
+
 #include <dt_object.h>
 #include <llog_swab.h>
 #include <lustre_fid.h>
@@ -124,8 +126,7 @@ static int llog_osd_create_new_object(const struct lu_env *env,
 static int llog_osd_exist(struct llog_handle *handle)
 {
        LASSERT(handle->lgh_obj);
-       return dt_object_exists(handle->lgh_obj) &&
-               !lu_object_is_dying(handle->lgh_obj->do_lu.lo_header);
+       return dt_object_exists(handle->lgh_obj) && !handle->lgh_destroyed;
 }
 
 static void *rec_tail(struct llog_rec_hdr *rec)
@@ -590,6 +591,7 @@ static int llog_osd_write_rec(const struct lu_env *env,
                        RETURN(-ENOSPC);
        }
 
+       down_write(&loghandle->lgh_last_sem);
        /* increment the last_idx along with llh_tail index, they should
         * be equal for a llog lifetime */
        loghandle->lgh_last_idx++;
@@ -673,6 +675,12 @@ out_unlock:
        if (rc)
                GOTO(out, rc);
 
+       if (OBD_FAIL_PRECHECK(OBD_FAIL_LLOG_PROCESS_TIMEOUT) &&
+          cfs_fail_val == (unsigned int)(loghandle->lgh_id.lgl_oi.oi.oi_id &
+                                         0xFFFFFFFF)) {
+               OBD_RACE(OBD_FAIL_LLOG_PROCESS_TIMEOUT);
+               msleep(1 * MSEC_PER_SEC);
+       }
        /* computed index can be used to determine offset for fixed-size
         * records. This also allows to handle Catalog wrap around case */
        if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
@@ -693,6 +701,8 @@ out_unlock:
        if (rc < 0)
                GOTO(out, rc);
 
+       up_write(&loghandle->lgh_last_sem);
+
        CDEBUG(D_HA, "added record "DFID".%u, %u off%llu\n",
               PFID(lu_object_fid(&o->do_lu)), index, rec->lrh_len,
               lgi->lgi_off);
@@ -726,6 +736,7 @@ out:
        }
 
        LLOG_HDR_TAIL(llh)->lrt_index = loghandle->lgh_last_idx;
+       up_write(&loghandle->lgh_last_sem);
 
        RETURN(rc);
 }
@@ -793,6 +804,8 @@ static void changelog_block_trim_ext(struct llog_rec_hdr *hdr,
        enum changelog_rec_flags flags = CLF_SUPPORTED;
        enum changelog_rec_extra_flags extra_flags = CLFE_SUPPORTED;
 
+       if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_XATTR))
+               extra_flags &= ~CLFE_XATTR;
        if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_OMODE))
                extra_flags &= ~CLFE_OPEN;
        if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_NID))
@@ -820,8 +833,26 @@ static void changelog_block_trim_ext(struct llog_rec_hdr *hdr,
                                extra_flags;
                }
 
+               if (unlikely(hdr->lrh_len == 0)) {
+                       /* It is corruption case, we cannot know the next rec,
+                        * jump to the last one directly to avoid dead loop. */
+                       LCONSOLE(D_WARNING, "Hit invalid llog record: "
+                                "idx %u, type %u, id %u\n",
+                                hdr->lrh_index, hdr->lrh_type, hdr->lrh_id);
+                       hdr = llog_rec_hdr_next(last_hdr);
+                       if (unlikely(hdr == last_hdr))
+                               LCONSOLE(D_WARNING, "The last record crashed: "
+                                        "idx %u, type %u, id %u\n",
+                                        hdr->lrh_index, hdr->lrh_type,
+                                        hdr->lrh_id);
+                       break;
+               }
+
                changelog_remap_rec(rec, rec->cr_flags & flags, xflag);
                hdr = llog_rec_hdr_next(hdr);
+               /* Yield CPU to avoid soft-lockup if there are too many records
+                * to be handled. */
+               cond_resched();
        } while ((char *)hdr <= (char *)last_hdr);
 }
 
@@ -871,7 +902,7 @@ static int llog_osd_next_block(const struct lu_env *env,
 
        o = loghandle->lgh_obj;
        LASSERT(o);
-       LASSERT(dt_object_exists(o));
+       LASSERT(llog_osd_exist(loghandle));
        dt = lu2dt_dev(o->do_lu.lo_dev);
        LASSERT(dt);
 
@@ -1045,7 +1076,7 @@ static int llog_osd_prev_block(const struct lu_env *env,
 
        o = loghandle->lgh_obj;
        LASSERT(o);
-       LASSERT(dt_object_exists(o));
+       LASSERT(llog_osd_exist(loghandle));
        dt = lu2dt_dev(o->do_lu.lo_dev);
        LASSERT(dt);
 
@@ -1411,7 +1442,7 @@ llog_osd_regular_fid_add_name_entry(const struct lu_env *env,
                               (struct dt_key *)name, th);
        } else {
                rc = dt_insert(env, dir, (struct dt_rec *)rec,
-                              (struct dt_key *)name, th, 1);
+                              (struct dt_key *)name, th);
        }
        dt_write_unlock(env, dir);
 
@@ -1578,8 +1609,7 @@ static int llog_osd_create(const struct lu_env *env, struct llog_handle *res,
                rec->rec_type = S_IFREG;
                dt_read_lock(env, llog_dir, 0);
                rc = dt_insert(env, llog_dir, (struct dt_rec *)rec,
-                              (struct dt_key *)res->lgh_name,
-                              th, 1);
+                              (struct dt_key *)res->lgh_name, th);
                dt_read_unlock(env, llog_dir);
                dt_object_put(env, llog_dir);
                if (rc)
@@ -1769,7 +1799,7 @@ static int llog_osd_destroy(const struct lu_env *env,
        LASSERT(o != NULL);
 
        dt_write_lock(env, o, 0);
-       if (!dt_object_exists(o))
+       if (!llog_osd_exist(loghandle))
                GOTO(out_unlock, rc = 0);
 
        if (loghandle->lgh_name) {
@@ -1795,6 +1825,7 @@ static int llog_osd_destroy(const struct lu_env *env,
        if (rc < 0)
                GOTO(out_unlock, rc);
 
+       loghandle->lgh_destroyed = true;
        if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
                rc = llog_osd_regular_fid_del_name_entry(env, o, th, false);
                if (rc < 0)