ocfs2_dlm: Check for migrateable lockres in dlm_empty_lockres()

In dlm_migrate_lockres(), we check upfront whether the lockres is a
candidate for migration. This patch encapsulates that code in a separate
function so that dlm_empty_lockres() can also use it during umount. This
patch addresses the umount process spinning problem.

Signed-off-by: Sunil Mushran <sunil.mushran@oracle.com>
Signed-off-by: Mark Fasheh <mark.fasheh@oracle.com>

fs/ocfs2/dlm/dlmdomain.c

@@ -657,6 +657,8 @@ void dlm_unregister_domain(struct dlm_ctxt *dlm)
 		dlm_kick_thread(dlm, NULL);
 
 		while (dlm_migrate_all_locks(dlm)) {
+			/* Give dlm_thread time to purge the lockres' */
+			msleep(500);
 			mlog(0, "%s: more migration to do\n", dlm->name);
 		}
 		dlm_mark_domain_leaving(dlm);

fs/ocfs2/dlm/dlmmaster.c

@@ -2424,6 +2424,57 @@ static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data)
 	dlm_lockres_put(res);
 }
 
+/* Checks whether the lockres can be migrated. Returns 0 if yes, < 0
+ * if not. If 0, numlocks is set to the number of locks in the lockres.
+ */
+static int dlm_is_lockres_migrateable(struct dlm_ctxt *dlm,
+				      struct dlm_lock_resource *res,
+				      int *numlocks)
+{
+	int ret;
+	int i;
+	int count = 0;
+	struct list_head *queue, *iter;
+	struct dlm_lock *lock;
+
+	assert_spin_locked(&res->spinlock);
+
+	ret = -EINVAL;
+	if (res->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
+		mlog(0, "cannot migrate lockres with unknown owner!\n");
+		goto leave;
+	}
+
+	if (res->owner != dlm->node_num) {
+		mlog(0, "cannot migrate lockres this node doesn't own!\n");
+		goto leave;
+	}
+
+	ret = 0;
+	queue = &res->granted;
+	for (i = 0; i < 3; i++) {
+		list_for_each(iter, queue) {
+			lock = list_entry(iter, struct dlm_lock, list);
+			++count;
+			if (lock->ml.node == dlm->node_num) {
+				mlog(0, "found a lock owned by this node still "
+				     "on the %s queue!  will not migrate this "
+				     "lockres\n", (i == 0 ? "granted" :
+						   (i == 1 ? "converting" :
+						    "blocked")));
+				ret = -ENOTEMPTY;
+				goto leave;
+			}
+		}
+		queue++;
+	}
+
+	*numlocks = count;
+	mlog(0, "migrateable lockres having %d locks\n", *numlocks);
+
+leave:
+	return ret;
+}
 
 /*
  * DLM_MIGRATE_LOCKRES
@@ -2437,14 +2488,12 @@ static int dlm_migrate_lockres(struct dlm_ctxt *dlm,
 	struct dlm_master_list_entry *mle = NULL;
 	struct dlm_master_list_entry *oldmle = NULL;
  	struct dlm_migratable_lockres *mres = NULL;
-	int ret = -EINVAL;
+	int ret = 0;
 	const char *name;
 	unsigned int namelen;
 	int mle_added = 0;
-	struct list_head *queue, *iter;
-	int i;
-	struct dlm_lock *lock;
-	int empty = 1, wake = 0;
+	int numlocks;
+	int wake = 0;
 
 	if (!dlm_grab(dlm))
 		return -EINVAL;
@@ -2458,42 +2507,16 @@ static int dlm_migrate_lockres(struct dlm_ctxt *dlm,
 	 * ensure this lockres is a proper candidate for migration
 	 */
 	spin_lock(&res->spinlock);
-	if (res->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
-		mlog(0, "cannot migrate lockres with unknown owner!\n");
-		spin_unlock(&res->spinlock);
-		goto leave;
-	}
-	if (res->owner != dlm->node_num) {
-		mlog(0, "cannot migrate lockres this node doesn't own!\n");
+	ret = dlm_is_lockres_migrateable(dlm, res, &numlocks);
+	if (ret < 0) {
 		spin_unlock(&res->spinlock);
 		goto leave;
 	}
-	mlog(0, "checking queues...\n");
-	queue = &res->granted;
-	for (i=0; i<3; i++) {
-		list_for_each(iter, queue) {
-			lock = list_entry (iter, struct dlm_lock, list);
-			empty = 0;
-			if (lock->ml.node == dlm->node_num) {
-				mlog(0, "found a lock owned by this node "
-				     "still on the %s queue!  will not "
-				     "migrate this lockres\n",
-				     i==0 ? "granted" :
-				     (i==1 ? "converting" : "blocked"));
-				spin_unlock(&res->spinlock);
-				ret = -ENOTEMPTY;
-				goto leave;
-			}
-		}
-		queue++;
-	}
-	mlog(0, "all locks on this lockres are nonlocal.  continuing\n");
 	spin_unlock(&res->spinlock);
 
 	/* no work to do */
-	if (empty) {
+	if (numlocks == 0) {
 		mlog(0, "no locks were found on this lockres! done!\n");
-		ret = 0;
 		goto leave;
 	}
 
@@ -2729,6 +2752,7 @@ int dlm_empty_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
 {
 	int ret;
 	int lock_dropped = 0;
+	int numlocks;
 
 	spin_lock(&res->spinlock);
 	if (res->owner != dlm->node_num) {
@@ -2740,6 +2764,13 @@ int dlm_empty_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
 		spin_unlock(&res->spinlock);
 		goto leave;
 	}
+
+	/* No need to migrate a lockres having no locks */
+	ret = dlm_is_lockres_migrateable(dlm, res, &numlocks);
+	if (ret >= 0 && numlocks == 0) {
+		spin_unlock(&res->spinlock);
+		goto leave;
+	}
 	spin_unlock(&res->spinlock);
 
 	/* Wheee! Migrate lockres here! Will sleep so drop spinlock. */