[Ocfs2-commits] smushran commits r2968 - branches/ocfs2-1.2/fs/ocfs2/dlm

[svn-commits at oss.oracle.com]

Author: smushran
Date: 2007-01-29 13:57:01 -0800 (Mon, 29 Jan 2007)
New Revision: 2968

Modified:
   branches/ocfs2-1.2/fs/ocfs2/dlm/dlmcommon.h
   branches/ocfs2-1.2/fs/ocfs2/dlm/dlmmaster.c
Log:
ocfs2_dlm: Ensure correct ordering of set/clear refmap bit on lockres

Eventhough the set refmap bit message is sent before the clear refmap
message, currently there is no guarentee that the set message will be
handled before the clear. This patch prevents the clear refmap to be
processed while the node is sending assert master messages to other
nodes. (The set refmap message is sent as a response to the assert
master request).

Signed-off-by: mfasheh

Modified: branches/ocfs2-1.2/fs/ocfs2/dlm/dlmcommon.h
===================================================================
--- branches/ocfs2-1.2/fs/ocfs2/dlm/dlmcommon.h	2007-01-29 21:55:05 UTC (rev 2967)
+++ branches/ocfs2-1.2/fs/ocfs2/dlm/dlmcommon.h	2007-01-29 21:57:01 UTC (rev 2968)
@@ -168,6 +168,12 @@
 };
 
 
+struct dlm_deref_lockres_priv
+{
+	struct dlm_lock_resource *deref_res;
+	u8 deref_node;
+};
+
 struct dlm_work_item
 {
 	struct list_head list;
@@ -178,6 +184,7 @@
 		struct dlm_request_all_locks_priv ral;
 		struct dlm_mig_lockres_priv ml;
 		struct dlm_assert_master_priv am;
+		struct dlm_deref_lockres_priv dl;
 	} u;
 };
 
@@ -882,6 +889,7 @@
 void dlm_assert_master_post_handler(int status, void *data, void *ret_data);
 int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
 			      void **ret_data);
+void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data);
 int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data,
 				void **ret_data);
 int dlm_mig_lockres_handler(struct o2net_msg *msg, u32 len, void *data,

Modified: branches/ocfs2-1.2/fs/ocfs2/dlm/dlmmaster.c
===================================================================
--- branches/ocfs2-1.2/fs/ocfs2/dlm/dlmmaster.c	2007-01-29 21:55:05 UTC (rev 2967)
+++ branches/ocfs2-1.2/fs/ocfs2/dlm/dlmmaster.c	2007-01-29 21:57:01 UTC (rev 2968)
@@ -1715,6 +1715,11 @@
 	unsigned int namelen = res->lockname.len;
 
 	BUG_ON(namelen > O2NM_MAX_NAME_LEN);
+
+	spin_lock(&res->spinlock);
+	res->state |= DLM_LOCK_RES_SETREF_INPROG;
+	spin_unlock(&res->spinlock);
+
 again:
 	reassert = 0;
 
@@ -1787,6 +1792,11 @@
 	if (reassert)
 		goto again;
 
+	spin_lock(&res->spinlock);
+	res->state &= ~DLM_LOCK_RES_SETREF_INPROG;
+	spin_unlock(&res->spinlock);
+	wake_up(&res->wq);
+
 	return ret;
 }
 
@@ -2292,6 +2302,9 @@
 	unsigned int namelen;
 	int ret = -EINVAL;
 	u8 node;
+	struct dlm_work_item *item;
+	int cleared = 0;
+	int dispatch = 0;
 
 	if (!dlm_grab(dlm))
 		return 0;
@@ -2309,7 +2322,6 @@
 		goto done;
 	}
 
-
 	spin_lock(&dlm->spinlock);
 	res = __dlm_lookup_lockres_full(dlm, name, namelen);
 	if (!res) {
@@ -2321,28 +2333,91 @@
 	spin_unlock(&dlm->spinlock);
 
 	spin_lock(&res->spinlock);
-	BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
-	if (test_bit(node, res->refmap)) {
-		ret = 0;
-		dlm_lockres_clear_refmap_bit(node, res);
-	} else {
-		mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
-		     "but it is already dropped!\n", dlm->name, namelen,
-		     name, node);
-		__dlm_print_one_lock_resource(res);
+	if (res->state & DLM_LOCK_RES_SETREF_INPROG)
+		dispatch = 1;
+	else {
+		BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
+		if (test_bit(node, res->refmap)) {
+			dlm_lockres_clear_refmap_bit(node, res);
+			cleared = 1;
+		}
 	}
 	spin_unlock(&res->spinlock);
 
-	if (!ret)
-		dlm_lockres_calc_usage(dlm, res);
+	if (!dispatch) {
+		if (cleared)
+			dlm_lockres_calc_usage(dlm, res);
+		else {
+			mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
+		     	"but it is already dropped!\n", dlm->name,
+		     	res->lockname.len, res->lockname.name, node);
+			__dlm_print_one_lock_resource(res);
+		}
+		ret = 0;
+		goto done;
+	}
+
+	item = kzalloc(sizeof(*item), GFP_NOFS);
+	if (!item) {
+		ret = -ENOMEM;
+		mlog_errno(ret);
+		goto done;
+	}
+
+	dlm_init_work_item(dlm, item, dlm_deref_lockres_worker, NULL);
+	item->u.dl.deref_res = res;
+	item->u.dl.deref_node = node;
+
+	spin_lock(&dlm->work_lock);
+	list_add_tail(&item->list, &dlm->work_list);
+	spin_unlock(&dlm->work_lock);
+
+	queue_work(dlm->dlm_worker, &dlm->dispatched_work);
+	return 0;
+
 done:
 	if (res)
 		dlm_lockres_put(res);
 	dlm_put(dlm);
+
 	return ret;
 }
 
+void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data)
+{
+	struct dlm_ctxt *dlm;
+	struct dlm_lock_resource *res;
+	u8 node;
+	u8 cleared = 0;
 
+	dlm = item->dlm;
+	res = item->u.dl.deref_res;
+	node = item->u.dl.deref_node;
+
+	spin_lock(&res->spinlock);
+	BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
+	if (test_bit(node, res->refmap)) {
+		__dlm_wait_on_lockres_flags(res, DLM_LOCK_RES_SETREF_INPROG);
+		dlm_lockres_clear_refmap_bit(node, res);
+		cleared = 1;
+	}
+	spin_unlock(&res->spinlock);
+
+	if (cleared) {
+		mlog(0, "%s:%.*s node %u ref dropped in dispatch\n",
+		     dlm->name, res->lockname.len, res->lockname.name, node);
+		dlm_lockres_calc_usage(dlm, res);
+	} else {
+		mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
+		     "but it is already dropped!\n", dlm->name,
+		     res->lockname.len, res->lockname.name, node);
+		__dlm_print_one_lock_resource(res);
+	}
+
+	dlm_lockres_put(res);
+}
+
+
 /*
  * DLM_MIGRATE_LOCKRES
  */