LU-7340 mdd: changelogs garbage collection

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

diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c
index 950445a..1628cc8 100644 (file)
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -85,8 +85,18 @@ static int llog_cat_new_log(const struct lu_env *env,
         * last_idx and cat_idx */
        if ((index == llh->llh_cat_idx + 1 && llh->llh_count > 1) ||
            (index == 0 && llh->llh_cat_idx == 0)) {
-               CWARN("%s: there are no more free slots in catalog\n",
-                     loghandle->lgh_ctxt->loc_obd->obd_name);
+               if (cathandle->lgh_name == NULL) {
+                       CWARN("%s: there are no more free slots in catalog "
+                             DFID":%x\n",
+                             loghandle->lgh_ctxt->loc_obd->obd_name,
+                             PFID(&cathandle->lgh_id.lgl_oi.oi_fid),
+                             cathandle->lgh_id.lgl_ogen);
+               } else {
+                       CWARN("%s: there are no more free slots in "
+                             "catalog %s\n",
+                             loghandle->lgh_ctxt->loc_obd->obd_name,
+                             cathandle->lgh_name);
+               }
                RETURN(-ENOSPC);
        }
 
@@ -744,16 +754,38 @@ int llog_cat_cancel_records(const struct lu_env *env,
                        continue;
                }
 
+               if ((cathandle->lgh_ctxt->loc_flags &
+                    LLOG_CTXT_FLAG_NORMAL_FID) && !llog_exist(loghandle)) {
+                       /* For update log, some of loghandles of cathandle
+                        * might not exist because remote llog creation might
+                        * be failed, so let's skip the record cancellation
+                        * for these non-exist llogs.
+                        */
+                       lrc = -ENOENT;
+                       CDEBUG(D_HA, "%s: llog "DFID":%x does not exist"
+                              ": rc = %d\n",
+                              cathandle->lgh_ctxt->loc_obd->obd_name,
+                              PFID(&lgl->lgl_oi.oi_fid), lgl->lgl_ogen, lrc);
+                       failed++;
+                       if (rc == 0)
+                               rc = lrc;
+                       continue;
+               }
+
                lrc = llog_cancel_rec(env, loghandle, cookies->lgc_index);
                if (lrc == LLOG_DEL_PLAIN) { /* log has been destroyed */
                        index = loghandle->u.phd.phd_cookie.lgc_index;
-                       rc = llog_cat_cleanup(env, cathandle, loghandle, index);
+                       lrc = llog_cat_cleanup(env, cathandle, loghandle,
+                                              index);
+                       if (rc == 0)
+                               rc = lrc;
                } else if (lrc == -ENOENT) {
                        if (rc == 0) /* ENOENT shouldn't rewrite any error */
                                rc = lrc;
                } else if (lrc < 0) {
                        failed++;
-                       rc = lrc;
+                       if (rc == 0)
+                               rc = lrc;
                }
                llog_handle_put(loghandle);
        }
@@ -796,9 +828,8 @@ static int llog_cat_process_common(const struct lu_env *env,
                 * deletion are not atomic. So we end up with an index but no
                 * actual record. Destroy the index and move on. */
                if (rc == -ENOENT || rc == -ESTALE)
-                       rc = llog_cat_cleanup(env, cat_llh, NULL,
-                                             rec->lrh_index);
-               if (rc)
+                       rc = LLOG_DEL_RECORD;
+               else if (rc)
                        CWARN("%s: can't find llog handle "DFID":%x: rc = %d\n",
                              cat_llh->lgh_ctxt->loc_obd->obd_name,
                              PFID(&lir->lid_id.lgl_oi.oi_fid),
@@ -856,9 +887,14 @@ static int llog_cat_process_cb(const struct lu_env *env,
 
 out:
        /* The empty plain log was destroyed while processing */
-       if (rc == LLOG_DEL_PLAIN)
+       if (rc == LLOG_DEL_PLAIN) {
                rc = llog_cat_cleanup(env, cat_llh, llh,
                                      llh->u.phd.phd_cookie.lgc_index);
+       } else if (rc == LLOG_DEL_RECORD) {
+               /* clear wrong catalog entry */
+               rc = llog_cat_cleanup(env, cat_llh, NULL, rec->lrh_index);
+       }
+
        if (llh)
                llog_handle_put(llh);
 
@@ -928,13 +964,14 @@ static int llog_cat_size_cb(const struct lu_env *env,
 
        ENTRY;
        rc = llog_cat_process_common(env, cat_llh, rec, &llh);
-       if (llh == NULL)
-               RETURN(0);
 
        if (rc == LLOG_DEL_PLAIN) {
                /* empty log was deleted, don't count it */
                rc = llog_cat_cleanup(env, cat_llh, llh,
                                      llh->u.phd.phd_cookie.lgc_index);
+       } else if (rc == LLOG_DEL_RECORD) {
+               /* clear wrong catalog entry */
+               rc = llog_cat_cleanup(env, cat_llh, NULL, rec->lrh_index);
        } else {
                size = llog_size(env, llh);
                *cum_size += size;
@@ -943,7 +980,8 @@ static int llog_cat_size_cb(const struct lu_env *env,
                       PFID(&llh->lgh_id.lgl_oi.oi_fid), size, *cum_size);
        }
 
-       llog_handle_put(llh);
+       if (llh != NULL)
+               llog_handle_put(llh);
 
        RETURN(0);
 }
@@ -959,6 +997,28 @@ __u64 llog_cat_size(const struct lu_env *env, struct llog_handle *cat_llh)
 }
 EXPORT_SYMBOL(llog_cat_size);
 
+/* currently returns the number of "free" entries in catalog,
+ * ie the available entries for a new plain LLOG file creation,
+ * even if catalog has wrapped
+ */
+__u32 llog_cat_free_space(struct llog_handle *cat_llh)
+{
+       /* simulate almost full Catalog */
+       if (OBD_FAIL_CHECK(OBD_FAIL_CAT_FREE_RECORDS))
+               return cfs_fail_val;
+
+       if (cat_llh->lgh_hdr->llh_count == 1)
+               return LLOG_HDR_BITMAP_SIZE(cat_llh->lgh_hdr) - 1;
+
+       if (cat_llh->lgh_last_idx > cat_llh->lgh_hdr->llh_cat_idx)
+               return LLOG_HDR_BITMAP_SIZE(cat_llh->lgh_hdr) - 1 +
+                      cat_llh->lgh_hdr->llh_cat_idx - cat_llh->lgh_last_idx;
+
+       /* catalog is presently wrapped */
+       return cat_llh->lgh_hdr->llh_cat_idx - cat_llh->lgh_last_idx;
+}
+EXPORT_SYMBOL(llog_cat_free_space);
+
 static int llog_cat_reverse_process_cb(const struct lu_env *env,
                                       struct llog_handle *cat_llh,
                                       struct llog_rec_hdr *rec, void *data)
@@ -969,6 +1029,15 @@ static int llog_cat_reverse_process_cb(const struct lu_env *env,
 
        ENTRY;
        rc = llog_cat_process_common(env, cat_llh, rec, &llh);
+
+       /* The empty plain log was destroyed while processing */
+       if (rc == LLOG_DEL_PLAIN) {
+               rc = llog_cat_cleanup(env, cat_llh, llh,
+                                     llh->u.phd.phd_cookie.lgc_index);
+       } else if (rc == LLOG_DEL_RECORD) {
+               /* clear wrong catalog entry */
+               rc = llog_cat_cleanup(env, cat_llh, NULL, rec->lrh_index);
+       }
        if (rc)
                RETURN(rc);