X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fobdclass%2Fllog.c;h=12e179eff74f575f1896be70a7558fc6a0def300;hp=80e4f12b746df7bfec2df2e0af34371f4868c1d5;hb=d813c75df6798efbf3228347628c0d671ca7269c;hpb=f36daac69fe6e0cd35e2369967f4bae11bd2666f

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index 80e4f12..12e179e 100644
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -23,11 +23,10 @@
  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2012, 2016, Intel Corporation.
+ * Copyright (c) 2012, 2017, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
- * Lustre is a trademark of Sun Microsystems, Inc.
  *
  * lustre/obdclass/llog.c
  *
@@ -47,8 +46,10 @@
 #include <linux/kthread.h>
 #include <llog_swab.h>
 #include <lustre_log.h>
+#include <obd_support.h>
 #include <obd_class.h>
 #include "llog_internal.h"
+
 /*
  * Allocate a new log or catalog handle
  * Used inside llog_open().
@@ -63,8 +64,9 @@ static struct llog_handle *llog_alloc_handle(void)
 
 	init_rwsem(&loghandle->lgh_lock);
 	mutex_init(&loghandle->lgh_hdr_mutex);
+	init_rwsem(&loghandle->lgh_last_sem);
 	INIT_LIST_HEAD(&loghandle->u.phd.phd_entry);
-	atomic_set(&loghandle->lgh_refcount, 1);
+	refcount_set(&loghandle->lgh_refcount, 1);
 
 	return loghandle;
 }
@@ -89,23 +91,37 @@ out:
 	OBD_FREE_PTR(loghandle);
 }
 
-void llog_handle_get(struct llog_handle *loghandle)
+struct llog_handle *llog_handle_get(struct llog_handle *loghandle)
 {
-	atomic_inc(&loghandle->lgh_refcount);
+	if (refcount_inc_not_zero(&loghandle->lgh_refcount))
+		return loghandle;
+	return NULL;
 }
 
-void llog_handle_put(struct llog_handle *loghandle)
+int llog_handle_put(const struct lu_env *env, struct llog_handle *loghandle)
 {
-	LASSERT(atomic_read(&loghandle->lgh_refcount) > 0);
-	if (atomic_dec_and_test(&loghandle->lgh_refcount))
+	int rc = 0;
+
+	if (refcount_dec_and_test(&loghandle->lgh_refcount)) {
+		const struct llog_operations *lop;
+
+		rc = llog_handle2ops(loghandle, &lop);
+		if (!rc) {
+			if (lop->lop_close)
+				rc = lop->lop_close(env, loghandle);
+			else
+				rc = -EOPNOTSUPP;
+		}
 		llog_free_handle(loghandle);
+	}
+	return rc;
 }
 
 static int llog_declare_destroy(const struct lu_env *env,
 				struct llog_handle *handle,
 				struct thandle *th)
 {
-	struct llog_operations *lop;
+	const struct llog_operations *lop;
 	int rc;
 
 	ENTRY;
@@ -124,7 +140,7 @@ static int llog_declare_destroy(const struct lu_env *env,
 int llog_trans_destroy(const struct lu_env *env, struct llog_handle *handle,
 		       struct thandle *th)
 {
-	struct llog_operations	*lop;
+	const struct llog_operations *lop;
 	int rc;
 	ENTRY;
 
@@ -135,7 +151,7 @@ int llog_trans_destroy(const struct lu_env *env, struct llog_handle *handle,
 		RETURN(-EOPNOTSUPP);
 
 	LASSERT(handle->lgh_obj != NULL);
-	if (!dt_object_exists(handle->lgh_obj))
+	if (!llog_exist(handle))
 		RETURN(0);
 
 	rc = lop->lop_destroy(env, handle, th);
@@ -145,9 +161,9 @@ int llog_trans_destroy(const struct lu_env *env, struct llog_handle *handle,
 
 int llog_destroy(const struct lu_env *env, struct llog_handle *handle)
 {
-	struct llog_operations	*lop;
-	struct dt_device	*dt;
-	struct thandle		*th;
+	const struct llog_operations *lop;
+	struct dt_device *dt;
+	struct thandle *th;
 	int rc;
 
 	ENTRY;
@@ -164,11 +180,14 @@ int llog_destroy(const struct lu_env *env, struct llog_handle *handle)
 		RETURN(rc);
 	}
 
-	if (!dt_object_exists(handle->lgh_obj))
+	if (!llog_exist(handle))
 		RETURN(0);
 
 	dt = lu2dt_dev(handle->lgh_obj->do_lu.lo_dev);
 
+	if (unlikely(unlikely(dt->dd_rdonly)))
+		RETURN(-EROFS);
+
 	th = dt_trans_create(env, dt);
 	if (IS_ERR(th))
 		RETURN(PTR_ERR(th));
@@ -191,38 +210,39 @@ out_trans:
 EXPORT_SYMBOL(llog_destroy);
 
 /* returns negative on error; 0 if success; 1 if success & log destroyed */
