LU-4778 llog: update catlog when plain log destroyed

When the empty plain log is destroyed while processing
catlog, we should call llog_cat_cleanup() to close the
plain log and update in catlog.

Intel-bug-id: INTL-64

Signed-off-by: Niu Yawei <yawei.niu@intel.com>
Change-Id: I97b1d5cc59e84fdb4888274b48f223a063293647
Reviewed-on: http://review.whamcloud.com/9702
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Mike Pershin <mike.pershin@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index 6f490eb..2dd6ad4 100644 (file)
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -355,6 +355,7 @@ struct llog_ctxt {
 
 #define LLOG_PROC_BREAK 0x0001
 #define LLOG_DEL_RECORD 0x0002
+#define LLOG_DEL_PLAIN  0x0003
 
 static inline int llog_obd2ops(struct llog_ctxt *ctxt,
                                struct llog_operations **lop)

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index 7ff0dd7..e972fcd 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -146,7 +146,7 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
                               loghandle->lgh_id.lgl_ogen, rc);
                        GOTO(out_err, rc);
                }
-               RETURN(1);
+               RETURN(LLOG_DEL_PLAIN);
        }
        spin_unlock(&loghandle->lgh_hdr_lock);
 
@@ -378,10 +378,9 @@ repeat:
                                if (rc == LLOG_PROC_BREAK) {
                                        GOTO(out, rc);
                                } else if (rc == LLOG_DEL_RECORD) {
-                                       llog_cancel_rec(lpi->lpi_env,
-                                                       loghandle,
-                                                       rec->lrh_index);
-                                        rc = 0;
+                                       rc = llog_cancel_rec(lpi->lpi_env,
+                                                            loghandle,
+                                                            rec->lrh_index);
                                 }
                                 if (rc)
                                         GOTO(out, rc);
@@ -494,7 +493,9 @@ EXPORT_SYMBOL(llog_process_or_fork);
 int llog_process(const struct lu_env *env, struct llog_handle *loghandle,
                 llog_cb_t cb, void *data, void *catdata)
 {
-       return llog_process_or_fork(env, loghandle, cb, data, catdata, true);
+       int rc;
+       rc = llog_process_or_fork(env, loghandle, cb, data, catdata, true);
+       return rc == LLOG_DEL_PLAIN ? 0 : rc;
 }
 EXPORT_SYMBOL(llog_process);
 
@@ -565,9 +566,8 @@ int llog_reverse_process(const struct lu_env *env,
                                if (rc == LLOG_PROC_BREAK) {
                                        GOTO(out, rc);
                                } else if (rc == LLOG_DEL_RECORD) {
-                                       llog_cancel_rec(env, loghandle,
-                                                       tail->lrt_index);
-                                       rc = 0;
+                                       rc = llog_cancel_rec(env, loghandle,
+                                                            tail->lrt_index);
                                }
                                 if (rc)
                                         GOTO(out, rc);

diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c
index a0d7854..fc7cd13 100644 (file)
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -510,7 +510,7 @@ int llog_cat_cancel_records(const struct lu_env *env,
                }
 
                lrc = llog_cancel_rec(env, loghandle, cookies->lgc_index);
-               if (lrc == 1) {          /* log has been destroyed */
+               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);
@@ -574,6 +574,10 @@ int llog_cat_process_cb(const struct lu_env *env, struct llog_handle *cat_llh,
                                          NULL, false);
        }
 
+       /* 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);
        llog_handle_put(llh);
 
        RETURN(rc);
@@ -605,11 +609,11 @@ int llog_cat_process_or_fork(const struct lu_env *env,
                 cd.lpcd_last_idx = 0;
                rc = llog_process_or_fork(env, cat_llh, llog_cat_process_cb,
                                          &d, &cd, fork);
-                if (rc != 0)
-                        RETURN(rc);
+               if (rc != 0)
+                       RETURN(rc);
 
-                cd.lpcd_first_idx = 0;
-                cd.lpcd_last_idx = cat_llh->lgh_last_idx;
+               cd.lpcd_first_idx = 0;
+               cd.lpcd_last_idx = cat_llh->lgh_last_idx;
                rc = llog_process_or_fork(env, cat_llh, llog_cat_process_cb,
                                          &d, &cd, fork);
         } else {
@@ -655,6 +659,12 @@ static int llog_cat_reverse_process_cb(const struct lu_env *env,
        }
 
        rc = llog_reverse_process(env, llh, d->lpd_cb, d->lpd_data, NULL);
+
+       /* The empty plain was destroyed while processing */
+       if (rc == LLOG_DEL_PLAIN)
+               rc = llog_cat_cleanup(env, cat_llh, llh,
+                                     llh->u.phd.phd_cookie.lgc_index);
+
        llog_handle_put(llh);
        RETURN(rc);
 }
@@ -682,11 +692,11 @@ int llog_cat_reverse_process(const struct lu_env *env,
                rc = llog_reverse_process(env, cat_llh,
                                          llog_cat_reverse_process_cb,
                                          &d, &cd);
-                if (rc != 0)
-                        RETURN(rc);
+               if (rc != 0)
+                       RETURN(rc);
 
-                cd.lpcd_first_idx = le32_to_cpu(llh->llh_cat_idx);
-                cd.lpcd_last_idx = 0;
+               cd.lpcd_first_idx = le32_to_cpu(llh->llh_cat_idx);
+               cd.lpcd_last_idx = 0;
                rc = llog_reverse_process(env, cat_llh,
                                          llog_cat_reverse_process_cb,
                                          &d, &cd);

diff --git a/lustre/obdclass/llog_test.c b/lustre/obdclass/llog_test.c
index 3a91918..a1051c2 100644 (file)
--- a/lustre/obdclass/llog_test.c
+++ b/lustre/obdclass/llog_test.c
@@ -784,7 +784,7 @@ static int llog_test_7_sub(const struct lu_env *env, struct llog_ctxt *ctxt)
 
        plain_counter = 0;
        rc = llog_reverse_process(env, llh, test_7_cancel_cb, "test 7", NULL);
-       if (rc) {
+       if (rc && rc != LLOG_DEL_PLAIN) {
                CERROR("7_sub: reverse llog process failed: %d\n", rc);
                GOTO(out_close, rc);
        }