[PATCH] JBD: implement j_running_transaction locking

Implement the designed locking around journal->j_running_transaction.

A lot more of the new locking scheme falls into place.

fs/jbd/checkpoint.c

@@ -399,7 +399,7 @@ int cleanup_journal_tail(journal_t *journal)
 	 * next transaction ID we will write, and where it will
 	 * start. */
 
-	/* j_checkpoint_transactions needs locking */
+	spin_lock(&journal->j_state_lock);
 	spin_lock(&journal->j_list_lock);
 	transaction = journal->j_checkpoint_transactions;
 	if (transaction) {
@@ -416,6 +416,7 @@ int cleanup_journal_tail(journal_t *journal)
 		blocknr = journal->j_head;
 	}
 	spin_unlock(&journal->j_list_lock);
+	spin_unlock(&journal->j_state_lock);
 	J_ASSERT (blocknr != 0);
 
 	/* If the oldest pinned transaction is at the tail of the log

fs/jbd/commit.c

@@ -62,7 +62,7 @@ void journal_commit_transaction(journal_t *journal)
 	 * all outstanding updates to complete.
 	 */
 
-	lock_journal(journal); /* Protect journal->j_running_transaction */
+	lock_journal(journal);
 
 #ifdef COMMIT_STATS
 	spin_lock(&journal->j_list_lock);
@@ -72,13 +72,13 @@ void journal_commit_transaction(journal_t *journal)
 
 	lock_kernel();
 
-	J_ASSERT (journal->j_running_transaction != NULL);
-	J_ASSERT (journal->j_committing_transaction == NULL);
+	J_ASSERT(journal->j_running_transaction != NULL);
+	J_ASSERT(journal->j_committing_transaction == NULL);
 
 	commit_transaction = journal->j_running_transaction;
-	J_ASSERT (commit_transaction->t_state == T_RUNNING);
+	J_ASSERT(commit_transaction->t_state == T_RUNNING);
 
