LU-15481 llog: Add LLOG_SKIP_PLAIN to skip llog plain

Add the catalog callback return LLOG_SKIP_PLAIN to conditionally skip
an entire llog plain.

This could speedup the catalog processing for specific usages when a
record need to be access in the "middle" of the catalog. This could
be usefull for changelog with several users or HSM.

This patch modify chlg_read_cat_process_cb() to use LLOG_SKIP_PLAIN.
The main idea came from: d813c75d ("LU-14688 mdt: changelog purge
deletes plain llog")

**Performance test:**

* Environement:
2474195 changelogs record store on the mds0 (40 llog plain):
mds# lctl get_param -n mdd.lustrefs-MDT0000.changelog_users
current index: 2474195
ID    index (idle seconds)
cl1   0 (3509)

* Test
Access to records at the end of the catalog (offset: 2474194):
client# time lfs changelog lustrefs-MDT0000 2474194 >/dev/null

* Results
- with the patch: 	real    0m0.592s
- without the patch:	real    0m17.835s (x30)

Signed-off-by: Etienne AUJAMES <etienne.aujames@cea.fr>
Change-Id: I887d5bef1f3a6a31c46bc58959e0f508266c53d2
Reviewed-on: https://review.whamcloud.com/46310
Reviewed-by: Alexander Boyko <alexander.boyko@hpe.com>
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index 360ba26..1fdca73 100644 (file)
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -340,6 +340,7 @@ struct llog_ctxt {
 #define LLOG_PROC_BREAK 0x0001
 #define LLOG_DEL_RECORD 0x0002
 #define LLOG_DEL_PLAIN  0x0003
+#define LLOG_SKIP_PLAIN 0x0004
 
 static inline int llog_obd2ops(struct llog_ctxt *ctxt,
                               const struct llog_operations **lop)
@@ -519,6 +520,22 @@ static inline int llog_is_full(struct llog_handle *llh)
        return llh->lgh_last_idx >= LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - 1;
 }
 
+/* Determine if a llog plain of a catalog could be skiped based on record
+ * custom indexes.
+ * This assumes that indexes follow each other. The number of records to skip
+ * can be computed base on a starting offset and the index of the current
+ * record (in llog catalog callback).
+ */
+static inline int llog_is_plain_skipable(struct llog_log_hdr *lh,
+                                        struct llog_rec_hdr *rec,
+                                        __u64 curr, __u64 start)
+{
+       if (start == 0 || curr >= start)
+               return 0;
+
+       return (LLOG_HDR_BITMAP_SIZE(lh) - rec->lrh_index) < (start - curr);
+}
+
 struct llog_cfg_rec {
        struct llog_rec_hdr     lcr_hdr;
        struct lustre_cfg       lcr_cfg;

diff --git a/lustre/mdc/mdc_changelog.c b/lustre/mdc/mdc_changelog.c
index 853bee2..c3a65f0 100644 (file)
--- a/lustre/mdc/mdc_changelog.c
+++ b/lustre/mdc/mdc_changelog.c
@@ -228,6 +228,11 @@ static int chlg_read_cat_process_cb(const struct lu_env *env,
                RETURN(rc);
        }
 
+       /* Check if we can skip the entire llog plain */
+       if (llog_is_plain_skipable(llh->lgh_hdr, hdr, rec->cr.cr_index,
+                                  crs->crs_start_offset))
+               RETURN(LLOG_SKIP_PLAIN);
+
        /* Skip undesired records */
        if (rec->cr.cr_index < crs->crs_start_offset)
                RETURN(0);

diff --git a/lustre/mdd/mdd_device.c b/lustre/mdd/mdd_device.c
index 37667af..94fb1f6 100644 (file)
--- a/lustre/mdd/mdd_device.c
+++ b/lustre/mdd/mdd_device.c
@@ -393,8 +393,8 @@ static int llog_changelog_cancel_cb(const struct lu_env *env,
         * last cr_index at this plain llog. And if it less then cookie endrec
         * cancel the whole file.
         */
-       if ((LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - hdr->lrh_index +
-            rec->cr.cr_index) < cl_cookie->endrec) {
+       if (llog_is_plain_skipable(llh->lgh_hdr, hdr, rec->cr.cr_index,
+                                  cl_cookie->endrec)) {
                int rc;
 
                if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_CHANGELOG_DEL))) {

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index 586ce9e..1af4ae5 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -735,7 +735,8 @@ repeat:
                                        lgc->lgc_index = tmp_idx;
                                }
 
-                               if (rc == LLOG_PROC_BREAK) {
+                               if (rc == LLOG_PROC_BREAK ||
+                                   rc == LLOG_SKIP_PLAIN) {
                                        GOTO(out, rc);
                                } else if (rc == LLOG_DEL_RECORD) {
                                        rc = llog_cancel_rec(lpi->lpi_env,
@@ -1003,7 +1004,8 @@ int llog_reverse_process(const struct lu_env *env,
                                      sizeof(*tail);
 
                                rc = cb(env, loghandle, rec, data);
-                               if (rc == LLOG_PROC_BREAK) {
+                               if (rc == LLOG_PROC_BREAK ||
+                                   rc == LLOG_SKIP_PLAIN) {
                                        GOTO(out, rc);
                                } else if (rc == LLOG_DEL_RECORD) {
                                        rc = llog_cancel_rec(env, loghandle,

diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c
index ba44ad3..4f5b0b1 100644 (file)
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -895,6 +895,9 @@ out:
        if (rc == LLOG_DEL_PLAIN || rc == LLOG_DEL_RECORD)
                /* clear wrong catalog entry */
                rc = llog_cat_cleanup(env, cat_llh, llh, rec->lrh_index);
+       else if (rc == LLOG_SKIP_PLAIN)
+               /* processing callback ask to skip the llog -> continue */
+               rc = 0;
 
        if (llh)
                llog_handle_put(env, llh);
@@ -1060,6 +1063,9 @@ static int llog_cat_reverse_process_cb(const struct lu_env *env,
        } else if (rc == LLOG_DEL_RECORD) {
                /* clear wrong catalog entry */
                rc = llog_cat_cleanup(env, cat_llh, NULL, rec->lrh_index);
+       } else if (rc == LLOG_SKIP_PLAIN) {
+               /* processing callback ask to skip the llog -> continue */
+               rc = 0;
        }
        if (rc)
                RETURN(rc);