X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Fllog.c;h=2cd959883fc99d4a7094ba6a2e1bbe700eca1b78;hb=848f9e20320cb7c01eaf7f1b5c27f5efd54e4818;hp=0c643b2069da3577303974d04bf156fff8334027;hpb=abe5c61764860431d437b453310aea7742f68064;p=fs%2Flustre-release.git

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index 0c643b2..2cd9598 100644
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -26,6 +26,8 @@
 /*
  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
+ *
+ * Copyright (c) 2012, 2013, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -65,9 +67,10 @@ struct llog_handle *llog_alloc_handle(void)
 	if (loghandle == NULL)
 		return ERR_PTR(-ENOMEM);
 
-	cfs_init_rwsem(&loghandle->lgh_lock);
-	cfs_spin_lock_init(&loghandle->lgh_hdr_lock);
+	init_rwsem(&loghandle->lgh_lock);
+	spin_lock_init(&loghandle->lgh_hdr_lock);
 	CFS_INIT_LIST_HEAD(&loghandle->u.phd.phd_entry);
+	cfs_atomic_set(&loghandle->lgh_refcount, 1);
 
 	return loghandle;
 }
@@ -77,22 +80,34 @@ struct llog_handle *llog_alloc_handle(void)
  */
 void llog_free_handle(struct llog_handle *loghandle)
 {
-	if (!loghandle)
-		return;
+	LASSERT(loghandle != NULL);
 
+	/* failed llog_init_handle */
 	if (!loghandle->lgh_hdr)
 		goto out;
+
 	if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)
-		cfs_list_del_init(&loghandle->u.phd.phd_entry);
-	if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
+		LASSERT(cfs_list_empty(&loghandle->u.phd.phd_entry));
+	else if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
 		LASSERT(cfs_list_empty(&loghandle->u.chd.chd_head));
 	LASSERT(sizeof(*(loghandle->lgh_hdr)) == LLOG_CHUNK_SIZE);
 	OBD_FREE(loghandle->lgh_hdr, LLOG_CHUNK_SIZE);
-
 out:
 	OBD_FREE_PTR(loghandle);
 }
 
+void llog_handle_get(struct llog_handle *loghandle)
+{
+	cfs_atomic_inc(&loghandle->lgh_refcount);
+}
+
+void llog_handle_put(struct llog_handle *loghandle)
+{
+	LASSERT(cfs_atomic_read(&loghandle->lgh_refcount) > 0);
+	if (cfs_atomic_dec_and_test(&loghandle->lgh_refcount))
+		llog_free_handle(loghandle);
+}
+
 /* 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)
@@ -101,57 +116,55 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
         int rc = 0;
         ENTRY;
 
-        CDEBUG(D_RPCTRACE, "Canceling %d in log "LPX64"\n",
-               index, loghandle->lgh_id.lgl_oid);
+        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);
         }
 
-	cfs_spin_lock(&loghandle->lgh_hdr_lock);
+	spin_lock(&loghandle->lgh_hdr_lock);
 	if (!ext2_clear_bit(index, llh->llh_bitmap)) {
-		cfs_spin_unlock(&loghandle->lgh_hdr_lock);
-                CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n", index);
-                RETURN(-ENOENT);
-        }
+		spin_unlock(&loghandle->lgh_hdr_lock);
+		CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n", index);
+		RETURN(-ENOENT);
+	}
 
-        llh->llh_count--;
+	llh->llh_count--;
 
-        if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
-            (llh->llh_count == 1) &&
-            (loghandle->lgh_last_idx == (LLOG_BITMAP_BYTES * 8) - 1)) {
-		cfs_spin_unlock(&loghandle->lgh_hdr_lock);
+	if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
+	    (llh->llh_count == 1) &&
+	    (loghandle->lgh_last_idx == (LLOG_BITMAP_BYTES * 8) - 1)) {
+		spin_unlock(&loghandle->lgh_hdr_lock);
 		rc = llog_destroy(env, loghandle);
 		if (rc < 0) {
-			CERROR("%s: can't destroy empty llog #"LPX64"#"LPX64
+			CERROR("%s: can't destroy empty llog #"DOSTID
 			       "#%08x: rc = %d\n",
 			       loghandle->lgh_ctxt->loc_obd->obd_name,
-			       loghandle->lgh_id.lgl_oid,
-			       loghandle->lgh_id.lgl_oseq,
+			       POSTID(&loghandle->lgh_id.lgl_oi),
 			       loghandle->lgh_id.lgl_ogen, rc);
 			GOTO(out_err, rc);
 		}
 		RETURN(1);
 	}
-	cfs_spin_unlock(&loghandle->lgh_hdr_lock);
+	spin_unlock(&loghandle->lgh_hdr_lock);
 
 	rc = llog_write(env, loghandle, &llh->llh_hdr, NULL, 0, NULL, 0);
 	if (rc < 0) {
-		CERROR("%s: fail to write header for llog #"LPX64"#"LPX64
+		CERROR("%s: fail to write header for llog #"DOSTID
 		       "#%08x: rc = %d\n",
 		       loghandle->lgh_ctxt->loc_obd->obd_name,
-		       loghandle->lgh_id.lgl_oid,
-		       loghandle->lgh_id.lgl_oseq,
+		       POSTID(&loghandle->lgh_id.lgl_oi),
 		       loghandle->lgh_id.lgl_ogen, rc);
 		GOTO(out_err, rc);
 	}
 	RETURN(0);
 out_err:
-	cfs_spin_lock(&loghandle->lgh_hdr_lock);
+	spin_lock(&loghandle->lgh_hdr_lock);
 	ext2_set_bit(index, llh->llh_bitmap);
 	llh->llh_count++;
-	cfs_spin_unlock(&loghandle->lgh_hdr_lock);
+	spin_unlock(&loghandle->lgh_hdr_lock);
 	return rc;
 }
 EXPORT_SYMBOL(llog_cancel_rec);
@@ -257,6 +270,32 @@ out:
 }
 EXPORT_SYMBOL(llog_init_handle);
 
+int llog_copy_handler(const struct lu_env *env,
+		      struct llog_handle *llh,
+		      struct llog_rec_hdr *rec,
+		      void *data)
+{
+	struct llog_rec_hdr local_rec = *rec;
+	struct llog_handle *local_llh = (struct llog_handle *)data;
+	char *cfg_buf = (char*) (rec + 1);
+	struct lustre_cfg *lcfg;
+	int rc = 0;
+	ENTRY;
+
+	/* Append all records */
+	local_rec.lrh_len -= sizeof(*rec) + sizeof(struct llog_rec_tail);
+	rc = llog_write(env, local_llh, &local_rec, NULL, 0,
+			(void *)cfg_buf, -1);
+
+	lcfg = (struct lustre_cfg *)cfg_buf;
+	CDEBUG(D_INFO, "idx=%d, rc=%d, len=%d, cmd %x %s %s\n",
+	       rec->lrh_index, rc, rec->lrh_len, lcfg->lcfg_command,
+	       lustre_cfg_string(lcfg, 0), lustre_cfg_string(lcfg, 1));
+
+	RETURN(rc);
+}
+EXPORT_SYMBOL(llog_copy_handler);
+
 static int llog_process_thread(void *arg)
 {
 	struct llog_process_info	*lpi = arg;
@@ -399,10 +438,10 @@ static int llog_process_thread_daemonize(void *arg)
 	struct lu_env			 env;
 	int				 rc;
 
-	cfs_daemonize_ctxt("llog_process_thread");
+	unshare_fs_struct();
 
 	/* client env has no keys, tags is just 0 */
-	rc = lu_env_init(&env, LCT_LOCAL);
+	rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
 	if (rc)
 		goto out;
 	lpi->lpi_env = &env;
@@ -411,7 +450,7 @@ static int llog_process_thread_daemonize(void *arg)
 
 	lu_env_fini(&env);
 out:
-	cfs_complete(&lpi->lpi_completion);
+	complete(&lpi->lpi_completion);
 	return rc;
 }
 #endif
