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 8771d68..1628cc8 100644 (file)
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -754,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);
        }
@@ -975,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)