-int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
-		    int index)
+int llog_cancel_arr_rec(const struct lu_env *env, struct llog_handle *loghandle,
+			int num, int *index)
 {
 	struct llog_thread_info *lgi = llog_info(env);
 	struct dt_device	*dt;
-	struct llog_log_hdr	*llh = loghandle->lgh_hdr;
+	struct llog_log_hdr	*llh;
 	struct thandle		*th;
-	int			 rc;
+	__u32			 tmp_lgc_index;
+	int			 rc, i = 0;
 	int rc1;
 	bool subtract_count = false;
 
 	ENTRY;
 
-	CDEBUG(D_RPCTRACE, "Canceling %d in log "DOSTID"\n", index,
-	       POSTID(&loghandle->lgh_id.lgl_oi));
-
-	if (index == 0) {
-		CERROR("Can't cancel index 0 which is header\n");
-		RETURN(-EINVAL);
-	}
-
 	LASSERT(loghandle != NULL);
 	LASSERT(loghandle->lgh_ctxt != NULL);
 	LASSERT(loghandle->lgh_obj != NULL);
 
+	llh = loghandle->lgh_hdr;
+
+	CDEBUG(D_RPCTRACE, "Canceling %d records, first %d in log "DFID"\n",
+	       num, index[0], PFID(&loghandle->lgh_id.lgl_oi.oi_fid));
+
 	dt = lu2dt_dev(loghandle->lgh_obj->do_lu.lo_dev);
 
+	if (unlikely(unlikely(dt->dd_rdonly)))
+		RETURN(0);
+
 	th = dt_trans_create(env, dt);
 	if (IS_ERR(th))
 		RETURN(PTR_ERR(th));
 
-	rc = llog_declare_write_rec(env, loghandle, &llh->llh_hdr, index, th);
+	rc = llog_declare_write_rec(env, loghandle, &llh->llh_hdr, 0, th);
 	if (rc < 0)
 		GOTO(out_trans, rc);
 
@@ -240,19 +260,32 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
 	down_write(&loghandle->lgh_lock);
 	/* clear bitmap */
 	mutex_lock(&loghandle->lgh_hdr_mutex);
-	if (!ext2_clear_bit(index, LLOG_HDR_BITMAP(llh))) {
-		CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n", index);
-		GOTO(out_unlock, rc);
+	for (i = 0; i < num; ++i) {
+		if (index[i] == 0) {
+			CERROR("Can't cancel index 0 which is header\n");
+			GOTO(out_unlock, rc = -EINVAL);
+		}
+		if (!__test_and_clear_bit_le(index[i], LLOG_HDR_BITMAP(llh))) {
+			CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n",
+			       index[i]);
+			GOTO(out_unlock, rc = -ENOENT);
+		}
 	}
-
-	loghandle->lgh_hdr->llh_count--;
+	loghandle->lgh_hdr->llh_count -= num;
 	subtract_count = true;
+
+	/* Since llog_process_thread use lgi_cookie, it`s better to save them
+	 * and restore after using
+	 */
+	tmp_lgc_index = lgi->lgi_cookie.lgc_index;
 	/* Pass this index to llog_osd_write_rec(), which will use the index
 	 * to only update the necesary bitmap. */
-	lgi->lgi_cookie.lgc_index = index;
+	lgi->lgi_cookie.lgc_index = index[0];
 	/* update header */
-	rc = llog_write_rec(env, loghandle, &llh->llh_hdr, &lgi->lgi_cookie,
-			    LLOG_HEADER_IDX, th);
+	rc = llog_write_rec(env, loghandle, &llh->llh_hdr, (num != 1 ? NULL :
+			    &lgi->lgi_cookie), LLOG_HEADER_IDX, th);
+	lgi->lgi_cookie.lgc_index = tmp_lgc_index;
+
 	if (rc != 0)
 		GOTO(out_unlock, rc);
 
@@ -270,36 +303,51 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
 			 * the bitmap has been clearly, so the record can not
 			 * be accessed anymore, let's return 0 for now, and
 			 * the orphan will be handled by LFSCK. */
-			CERROR("%s: can't destroy empty llog #"DOSTID
-			       "#%08x: rc = %d\n",
-			       loghandle->lgh_ctxt->loc_obd->obd_name,
-			       POSTID(&loghandle->lgh_id.lgl_oi),
-			       loghandle->lgh_id.lgl_ogen, rc);
-			GOTO(out_unlock, rc);
+			CERROR("%s: can't destroy empty llog "DFID": rc = %d\n",
+			       loghandle2name(loghandle),
+			       PFID(&loghandle->lgh_id.lgl_oi.oi_fid), rc);
+			GOTO(out_unlock, rc = 0);
 		}
 		rc = LLOG_DEL_PLAIN;
 	}
 
 out_unlock:
+	if (rc < 0) {
+		/* restore bitmap while holding a mutex */
+		if (subtract_count) {
+			loghandle->lgh_hdr->llh_count += num;
+			subtract_count = false;
+		}
+		for (i = i - 1; i >= 0; i--)
+			set_bit_le(index[i], LLOG_HDR_BITMAP(llh));
+	}
 	mutex_unlock(&loghandle->lgh_hdr_mutex);
 	up_write(&loghandle->lgh_lock);
 out_trans:
 	rc1 = dt_trans_stop(env, dt, th);
 	if (rc == 0)
 		rc = rc1;
-	if (rc < 0 && subtract_count) {
+	if (rc1 < 0) {
 		mutex_lock(&loghandle->lgh_hdr_mutex);
-		loghandle->lgh_hdr->llh_count++;
-		ext2_set_bit(index, LLOG_HDR_BITMAP(llh));
+		if (subtract_count)
+			loghandle->lgh_hdr->llh_count += num;
+		for (i = i - 1; i >= 0; i--)
+			set_bit_le(index[i], LLOG_HDR_BITMAP(llh));
 		mutex_unlock(&loghandle->lgh_hdr_mutex);
 	}
 	RETURN(rc);
 }
 
