X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fquota%2Fqsd_reint.c;h=f50862ede4269a8808644db0e5edb5edd11c0e38;hb=049a96e78011c7a908596b809044dae1963e6fda;hp=261d5dcea389a58ae7545eabf6edfc58583b5992;hpb=984f4ce51fd38caaf0bd2b706a130f7f17c51638;p=fs%2Flustre-release.git

diff --git a/lustre/quota/qsd_reint.c b/lustre/quota/qsd_reint.c
index 261d5dc..f50862e 100644
--- a/lustre/quota/qsd_reint.c
+++ b/lustre/quota/qsd_reint.c
@@ -21,19 +21,17 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2011, 2012, Intel, Inc.
+ * Copyright (c) 2012, 2014, Intel Corporation.
  * Use is subject to license terms.
  *
  * Author: Johann Lombardi <johann.lombardi@intel.com>
  * Author: Niu    Yawei    <yawei.niu@intel.com>
  */
 
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
-
 #define DEBUG_SUBSYSTEM S_LQUOTA
 
+#include <linux/kthread.h>
+#include <lustre_swab.h>
 #include "qsd_internal.h"
 
 /*
@@ -44,7 +42,7 @@ static void qsd_reint_completion(const struct lu_env *env,
 				 struct quota_body *req_qbody,
 				 struct quota_body *rep_qbody,
 				 struct lustre_handle *lockh,
-				 union ldlm_wire_lvb *lvb,
+				 struct lquota_lvb *lvb,
 				 void *arg, int rc)
 {
 	struct qsd_instance	*qsd = qqi->qqi_qsd;
@@ -60,9 +58,9 @@ static void qsd_reint_completion(const struct lu_env *env,
 	}
 
 	CDEBUG(D_QUOTA, "%s: global quota lock successfully acquired, glb "
-	       "fid:"DFID", glb ver:"LPU64", slv fid:"DFID", slv ver:"LPU64"\n",
+	       "fid:"DFID", glb ver:%llu, slv fid:"DFID", slv ver:%llu\n",
 	       qsd->qsd_svname, PFID(&req_qbody->qb_fid),
-	       lvb->l_lquota.lvb_glb_ver, PFID(&rep_qbody->qb_slv_fid),
+	       lvb->lvb_glb_ver, PFID(&rep_qbody->qb_slv_fid),
 	       rep_qbody->qb_slv_ver);
 
 	*slv_ver = rep_qbody->qb_slv_ver;
@@ -98,7 +96,7 @@ out:
 static int qsd_reint_entries(const struct lu_env *env,
 			     struct qsd_qtype_info *qqi,
 			     struct idx_info *ii, bool global,
-			     cfs_page_t **pages,
+			     struct page **pages,
 			     unsigned int npages, bool need_swab)
 {
 	struct qsd_thread_info	*qti = qsd_info(env);
@@ -123,7 +121,7 @@ static int qsd_reint_entries(const struct lu_env *env,
 	size = ii->ii_recsize + ii->ii_keysize;
 
 	for (i = 0; i < npages; i++) {
-		union lu_page	*lip = cfs_kmap(pages[i]);
+		union lu_page	*lip = kmap(pages[i]);
 
 		for (j = 0; j < LU_PAGE_COUNT; j++) {
 			if (need_swab)
@@ -173,7 +171,7 @@ static int qsd_reint_entries(const struct lu_env *env,
 			lip++;
 		}
 out:
-		cfs_kunmap(pages[i]);
+		kunmap(pages[i]);
 		if (rc)
 			break;
 	}
@@ -187,7 +185,7 @@ static int qsd_reint_index(const struct lu_env *env, struct qsd_qtype_info *qqi,
 	struct qsd_instance	*qsd = qqi->qqi_qsd;
 	struct idx_info		*ii = &qti->qti_ii;
 	struct lu_fid		*fid;
-	cfs_page_t		**pages = NULL;
+	struct page		**pages = NULL;
 	unsigned int		 npages, pg_cnt;
 	__u64			 start_hash = 0, ver = 0;
 	bool			 need_swab = false;
@@ -197,15 +195,15 @@ static int qsd_reint_index(const struct lu_env *env, struct qsd_qtype_info *qqi,
 	fid = global ? &qqi->qqi_fid : &qqi->qqi_slv_fid;
 
 	/* let's do a 1MB bulk */
