X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fquota%2Fqsd_reint.c;h=59e38777942ad421aee34e2802e28c02b52df3dc;hp=040b88518d5df44d6120a6009e767cecd6e4f4ed;hb=826a8ba3e9bacbd48a7b52f715640287f4a9b952;hpb=b2cb6fd1095f9c483b7bc1ebbbfdaef719aea87c

diff --git a/lustre/quota/qsd_reint.c b/lustre/quota/qsd_reint.c
index 040b885..59e3877 100644
--- a/lustre/quota/qsd_reint.c
+++ b/lustre/quota/qsd_reint.c
@@ -21,7 +21,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2011, 2012, Intel, Inc.
+ * Copyright (c) 2012, 2013, Intel Corporation.
  * Use is subject to license terms.
  *
  * Author: Johann Lombardi <johann.lombardi@intel.com>
@@ -44,7 +44,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;
@@ -52,16 +52,17 @@ static void qsd_reint_completion(const struct lu_env *env,
 	ENTRY;
 
 	if (rc) {
-		CERROR("%s: failed to enqueue global quota lock, glb "
-		       "fid:"DFID", rc:%d\n", qsd->qsd_svname,
-		       PFID(&req_qbody->qb_fid), rc);
+		CDEBUG_LIMIT(rc != -EAGAIN ? D_ERROR : D_QUOTA,
+			     "%s: failed to enqueue global quota lock, glb fid:"
+			     DFID", rc:%d\n", qsd->qsd_svname,
+			     PFID(&req_qbody->qb_fid), rc);
 		RETURN_EXIT;
 	}
 
 	CDEBUG(D_QUOTA, "%s: global quota lock successfully acquired, glb "
 	       "fid:"DFID", glb ver:"LPU64", slv fid:"DFID", slv ver:"LPU64"\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;
@@ -82,6 +83,8 @@ static int qsd_reint_qid(const struct lu_env *env, struct qsd_qtype_info *qqi,
 	if (IS_ERR(lqe))
 		RETURN(PTR_ERR(lqe));
 
+	LQUOTA_DEBUG(lqe, "reintegrating entry");
+
 	rc = qsd_update_lqe(env, lqe, global, rec);
 	if (rc)
 		GOTO(out, rc);
@@ -95,29 +98,52 @@ 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);
+	struct qsd_instance	*qsd = qqi->qqi_qsd;
 	union lquota_id		*qid = &qti->qti_id;
 	int			 i, j, k, size;
 	int			 rc = 0;
 	ENTRY;
 
+	CDEBUG(D_QUOTA, "%s: processing %d pages for %s index\n",
+	       qsd->qsd_svname, npages, global ? "global" : "slave");
+
 	/* sanity check on the record size */
 	if ((global && ii->ii_recsize != sizeof(struct lquota_glb_rec)) ||
 	    (!global && ii->ii_recsize != sizeof(struct lquota_slv_rec))) {
-		CERROR("Invalid record size:%d, global:%s\n",
-		       ii->ii_recsize, global ? "true" : "false");
+		CERROR("%s: invalid record size (%d) for %s index\n",
+		       qsd->qsd_svname, ii->ii_recsize,
+		       global ? "global" : "slave");
 		RETURN(-EINVAL);
 	}
 
-	size = ii->ii_recsize + ii->ii_keysize + sizeof(__u64);
+	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)
+				/* swab header */
+				lustre_swab_lip_header(&lip->lp_idx);
+
+			if (lip->lp_idx.lip_magic != LIP_MAGIC) {
+				CERROR("%s: invalid magic (%x != %x) for page "
+				       "%d/%d while transferring %s index\n",
+				       qsd->qsd_svname, lip->lp_idx.lip_magic,
+				       LIP_MAGIC, i + 1, npages,
+				       global ? "global" : "slave");
+				GOTO(out, rc = -EINVAL);
+			}
+
+			CDEBUG(D_QUOTA, "%s: processing page %d/%d with %d "
+			       "entries for %s index\n", qsd->qsd_svname, i + 1,
+			       npages, lip->lp_idx.lip_nr,
+			       global ? "global" : "slave");
+
 			for (k = 0; k < lip->lp_idx.lip_nr; k++) {
 				char *entry;
 
@@ -147,7 +173,7 @@ static int qsd_reint_entries(const struct lu_env *env,
 			lip++;
 		}
 out:
-		cfs_kunmap(pages[i]);
+		kunmap(pages[i]);
 		if (rc)
 			break;
 	}
@@ -161,7 +187,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;
@@ -171,15 +197,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_CACHE_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_IOFS);
 		if (pages[i] == NULL)
 			GOTO(out, rc = -ENOMEM);
 	}
@@ -232,7 +258,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_CACHE_SHIFT - LU_PAGE_SHIFT;
 
 	if (pg_cnt > npages) {
 		CERROR("%s: master returned more pages than expected, %u > %u"
@@ -252,7 +278,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));
 	}
 
