LU-6838 llog: limit file size of plain logs

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

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index 68ee8fa..f2f2fc6 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -15,11 +15,7 @@
  *
  * You should have received a copy of the GNU General Public License
  * version 2 along with this program; If not, see
- * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
- *
- * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
- * CA 95054 USA or visit www.sun.com if you need additional information or
- * have any questions.
+ * http://www.gnu.org/licenses/gpl-2.0.html
  *
  * GPL HEADER END
  */
@@ -229,8 +225,11 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
        if (rc < 0)
                GOTO(out_trans, rc);
 
-       if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY))
+       if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY)) {
                rc = llog_declare_destroy(env, loghandle, th);
+               if (rc < 0)
+                       GOTO(out_trans, rc);
+       }
 
        th->th_wait_submit = 1;
        rc = dt_trans_start_local(env, dt, th);
@@ -262,6 +261,8 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
             (loghandle->u.phd.phd_cat_handle != NULL &&
              loghandle->u.phd.phd_cat_handle->u.chd.chd_current_log !=
                loghandle))) {
+               /* never try to destroy it again */
+               llh->llh_flags &= ~LLOG_F_ZAP_WHEN_EMPTY;
                rc = llog_trans_destroy(env, loghandle, th);
                if (rc < 0) {
                        /* Sigh, can not destroy the final plain llog, but
@@ -422,7 +423,7 @@ static int llog_process_thread(void *arg)
        struct llog_process_cat_data    *cd  = lpi->lpi_catdata;
        char                            *buf;
        size_t                           chunk_size;
-       __u64                            cur_offset;
+       __u64                            cur_offset, tmp_offset;
        int                              rc = 0, index = 1, last_index;
        int                              saved_index = 0;
        int                              last_called_index = 0;
@@ -482,7 +483,8 @@ repeat:
                 * The absolute offset of the current chunk is calculated
                 * from cur_offset value and stored in chunk_offset variable.
                 */
-               if (cur_offset % chunk_size != 0) {
+               tmp_offset = cur_offset;
+               if (do_div(tmp_offset, chunk_size) != 0) {
                        partial_chunk = true;
                        chunk_offset = cur_offset & ~(chunk_size - 1);
                } else {
@@ -513,8 +515,19 @@ repeat:
                                 * while llog_processing, check this is not
                                 * the case and re-read the current chunk
                                 * otherwise. */
+                               int records;
                                if (index > loghandle->lgh_last_idx)
                                        GOTO(out, rc = 0);
+                               /* <2 records means no more records
+                                * if the last record we processed was
+                                * the final one, then the underlying
+                                * object might have been destroyed yet.
+                                * we better don't access that.. */
+                               mutex_lock(&loghandle->lgh_hdr_mutex);
+                               records = loghandle->lgh_hdr->llh_count;
+                               mutex_unlock(&loghandle->lgh_hdr_mutex);
+                               if (records <= 1)
+                                       GOTO(out, rc = 0);
                                CDEBUG(D_OTHER, "Re-read last llog buffer for "
                                       "new records, index %u, last %u\n",
                                       index, loghandle->lgh_last_idx);
@@ -892,11 +905,29 @@ int llog_write_rec(const struct lu_env *env, struct llog_handle *handle,
 
        ENTRY;
 
+       /* API sanity checks */
+       if (handle == NULL) {
+               CERROR("loghandle is missed\n");
+               RETURN(-EPROTO);
+       } else if (handle->lgh_obj == NULL) {
+               CERROR("loghandle %p with NULL object\n",
+                       handle);
+               RETURN(-EPROTO);
+       } else if (th == NULL) {
+               CERROR("%s: missed transaction handle\n",
+                       handle->lgh_obj->do_lu.lo_dev->ld_obd->obd_name);
+               RETURN(-EPROTO);
+       } else if (handle->lgh_hdr == NULL) {
+               CERROR("%s: loghandle %p with no header\n",
+                       handle->lgh_obj->do_lu.lo_dev->ld_obd->obd_name,
+                       handle);
+               RETURN(-EPROTO);
+       }
+
        rc = llog_handle2ops(handle, &lop);
        if (rc)
                RETURN(rc);
 
-       LASSERT(lop);
        if (lop->lop_write_rec == NULL)
                RETURN(-EOPNOTSUPP);