+int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
+		    int index)
+{
+	return llog_cancel_arr_rec(env, loghandle, 1, &index);
+}
+
 int llog_read_header(const struct lu_env *env, struct llog_handle *handle,
 		     const struct obd_uuid *uuid)
 {
-	struct llog_operations *lop;
+	const struct llog_operations *lop;
 	int rc;
 	ENTRY;
 
@@ -322,7 +370,7 @@ int llog_read_header(const struct lu_env *env, struct llog_handle *handle,
 						LLOG_MIN_CHUNK_SIZE);
 		llh->llh_hdr.lrh_len = handle->lgh_ctxt->loc_chunk_size;
 		llh->llh_hdr.lrh_index = 0;
-		llh->llh_timestamp = cfs_time_current_sec();
+		llh->llh_timestamp = ktime_get_real_seconds();
 		if (uuid)
 			memcpy(&llh->llh_tgtuuid, uuid,
 			       sizeof(llh->llh_tgtuuid));
@@ -332,7 +380,7 @@ int llog_read_header(const struct lu_env *env, struct llog_handle *handle,
 		memset(LLOG_HDR_BITMAP(llh), 0, llh->llh_hdr.lrh_len -
 						llh->llh_bitmap_offset -
 						sizeof(llh->llh_tail));
-		ext2_set_bit(0, LLOG_HDR_BITMAP(llh));
+		set_bit_le(0, LLOG_HDR_BITMAP(llh));
 		LLOG_HDR_TAIL(llh)->lrt_len = llh->llh_hdr.lrh_len;
 		LLOG_HDR_TAIL(llh)->lrt_index = llh->llh_hdr.lrh_index;
 		rc = 0;
@@ -368,7 +416,7 @@ int llog_init_handle(const struct lu_env *env, struct llog_handle *handle,
 			     (llh->llh_flags & LLOG_F_IS_CAT &&
 			      flags & LLOG_F_IS_PLAIN))) {
 			CERROR("%s: llog type is %s but initializing %s\n",
-			       handle->lgh_ctxt->loc_obd->obd_name,
+			       loghandle2name(handle),
 			       llh->llh_flags & LLOG_F_IS_CAT ?
 			       "catalog" : "plain",
 			       flags & LLOG_F_IS_CAT ? "catalog" : "plain");
@@ -388,7 +436,7 @@ int llog_init_handle(const struct lu_env *env, struct llog_handle *handle,
 		if (unlikely(uuid &&
 			     !obd_uuid_equals(uuid, &llh->llh_tgtuuid))) {
 			CERROR("%s: llog uuid mismatch: %s/%s\n",
-			       handle->lgh_ctxt->loc_obd->obd_name,
+			       loghandle2name(handle),
 			       (char *)uuid->uuid,
 			       (char *)llh->llh_tgtuuid.uuid);
 			GOTO(out, rc = -EEXIST);
@@ -401,8 +449,8 @@ int llog_init_handle(const struct lu_env *env, struct llog_handle *handle,
 		llh->llh_flags |= LLOG_F_IS_FIXSIZE;
 	} else if (!(flags & LLOG_F_IS_PLAIN)) {
 		CERROR("%s: unknown flags: %#x (expected %#x or %#x)\n",
-		       handle->lgh_ctxt->loc_obd->obd_name,
-		       flags, LLOG_F_IS_CAT, LLOG_F_IS_PLAIN);
+		       loghandle2name(handle), flags, LLOG_F_IS_CAT,
+		       LLOG_F_IS_PLAIN);
 		rc = -EINVAL;
 	}
 	llh->llh_flags |= fmt;
@@ -415,24 +463,53 @@ out:
 }
 EXPORT_SYMBOL(llog_init_handle);
 
+int llog_verify_record(const struct llog_handle *llh, struct llog_rec_hdr *rec)
+{
+	int chunk_size = llh->lgh_hdr->llh_hdr.lrh_len;
+
+	if (rec->lrh_len == 0 || rec->lrh_len > chunk_size) {
+		CERROR("%s: record is too large: %d > %d\n",
+		       loghandle2name(llh), rec->lrh_len, chunk_size);
+		return -EINVAL;
+	}
+	if (rec->lrh_index >= LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr)) {
+		CERROR("%s: index is too high: %d\n",
+		       loghandle2name(llh), rec->lrh_index);
+		return -EINVAL;
+	}
+	if ((rec->lrh_type & LLOG_OP_MASK) != LLOG_OP_MAGIC) {
+		CERROR("%s: magic %x is bad\n",
+		       loghandle2name(llh), rec->lrh_type);
+		return -EINVAL;
+	}
+
+	return 0;
+}
+EXPORT_SYMBOL(llog_verify_record);
+
 static int llog_process_thread(void *arg)
 {
 	struct llog_process_info	*lpi = arg;
 	struct llog_handle		*loghandle = lpi->lpi_loghandle;
 	struct llog_log_hdr		*llh = loghandle->lgh_hdr;
 	struct llog_process_cat_data	*cd  = lpi->lpi_catdata;
+	struct llog_thread_info		*lti;
 	char				*buf;
 	size_t				 chunk_size;
-	__u64				 cur_offset, tmp_offset;
+	__u64				 cur_offset;
 	int				 rc = 0, index = 1, last_index;
 	int				 saved_index = 0;
 	int				 last_called_index = 0;
+	bool				 repeated = false;
+	bool				refresh_idx = false;
 
 	ENTRY;
 
 	if (llh == NULL)
 		RETURN(-EINVAL);
 
+	lti = lpi->lpi_env == NULL ? NULL : llog_info(lpi->lpi_env);
+
 	cur_offset = chunk_size = llh->llh_hdr.lrh_len;
 	/* expect chunk_size to be power of two */
 	LASSERT(is_power_of_2(chunk_size));
@@ -454,13 +531,14 @@ static int llog_process_thread(void *arg)
 
 	while (rc == 0) {
 		struct llog_rec_hdr *rec;
-		off_t chunk_offset;
+		off_t chunk_offset = 0;
 		unsigned int buf_offset = 0;
-		bool partial_chunk;
+		int	lh_last_idx;
+		int	synced_idx = 0;
 
 		/* skip records not set in bitmap */
 		while (index <= last_index &&
-		       !ext2_test_bit(index, LLOG_HDR_BITMAP(llh)))
+		       !test_bit_le(index, LLOG_HDR_BITMAP(llh)))
 			++index;
 
 		/* There are no indices prior the last_index */
@@ -473,8 +551,22 @@ static int llog_process_thread(void *arg)
 repeat:
 		/* get the buf with our target record; avoid old garbage */
 		memset(buf, 0, chunk_size);
+		/* the record index for outdated chunk data */
+		/* it is safe to process buffer until saved lgh_last_idx */
+		lh_last_idx = LLOG_HDR_TAIL(llh)->lrt_index;
 		rc = llog_next_block(lpi->lpi_env, loghandle, &saved_index,
 				     index, &cur_offset, buf, chunk_size);
+		if (repeated && rc)
+			CDEBUG(D_OTHER, "cur_offset %llu, chunk_offset %llu,"
+			       " buf_offset %u, rc = %d\n", cur_offset,
+			       (__u64)chunk_offset, buf_offset, rc);
+		if (rc == -ESTALE)
+			GOTO(out, rc = 0);
+		/* we`ve tried to reread the chunk, but there is no
+		 * new records */
+		if (rc == -EIO && repeated && (chunk_offset + buf_offset) ==
+		    cur_offset)
+			GOTO(out, rc = 0);
 		if (rc != 0)
 			GOTO(out, rc);
 
@@ -483,14 +575,10 @@ repeat:
 		 * The absolute offset of the current chunk is calculated
 		 * from cur_offset value and stored in chunk_offset variable.
 		 */
-		tmp_offset = cur_offset;
-		if (do_div(tmp_offset, chunk_size) != 0) {
-			partial_chunk = true;
+		if ((cur_offset & (chunk_size - 1)) != 0)
 			chunk_offset = cur_offset & ~(chunk_size - 1);
-		} else {
-			partial_chunk = false;
+		else
 			chunk_offset = cur_offset - chunk_size;
-		}
 
 		/* NB: when rec->lrh_len is accessed it is already swabbed
 		 * since it is used at the "end" of the loop and the rec
@@ -508,40 +596,72 @@ repeat:
 			CDEBUG(D_OTHER, "after swabbing, type=%#x idx=%d\n",
 			       rec->lrh_type, rec->lrh_index);
 
-			/* for partial chunk the end of it is zeroed, check
-			 * for index 0 to distinguish it. */
-			if (partial_chunk && rec->lrh_index == 0) {
-				/* concurrent llog_add() might add new records
-				 * 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);
+			if (index == (synced_idx + 1) &&
+			    synced_idx == LLOG_HDR_TAIL(llh)->lrt_index)
+				GOTO(out, rc = 0);
+
+			if (OBD_FAIL_PRECHECK(OBD_FAIL_LLOG_PROCESS_TIMEOUT) &&
+				cfs_fail_val == (unsigned int)
+					(loghandle->lgh_id.lgl_oi.oi.oi_id &
+					 0xFFFFFFFF)) {
+				OBD_RACE(OBD_FAIL_LLOG_PROCESS_TIMEOUT);
+			}
+
+			/* the bitmap could be changed during processing
+			 * records from the chunk. For wrapped catalog
+			 * it means we can read deleted record and try to
+			 * process it. Check this case and reread the chunk.
+			 * It is safe to process to lh_last_idx, including
+			 * lh_last_idx if it was synced. We can not do <=
+			 * comparison, cause for wrapped catalog lgh_last_idx
+			 * could be less than index. So we detect last index
+			 * for processing as index == lh_last_idx+1. But when
+			 * catalog is wrapped and full lgh_last_idx=llh_cat_idx,
+			 * the first processing index is llh_cat_idx+1.The
+			 * exception is !(lgh_last_idx == llh_cat_idx &&
+			 * index == llh_cat_idx + 1), and after simplification
+			 * it turns to
+			 * lh_last_idx != LLOG_HDR_TAIL(llh)->lrt_index
+			 * This exception is working for catalog only.
+			 */
+
+			if ((index == lh_last_idx && synced_idx != index) ||
+			    (index == (lh_last_idx + 1) &&
+			     lh_last_idx != LLOG_HDR_TAIL(llh)->lrt_index) ||
+			    (rec->lrh_index == 0 && !repeated)) {
+
 				/* save offset inside buffer for the re-read */
 				buf_offset = (char *)rec - (char *)buf;
 				cur_offset = chunk_offset;
+				repeated = true;
+				/* We need to be sure lgh_last_idx
+				 * record was saved to disk
+				 */
+				down_read(&loghandle->lgh_last_sem);
+				synced_idx = LLOG_HDR_TAIL(llh)->lrt_index;
+				up_read(&loghandle->lgh_last_sem);
+				CDEBUG(D_OTHER, "synced_idx: %d\n", synced_idx);
 				goto repeat;
+
 			}
 
-			if (rec->lrh_len == 0 || rec->lrh_len > chunk_size) {
-				CWARN("invalid length %d in llog record for "
-				      "index %d/%d\n", rec->lrh_len,
-				      rec->lrh_index, index);
-				GOTO(out, rc = -EINVAL);
+			repeated = false;
+
+			rc = llog_verify_record(loghandle, rec);
+			if (rc) {
+				CERROR("%s: invalid record in llog "DFID
+				       " record for index %d/%d: rc = %d\n",
+				       loghandle2name(loghandle),
+				       PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
+				       rec->lrh_index, index, rc);
+				/*
+				 * the block seem to be corrupted, let's try
+				 * with the next one. reset rc to go to the
+				 * next chunk.
+				 */
+				refresh_idx = true;
+				index = 0;
+				GOTO(repeat, rc = 0);
 			}
 
 			if (rec->lrh_index < index) {
@@ -551,11 +671,22 @@ repeat:
 			}
 
 			if (rec->lrh_index != index) {
-				CERROR("%s: Invalid record: index %u but "
-				       "expected %u\n",
-				       loghandle->lgh_ctxt->loc_obd->obd_name,
-				       rec->lrh_index, index);
-				GOTO(out, rc = -ERANGE);
+				/*
+				 * the last time we couldn't parse the block due
+				 * to corruption, thus has no idea about the
+				 * next index, take it from the block, once.
+				 */
+				if (refresh_idx) {
+					refresh_idx = false;
+					index = rec->lrh_index;
+				} else {
+					CERROR("%s: "DFID" Invalid record: index"
+					       " %u but expected %u\n",
+					       loghandle2name(loghandle),
+					       PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
+					       rec->lrh_index, index);
+					GOTO(out, rc = -ERANGE);
+				}
 			}
 
 			CDEBUG(D_OTHER,
@@ -563,24 +694,65 @@ repeat:
 			       rec->lrh_index, rec->lrh_len,
 			       (int)(buf + chunk_size - (char *)rec));
 
-			loghandle->lgh_cur_idx = rec->lrh_index;
+			/* lgh_cur_offset is used only at llog_test_3 */
 			loghandle->lgh_cur_offset = (char *)rec - (char *)buf +
 						    chunk_offset;
 
 			/* if set, process the callback on this record */
-			if (ext2_test_bit(index, LLOG_HDR_BITMAP(llh))) {
+			if (test_bit_le(index, LLOG_HDR_BITMAP(llh))) {
+				struct llog_cookie *lgc;
+				__u64	tmp_off;
+				int	tmp_idx;
+
+				CDEBUG((llh->llh_flags & LLOG_F_IS_CAT ?
+					D_HA : D_OTHER),
+				       "index: %d, lh_last_idx: %d "
+				       "synced_idx: %d lgh_last_idx: %d\n",
+				       index, lh_last_idx, synced_idx,
+				       loghandle->lgh_last_idx);
+
+				if (lti != NULL) {
+					lgc = &lti->lgi_cookie;
+					/* store lu_env for recursive calls */
+					tmp_off = lgc->lgc_offset;
+					tmp_idx = lgc->lgc_index;
+
+					lgc->lgc_offset = (char *)rec -
+						(char *)buf + chunk_offset;
+					lgc->lgc_index = rec->lrh_index;
+				}
+				/* using lu_env for passing record offset to
+				 * llog_write through various callbacks */
 				rc = lpi->lpi_cb(lpi->lpi_env, loghandle, rec,
 						 lpi->lpi_cbdata);
 				last_called_index = index;
+
+				if (lti != NULL) {
+					lgc->lgc_offset = tmp_off;
+					lgc->lgc_index = tmp_idx;
+				}
+
 				if (rc == LLOG_PROC_BREAK) {
 					GOTO(out, rc);
 				} else if (rc == LLOG_DEL_RECORD) {
 					rc = llog_cancel_rec(lpi->lpi_env,
 							     loghandle,
 							     rec->lrh_index);
+					/* Allow parallel cancelling, ENOENT
+					 * means record was canceled at another
+					 * processing thread or callback
+					 */
+					if (rc == -ENOENT)
+						rc = 0;
 				}
 				if (rc)
 					GOTO(out, rc);
+				/* some stupid callbacks directly cancel records
+				 * and delete llog. Check it and stop
+				 * processing. */
+				if (loghandle->lgh_hdr == NULL ||
+				    loghandle->lgh_hdr->llh_count == 1)
+					GOTO(out, rc = 0);
 			}
 			/* exit if the last index is reached */
 			if (index >= last_index)
@@ -590,6 +762,11 @@ repeat:
 	}
 
 out:
+	CDEBUG(D_HA, "stop processing %s "DOSTID":%x index %d count %d\n",
+	       ((llh->llh_flags & LLOG_F_IS_CAT) ? "catalog" : "plain"),
+	       POSTID(&loghandle->lgh_id.lgl_oi), loghandle->lgh_id.lgl_ogen,
+	       index, llh->llh_count);
+
 	if (cd != NULL)
 		cd->lpcd_last_idx = last_called_index;
 
@@ -601,16 +778,23 @@ out:
 			 * retry until the umount or abort recovery, see
 			 * lod_sub_recovery_thread() */
 			CERROR("%s retry remote llog process\n",
-			       loghandle->lgh_ctxt->loc_obd->obd_name);
+			       loghandle2name(loghandle));
 			rc = -EAGAIN;
 		} else {
 			/* something bad happened to the processing of a local
 			 * llog file, probably I/O error or the log got
 			 * corrupted to be able to finally release the log we
 			 * discard any remaining bits in the header */
-			CERROR("Local llog found corrupted\n");
+			CERROR("%s: Local llog found corrupted #"DOSTID":%x"
+			       " %s index %d count %d\n",
+			       loghandle2name(loghandle),
+			       POSTID(&loghandle->lgh_id.lgl_oi),
+			       loghandle->lgh_id.lgl_ogen,
+			       ((llh->llh_flags & LLOG_F_IS_CAT) ? "catalog" :
+				"plain"), index, llh->llh_count);
+
 			while (index <= last_index) {
-				if (ext2_test_bit(index,
+				if (test_bit_le(index,
 						  LLOG_HDR_BITMAP(llh)) != 0)
 					llog_cancel_rec(lpi->lpi_env, loghandle,
 							index);
@@ -621,8 +805,8 @@ out:
 	}
 
 	OBD_FREE_LARGE(buf, chunk_size);
-        lpi->lpi_rc = rc;
-        return 0;
+	lpi->lpi_rc = rc;
+	return 0;
 }
 
 static int llog_process_thread_daemonize(void *arg)
@@ -643,12 +827,11 @@ static int llog_process_thread_daemonize(void *arg)
 		 * used outside of the kernel itself, because it calls
 		 * free_nsproxy() which is not exported by the kernel
 		 * (defined in kernel/nsproxy.c) */
-		atomic_dec(&curr_ns->count);
+		if (curr_ns)
+			atomic_dec(&curr_ns->count);
 	}
 	task_unlock(lpi->lpi_reftask);
 
-	unshare_fs_struct();
-
 	/* client env has no keys, tags is just 0 */
 	rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
 	if (rc)
@@ -667,10 +850,13 @@ int llog_process_or_fork(const struct lu_env *env,
 			 struct llog_handle *loghandle,
 			 llog_cb_t cb, void *data, void *catdata, bool fork)
 {
-        struct llog_process_info *lpi;
-        int                      rc;
+	struct llog_process_info *lpi;
+	struct llog_process_data *d = data;
+	struct llog_process_cat_data *cd = catdata;
+	__u32 flags = loghandle->lgh_hdr->llh_flags;
+	int rc;
 
-        ENTRY;
+	ENTRY;
 
 	OBD_ALLOC_PTR(lpi);
 	if (lpi == NULL) {
@@ -682,6 +868,11 @@ int llog_process_or_fork(const struct lu_env *env,
 	lpi->lpi_cbdata    = data;
 	lpi->lpi_catdata   = catdata;
 
+	CDEBUG(D_OTHER, "Processing "DFID" flags 0x%03x startcat %d startidx %d first_idx %d last_idx %d\n",
+	       PFID(&loghandle->lgh_id.lgl_oi.oi_fid), flags,
+	       (flags & LLOG_F_IS_CAT) && d ? d->lpd_startcat : -1,
+	       (flags & LLOG_F_IS_CAT) && d ? d->lpd_startidx : -1,
+	       cd ? cd->lpcd_first_idx : -1, cd ? cd->lpcd_last_idx : -1);
 	if (fork) {
 		struct task_struct *task;
 
@@ -698,7 +889,7 @@ int llog_process_or_fork(const struct lu_env *env,
 		if (IS_ERR(task)) {
 			rc = PTR_ERR(task);
 			CERROR("%s: cannot start thread: rc = %d\n",
-			       loghandle->lgh_ctxt->loc_obd->obd_name, rc);
+			       loghandle2name(loghandle), rc);
 			GOTO(out_lpi, rc);
 		}
 		wait_for_completion(&lpi->lpi_completion);
@@ -723,6 +914,27 @@ int llog_process(const struct lu_env *env, struct llog_handle *loghandle,
 }
 EXPORT_SYMBOL(llog_process);
 
+static inline const struct cred *llog_raise_resource(void)
+{
+	struct cred *cred = NULL;
+
+	if (cap_raised(current_cap(), CAP_SYS_RESOURCE))
+		return cred;
+
+	cred = prepare_creds();
+	if (!cred)
+		return cred;
+
+	cap_raise(cred->cap_effective, CAP_SYS_RESOURCE);
+	return override_creds(cred);
+}
+
+static inline void llog_restore_resource(const struct cred *old_cred)
+{
+	if (old_cred)
+		revert_creds(old_cred);
+}
+
 int llog_reverse_process(const struct lu_env *env,
 			 struct llog_handle *loghandle, llog_cb_t cb,
 			 void *data, void *catdata)
@@ -751,7 +963,7 @@ int llog_reverse_process(const struct lu_env *env,
 
 		/* skip records not set in bitmap */
 		while (index >= first_index &&
-		       !ext2_test_bit(index, LLOG_HDR_BITMAP(llh)))
+		       !test_bit_le(index, LLOG_HDR_BITMAP(llh)))
 			--index;
 
 		LASSERT(index >= first_index - 1);
@@ -782,7 +994,7 @@ int llog_reverse_process(const struct lu_env *env,
 				GOTO(out, rc = 0); /* no more records */
 
 			/* if set, process the callback on this record */
-			if (ext2_test_bit(index, LLOG_HDR_BITMAP(llh))) {
+			if (test_bit_le(index, LLOG_HDR_BITMAP(llh))) {
 				rec = (void *)tail - tail->lrt_len +
 				      sizeof(*tail);
 
@@ -828,8 +1040,8 @@ EXPORT_SYMBOL(llog_reverse_process);
  */
 int llog_exist(struct llog_handle *loghandle)
 {
-	struct llog_operations	*lop;
-	int			 rc;
+	const struct llog_operations *lop;
+	int rc;
 
 	ENTRY;
 
@@ -847,8 +1059,9 @@ EXPORT_SYMBOL(llog_exist);
 int llog_declare_create(const struct lu_env *env,
 			struct llog_handle *loghandle, struct thandle *th)
 {
-	struct llog_operations	*lop;
-	int			 raised, rc;
+	const struct cred *old_cred;
+	const struct llog_operations *lop;
+	int rc;
 
 	ENTRY;
 
@@ -858,20 +1071,18 @@ int llog_declare_create(const struct lu_env *env,
 	if (lop->lop_declare_create == NULL)
 		RETURN(-EOPNOTSUPP);
 
-	raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
-	if (!raised)
-		cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
+	old_cred = llog_raise_resource();
 	rc = lop->lop_declare_create(env, loghandle, th);
-	if (!raised)
-		cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
+	llog_restore_resource(old_cred);
 	RETURN(rc);
 }
 
 int llog_create(const struct lu_env *env, struct llog_handle *handle,
 		struct thandle *th)
 {
-	struct llog_operations	*lop;
-	int			 raised, rc;
+	const struct cred *old_cred;
+	const struct llog_operations *lop;
+	int rc;
 
 	ENTRY;
 
@@ -881,12 +1092,9 @@ int llog_create(const struct lu_env *env, struct llog_handle *handle,
 	if (lop->lop_create == NULL)
 		RETURN(-EOPNOTSUPP);
 
-	raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
-	if (!raised)
-		cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
+	old_cred = llog_raise_resource();
 	rc = lop->lop_create(env, handle, th);
-	if (!raised)
-		cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
+	llog_restore_resource(old_cred);
 	RETURN(rc);
 }
 
@@ -895,8 +1103,9 @@ int llog_declare_write_rec(const struct lu_env *env,
 			   struct llog_rec_hdr *rec, int idx,
 			   struct thandle *th)
 {
-	struct llog_operations	*lop;
-	int			 raised, rc;
+	const struct cred *old_cred;
+	const struct llog_operations *lop;
+	int rc;
 
 	ENTRY;
 
@@ -907,12 +1116,9 @@ int llog_declare_write_rec(const struct lu_env *env,
 	if (lop->lop_declare_write_rec == NULL)
 		RETURN(-EOPNOTSUPP);
 
-	raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
-	if (!raised)
-		cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
+	old_cred = llog_raise_resource();
 	rc = lop->lop_declare_write_rec(env, handle, rec, idx, th);
-	if (!raised)
-		cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
+	llog_restore_resource(old_cred);
 	RETURN(rc);
 }
 
@@ -920,8 +1126,9 @@ int llog_write_rec(const struct lu_env *env, struct llog_handle *handle,
 		   struct llog_rec_hdr *rec, struct llog_cookie *logcookies,
 		   int idx, struct thandle *th)
 {
-	struct llog_operations	*lop;
-	int			 raised, rc, buflen;
+	const struct cred *old_cred;
+	const struct llog_operations *lop;
+	int rc, buflen;
 
 	ENTRY;
 
@@ -935,12 +1142,11 @@ int llog_write_rec(const struct lu_env *env, struct llog_handle *handle,
 		RETURN(-EPROTO);
 	} else if (th == NULL) {
 		CERROR("%s: missed transaction handle\n",
-			handle->lgh_obj->do_lu.lo_dev->ld_obd->obd_name);
+		       loghandle2name(handle));
 		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);
+		       loghandle2name(handle), handle);
 		RETURN(-EPROTO);
 	}
 
@@ -954,12 +1160,9 @@ int llog_write_rec(const struct lu_env *env, struct llog_handle *handle,
 	buflen = rec->lrh_len;
 	LASSERT(cfs_size_round(buflen) == buflen);
 
-	raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
-	if (!raised)
-		cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
+	old_cred = llog_raise_resource();
 	rc = lop->lop_write_rec(env, handle, rec, logcookies, idx, th);
-	if (!raised)
-		cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
+	llog_restore_resource(old_cred);
 	RETURN(rc);
 }
 
@@ -967,19 +1170,17 @@ int llog_add(const struct lu_env *env, struct llog_handle *lgh,
 	     struct llog_rec_hdr *rec, struct llog_cookie *logcookies,
 	     struct thandle *th)
 {
-	int raised, rc;
+	const struct cred *old_cred;
+	int rc;
 
 	ENTRY;
 
 	if (lgh->lgh_logops->lop_add == NULL)
 		RETURN(-EOPNOTSUPP);
 
-	raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
-	if (!raised)
-		cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
+	old_cred = llog_raise_resource();
 	rc = lgh->lgh_logops->lop_add(env, lgh, rec, logcookies, th);
-	if (!raised)
-		cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
+	llog_restore_resource(old_cred);
 	RETURN(rc);
 }
 EXPORT_SYMBOL(llog_add);
@@ -987,19 +1188,17 @@ EXPORT_SYMBOL(llog_add);
 int llog_declare_add(const struct lu_env *env, struct llog_handle *lgh,
 		     struct llog_rec_hdr *rec, struct thandle *th)
 {
-	int raised, rc;
+	const struct cred *old_cred;
+	int rc;
 
 	ENTRY;
 
 	if (lgh->lgh_logops->lop_declare_add == NULL)
 		RETURN(-EOPNOTSUPP);
 
-	raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
-	if (!raised)
-		cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
+	old_cred = llog_raise_resource();
 	rc = lgh->lgh_logops->lop_declare_add(env, lgh, rec, th);
-	if (!raised)
-		cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
+	llog_restore_resource(old_cred);
 	RETURN(rc);
 }
 EXPORT_SYMBOL(llog_declare_add);
@@ -1029,6 +1228,9 @@ int llog_open_create(const struct lu_env *env, struct llog_ctxt *ctxt,
 
 	d = lu2dt_dev((*res)->lgh_obj->do_lu.lo_dev);
 
+	if (unlikely(unlikely(d->dd_rdonly)))
+		RETURN(-EROFS);
+
 	th = dt_trans_create(env, d);
 	if (IS_ERR(th))
 		GOTO(out, rc = PTR_ERR(th));
@@ -1096,7 +1298,8 @@ int llog_write(const struct lu_env *env, struct llog_handle *loghandle,
 {
 	struct dt_device	*dt;
 	struct thandle		*th;
-	int			 rc;
+	bool			need_cookie;
+	int			rc;
 
 	ENTRY;
 
@@ -1106,6 +1309,9 @@ int llog_write(const struct lu_env *env, struct llog_handle *loghandle,
 
 	dt = lu2dt_dev(loghandle->lgh_obj->do_lu.lo_dev);
 
+	if (unlikely(unlikely(dt->dd_rdonly)))
+		RETURN(-EROFS);
+
 	th = dt_trans_create(env, dt);
 	if (IS_ERR(th))
 		RETURN(PTR_ERR(th));
@@ -1119,8 +1325,21 @@ int llog_write(const struct lu_env *env, struct llog_handle *loghandle,
 	if (rc)
 		GOTO(out_trans, rc);
 
+	need_cookie = !(idx == LLOG_HEADER_IDX || idx == LLOG_NEXT_IDX);
+
 	down_write(&loghandle->lgh_lock);
-	rc = llog_write_rec(env, loghandle, rec, NULL, idx, th);
+	if (need_cookie) {
+		struct llog_thread_info *lti = llog_info(env);
+
+		/* cookie comes from llog_process_thread */
+		rc = llog_write_rec(env, loghandle, rec, &lti->lgi_cookie,
+				    rec->lrh_index, th);
+		/* upper layer didn`t pass cookie so change rc */
+		rc = (rc == 1 ? 0 : rc);
+	} else {
+		rc = llog_write_rec(env, loghandle, rec, NULL, idx, th);
+	}
+
 	up_write(&loghandle->lgh_lock);
 out_trans:
 	dt_trans_stop(env, dt, th);
@@ -1132,7 +1351,7 @@ int llog_open(const struct lu_env *env, struct llog_ctxt *ctxt,
 	      struct llog_handle **lgh, struct llog_logid *logid,
 	      char *name, enum llog_open_param open_param)
 {
-	int	 raised;
+	const struct cred *old_cred;
 	int	 rc;
 
 	ENTRY;
@@ -1151,12 +1370,9 @@ int llog_open(const struct lu_env *env, struct llog_ctxt *ctxt,
 	(*lgh)->lgh_ctxt = ctxt;
 	(*lgh)->lgh_logops = ctxt->loc_logops;
 
-	raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
-	if (!raised)
-		cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
+	old_cred = llog_raise_resource();
 	rc = ctxt->loc_logops->lop_open(env, *lgh, logid, name, open_param);
-	if (!raised)
-		cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
+	llog_restore_resource(old_cred);
 	if (rc) {
 		llog_free_handle(*lgh);
 		*lgh = NULL;
@@ -1167,20 +1383,7 @@ EXPORT_SYMBOL(llog_open);
 
 int llog_close(const struct lu_env *env, struct llog_handle *loghandle)
 {
-	struct llog_operations	*lop;
-	int			 rc;
-
-	ENTRY;
-
-	rc = llog_handle2ops(loghandle, &lop);
-	if (rc)
-		GOTO(out, rc);
-	if (lop->lop_close == NULL)
-		GOTO(out, rc = -EOPNOTSUPP);
-	rc = lop->lop_close(env, loghandle);
-out:
-	llog_handle_put(loghandle);
-	RETURN(rc);
+	return llog_handle_put(env, loghandle);
 }
 EXPORT_SYMBOL(llog_close);
 
@@ -1304,8 +1507,9 @@ __u64 llog_size(const struct lu_env *env, struct llog_handle *llh)
 
 	rc = llh->lgh_obj->do_ops->do_attr_get(env, llh->lgh_obj, &la);
 	if (rc) {
-		CERROR("%s: attr_get failed, rc = %d\n",
-		       llh->lgh_ctxt->loc_obd->obd_name, rc);
+		CERROR("%s: attr_get failed for "DFID": rc = %d\n",
+		       loghandle2name(llh), PFID(&llh->lgh_id.lgl_oi.oi_fid),
+		       rc);
 		return 0;
 	}