@@ -318,38 +344,32 @@ static int qsd_reconciliation(const struct lu_env *env,
 
 		lqe = lqe_locate(env, qqi->qqi_site, qid);
 		if (IS_ERR(lqe)) {
-			CWARN("%s: Fail to locate lqe. "DFID", %ld\n",
+			CWARN("%s: failed to locate lqe. "DFID", %ld\n",
 			      qsd->qsd_svname, PFID(&qqi->qqi_fid),
 			      PTR_ERR(lqe));
 			GOTO(out, rc = PTR_ERR(lqe));
 		}
 
-		if (!lqe->lqe_enforced) {
-			lqe_putref(lqe);
-			goto next;
-		}
-
 		rc = qsd_refresh_usage(env, lqe);
 		if (rc) {
-			CWARN("%s: Fail to get usage. "DFID", %d\n",
+			CWARN("%s: failed to get usage. "DFID", %d\n",
 			      qsd->qsd_svname, PFID(&qqi->qqi_fid), rc);
 			lqe_putref(lqe);
 			GOTO(out, rc);
 		}
 
-		rc = qsd_dqacq(env, lqe, QSD_REP);
+		rc = qsd_adjust(env, lqe);
 		lqe_putref(lqe);
-
 		if (rc) {
-			CWARN("%s: Fail to report quota. "DFID", %d\n",
+			CWARN("%s: failed to report quota. "DFID", %d\n",
 			      qsd->qsd_svname, PFID(&qqi->qqi_fid), rc);
 			GOTO(out, rc);
 		}
 next:
 		rc = iops->next(env, it);
 		if (rc < 0)
-			CWARN("%s: Error next "DFID". %d\n", qsd->qsd_svname,
-			      PFID(&qqi->qqi_fid), rc);
+			CWARN("%s: failed to parse index, ->next error:%d "DFID
+			      "\n", qsd->qsd_svname, rc, PFID(&qqi->qqi_fid));
 	} while (rc == 0);
 
 	/* reach the end */
@@ -365,9 +385,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;
 }
@@ -376,9 +396,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;
 }
@@ -397,8 +417,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));
 
@@ -431,17 +449,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,
@@ -455,7 +473,7 @@ static int qsd_reint_main(void *args)
 
 		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,
+		       qti->qti_lvb.lvb_glb_ver, qqi->qqi_glb_ver,
 		       qti->qti_slv_ver, qqi->qqi_slv_ver);
 	}
 
@@ -465,7 +483,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",
@@ -491,7 +509,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);
 
@@ -512,9 +530,9 @@ 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);
 
 	qqi_putref(qqi);
 	lu_ref_del(&qqi->qqi_reference, "reint_thread", thread);
@@ -538,36 +556,126 @@ void qsd_stop_reint_thread(struct qsd_qtype_info *qqi)
 	}
 }
 
+static int qsd_entry_iter_cb(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+			     cfs_hlist_node_t *hnode, void *data)
+{
+	struct lquota_entry	*lqe;
+	int			*pending = (int *)data;
+
+	lqe = cfs_hlist_entry(hnode, struct lquota_entry, lqe_hash);
+	LASSERT(atomic_read(&lqe->lqe_ref) > 0);
+
+	lqe_read_lock(lqe);
+	*pending += lqe->lqe_pending_req;
+	lqe_read_unlock(lqe);
+
+	return 0;
+}
+
+static bool qsd_pending_updates(struct qsd_qtype_info *qqi)
+{
+	struct qsd_instance	*qsd = qqi->qqi_qsd;
+	struct qsd_upd_rec	*upd;
+	struct lquota_entry	*lqe, *n;
+	int			 dqacq = 0;
+	bool			 updates = false;
+	ENTRY;
+
+	/* any pending quota adjust? */
+	spin_lock(&qsd->qsd_adjust_lock);
+	cfs_list_for_each_entry_safe(lqe, n, &qsd->qsd_adjust_list, lqe_link) {
+		if (lqe2qqi(lqe) == qqi) {
+			cfs_list_del_init(&lqe->lqe_link);
+			lqe_putref(lqe);
+		}
+	}
+	spin_unlock(&qsd->qsd_adjust_lock);
+
+	/* any pending updates? */
+	read_lock(&qsd->qsd_lock);
+	cfs_list_for_each_entry(upd, &qsd->qsd_upd_list, qur_link) {
+		if (upd->qur_qqi == qqi) {
+			read_unlock(&qsd->qsd_lock);
+			CDEBUG(D_QUOTA, "%s: pending %s updates for type:%d.\n",
+			       qsd->qsd_svname,
+			       upd->qur_global ? "global" : "slave",
+			       qqi->qqi_qtype);
+			GOTO(out, updates = true);
+		}
+	}
+	read_unlock(&qsd->qsd_lock);
+
+	/* any pending quota request? */
+	cfs_hash_for_each_safe(qqi->qqi_site->lqs_hash, qsd_entry_iter_cb,
+			       &dqacq);
+	if (dqacq) {
+		CDEBUG(D_QUOTA, "%s: pending dqacq for type:%d.\n",
+		       qsd->qsd_svname, qqi->qqi_qtype);
+		updates = true;
+	}
+	EXIT;
+out:
+	if (updates)
+		CERROR("%s: Delaying reintegration for qtype:%d until pending "
+		       "updates are flushed.\n",
+		       qsd->qsd_svname, qqi->qqi_qtype);
+	return updates;
+}
+
 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 };
+	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);
 
-	rc = cfs_create_thread(qsd_reint_main, (void *)qqi, 0);
-	if (rc < 0) {
+	/* 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)) {
+		write_lock(&qsd->qsd_lock);
+		qqi->qqi_reint = 0;
+		write_unlock(&qsd->qsd_lock);
+		RETURN(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);
+
+	rc = PTR_ERR(kthread_run(qsd_reint_main, (void *)qqi, name));
+	OBD_FREE(name, MTI_NAME_MAXLEN);
+
+	if (IS_ERR_VALUE(rc)) {
 		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);
 	}