From: Di Wang <di.wang@intel.com>
Date: Fri, 19 Feb 2016 09:29:26 +0000 (-0500)
Subject: LU-7800 llog: check if next llog exists
X-Git-Tag: 2.8.57~40
X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=commitdiff_plain;h=96509385abe2d03bd9212edef26e4aabe0ebf8bd

LU-7800 llog: check if next llog exists

Because next llog creation will only be checked
and created in declare phase, and it does not
serialize the catllog accessing in the whole
declare_add and add process, so if there are
mulitple threads access the catlog at the same
time, and if the llog creation did not succeeds,
then next_log (in catllog) might be NULL, and
it will cause panic in llog_cat_current_log().

Signed-off-by: Di Wang <di.wang@intel.com>
Change-Id: I2343023c1f3109c077c98d78d3669377d95ed42f
Reviewed-on: http://review.whamcloud.com/18542
Tested-by: Jenkins
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: James Simmons <uja.ornl@yahoo.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>
---

diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c
index 921c9cb..1e0c1fd 100644
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -363,21 +363,40 @@ static struct llog_handle *llog_cat_current_log(struct llog_handle *cathandle,
 		down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);
 		llh = loghandle->lgh_hdr;
 		LASSERT(llh);
-		if (!llog_is_full(loghandle)) {
-			up_write(&cathandle->lgh_lock);
-			RETURN(loghandle);
-		} else {
+		if (!llog_is_full(loghandle))
+			GOTO(out_unlock, loghandle);
+		else
 			up_write(&loghandle->lgh_lock);
-		}
-        }
+	}
 
 next:
+	/* Sigh, the chd_next_log and chd_current_log is initialized
+	 * in declare phase, and we do not serialize the catlog
+	 * accessing, so it might be possible the llog creation
+	 * thread (see llog_cat_declare_add_rec()) did not create
+	 * llog successfully, then the following thread might
+	 * meet this situation. */
+	if (IS_ERR_OR_NULL(cathandle->u.chd.chd_next_log)) {
+		CERROR("%s: next log does not exist!\n",
+		       cathandle->lgh_ctxt->loc_obd->obd_name);
+		loghandle = ERR_PTR(-EIO);
+		if (cathandle->u.chd.chd_next_log == NULL) {
+			/* Store the error in chd_next_log, so
+			 * the following process can get correct
+			 * failure value */
+			cathandle->u.chd.chd_next_log = loghandle;
+		}
+		GOTO(out_unlock, loghandle);
+	}
+
 	CDEBUG(D_INODE, "use next log\n");
 
 	loghandle = cathandle->u.chd.chd_next_log;
 	cathandle->u.chd.chd_current_log = loghandle;
 	cathandle->u.chd.chd_next_log = NULL;
 	down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);
+
+out_unlock:
 	up_write(&cathandle->lgh_lock);
 	LASSERT(loghandle);
 	RETURN(loghandle);
@@ -434,7 +453,8 @@ int llog_cat_add_rec(const struct lu_env *env, struct llog_handle *cathandle,
 
 retry:
 	loghandle = llog_cat_current_log(cathandle, th);
-	LASSERT(!IS_ERR(loghandle));
+	if (IS_ERR(loghandle))
+		RETURN(PTR_ERR(loghandle));
 
 	/* loghandle is already locked by llog_cat_current_log() for us */
 	if (!llog_exist(loghandle)) {
@@ -500,10 +520,12 @@ int llog_cat_declare_add_rec(const struct lu_env *env,
 			}
 		}
 		up_write(&cathandle->lgh_lock);
-	} else if (cathandle->u.chd.chd_next_log == NULL) {
+	} else if (cathandle->u.chd.chd_next_log == NULL ||
+		   IS_ERR(cathandle->u.chd.chd_next_log)) {
 		/* declare next plain llog */
 		down_write(&cathandle->lgh_lock);
-		if (cathandle->u.chd.chd_next_log == NULL) {
+		if (cathandle->u.chd.chd_next_log == NULL ||
+		    IS_ERR(cathandle->u.chd.chd_next_log)) {
 			rc = llog_open(env, cathandle->lgh_ctxt, &loghandle,
 				       NULL, NULL, LLOG_OPEN_NEW);
 			if (rc == 0) {
@@ -587,7 +609,7 @@ write_again:
 	}
 
 	next = cathandle->u.chd.chd_next_log;
-	if (next) {
+	if (!IS_ERR_OR_NULL(next)) {
 		if (!llog_exist(next)) {
 			if (dt_object_remote(cathandle->lgh_obj)) {
 				/* For remote operation, if we put the llog
@@ -600,15 +622,30 @@ write_again:
 				 * this transaction. So let's create the llog
 				 * object synchronously here to remove the
 				 * dependency. */
-				down_read_nested(&cathandle->lgh_lock,
+				down_write_nested(&cathandle->lgh_lock,
 						 LLOGH_CAT);
 				next = cathandle->u.chd.chd_next_log;
+				if (IS_ERR_OR_NULL(next)) {
+					/* Sigh, another thread just tried,
+					 * let's fail as well */
+					up_write(&cathandle->lgh_lock);
+					if (next == NULL)
+						rc = -EIO;
+					else
+						rc = PTR_ERR(next);
+					GOTO(out, rc);
+				}
+
 				down_write_nested(&next->lgh_lock, LLOGH_LOG);
-				if (!llog_exist(next))
+				if (!llog_exist(next)) {
 					rc = llog_cat_new_log(env, cathandle,
 							      next, NULL);
+					if (rc < 0)
+						cathandle->u.chd.chd_next_log =
+								ERR_PTR(rc);
+				}
 				up_write(&next->lgh_lock);
-				up_read(&cathandle->lgh_lock);
+				up_write(&cathandle->lgh_lock);
 				if (rc < 0)
 					GOTO(out, rc);
 			} else {