[Ocfs2-commits] zab commits r2033 - trunk/fs/ocfs2/cluster

[svn-commits at oss.oracle.com]

Author: zab
Signed-off-by: mfasheh
Date: 2005-03-21 19:22:04 -0600 (Mon, 21 Mar 2005)
New Revision: 2033

Modified:
   trunk/fs/ocfs2/cluster/heartbeat.c
Log:
o do generate events for this node
o make hb starting wait until the region is stable for a few iterations
o make sure to get the slots of the thread's tmp lists after each iteration
                                                                                
Signed-off-by: mfasheh


Modified: trunk/fs/ocfs2/cluster/heartbeat.c
===================================================================
--- trunk/fs/ocfs2/cluster/heartbeat.c	2005-03-22 00:58:10 UTC (rev 2032)
+++ trunk/fs/ocfs2/cluster/heartbeat.c	2005-03-22 01:22:04 UTC (rev 2033)
@@ -86,6 +86,8 @@
  */
 static LIST_HEAD(hb_active_regions);
 
+static DECLARE_WAIT_QUEUE_HEAD(hb_steady_queue);
+
 static struct hb_callback {
 	struct list_head list;
 	struct semaphore sem;
@@ -129,6 +131,11 @@
 	/* a single hb-thread writer and many fill_node readers are protected */
 	rwlock_t		hr_slot_list_lock;
 	struct list_head	hr_live_list;
+
+	/* let the person setting up hb wait for it to return until it
+	 * has reached a 'steady' state.  This will be fixed when we have
+	 * a more complete api that doesn't lead to this sort of fragility. */
+	atomic_t		hr_steady_iterations;
 };
 
 static void hb_end_buffer_io_sync(struct buffer_head *bh, int uptodate)
@@ -234,6 +241,7 @@
 	struct nm_node *node;
 	struct buffer_head *bh;
 	struct hb_disk_slot *slot;
+	struct list_head *pos, *tmp;
 	hb_disk_heartbeat_block *hb_block;
 	/* only need to worry about locking when we touch the reg lists
 	 * which fill_node_map sees.  otherwise only we touch these
@@ -278,9 +286,6 @@
 
 	/* now we read again and see what other nodes have done */
 	for(i = 0; i < reg->hr_blocks; i++) {
-		/* never, ever, generate events for our node */
-		if (i == nm_this_node())
-			continue;
 		slot = &reg->hr_slots[i];
 
 		bh = slot->ds_bh;
@@ -326,7 +331,21 @@
 		}
 	}
 
-	list_for_each_entry(slot, &newborn, ds_live_item) {
+	/* let the person who launched us know when things are steady */
+	if (list_empty(&newborn) && list_empty(&deceased) &&
+	    (atomic_read(&reg->hr_steady_iterations) != 0)) {
+		if (atomic_dec_and_test(&reg->hr_steady_iterations))
+			wake_up(&hb_steady_queue);
+	}
+
+	/* record our new live guys in the live list and call callbacks */
+	list_for_each_safe(pos, tmp, &newborn) {
+		slot = list_entry(pos, struct hb_disk_slot, ds_live_item);
+
+		write_lock(&reg->hr_slot_list_lock);
+		list_move(&slot->ds_live_item, &reg->hr_live_list);
+		write_unlock(&reg->hr_slot_list_lock);
+
 		node = nm_get_node_by_num(slot->ds_node_num);
 		if (node == NULL) {
 			hbprintk("saw hb for node %d but don't have a node\n",
@@ -336,7 +355,13 @@
 		hb_do_node_up(node, slot->ds_node_num);
 		nm_node_put(node);
 	}
-	list_for_each_entry(slot, &deceased, ds_live_item) {
+
+	/* drop our temporary live_item linkage and call callbacks */ 
+	list_for_each_safe(pos, tmp, &deceased) {
+		slot = list_entry(pos, struct hb_disk_slot, ds_live_item);
+
+		list_del_init(&slot->ds_live_item);
+
 		node = nm_get_node_by_num(slot->ds_node_num);
 		if (node == NULL) {
 			hbprintk("node %d went down but don't have a node\n",
@@ -346,12 +371,6 @@
 		hb_do_node_down(node, slot->ds_node_num);
 		nm_node_put(node);
 	}
-
-	if (!list_empty(&newborn)) {
-		write_lock(&reg->hr_slot_list_lock);
-		list_splice_init(&newborn, &reg->hr_live_list);
-		write_unlock(&reg->hr_slot_list_lock);
-	}
 }
 
 /*
@@ -410,9 +429,6 @@
 	}
 
 	up_read(&hb_callback_sem);
-
-	/* our node is Always Up */
-	set_bit(nm_this_node(), map);
 }
 EXPORT_SYMBOL(hb_fill_node_map);
 
@@ -569,6 +585,8 @@
 		INIT_LIST_HEAD(&slot->ds_live_item);
 	}
 
+	atomic_set(&reg->hr_steady_iterations, 2);
+
 	reg->hr_task = kthread_run(hb_thread, reg, "hb-%s",
 				   reg->hr_kobj.k_name);
 	if (IS_ERR(reg->hr_task)) {
@@ -580,7 +598,11 @@
 	list_add_tail(&reg->hr_active_item, &hb_active_regions);
 	up_write(&hb_callback_sem);
 
-	ret = count;
+	ret = wait_event_interruptible(hb_steady_queue,
+				atomic_read(&reg->hr_steady_iterations) == 0);
+	if (ret == 0)
+		ret = count;
+
 out:
 	if (filp)
 		fput(filp);