-	jbd_debug (1, "JBD: starting commit of transaction %d\n",
+	jbd_debug(1, "JBD: starting commit of transaction %d\n",
 			commit_transaction->t_tid);
 
 	commit_transaction->t_state = T_LOCKED;
@@ -158,14 +158,13 @@ void journal_commit_transaction(journal_t *journal)
 	 * get a new running transaction for incoming filesystem updates
 	 */
 
+	spin_lock(&journal->j_state_lock);
 	commit_transaction->t_state = T_FLUSH;
-
-	wake_up(&journal->j_wait_transaction_locked);
-
 	journal->j_committing_transaction = commit_transaction;
 	journal->j_running_transaction = NULL;
-
 	commit_transaction->t_log_start = journal->j_head;
+	wake_up(&journal->j_wait_transaction_locked);
+	spin_unlock(&journal->j_state_lock);
 
 	unlock_kernel();
 	

fs/jbd/journal.c

@@ -417,13 +417,9 @@ int log_space_left (journal_t *journal)
 /*
  * This function must be non-allocating for PF_MEMALLOC tasks
  */
-tid_t log_start_commit (journal_t *journal, transaction_t *transaction)
+static tid_t __log_start_commit(journal_t *journal, transaction_t *transaction)
 {
-	tid_t target;
-
-	lock_kernel(); /* Protect journal->j_running_transaction */
-
-	target = journal->j_commit_request;
+	tid_t target = journal->j_commit_request;
 
 	/*
 	 * A NULL transaction asks us to commit the currently running
@@ -456,10 +452,19 @@ tid_t log_start_commit (journal_t *journal, transaction_t *transaction)
 	wake_up(&journal->j_wait_commit);
 
 out:
-	unlock_kernel();
 	return target;
 }
 
+tid_t log_start_commit(journal_t *journal, transaction_t *transaction)
+{
+	tid_t ret;
+
+	spin_lock(&journal->j_state_lock);
+	ret = __log_start_commit(journal, transaction);
+	spin_unlock(&journal->j_state_lock);
+	return ret;
+}
+
 /*
  * Wait for a specified commit to complete.
  * The caller may not hold the journal lock.
@@ -1205,24 +1210,30 @@ static int journal_convert_superblock_v1(journal_t *journal,
  * recovery does not need to happen on remount.
  */
 
-int journal_flush (journal_t *journal)
+int journal_flush(journal_t *journal)
 {
 	int err = 0;
 	transaction_t *transaction = NULL;
 	unsigned long old_tail;
 
-	lock_kernel();
+	spin_lock(&journal->j_state_lock);
 	
 	/* Force everything buffered to the log... */
 	if (journal->j_running_transaction) {
 		transaction = journal->j_running_transaction;
-		log_start_commit(journal, transaction);
+		__log_start_commit(journal, transaction);
 	} else if (journal->j_committing_transaction)
 		transaction = journal->j_committing_transaction;
 
 	/* Wait for the log commit to complete... */
-	if (transaction)
-		log_wait_commit(journal, transaction->t_tid);
+	if (transaction) {
+		tid_t tid = transaction->t_tid;
+
+		spin_unlock(&journal->j_state_lock);
+		log_wait_commit(journal, tid);
+	} else {
+		spin_unlock(&journal->j_state_lock);
+	}
 
 	/* ...and flush everything in the log out to disk. */
 	lock_journal(journal);
@@ -1248,8 +1259,6 @@ int journal_flush (journal_t *journal)
 	J_ASSERT(journal->j_head == journal->j_tail);
 	J_ASSERT(journal->j_tail_sequence == journal->j_transaction_sequence);
 	
-	unlock_kernel();
-	
 	return err;
 }
 
@@ -1318,10 +1327,12 @@ const char *journal_dev_name(journal_t *journal, char *buffer)
  * itself are here.
  */
 
-/* Quick version for internal journal use (doesn't lock the journal).
+/*
+ * Quick version for internal journal use (doesn't lock the journal).
  * Aborts hard --- we mark the abort as occurred, but do _nothing_ else,
- * and don't attempt to make any other journal updates. */
-void __journal_abort_hard (journal_t *journal)
+ * and don't attempt to make any other journal updates.
+ */
+void __journal_abort_hard(journal_t *journal)
 {
 	transaction_t *transaction;
 	char b[BDEVNAME_SIZE];
@@ -1329,15 +1340,15 @@ void __journal_abort_hard (journal_t *journal)
 	if (journal->j_flags & JFS_ABORT)
 		return;
 
-	printk (KERN_ERR "Aborting journal on device %s.\n",
+	printk(KERN_ERR "Aborting journal on device %s.\n",
 		journal_dev_name(journal, b));
 
-	lock_kernel();
+	spin_lock(&journal->j_state_lock);
 	journal->j_flags |= JFS_ABORT;
 	transaction = journal->j_running_transaction;
 	if (transaction)
-		log_start_commit(journal, transaction);
-	unlock_kernel();
+		__log_start_commit(journal, transaction);
+	spin_unlock(&journal->j_state_lock);
 }
 
 /* Soft abort: record the abort error status in the journal superblock,

fs/jbd/transaction.c

@@ -39,6 +39,8 @@
  *	The journal MUST be locked.  We don't perform atomic mallocs on the
  *	new transaction	and we can't block without protecting against other
  *	processes trying to touch the journal while it is in transition.
+ *
+ * Called under j_state_lock
  */
 
 static transaction_t *
@@ -443,13 +445,14 @@ void journal_lock_updates(journal_t *journal)
 
 	spin_lock(&journal->j_state_lock);
 	++journal->j_barrier_count;
-	spin_unlock(&journal->j_state_lock);
 
 	/* Wait until there are no running updates */
 	while (1) {
 		transaction_t *transaction = journal->j_running_transaction;
+
 		if (!transaction)
 			break;
+
 		spin_lock(&transaction->t_handle_lock);
 		if (!transaction->t_updates) {
 			spin_unlock(&transaction->t_handle_lock);
@@ -458,12 +461,14 @@ void journal_lock_updates(journal_t *journal)
 		prepare_to_wait(&journal->j_wait_updates, &wait,
 				TASK_UNINTERRUPTIBLE);
 		spin_unlock(&transaction->t_handle_lock);
+		spin_unlock(&journal->j_state_lock);
 		unlock_journal(journal);
 		schedule();
 		finish_wait(&journal->j_wait_updates, &wait);
 		lock_journal(journal);
+		spin_lock(&journal->j_state_lock);
 	}
-
+	spin_unlock(&journal->j_state_lock);
 	unlock_journal(journal);
 
 	/*
@@ -1781,6 +1786,7 @@ static int journal_unmap_buffer(journal_t *journal, struct buffer_head *bh)
 	if (!buffer_jbd(bh))
 		goto zap_buffer_unlocked;
 
+	spin_lock(&journal->j_state_lock);
 	jbd_lock_bh_state(bh);
 	spin_lock(&journal->j_list_lock);
 	jh = bh2jh(bh);
@@ -1813,6 +1819,7 @@ static int journal_unmap_buffer(journal_t *journal, struct buffer_head *bh)
 					journal->j_running_transaction);
 			spin_unlock(&journal->j_list_lock);
 			jbd_unlock_bh_state(bh);
+			spin_unlock(&journal->j_state_lock);
 			return ret;
 		} else {
 			/* There is no currently-running transaction. So the
@@ -1825,6 +1832,7 @@ static int journal_unmap_buffer(journal_t *journal, struct buffer_head *bh)
 					journal->j_committing_transaction);
 				spin_unlock(&journal->j_list_lock);
 				jbd_unlock_bh_state(bh);
+				spin_unlock(&journal->j_state_lock);
 				return ret;
 			} else {
 				/* The orphan record's transaction has
@@ -1847,6 +1855,7 @@ static int journal_unmap_buffer(journal_t *journal, struct buffer_head *bh)
 		}
 		spin_unlock(&journal->j_list_lock);
 		jbd_unlock_bh_state(bh);
+		spin_unlock(&journal->j_state_lock);
 		return 0;
 	} else {
 		/* Good, the buffer belongs to the running transaction.
@@ -1862,6 +1871,7 @@ static int journal_unmap_buffer(journal_t *journal, struct buffer_head *bh)
 zap_buffer:
 	spin_unlock(&journal->j_list_lock);
 	jbd_unlock_bh_state(bh);
+	spin_unlock(&journal->j_state_lock);
 zap_buffer_unlocked:
 	clear_buffer_dirty(bh);
 	J_ASSERT_BH(bh, !buffer_jbddirty(bh));