OCFS2 1.4: Proposed fix for mount/recovery race

index 3633edd..e0f5459 100644
@@ -585,7 +585,7 @@ struct ocfs2_dinode {
 		struct {		/* Info for journal system
 					   inodes */
 			__le32 ij_flags;	/* Mounted, version, etc. */
-			__le32 ij_pad;
+			__le32 ij_reco_generation;
 		} journal1;
 	} id1;				/* Inode type dependant 1 */
 /*C0*/	union {

index 741a4e2..2344f8b 100644
@@ -57,7 +57,7 @@ static int __ocfs2_recovery_thread(void *arg);
 static int ocfs2_commit_cache(struct ocfs2_super *osb);
 static int ocfs2_wait_on_mount(struct ocfs2_super *osb);
 static int ocfs2_journal_toggle_dirty(struct ocfs2_super *osb,
-				      int dirty);
+				      int dirty, int replayed);
 static int ocfs2_trylock_journal(struct ocfs2_super *osb,
 				 int slot_num);
 static int ocfs2_recover_orphans(struct ocfs2_super *osb,

@@ -437,7 +437,7 @@ done:
 }
 
 static int ocfs2_journal_toggle_dirty(struct ocfs2_super *osb,
-				      int dirty)
+				      int dirty, int replayed)
 {
 	int status;
 	unsigned int flags;

@@ -467,6 +467,9 @@ static int ocfs2_journal_toggle_dirty(struct ocfs2_super *osb,
 		flags &= ~OCFS2_JOURNAL_DIRTY_FL;
 	fe->id1.journal1.ij_flags = cpu_to_le32(flags);
 
+	if (replayed)
+		le32_add_cpu(&(fe->id1.journal1.ij_reco_generation), 1);
+
 	status = ocfs2_write_block(osb, bh, journal->j_inode);
 	if (status < 0)
 		mlog_errno(status);

@@ -541,7 +544,7 @@ void ocfs2_journal_shutdown(struct ocfs2_super *osb)
 		 * Do not toggle if flush was unsuccessful otherwise
 		 * will leave dirty metadata in a "clean" journal
 		 */
-		status = ocfs2_journal_toggle_dirty(osb, 0);
+		status = ocfs2_journal_toggle_dirty(osb, 0, 0);
 		if (status < 0)
 			mlog_errno(status);
 	}

@@ -584,7 +587,7 @@ static void ocfs2_clear_journal_error(struct super_block *sb,
 	}
 }
 
-int ocfs2_journal_load(struct ocfs2_journal *journal, int local)
+int ocfs2_journal_load(struct ocfs2_journal *journal, int local, int replay)
 {
 	int status = 0;
 	struct ocfs2_super *osb;

@@ -603,7 +606,7 @@ int ocfs2_journal_load(struct ocfs2_journal *journal, int local)
 
 	ocfs2_clear_journal_error(osb->sb, journal->j_journal, osb->slot_num);
 
-	status = ocfs2_journal_toggle_dirty(osb, 1);
+	status = ocfs2_journal_toggle_dirty(osb, 1, replay);
 	if (status < 0) {
 		mlog_errno(status);
 		goto done;

@@ -645,7 +648,7 @@ int ocfs2_journal_wipe(struct ocfs2_journal *journal, int full)
 		goto bail;
 	}
 
-	status = ocfs2_journal_toggle_dirty(journal->j_osb, 0);
+	status = ocfs2_journal_toggle_dirty(journal->j_osb, 0, 0);
 	if (status < 0)
 		mlog_errno(status);
 

@@ -950,11 +953,57 @@ out:
 	mlog_exit_void();
 }
 