-	npages = min_t(unsigned int, PTLRPC_MAX_BRW_SIZE, 1 << 20);
-	npages /= CFS_PAGE_SIZE;
+	npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
+	npages /= PAGE_SIZE;
 
 	/* allocate pages for bulk index read */
 	OBD_ALLOC(pages, npages * sizeof(*pages));
 	if (pages == NULL)
 		GOTO(out, rc = -ENOMEM);
 	for (i = 0; i < npages; i++) {
-		pages[i] = cfs_alloc_page(CFS_ALLOC_STD);
+		pages[i] = alloc_page(GFP_NOFS);
 		if (pages[i] == NULL)
 			GOTO(out, rc = -ENOMEM);
 	}
@@ -258,7 +256,7 @@ repeat:
 		ver = ii->ii_version;
 
 	pg_cnt = (ii->ii_count + (LU_PAGE_COUNT) - 1);
-	pg_cnt >>= CFS_PAGE_SHIFT - LU_PAGE_SHIFT;
+	pg_cnt >>= PAGE_SHIFT - LU_PAGE_SHIFT;
 
 	if (pg_cnt > npages) {
 		CERROR("%s: master returned more pages than expected, %u > %u"
@@ -278,7 +276,7 @@ out:
 	if (pages != NULL) {
 		for (i = 0; i < npages; i++)
 			if (pages[i] != NULL)
-				cfs_free_page(pages[i]);
+				__free_page(pages[i]);
 		OBD_FREE(pages, npages * sizeof(*pages));
 	}
 
@@ -286,8 +284,8 @@ out:
 	if (rc == 0) {
 		rc = qsd_write_version(env, qqi, ver, global);
 		if (rc)
-			CERROR("%s: write version "LPU64" to "DFID" failed. "
-			       "%d\n", qsd->qsd_svname, ver, PFID(fid), rc);
+			CERROR("%s: write version %llu to "DFID" failed : rc = %d\n",
+			       qsd->qsd_svname, ver, PFID(fid), rc);
 	}
 
 	RETURN(rc);
@@ -309,7 +307,7 @@ static int qsd_reconciliation(const struct lu_env *env,
 	LASSERT(qqi->qqi_glb_obj != NULL);
 	iops = &qqi->qqi_glb_obj->do_index_ops->dio_it;
 
-	it = iops->init(env, qqi->qqi_glb_obj, 0, BYPASS_CAPA);
+	it = iops->init(env, qqi->qqi_glb_obj, 0);
 	if (IS_ERR(it)) {
 		CWARN("%s: Initialize it for "DFID" failed. %ld\n",
 		      qsd->qsd_svname, PFID(&qqi->qqi_fid), PTR_ERR(it));
@@ -385,9 +383,9 @@ static int qsd_connected(struct qsd_instance *qsd)
 {
 	int	connected;
 
-	cfs_read_lock(&qsd->qsd_lock);
+	read_lock(&qsd->qsd_lock);
 	connected = qsd->qsd_exp_valid ? 1 : 0;
-	cfs_read_unlock(&qsd->qsd_lock);
+	read_unlock(&qsd->qsd_lock);
 
 	return connected;
 }
@@ -396,9 +394,9 @@ static int qsd_started(struct qsd_instance *qsd)
 {
 	int	started;
 
-	cfs_read_lock(&qsd->qsd_lock);
+	read_lock(&qsd->qsd_lock);
 	started = qsd->qsd_started ? 1 : 0;
-	cfs_read_unlock(&qsd->qsd_lock);
+	read_unlock(&qsd->qsd_lock);
 
 	return started;
 }
@@ -417,8 +415,6 @@ static int qsd_reint_main(void *args)
 	int			 rc;
 	ENTRY;
 
-	cfs_daemonize("qsd_reint");
-
 	CDEBUG(D_QUOTA, "%s: Starting reintegration thread for "DFID"\n",
 	       qsd->qsd_svname, PFID(&qqi->qqi_fid));
 
@@ -426,7 +422,7 @@ static int qsd_reint_main(void *args)
 	lu_ref_add(&qqi->qqi_reference, "reint_thread", thread);
 
 	thread_set_flags(thread, SVC_RUNNING);
-	cfs_waitq_signal(&thread->t_ctl_waitq);
+	wake_up(&thread->t_ctl_waitq);
 
 	OBD_ALLOC_PTR(env);
 	if (env == NULL)
@@ -451,17 +447,17 @@ static int qsd_reint_main(void *args)
 
 	memset(&qti->qti_lvb, 0, sizeof(qti->qti_lvb));
 
-	cfs_read_lock(&qsd->qsd_lock);
+	read_lock(&qsd->qsd_lock);
 	/* check whether we already own a global quota lock for this type */
 	if (lustre_handle_is_used(&qqi->qqi_lockh) &&
 	    ldlm_lock_addref_try(&qqi->qqi_lockh, qsd_glb_einfo.ei_mode) == 0) {
-		cfs_read_unlock(&qsd->qsd_lock);
+		read_unlock(&qsd->qsd_lock);
 		/* force refresh of global & slave index copy */
-		qti->qti_lvb.l_lquota.lvb_glb_ver = ~0ULL;
+		qti->qti_lvb.lvb_glb_ver = ~0ULL;
 		qti->qti_slv_ver = ~0ULL;
 	} else {
 		/* no valid lock found, let's enqueue a new one */
-		cfs_read_unlock(&qsd->qsd_lock);
+		read_unlock(&qsd->qsd_lock);
 
 		memset(&qti->qti_body, 0, sizeof(qti->qti_body));
 		memcpy(&qti->qti_body.qb_fid, &qqi->qqi_fid,
@@ -473,9 +469,9 @@ static int qsd_reint_main(void *args)
 		if (rc)
 			GOTO(out_env_init, rc);
 
-		CDEBUG(D_QUOTA, "%s: glb_ver:"LPU64"/"LPU64",slv_ver:"LPU64"/"
-		       LPU64"\n", qsd->qsd_svname,
-		       qti->qti_lvb.l_lquota.lvb_glb_ver, qqi->qqi_glb_ver,
+		CDEBUG(D_QUOTA, "%s: glb_ver:%llu/%llu,slv_ver:%llu/"
+		       "%llu\n", qsd->qsd_svname,
+		       qti->qti_lvb.lvb_glb_ver, qqi->qqi_glb_ver,
 		       qti->qti_slv_ver, qqi->qqi_slv_ver);
 	}
 
@@ -485,7 +481,7 @@ static int qsd_reint_main(void *args)
 
 	OBD_FAIL_TIMEOUT(OBD_FAIL_QUOTA_DELAY_REINT, 10);
 
-	if (qqi->qqi_glb_ver != qti->qti_lvb.l_lquota.lvb_glb_ver) {
+	if (qqi->qqi_glb_ver != qti->qti_lvb.lvb_glb_ver) {
 		rc = qsd_reint_index(env, qqi, true);
 		if (rc) {
 			CWARN("%s: reint global for "DFID" failed. %d\n",
@@ -503,7 +499,7 @@ static int qsd_reint_main(void *args)
 	if (qqi->qqi_slv_ver != qti->qti_slv_ver) {
 		rc = qsd_reint_index(env, qqi, false);
 		if (rc) {
-			CWARN("%s: Reint slave for "DFID" failed. %d\n",
+			CWARN("%s: reintegration for "DFID" failed with %d\n",
 			      qsd->qsd_svname, PFID(&qqi->qqi_slv_fid), rc);
 			GOTO(out_lock, rc);
 		}
@@ -511,7 +507,7 @@ static int qsd_reint_main(void *args)
 		qsd_bump_version(qqi, qqi->qqi_slv_ver, false);
 	}
 
-	/* wait for the connection to master established */
+	/* wait for the qsd instance started (target recovery done) */
 	l_wait_event(thread->t_ctl_waitq,
 		     qsd_started(qsd) || !thread_is_running(thread), &lwi);
 
@@ -521,8 +517,8 @@ static int qsd_reint_main(void *args)
 	/* Step 4: start reconciliation for each enforced ID */
 	rc = qsd_reconciliation(env, qqi);
 	if (rc)
-		CWARN("%s: reconciliation failed. "DFID", %d\n",
-		      qsd->qsd_svname, PFID(&qti->qti_fid), rc);
+		CWARN("%s: reconciliation for "DFID" failed with %d\n",
+		      qsd->qsd_svname, PFID(&qqi->qqi_slv_fid), rc);
 
 	EXIT;
 out_lock:
@@ -532,15 +528,16 @@ out_env_init:
 out_env:
 	OBD_FREE_PTR(env);
 out:
-	cfs_write_lock(&qsd->qsd_lock);
+	write_lock(&qsd->qsd_lock);
 	qqi->qqi_reint = 0;
-	cfs_write_unlock(&qsd->qsd_lock);
+	write_unlock(&qsd->qsd_lock);
+
+	thread_set_flags(thread, SVC_STOPPED);
+	wake_up(&thread->t_ctl_waitq);
 
-	qqi_putref(qqi);
 	lu_ref_del(&qqi->qqi_reference, "reint_thread", thread);
+	qqi_putref(qqi);
 
-	thread_set_flags(thread, SVC_STOPPED);
-	cfs_waitq_signal(&thread->t_ctl_waitq);
 	return rc;
 }
 
@@ -551,20 +548,20 @@ void qsd_stop_reint_thread(struct qsd_qtype_info *qqi)
 
 	if (!thread_is_stopped(thread)) {
 		thread_set_flags(thread, SVC_STOPPING);
-		cfs_waitq_signal(&thread->t_ctl_waitq);
+		wake_up(&thread->t_ctl_waitq);
 
 		l_wait_event(thread->t_ctl_waitq,
 			     thread_is_stopped(thread), &lwi);
 	}
 }
 
-static int qsd_entry_iter_cb(cfs_hash_t *hs, cfs_hash_bd_t *bd,
-			     cfs_hlist_node_t *hnode, void *data)
+static int qsd_entry_iter_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
+			     struct hlist_node *hnode, void *data)
 {
 	struct lquota_entry	*lqe;
 	int			*pending = (int *)data;
 
-	lqe = cfs_hlist_entry(hnode, struct lquota_entry, lqe_hash);
+	lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
 	LASSERT(atomic_read(&lqe->lqe_ref) > 0);
 
 	lqe_read_lock(lqe);
@@ -584,20 +581,20 @@ static bool qsd_pending_updates(struct qsd_qtype_info *qqi)
 	ENTRY;
 
 	/* any pending quota adjust? */
-	cfs_spin_lock(&qsd->qsd_adjust_lock);
-	cfs_list_for_each_entry_safe(lqe, n, &qsd->qsd_adjust_list, lqe_link) {
+	spin_lock(&qsd->qsd_adjust_lock);
+	list_for_each_entry_safe(lqe, n, &qsd->qsd_adjust_list, lqe_link) {
 		if (lqe2qqi(lqe) == qqi) {
-			cfs_list_del_init(&lqe->lqe_link);
+			list_del_init(&lqe->lqe_link);
 			lqe_putref(lqe);
 		}
 	}
-	cfs_spin_unlock(&qsd->qsd_adjust_lock);
+	spin_unlock(&qsd->qsd_adjust_lock);
 
 	/* any pending updates? */
-	cfs_read_lock(&qsd->qsd_lock);
-	cfs_list_for_each_entry(upd, &qsd->qsd_upd_list, qur_link) {
+	read_lock(&qsd->qsd_lock);
+	list_for_each_entry(upd, &qsd->qsd_upd_list, qur_link) {
 		if (upd->qur_qqi == qqi) {
-			cfs_read_unlock(&qsd->qsd_lock);
+			read_unlock(&qsd->qsd_lock);
 			CDEBUG(D_QUOTA, "%s: pending %s updates for type:%d.\n",
 			       qsd->qsd_svname,
 			       upd->qur_global ? "global" : "slave",
@@ -605,7 +602,7 @@ static bool qsd_pending_updates(struct qsd_qtype_info *qqi)
 			GOTO(out, updates = true);
 		}
 	}
-	cfs_read_unlock(&qsd->qsd_lock);
+	read_unlock(&qsd->qsd_lock);
 
 	/* any pending quota request? */
 	cfs_hash_for_each_safe(qqi->qqi_site->lqs_hash, qsd_entry_iter_cb,
@@ -628,42 +625,58 @@ int qsd_start_reint_thread(struct qsd_qtype_info *qqi)
 {
 	struct ptlrpc_thread	*thread = &qqi->qqi_reint_thread;
 	struct qsd_instance	*qsd = qqi->qqi_qsd;
-	struct l_wait_info	lwi = { 0 };
-	int			rc;
+	struct l_wait_info	 lwi = { 0 };
+	struct task_struct	*task;
+	int			 rc;
+	char			*name;
 	ENTRY;
 
 	/* don't bother to do reintegration when quota isn't enabled */
-	if (!qsd_type_enabled(qqi->qqi_qsd, qqi->qqi_qtype))
+	if (!qsd_type_enabled(qsd, qqi->qqi_qtype))
+		RETURN(0);
+
+	if (qsd->qsd_acct_failed)
+		/* no space accounting support, can't enable enforcement */
 		RETURN(0);
 
 	/* check if the reintegration has already started or finished */
-	cfs_write_lock(&qsd->qsd_lock);
+	write_lock(&qsd->qsd_lock);
 
 	if ((qqi->qqi_glb_uptodate && qqi->qqi_slv_uptodate) ||
 	     qqi->qqi_reint || qsd->qsd_stopping) {
-		cfs_write_unlock(&qsd->qsd_lock);
+		write_unlock(&qsd->qsd_lock);
 		RETURN(0);
 	}
 	qqi->qqi_reint = 1;
 
-	cfs_write_unlock(&qsd->qsd_lock);
+	write_unlock(&qsd->qsd_lock);
 
 	/* there could be some unfinished global or index entry updates
 	 * (very unlikely), to avoid them messing up with the reint
 	 * procedure, we just return and try to re-start reint later. */
 	if (qsd_pending_updates(qqi)) {
-		cfs_write_lock(&qsd->qsd_lock);
+		write_lock(&qsd->qsd_lock);
 		qqi->qqi_reint = 0;
-		cfs_write_unlock(&qsd->qsd_lock);
+		write_unlock(&qsd->qsd_lock);
 		RETURN(0);
 	}
 
-	rc = cfs_create_thread(qsd_reint_main, (void *)qqi, 0);
-	if (rc < 0) {
+	OBD_ALLOC(name, MTI_NAME_MAXLEN);
+	if (name == NULL)
+		RETURN(-ENOMEM);
+
+	snprintf(name, MTI_NAME_MAXLEN, "qsd_reint_%d.%s",
+		 qqi->qqi_qtype, qsd->qsd_svname);
+
+	task = kthread_run(qsd_reint_main, qqi, name);
+	OBD_FREE(name, MTI_NAME_MAXLEN);
+
+	if (IS_ERR(task)) {
+		rc = PTR_ERR(task);
 		thread_set_flags(thread, SVC_STOPPED);
-		cfs_write_lock(&qsd->qsd_lock);
+		write_lock(&qsd->qsd_lock);
 		qqi->qqi_reint = 0;
-		cfs_write_unlock(&qsd->qsd_lock);
+		write_unlock(&qsd->qsd_lock);
 		RETURN(rc);
 	}