@@ -440,16 +479,16 @@ int llog_process_or_fork(const struct lu_env *env,
 		/* The new thread can't use parent env,
 		 * init the new one in llog_process_thread_daemonize. */
 		lpi->lpi_env = NULL;
-		cfs_init_completion(&lpi->lpi_completion);
-		rc = cfs_create_thread(llog_process_thread_daemonize, lpi,
-				       CFS_DAEMON_FLAGS);
-		if (rc < 0) {
+		init_completion(&lpi->lpi_completion);
+		rc = PTR_ERR(kthread_run(llog_process_thread_daemonize, lpi,
+					     "llog_process_thread"));
+		if (IS_ERR_VALUE(rc)) {
 			CERROR("%s: cannot start thread: rc = %d\n",
 			       loghandle->lgh_ctxt->loc_obd->obd_name, rc);
 			OBD_FREE_PTR(lpi);
 			RETURN(rc);
 		}
-		cfs_wait_for_completion(&lpi->lpi_completion);
+		wait_for_completion(&lpi->lpi_completion);
 	} else {
 		lpi->lpi_env = env;
 		llog_process_thread(lpi);
@@ -864,17 +903,17 @@ int llog_write(const struct lu_env *env, struct llog_handle *loghandle,
 		if (rc)
 			GOTO(out_trans, rc);
 
-		cfs_down_write(&loghandle->lgh_lock);
+		down_write(&loghandle->lgh_lock);
 		rc = llog_write_rec(env, loghandle, rec, reccookie,
 				    cookiecount, buf, idx, th);
-		cfs_up_write(&loghandle->lgh_lock);
+		up_write(&loghandle->lgh_lock);
 out_trans:
 		dt_trans_stop(env, dt, th);
 	} else { /* lvfs compatibility */
-		cfs_down_write(&loghandle->lgh_lock);
+		down_write(&loghandle->lgh_lock);
 		rc = llog_write_rec(env, loghandle, rec, reccookie,
 				    cookiecount, buf, idx, NULL);
-		cfs_up_write(&loghandle->lgh_lock);
+		up_write(&loghandle->lgh_lock);
 	}
 	RETURN(rc);
 }
@@ -928,10 +967,10 @@ int llog_close(const struct lu_env *env, struct llog_handle *loghandle)
 	if (rc)
 		GOTO(out, rc);
 	if (lop->lop_close == NULL)
-		GOTO(out, -EOPNOTSUPP);
+		GOTO(out, rc = -EOPNOTSUPP);
 	rc = lop->lop_close(env, loghandle);
 out:
-	llog_free_handle(loghandle);
+	llog_handle_put(loghandle);
 	RETURN(rc);
 }
 EXPORT_SYMBOL(llog_close);