+static int ocfs2_read_journal_recogen(struct ocfs2_super *osb,
+				      int slot_num,
+				      u32 *slot_reco_gen,
+				      struct inode **ret_in)
+{
+	int status;
+	struct ocfs2_dinode *fe;
+	struct buffer_head *bh = NULL;
+	struct inode *inode = NULL;
+
+	inode = ocfs2_get_system_file_inode(osb, JOURNAL_SYSTEM_INODE,
+					    slot_num);
+	if (inode == NULL) {
+		status = -EACCES;
+		mlog_errno(status);
+		goto bail;
+	}
+	if (is_bad_inode(inode)) {
+		status = -EACCES;
+		iput(inode);
+		mlog_errno(status);
+		goto bail;
+	}
+	SET_INODE_JOURNAL(inode);
+
+	status = ocfs2_read_block(osb, OCFS2_I(inode)->ip_blkno, &bh, 0, inode);
+	if (status < 0) {
+		iput(inode);
+		mlog_errno(status);
+		goto bail;
+	}
+
+	fe = (struct ocfs2_dinode *)bh->b_data;
+
+	*slot_reco_gen = le32_to_cpu(fe->id1.journal1.ij_reco_generation);
+	if (ret_in)
+		*ret_in = inode;
+	else
+		iput(inode);
+
+	brelse(bh);
+bail:
+	return status;
+}
+
 /* Does the actual journal replay and marks the journal inode as
  * clean. Will only replay if the journal inode is marked dirty. */
 static int ocfs2_replay_journal(struct ocfs2_super *osb,
 				int node_num,
-				int slot_num)
+				int slot_num,
+				int *busy)
 {
 	int status;
 	int got_lock = 0;

@@ -963,22 +1012,33 @@ static int ocfs2_replay_journal(struct ocfs2_super *osb,
 	struct ocfs2_dinode *fe;
 	journal_t *journal = NULL;
 	struct buffer_head *bh = NULL;
+	u32 recogen;
 
-	inode = ocfs2_get_system_file_inode(osb, JOURNAL_SYSTEM_INODE,
-					    slot_num);
-	if (inode == NULL) {
-		status = -EACCES;
+	*busy = 0;
+
+	status = ocfs2_read_journal_recogen(osb, slot_num, &recogen, &inode);
+	if (status) {
 		mlog_errno(status);
 		goto done;
 	}
-	if (is_bad_inode(inode)) {
-		status = -EACCES;
-		iput(inode);
-		inode = NULL;
-		mlog_errno(status);
+
+	/*
+	 * As the fs recovery is asynchronous, there is a small chance that another
+	 * node mounted (and recovered) the slot before the recovery thread could
+	 * get the lock. To handle that, we dirty read the journal inode for that
+	 * slot to get the recovery generation. If it is different than what we
+	 * expected, the slot has been recovered. If not, it needs recovery.
+	 */
+	if (osb->slot_reco_generation[slot_num] != recogen) {
+		mlog(0, "Slot %u already recovered (old/new=%u/%u)\n", slot_num,
+		     osb->slot_reco_generation[slot_num], recogen);
+		osb->slot_reco_generation[slot_num] = recogen;
+		*busy = 1;
+		status = 0;
 		goto done;
 	}
-	SET_INODE_JOURNAL(inode);
+
+	/* Continue with recovery as the journal has not yet been recovered */
 
 	status = ocfs2_inode_lock_full(inode, &bh, 1, OCFS2_META_LOCK_RECOVERY);
 	if (status < 0) {

@@ -992,9 +1052,12 @@ static int ocfs2_replay_journal(struct ocfs2_super *osb,
 	fe = (struct ocfs2_dinode *) bh->b_data;
 
 	flags = le32_to_cpu(fe->id1.journal1.ij_flags);
+	recogen = le32_to_cpu(fe->id1.journal1.ij_reco_generation);
 
 	if (!(flags & OCFS2_JOURNAL_DIRTY_FL)) {
 		mlog(0, "No recovery required for node %d\n", node_num);
+		/* Refresh recovery generation for the slot */
+		osb->slot_reco_generation[slot_num] = recogen;
 		goto done;
 	}
 

@@ -1042,6 +1105,11 @@ static int ocfs2_replay_journal(struct ocfs2_super *osb,
 	flags &= ~OCFS2_JOURNAL_DIRTY_FL;
 	fe->id1.journal1.ij_flags = cpu_to_le32(flags);
 
+	/* Increment recovery generation to indicate successful recovery */
+	le32_add_cpu(&(fe->id1.journal1.ij_reco_generation), 1);
+	osb->slot_reco_generation[slot_num] =
+		le32_to_cpu(fe->id1.journal1.ij_reco_generation);
+
 	status = ocfs2_write_block(osb, bh, inode);
 	if (status < 0)
 		mlog_errno(status);

@@ -1066,6 +1134,30 @@ done:
 	return status;
 }
 
+/* Refresh recovery generations of all slots */
+static int ocfs2_refresh_all_journal_recogens(struct ocfs2_super *osb)
+{
+	int i;
+	int status = 0;
+
+	for (i = 0; i < osb->max_slots; ++i) {
+		if (i == osb->slot_num)
+			continue;
+		status = ocfs2_read_journal_recogen(osb, i,
+						&(osb->slot_reco_generation[i]),
+						NULL);
+		if (status) {
+			mlog_errno(status);
+			goto bail;
+		}
+		mlog(0, "Slot %u recovery generation is %u\n", i,
+		     osb->slot_reco_generation[i]);
+	}
+
+bail:
+	return status;
+}
+
 /*
  * Do the most important parts of node recovery:
  *  - Replay it's journal

@@ -1081,7 +1173,7 @@ done:
 static int ocfs2_recover_node(struct ocfs2_super *osb,
 			      int node_num)
 {
-	int status = 0;
+	int status = 0, busy = 0;
 	int slot_num;
 	struct ocfs2_slot_info *si = osb->slot_info;
 	struct ocfs2_dinode *la_copy = NULL;

@@ -1098,19 +1190,28 @@ static int ocfs2_recover_node(struct ocfs2_super *osb,
 
 	slot_num = ocfs2_node_num_to_slot(si, node_num);
 	if (slot_num == OCFS2_INVALID_SLOT) {
-		status = 0;
 		mlog(0, "no slot for this node, so no recovery required.\n");
+		/* Refresh all journal recoevery generations from disk */
+		status = ocfs2_refresh_all_journal_recogens(osb);
+		if (status)
+			mlog_errno(status);
 		goto done;
 	}
 
 	mlog(0, "node %d was using slot %d\n", node_num, slot_num);
 
-	status = ocfs2_replay_journal(osb, node_num, slot_num);
+	status = ocfs2_replay_journal(osb, node_num, slot_num, &busy);
 	if (status < 0) {
 		mlog_errno(status);
 		goto done;
 	}
 
+	if (busy) {
+		mlog(0, "Skipping recovery for slot %u (node %u) "
+		     "as another node has recovered it\n", slot_num, node_num);
+		goto done;
+	}
+
 	/* Stamp a clean local alloc file AFTER recovering the journal... */
 	status = ocfs2_begin_local_alloc_recovery(osb, slot_num, &la_copy);
 	if (status < 0) {

@@ -1190,14 +1291,26 @@ int ocfs2_mark_dead_nodes(struct ocfs2_super *osb)
 {
 	int status, i, node_num;
 	struct ocfs2_slot_info *si = osb->slot_info;
+	u32 recogen;
 
 	/* This is called with the super block cluster lock, so we
 	 * know that the slot map can't change underneath us. */
 
 	spin_lock(&si->si_lock);
 	for(i = 0; i < si->si_num_slots; i++) {
+		/* Read the recovery generation */
+		status = ocfs2_read_journal_recogen(osb, i, &recogen, NULL);
+		if (status) {
+			mlog_errno(status);
+			goto bail;
+		}
+		osb->slot_reco_generation[i] = recogen;
+
+		mlog(0, "Slot %u recovery generation is %u\n", i, recogen);
+
 		if (i == osb->slot_num)
 			continue;
+
 		if (ocfs2_is_empty_slot(si, i))
 			continue;
 

@@ -1495,6 +1608,9 @@ int ocfs2_check_journals_nolocks(struct ocfs2_super *osb)
 
 		di = (struct ocfs2_dinode *) di_bh->b_data;
 
+		osb->slot_reco_generation[slot] =
+			le32_to_cpu(di->id1.journal1.ij_reco_generation);
+
 		if (le32_to_cpu(di->id1.journal1.ij_flags) &
 		    OCFS2_JOURNAL_DIRTY_FL)
 			ret = -EROFS;

index 52f02fe..9066bf0 100644
@@ -157,7 +157,7 @@ int    ocfs2_journal_init(struct ocfs2_journal *journal,
 void   ocfs2_journal_shutdown(struct ocfs2_super *osb);
 int    ocfs2_journal_wipe(struct ocfs2_journal *journal,
 			  int full);
-int    ocfs2_journal_load(struct ocfs2_journal *journal, int local);
+int    ocfs2_journal_load(struct ocfs2_journal *journal, int local, int replay);
 int    ocfs2_check_journals_nolocks(struct ocfs2_super *osb);
 void   ocfs2_recovery_thread(struct ocfs2_super *osb,
 			     int node_num);

index 9546470..4af81ee 100644
@@ -193,6 +193,8 @@ struct ocfs2_super
 
 	struct ocfs2_slot_info *slot_info;
 
+	u32 *slot_reco_generation;
+
 	spinlock_t node_map_lock;
 	struct ocfs2_node_map recovery_map;
 

index ff31722..8540ab6 100644
@@ -1497,6 +1497,15 @@ static int ocfs2_initialize_super(struct super_block *sb,
 	}
 	mlog(0, "max_slots for this device: %u\n", osb->max_slots);
 
+	osb->slot_reco_generation = kcalloc(osb->max_slots,
+					    sizeof(*osb->slot_reco_generation),
+					    GFP_KERNEL);
+	if (!osb->slot_reco_generation) {
+		status = -ENOMEM;
+		mlog_errno(status);
+		goto bail;
+	}
+
 	init_waitqueue_head(&osb->osb_wipe_event);
 	osb->osb_orphan_wipes = kcalloc(osb->max_slots,
 					sizeof(*osb->osb_orphan_wipes),

@@ -1739,7 +1748,7 @@ static int ocfs2_check_volume(struct ocfs2_super *osb)
 	local = ocfs2_mount_local(osb);
 
 	/* will play back anything left in the journal. */
-	ocfs2_journal_load(osb->journal, local);
+	ocfs2_journal_load(osb->journal, local, dirty);
 
 	if (dirty) {
 		/* recover my local alloc if we didn't unmount cleanly. */

@@ -1754,8 +1763,6 @@ static int ocfs2_check_volume(struct ocfs2_super *osb)
 		 * ourselves as mounted. */
 	}
 
-	mlog(0, "Journal loaded.\n");
-
 	status = ocfs2_load_local_alloc(osb);
 	if (status < 0) {
 		mlog_errno(status);

@@ -1801,6 +1808,7 @@ static void ocfs2_delete_osb(struct ocfs2_super *osb)
 		ocfs2_free_slot_info(osb->slot_info);
 
 	kfree(osb->osb_orphan_wipes);
+	kfree(osb->slot_reco_generation);
 	/* FIXME
 	 * This belongs in journal shutdown, but because we have to
 	 * allocate osb->journal at the start of ocfs2_initalize_osb(),