dlm: master - dlm_controld: ignore plocks until checkpoint time

[David Teigland]

Gitweb:       
http://git.fedorahosted.org/git/dlm.git?p=dlm.git;a=commitdiff;h=2a7c1503...
Commit:        2a7c15034a0101d9bb8948f1b40fc70728a79419
Parent:        b5e48c2ee2af3440bb9f348536042aff45660d3f
Author:        David Teigland <teigland(a)redhat.com&gt;
AuthorDate:    Fri Aug 13 15:05:18 2010 -0500
Committer:     David Teigland <teigland(a)redhat.com&gt;
CommitterDate: Fri Aug 13 15:20:19 2010 -0500

dlm_controld: ignore plocks until checkpoint time

After our join confchg, we need to ignore plock messages
until the point in time where the ckpt_node saves plock
state (final start message received).  At that time we need
to shift from ignoring plock messages to saving plock
messages that will be applied on top of the plock state that
is read from the checkpoint.  The code was not ignoring
plock messages during the first stage of the process, which
led to incorrect plock state.

bz 623810

Signed-off-by: David Teigland <teigland(a)redhat.com&gt;
---
 group/dlm_controld/cpg.c   |   35 +++++++++++++++++++++++++++++++++++
 group/dlm_controld/plock.c |   10 +++++++---
 2 files changed, 42 insertions(+), 3 deletions(-)

diff --git a/group/dlm_controld/cpg.c b/group/dlm_controld/cpg.c
index 2e5a1bb..9b0d223 100644
--- a/group/dlm_controld/cpg.c
+++ b/group/dlm_controld/cpg.c
@@ -1099,6 +1099,10 @@ static void receive_plocks_stored(struct lockspace *ls, struct
dlm_header *hd,
 		  "need_plocks %d", hd->nodeid, hd->msgdata, hd->flags,
 		  hd->msgdata2, ls->need_plocks);
 
+	log_plock(ls, "receive_plocks_stored %d:%u flags %x sig %x "
+		  "need_plocks %d", hd->nodeid, hd->msgdata, hd->flags,
+		  hd->msgdata2, ls->need_plocks);
+
 	ls->last_plock_sig = hd->msgdata2;
 
 	if (!ls->need_plocks)
@@ -1269,6 +1273,8 @@ static void prepare_plocks(struct lockspace *ls)
 	struct member *memb;
 	uint32_t sig;
 
+	log_plock(ls, "prepare_plocks");
+
 	if (!cfgd_enable_plock || ls->disable_plock)
 		return;
 
@@ -1569,6 +1575,12 @@ static void dlm_header_in(struct dlm_header *hd)
 	hd->msgdata2    = le32_to_cpu(hd->msgdata2);
 }
 
+/* after our join confchg, we want to ignore plock messages (see need_plocks
+   checks below) until the point in time where the ckpt_node saves plock
+   state (final start message received); at this time we want to shift from
+   ignoring plock messages to saving plock messages to apply on top of the
+   plock state that we read. */
+
 static void deliver_cb(cpg_handle_t handle,
 		       const struct cpg_name *group_name,
 		       uint32_t nodeid, uint32_t pid,
@@ -1576,6 +1588,7 @@ static void deliver_cb(cpg_handle_t handle,
 {
 	struct lockspace *ls;
 	struct dlm_header *hd;
+	int ignore_plock;
 
 	ls = find_ls_handle(handle);
 	if (!ls) {
@@ -1606,6 +1619,8 @@ static void deliver_cb(cpg_handle_t handle,
 		return;
 	}
 
+	ignore_plock = 0;
+
 	switch (hd->type) {
 	case DLM_MSG_START:
 		receive_start(ls, hd, len);
@@ -1614,6 +1629,10 @@ static void deliver_cb(cpg_handle_t handle,
 	case DLM_MSG_PLOCK:
 		if (ls->disable_plock)
 			break;
+		if (ls->need_plocks && !ls->save_plocks) {
+			ignore_plock = 1;
+			break;
+		}
 		if (cfgd_enable_plock)
 			receive_plock(ls, hd, len);
 		else
@@ -1624,6 +1643,10 @@ static void deliver_cb(cpg_handle_t handle,
 	case DLM_MSG_PLOCK_OWN:
 		if (ls->disable_plock)
 			break;
+		if (ls->need_plocks && !ls->save_plocks) {
+			ignore_plock = 1;
+			break;
+		}
 		if (cfgd_enable_plock && cfgd_plock_ownership)
 			receive_own(ls, hd, len);
 		else
@@ -1635,6 +1658,10 @@ static void deliver_cb(cpg_handle_t handle,
 	case DLM_MSG_PLOCK_DROP:
 		if (ls->disable_plock)
 			break;
+		if (ls->need_plocks && !ls->save_plocks) {
+			ignore_plock = 1;
+			break;
+		}
 		if (cfgd_enable_plock && cfgd_plock_ownership)
 			receive_drop(ls, hd, len);
 		else
@@ -1647,6 +1674,10 @@ static void deliver_cb(cpg_handle_t handle,
 	case DLM_MSG_PLOCK_SYNC_WAITER:
 		if (ls->disable_plock)
 			break;
+		if (ls->need_plocks && !ls->save_plocks) {
+			ignore_plock = 1;
+			break;
+		}
 		if (cfgd_enable_plock && cfgd_plock_ownership)
 			receive_sync(ls, hd, len);
 		else
@@ -1701,6 +1732,10 @@ static void deliver_cb(cpg_handle_t handle,
 		log_error("unknown msg type %d", hd->type);
 	}
 
+	if (ignore_plock)
+		log_plock(ls, "msg %s nodeid %d need_plock ignore",
+			  msg_name(hd->type), nodeid);
+
 	apply_changes(ls);
 }
 
diff --git a/group/dlm_controld/plock.c b/group/dlm_controld/plock.c
index e292e95..41b0423 100644
--- a/group/dlm_controld/plock.c
+++ b/group/dlm_controld/plock.c
@@ -1562,11 +1562,12 @@ void process_saved_plocks(struct lockspace *ls)
 {
 	struct save_msg *sm, *sm2;
 	struct dlm_header *hd;
+	int count = 0;
 
-	if (list_empty(&ls->saved_messages))
-		return;
+	log_plock(ls, "process_saved_plocks begin");
 
-	log_plock(ls, "process_saved_plocks");
+	if (list_empty(&ls->saved_messages))
+		goto out;
 
 	list_for_each_entry_safe(sm, sm2, &ls->saved_messages, list) {
 		hd = (struct dlm_header *)sm->buf;
@@ -1591,7 +1592,10 @@ void process_saved_plocks(struct lockspace *ls)
 
 		list_del(&sm->list);
 		free(sm);
+		count++;
 	}
+ out:
+	log_plock(ls, "process_saved_plocks %d end", count);
 }
 
 /* locks still marked SYNCING should not go into the ckpt; the new node