#include "ocfs2.h"
#include "alloc.h"
+#include "dir.h"
#include "dlmglue.h"
#include "extent_map.h"
#include "heartbeat.h"
#include "inode.h"
#include "journal.h"
#include "localalloc.h"
-#include "namei.h"
#include "slot_map.h"
#include "super.h"
-#include "vote.h"
#include "sysfile.h"
#include "buffer_head_io.h"
-spinlock_t trans_inc_lock = SPIN_LOCK_UNLOCKED;
+DEFINE_SPINLOCK(trans_inc_lock);
static int ocfs2_force_read_journal(struct inode *inode);
static int ocfs2_recover_node(struct ocfs2_super *osb,
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 void ocfs2_handle_cleanup_locks(struct ocfs2_journal *journal,
- struct ocfs2_journal_handle *handle);
-static void ocfs2_commit_unstarted_handle(struct ocfs2_journal_handle *handle);
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,
int slot);
static int ocfs2_commit_thread(void *arg);
+
+/*
+ * The recovery_list is a simple linked list of node numbers to recover.
+ * It is protected by the recovery_lock.
+ */
+
+struct ocfs2_recovery_map {
+ unsigned int rm_used;
+ unsigned int *rm_entries;
+};
+
+int ocfs2_recovery_init(struct ocfs2_super *osb)
+{
+ struct ocfs2_recovery_map *rm;
+
+ mutex_init(&osb->recovery_lock);
+ osb->disable_recovery = 0;
+ osb->recovery_thread_task = NULL;
+ init_waitqueue_head(&osb->recovery_event);
+
+ rm = kzalloc(sizeof(struct ocfs2_recovery_map) +
+ osb->max_slots * sizeof(unsigned int),
+ GFP_KERNEL);
+ if (!rm) {
+ mlog_errno(-ENOMEM);
+ return -ENOMEM;
+ }
+
+ rm->rm_entries = (unsigned int *)((char *)rm +
+ sizeof(struct ocfs2_recovery_map));
+ osb->recovery_map = rm;
+
+ return 0;
+}
+
+/* we can't grab the goofy sem lock from inside wait_event, so we use
+ * memory barriers to make sure that we'll see the null task before
+ * being woken up */
+static int ocfs2_recovery_thread_running(struct ocfs2_super *osb)
+{
+ mb();
+ return osb->recovery_thread_task != NULL;
+}
+
+void ocfs2_recovery_exit(struct ocfs2_super *osb)
+{
+ struct ocfs2_recovery_map *rm;
+
+ /* disable any new recovery threads and wait for any currently
+ * running ones to exit. Do this before setting the vol_state. */
+ mutex_lock(&osb->recovery_lock);
+ osb->disable_recovery = 1;
+ mutex_unlock(&osb->recovery_lock);
+ wait_event(osb->recovery_event, !ocfs2_recovery_thread_running(osb));
+
+ /* At this point, we know that no more recovery threads can be
+ * launched, so wait for any recovery completion work to
+ * complete. */
+ flush_workqueue(ocfs2_wq);
+
+ /*
+ * Now that recovery is shut down, and the osb is about to be
+ * freed, the osb_lock is not taken here.
+ */
+ rm = osb->recovery_map;
+ /* XXX: Should we bug if there are dirty entries? */
+
+ kfree(rm);
+}
+
+static int __ocfs2_recovery_map_test(struct ocfs2_super *osb,
+ unsigned int node_num)
+{
+ int i;
+ struct ocfs2_recovery_map *rm = osb->recovery_map;
+
+ assert_spin_locked(&osb->osb_lock);
+
+ for (i = 0; i < rm->rm_used; i++) {
+ if (rm->rm_entries[i] == node_num)
+ return 1;
+ }
+
+ return 0;
+}
+
+/* Behaves like test-and-set. Returns the previous value */
+static int ocfs2_recovery_map_set(struct ocfs2_super *osb,
+ unsigned int node_num)
+{
+ struct ocfs2_recovery_map *rm = osb->recovery_map;
+
+ spin_lock(&osb->osb_lock);
+ if (__ocfs2_recovery_map_test(osb, node_num)) {
+ spin_unlock(&osb->osb_lock);
+ return 1;
+ }
+
+ /* XXX: Can this be exploited? Not from o2dlm... */
+ BUG_ON(rm->rm_used >= osb->max_slots);
+
+ rm->rm_entries[rm->rm_used] = node_num;
+ rm->rm_used++;
+ spin_unlock(&osb->osb_lock);
+
+ return 0;
+}
+
+static void ocfs2_recovery_map_clear(struct ocfs2_super *osb,
+ unsigned int node_num)
+{
+ int i;
+ struct ocfs2_recovery_map *rm = osb->recovery_map;
+
+ spin_lock(&osb->osb_lock);
+
+ for (i = 0; i < rm->rm_used; i++) {
+ if (rm->rm_entries[i] == node_num)
+ break;
+ }
+
+ if (i < rm->rm_used) {
+ /* XXX: be careful with the pointer math */
+ memmove(&(rm->rm_entries[i]), &(rm->rm_entries[i + 1]),
+ (rm->rm_used - i - 1) * sizeof(unsigned int));
+ rm->rm_used--;
+ }
+
+ spin_unlock(&osb->osb_lock);
+}
+
static int ocfs2_commit_cache(struct ocfs2_super *osb)
{
int status = 0;
goto finally;
}
- journal_lock_updates(journal->j_journal);
- status = journal_flush(journal->j_journal);
- journal_unlock_updates(journal->j_journal);
+ jbd2_journal_lock_updates(journal->j_journal);
+ status = jbd2_journal_flush(journal->j_journal);
+ jbd2_journal_unlock_updates(journal->j_journal);
if (status < 0) {
up_write(&journal->j_trans_barrier);
mlog_errno(status);
mlog(0, "commit_thread: flushed transaction %lu (%u handles)\n",
journal->j_trans_id, flushed);
- ocfs2_kick_vote_thread(osb);
+ ocfs2_wake_downconvert_thread(osb);
wake_up(&journal->j_checkpointed);
finally:
mlog_exit(status);
return status;
}
-struct ocfs2_journal_handle *ocfs2_alloc_handle(struct ocfs2_super *osb)
-{
- struct ocfs2_journal_handle *retval = NULL;
-
- retval = kcalloc(1, sizeof(*retval), GFP_KERNEL);
- if (!retval) {
- mlog(ML_ERROR, "Failed to allocate memory for journal "
- "handle!\n");
- return NULL;
- }
-
- retval->max_buffs = 0;
- retval->num_locks = 0;
- retval->k_handle = NULL;
-
- INIT_LIST_HEAD(&retval->locks);
- INIT_LIST_HEAD(&retval->inode_list);
- retval->journal = osb->journal;
-
- return retval;
-}
-
/* pass it NULL and it will allocate a new handle object for you. If
* you pass it a handle however, it may still return error, in which
* case it has free'd the passed handle for you. */
-struct ocfs2_journal_handle *ocfs2_start_trans(struct ocfs2_super *osb,
- struct ocfs2_journal_handle *handle,
- int max_buffs)
+handle_t *ocfs2_start_trans(struct ocfs2_super *osb, int max_buffs)
{
- int ret;
journal_t *journal = osb->journal->j_journal;
-
- mlog_entry("(max_buffs = %d)\n", max_buffs);
+ handle_t *handle;
BUG_ON(!osb || !osb->journal->j_journal);
- if (ocfs2_is_hard_readonly(osb)) {
- ret = -EROFS;
- goto done_free;
- }
+ if (ocfs2_is_hard_readonly(osb))
+ return ERR_PTR(-EROFS);
BUG_ON(osb->journal->j_state == OCFS2_JOURNAL_FREE);
BUG_ON(max_buffs <= 0);
BUG();
}
- if (!handle)
- handle = ocfs2_alloc_handle(osb);
- if (!handle) {
- ret = -ENOMEM;
- mlog(ML_ERROR, "Failed to allocate memory for journal "
- "handle!\n");
- goto done_free;
- }
-
- handle->max_buffs = max_buffs;
-
down_read(&osb->journal->j_trans_barrier);
- /* actually start the transaction now */
- handle->k_handle = journal_start(journal, max_buffs);
- if (IS_ERR(handle->k_handle)) {
+ handle = jbd2_journal_start(journal, max_buffs);
+ if (IS_ERR(handle)) {
up_read(&osb->journal->j_trans_barrier);
- ret = PTR_ERR(handle->k_handle);
- handle->k_handle = NULL;
- mlog_errno(ret);
+ mlog_errno(PTR_ERR(handle));
if (is_journal_aborted(journal)) {
ocfs2_abort(osb->sb, "Detected aborted journal");
- ret = -EROFS;
+ handle = ERR_PTR(-EROFS);
}
- goto done_free;
+ } else {
+ if (!ocfs2_mount_local(osb))
+ atomic_inc(&(osb->journal->j_num_trans));
}
- atomic_inc(&(osb->journal->j_num_trans));
- handle->flags |= OCFS2_HANDLE_STARTED;
-
- mlog_exit_ptr(handle);
return handle;
-
-done_free:
- if (handle)
- ocfs2_commit_unstarted_handle(handle); /* will kfree handle */
-
- mlog_exit(ret);
- return ERR_PTR(ret);
}
-void ocfs2_handle_add_inode(struct ocfs2_journal_handle *handle,
- struct inode *inode)
+int ocfs2_commit_trans(struct ocfs2_super *osb,
+ handle_t *handle)
{
- BUG_ON(!handle);
- BUG_ON(!inode);
-
- atomic_inc(&inode->i_count);
-
- /* we're obviously changing it... */
- mutex_lock(&inode->i_mutex);
-
- /* sanity check */
- BUG_ON(OCFS2_I(inode)->ip_handle);
- BUG_ON(!list_empty(&OCFS2_I(inode)->ip_handle_list));
-
- OCFS2_I(inode)->ip_handle = handle;
- list_del(&(OCFS2_I(inode)->ip_handle_list));
- list_add_tail(&(OCFS2_I(inode)->ip_handle_list), &(handle->inode_list));
-}
-
-static void ocfs2_handle_unlock_inodes(struct ocfs2_journal_handle *handle)
-{
- struct list_head *p, *n;
- struct inode *inode;
- struct ocfs2_inode_info *oi;
-
- list_for_each_safe(p, n, &handle->inode_list) {
- oi = list_entry(p, struct ocfs2_inode_info,
- ip_handle_list);
- inode = &oi->vfs_inode;
-
- OCFS2_I(inode)->ip_handle = NULL;
- list_del_init(&OCFS2_I(inode)->ip_handle_list);
-
- mutex_unlock(&inode->i_mutex);
- iput(inode);
- }
-}
-
-/* This is trivial so we do it out of the main commit
- * paths. Beware, it can be called from start_trans too! */
-static void ocfs2_commit_unstarted_handle(struct ocfs2_journal_handle *handle)
-{
- mlog_entry_void();
-
- BUG_ON(handle->flags & OCFS2_HANDLE_STARTED);
-
- ocfs2_handle_unlock_inodes(handle);
- /* You are allowed to add journal locks before the transaction
- * has started. */
- ocfs2_handle_cleanup_locks(handle->journal, handle);
-
- kfree(handle);
-
- mlog_exit_void();
-}
-
-void ocfs2_commit_trans(struct ocfs2_journal_handle *handle)
-{
- handle_t *jbd_handle;
- int retval;
- struct ocfs2_journal *journal = handle->journal;
-
- mlog_entry_void();
+ int ret;
+ struct ocfs2_journal *journal = osb->journal;
BUG_ON(!handle);
- if (!(handle->flags & OCFS2_HANDLE_STARTED)) {
- ocfs2_commit_unstarted_handle(handle);
- mlog_exit_void();
- return;
- }
-
- /* release inode semaphores we took during this transaction */
- ocfs2_handle_unlock_inodes(handle);
-
- /* ocfs2_extend_trans may have had to call journal_restart
- * which will always commit the transaction, but may return
- * error for any number of reasons. If this is the case, we
- * clear k_handle as it's not valid any more. */
- if (handle->k_handle) {
- jbd_handle = handle->k_handle;
-
- if (handle->flags & OCFS2_HANDLE_SYNC)
- jbd_handle->h_sync = 1;
- else
- jbd_handle->h_sync = 0;
-
- /* actually stop the transaction. if we've set h_sync,
- * it'll have been committed when we return */
- retval = journal_stop(jbd_handle);
- if (retval < 0) {
- mlog_errno(retval);
- mlog(ML_ERROR, "Could not commit transaction\n");
- BUG();
- }
-
- handle->k_handle = NULL; /* it's been free'd in journal_stop */
- }
-
- ocfs2_handle_cleanup_locks(journal, handle);
+ ret = jbd2_journal_stop(handle);
+ if (ret < 0)
+ mlog_errno(ret);
up_read(&journal->j_trans_barrier);
- kfree(handle);
- mlog_exit_void();
+ return ret;
}
/*
* transaction. extend_trans will either extend the current handle by
* nblocks, or commit it and start a new one with nblocks credits.
*
+ * This might call jbd2_journal_restart() which will commit dirty buffers
+ * and then restart the transaction. Before calling
+ * ocfs2_extend_trans(), any changed blocks should have been
+ * dirtied. After calling it, all blocks which need to be changed must
+ * go through another set of journal_access/journal_dirty calls.
+ *
* WARNING: This will not release any semaphores or disk locks taken
* during the transaction, so make sure they were taken *before*
* start_trans or we'll have ordering deadlocks.
* good because transaction ids haven't yet been recorded on the
* cluster locks associated with this handle.
*/
-int ocfs2_extend_trans(struct ocfs2_journal_handle *handle,
- int nblocks)
+int ocfs2_extend_trans(handle_t *handle, int nblocks)
{
int status;
BUG_ON(!handle);
- BUG_ON(!(handle->flags & OCFS2_HANDLE_STARTED));
BUG_ON(!nblocks);
mlog_entry_void();
mlog(0, "Trying to extend transaction by %d blocks\n", nblocks);
- status = journal_extend(handle->k_handle, nblocks);
+#ifdef CONFIG_OCFS2_DEBUG_FS
+ status = 1;
+#else
+ status = jbd2_journal_extend(handle, nblocks);
if (status < 0) {
mlog_errno(status);
goto bail;
}
+#endif
if (status > 0) {
- mlog(0, "journal_extend failed, trying journal_restart\n");
- status = journal_restart(handle->k_handle, nblocks);
+ mlog(0,
+ "jbd2_journal_extend failed, trying "
+ "jbd2_journal_restart\n");
+ status = jbd2_journal_restart(handle, nblocks);
if (status < 0) {
- handle->k_handle = NULL;
mlog_errno(status);
goto bail;
}
- handle->max_buffs = nblocks;
- } else
- handle->max_buffs += nblocks;
+ }
status = 0;
bail:
return status;
}
-int ocfs2_journal_access(struct ocfs2_journal_handle *handle,
+int ocfs2_journal_access(handle_t *handle,
struct inode *inode,
struct buffer_head *bh,
int type)
BUG_ON(!inode);
BUG_ON(!handle);
BUG_ON(!bh);
- BUG_ON(!(handle->flags & OCFS2_HANDLE_STARTED));
- mlog_entry("bh->b_blocknr=%llu, type=%d (\"%s\"), bh->b_size = %hu\n",
+ mlog_entry("bh->b_blocknr=%llu, type=%d (\"%s\"), bh->b_size = %zu\n",
(unsigned long long)bh->b_blocknr, type,
(type == OCFS2_JOURNAL_ACCESS_CREATE) ?
"OCFS2_JOURNAL_ACCESS_CREATE" :
switch (type) {
case OCFS2_JOURNAL_ACCESS_CREATE:
case OCFS2_JOURNAL_ACCESS_WRITE:
- status = journal_get_write_access(handle->k_handle, bh);
+ status = jbd2_journal_get_write_access(handle, bh);
break;
case OCFS2_JOURNAL_ACCESS_UNDO:
- status = journal_get_undo_access(handle->k_handle, bh);
+ status = jbd2_journal_get_undo_access(handle, bh);
break;
default:
return status;
}
-int ocfs2_journal_dirty(struct ocfs2_journal_handle *handle,
+int ocfs2_journal_dirty(handle_t *handle,
struct buffer_head *bh)
{
int status;
- BUG_ON(!(handle->flags & OCFS2_HANDLE_STARTED));
-
mlog_entry("(bh->b_blocknr=%llu)\n",
(unsigned long long)bh->b_blocknr);
- status = journal_dirty_metadata(handle->k_handle, bh);
+ status = jbd2_journal_dirty_metadata(handle, bh);
if (status < 0)
mlog(ML_ERROR, "Could not dirty metadata buffer. "
"(bh->b_blocknr=%llu)\n",
return status;
}
+#ifdef CONFIG_OCFS2_COMPAT_JBD
int ocfs2_journal_dirty_data(handle_t *handle,
struct buffer_head *bh)
{
return err;
}
+#endif
-/* We always assume you're adding a metadata lock at level 'ex' */
-int ocfs2_handle_add_lock(struct ocfs2_journal_handle *handle,
- struct inode *inode)
-{
- int status;
- struct ocfs2_journal_lock *lock;
-
- BUG_ON(!inode);
-
- lock = kmem_cache_alloc(ocfs2_lock_cache, GFP_NOFS);
- if (!lock) {
- status = -ENOMEM;
- mlog_errno(-ENOMEM);
- goto bail;
- }
-
- if (!igrab(inode))
- BUG();
- lock->jl_inode = inode;
-
- list_add_tail(&(lock->jl_lock_list), &(handle->locks));
- handle->num_locks++;
-
- status = 0;
-bail:
- mlog_exit(status);
- return status;
-}
-
-static void ocfs2_handle_cleanup_locks(struct ocfs2_journal *journal,
- struct ocfs2_journal_handle *handle)
-{
- struct list_head *p, *n;
- struct ocfs2_journal_lock *lock;
- struct inode *inode;
-
- list_for_each_safe(p, n, &(handle->locks)) {
- lock = list_entry(p, struct ocfs2_journal_lock,
- jl_lock_list);
- list_del(&lock->jl_lock_list);
- handle->num_locks--;
-
- inode = lock->jl_inode;
- ocfs2_meta_unlock(inode, 1);
- if (atomic_read(&inode->i_count) == 1)
- mlog(ML_ERROR,
- "Inode %"MLFu64", I'm doing a last iput for!",
- OCFS2_I(inode)->ip_blkno);
- iput(inode);
- kmem_cache_free(ocfs2_lock_cache, lock);
- }
-}
-
-#define OCFS2_DEFAULT_COMMIT_INTERVAL (HZ * 5)
+#define OCFS2_DEFAULT_COMMIT_INTERVAL (HZ * JBD2_DEFAULT_MAX_COMMIT_AGE)
void ocfs2_set_journal_params(struct ocfs2_super *osb)
{
journal_t *journal = osb->journal->j_journal;
+ unsigned long commit_interval = OCFS2_DEFAULT_COMMIT_INTERVAL;
+
+ if (osb->osb_commit_interval)
+ commit_interval = osb->osb_commit_interval;
spin_lock(&journal->j_state_lock);
- journal->j_commit_interval = OCFS2_DEFAULT_COMMIT_INTERVAL;
+ journal->j_commit_interval = commit_interval;
if (osb->s_mount_opt & OCFS2_MOUNT_BARRIER)
- journal->j_flags |= JFS_BARRIER;
+ journal->j_flags |= JBD2_BARRIER;
else
- journal->j_flags &= ~JFS_BARRIER;
+ journal->j_flags &= ~JBD2_BARRIER;
spin_unlock(&journal->j_state_lock);
}
struct ocfs2_dinode *di = NULL;
struct buffer_head *bh = NULL;
struct ocfs2_super *osb;
- int meta_lock = 0;
+ int inode_lock = 0;
mlog_entry_void();
SET_INODE_JOURNAL(inode);
OCFS2_I(inode)->ip_open_count++;
- status = ocfs2_meta_lock(inode, NULL, &bh, 1);
+ /* Skip recovery waits here - journal inode metadata never
+ * changes in a live cluster so it can be considered an
+ * exception to the rule. */
+ status = ocfs2_inode_lock_full(inode, &bh, 1, OCFS2_META_LOCK_RECOVERY);
if (status < 0) {
if (status != -ERESTARTSYS)
mlog(ML_ERROR, "Could not get lock on journal!\n");
goto done;
}
- meta_lock = 1;
+ inode_lock = 1;
di = (struct ocfs2_dinode *)bh->b_data;
if (inode->i_size < OCFS2_MIN_JOURNAL_SIZE) {
}
mlog(0, "inode->i_size = %lld\n", inode->i_size);
- mlog(0, "inode->i_blocks = %lu\n", inode->i_blocks);
+ mlog(0, "inode->i_blocks = %llu\n",
+ (unsigned long long)inode->i_blocks);
mlog(0, "inode->ip_clusters = %u\n", OCFS2_I(inode)->ip_clusters);
/* call the kernels journal init function now */
- j_journal = journal_init_inode(inode);
+ j_journal = jbd2_journal_init_inode(inode);
if (j_journal == NULL) {
mlog(ML_ERROR, "Linux journal layer error\n");
status = -EINVAL;
goto done;
}
- mlog(0, "Returned from journal_init_inode\n");
+ mlog(0, "Returned from jbd2_journal_init_inode\n");
mlog(0, "j_journal->j_maxlen = %u\n", j_journal->j_maxlen);
*dirty = (le32_to_cpu(di->id1.journal1.ij_flags) &
status = 0;
done:
if (status < 0) {
- if (meta_lock)
- ocfs2_meta_unlock(inode, 1);
- if (bh != NULL)
- brelse(bh);
+ if (inode_lock)
+ ocfs2_inode_unlock(inode, 1);
+ brelse(bh);
if (inode) {
OCFS2_I(inode)->ip_open_count--;
iput(inode);
return status;
}
+static void ocfs2_bump_recovery_generation(struct ocfs2_dinode *di)
+{
+ le32_add_cpu(&(di->id1.journal1.ij_recovery_generation), 1);
+}
+
+static u32 ocfs2_get_recovery_generation(struct ocfs2_dinode *di)
+{
+ return le32_to_cpu(di->id1.journal1.ij_recovery_generation);
+}
+
static int ocfs2_journal_toggle_dirty(struct ocfs2_super *osb,
- int dirty)
+ int dirty, int replayed)
{
int status;
unsigned int flags;
/* This is called from startup/shutdown which will
* handle the errors in a specific manner, so no need
* to call ocfs2_error() here. */
- mlog(ML_ERROR, "Journal dinode %"MLFu64" has invalid "
- "signature: %.*s", fe->i_blkno, 7, fe->i_signature);
+ mlog(ML_ERROR, "Journal dinode %llu has invalid "
+ "signature: %.*s",
+ (unsigned long long)le64_to_cpu(fe->i_blkno), 7,
+ fe->i_signature);
status = -EIO;
goto out;
}
flags &= ~OCFS2_JOURNAL_DIRTY_FL;
fe->id1.journal1.ij_flags = cpu_to_le32(flags);
+ if (replayed)
+ ocfs2_bump_recovery_generation(fe);
+
status = ocfs2_write_block(osb, bh, journal->j_inode);
if (status < 0)
mlog_errno(status);
if (journal->j_state != OCFS2_JOURNAL_LOADED)
goto done;
- /* need to inc inode use count as journal_destroy will iput. */
+ /* need to inc inode use count - jbd2_journal_destroy will iput. */
if (!igrab(inode))
BUG();
BUG_ON(atomic_read(&(osb->journal->j_num_trans)) != 0);
- status = ocfs2_journal_toggle_dirty(osb, 0);
- if (status < 0)
- mlog_errno(status);
+ if (ocfs2_mount_local(osb)) {
+ jbd2_journal_lock_updates(journal->j_journal);
+ status = jbd2_journal_flush(journal->j_journal);
+ jbd2_journal_unlock_updates(journal->j_journal);
+ if (status < 0)
+ mlog_errno(status);
+ }
+
+ if (status == 0) {
+ /*
+ * Do not toggle if flush was unsuccessful otherwise
+ * will leave dirty metadata in a "clean" journal
+ */
+ status = ocfs2_journal_toggle_dirty(osb, 0, 0);
+ if (status < 0)
+ mlog_errno(status);
+ }
/* Shutdown the kernel journal system */
- journal_destroy(journal->j_journal);
+ jbd2_journal_destroy(journal->j_journal);
OCFS2_I(inode)->ip_open_count--;
/* unlock our journal */
- ocfs2_meta_unlock(inode, 1);
+ ocfs2_inode_unlock(inode, 1);
brelse(journal->j_bh);
journal->j_bh = NULL;
{
int olderr;
- olderr = journal_errno(journal);
+ olderr = jbd2_journal_errno(journal);
if (olderr) {
mlog(ML_ERROR, "File system error %d recorded in "
"journal %u.\n", olderr, slot);
mlog(ML_ERROR, "File system on device %s needs checking.\n",
sb->s_id);
- journal_ack_err(journal);
- journal_clear_err(journal);
+ jbd2_journal_ack_err(journal);
+ jbd2_journal_clear_err(journal);
}
}
-int ocfs2_journal_load(struct ocfs2_journal *journal)
+int ocfs2_journal_load(struct ocfs2_journal *journal, int local, int replayed)
{
int status = 0;
struct ocfs2_super *osb;
mlog_entry_void();
- if (!journal)
- BUG();
+ BUG_ON(!journal);
osb = journal->j_osb;
- status = journal_load(journal->j_journal);
+ status = jbd2_journal_load(journal->j_journal);
if (status < 0) {
mlog(ML_ERROR, "Failed to load journal!\n");
goto done;
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, replayed);
if (status < 0) {
mlog_errno(status);
goto done;
}
/* Launch the commit thread */
- osb->commit_task = kthread_run(ocfs2_commit_thread, osb, "ocfs2cmt-%d",
- osb->osb_id);
- if (IS_ERR(osb->commit_task)) {
- status = PTR_ERR(osb->commit_task);
+ if (!local) {
+ osb->commit_task = kthread_run(ocfs2_commit_thread, osb,
+ "ocfs2cmt");
+ if (IS_ERR(osb->commit_task)) {
+ status = PTR_ERR(osb->commit_task);
+ osb->commit_task = NULL;
+ mlog(ML_ERROR, "unable to launch ocfs2commit thread, "
+ "error=%d", status);
+ goto done;
+ }
+ } else
osb->commit_task = NULL;
- mlog(ML_ERROR, "unable to launch ocfs2commit thread, error=%d",
- status);
- goto done;
- }
done:
mlog_exit(status);
BUG_ON(!journal);
- status = journal_wipe(journal->j_journal, full);
+ status = jbd2_journal_wipe(journal->j_journal, full);
if (status < 0) {
mlog_errno(status);
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);
return status;
}
+static int ocfs2_recovery_completed(struct ocfs2_super *osb)
+{
+ int empty;
+ struct ocfs2_recovery_map *rm = osb->recovery_map;
+
+ spin_lock(&osb->osb_lock);
+ empty = (rm->rm_used == 0);
+ spin_unlock(&osb->osb_lock);
+
+ return empty;
+}
+
+void ocfs2_wait_for_recovery(struct ocfs2_super *osb)
+{
+ wait_event(osb->recovery_event, ocfs2_recovery_completed(osb));
+}
+
/*
* JBD Might read a cached version of another nodes journal file. We
* don't want this as this file changes often and we get no
static int ocfs2_force_read_journal(struct inode *inode)
{
int status = 0;
- int i, p_blocks;
- u64 v_blkno, p_blkno;
-#define CONCURRENT_JOURNAL_FILL 32
+ int i;
+ u64 v_blkno, p_blkno, p_blocks, num_blocks;
+#define CONCURRENT_JOURNAL_FILL 32ULL
struct buffer_head *bhs[CONCURRENT_JOURNAL_FILL];
mlog_entry_void();
- BUG_ON(inode->i_blocks !=
- ocfs2_align_bytes_to_sectors(i_size_read(inode)));
-
memset(bhs, 0, sizeof(struct buffer_head *) * CONCURRENT_JOURNAL_FILL);
- mlog(0, "Force reading %lu blocks\n",
- (inode->i_blocks >> (inode->i_sb->s_blocksize_bits - 9)));
-
+ num_blocks = ocfs2_blocks_for_bytes(inode->i_sb, inode->i_size);
v_blkno = 0;
- while (v_blkno <
- (inode->i_blocks >> (inode->i_sb->s_blocksize_bits - 9))) {
-
+ while (v_blkno < num_blocks) {
status = ocfs2_extent_map_get_blocks(inode, v_blkno,
- 1, &p_blkno,
- &p_blocks);
+ &p_blkno, &p_blocks, NULL);
if (status < 0) {
mlog_errno(status);
goto bail;
if (p_blocks > CONCURRENT_JOURNAL_FILL)
p_blocks = CONCURRENT_JOURNAL_FILL;
- status = ocfs2_read_blocks(OCFS2_SB(inode->i_sb),
- p_blkno, p_blocks, bhs, 0,
- inode);
+ /* We are reading journal data which should not
+ * be put in the uptodate cache */
+ status = ocfs2_read_blocks_sync(OCFS2_SB(inode->i_sb),
+ p_blkno, p_blocks, bhs);
if (status < 0) {
mlog_errno(status);
goto bail;
bail:
for(i = 0; i < CONCURRENT_JOURNAL_FILL; i++)
- if (bhs[i])
- brelse(bhs[i]);
+ brelse(bhs[i]);
mlog_exit(status);
return status;
}
* NOTE: This function can and will sleep on recovery of other nodes
* during cluster locking, just like any other ocfs2 process.
*/
-void ocfs2_complete_recovery(void *data)
+void ocfs2_complete_recovery(struct work_struct *work)
{
int ret;
- struct ocfs2_super *osb = data;
- struct ocfs2_journal *journal = osb->journal;
+ struct ocfs2_journal *journal =
+ container_of(work, struct ocfs2_journal, j_recovery_work);
+ struct ocfs2_super *osb = journal->j_osb;
struct ocfs2_dinode *la_dinode, *tl_dinode;
- struct ocfs2_la_recovery_item *item;
- struct list_head *p, *n;
+ struct ocfs2_la_recovery_item *item, *n;
LIST_HEAD(tmp_la_list);
mlog_entry_void();
list_splice_init(&journal->j_la_cleanups, &tmp_la_list);
spin_unlock(&journal->j_lock);
- list_for_each_safe(p, n, &tmp_la_list) {
- item = list_entry(p, struct ocfs2_la_recovery_item, lri_list);
+ list_for_each_entry_safe(item, n, &tmp_la_list, lri_list) {
list_del_init(&item->lri_list);
mlog(0, "Complete recovery for slot %d\n", item->lri_slot);
la_dinode = item->lri_la_dinode;
if (la_dinode) {
- mlog(0, "Clean up local alloc %"MLFu64"\n",
- la_dinode->i_blkno);
+ mlog(0, "Clean up local alloc %llu\n",
+ (unsigned long long)le64_to_cpu(la_dinode->i_blkno));
ret = ocfs2_complete_local_alloc_recovery(osb,
la_dinode);
tl_dinode = item->lri_tl_dinode;
if (tl_dinode) {
- mlog(0, "Clean up truncate log %"MLFu64"\n",
- tl_dinode->i_blkno);
+ mlog(0, "Clean up truncate log %llu\n",
+ (unsigned long long)le64_to_cpu(tl_dinode->i_blkno));
ret = ocfs2_complete_truncate_log_recovery(osb,
tl_dinode);
{
struct ocfs2_la_recovery_item *item;
- item = kmalloc(sizeof(struct ocfs2_la_recovery_item), GFP_KERNEL);
+ item = kmalloc(sizeof(struct ocfs2_la_recovery_item), GFP_NOFS);
if (!item) {
/* Though we wish to avoid it, we are in fact safe in
* skipping local alloc cleanup as fsck.ocfs2 is more
{
int status, node_num;
struct ocfs2_super *osb = arg;
+ struct ocfs2_recovery_map *rm = osb->recovery_map;
mlog_entry_void();
goto bail;
}
- while(!ocfs2_node_map_is_empty(osb, &osb->recovery_map)) {
- node_num = ocfs2_node_map_first_set_bit(osb,
- &osb->recovery_map);
- if (node_num == O2NM_INVALID_NODE_NUM) {
- mlog(0, "Out of nodes to recover.\n");
- break;
- }
+ spin_lock(&osb->osb_lock);
+ while (rm->rm_used) {
+ /* It's always safe to remove entry zero, as we won't
+ * clear it until ocfs2_recover_node() has succeeded. */
+ node_num = rm->rm_entries[0];
+ spin_unlock(&osb->osb_lock);
status = ocfs2_recover_node(osb, node_num);
- if (status < 0) {
+ if (!status) {
+ ocfs2_recovery_map_clear(osb, node_num);
+ } else {
mlog(ML_ERROR,
"Error %d recovering node %d on device (%u,%u)!\n",
status, node_num,
MAJOR(osb->sb->s_dev), MINOR(osb->sb->s_dev));
mlog(ML_ERROR, "Volume requires unmount.\n");
- continue;
}
- ocfs2_recovery_map_clear(osb, node_num);
+ spin_lock(&osb->osb_lock);
}
+ spin_unlock(&osb->osb_lock);
+ mlog(0, "All nodes recovered\n");
+
+ /* Refresh all journal recovery generations from disk */
+ status = ocfs2_check_journals_nolocks(osb);
+ status = (status == -EROFS) ? 0 : status;
+ if (status < 0)
+ mlog_errno(status);
+
ocfs2_super_unlock(osb, 1);
/* We always run recovery on our own orphan dir - the dead
- * node(s) may have voted "no" on an inode delete earlier. A
- * revote is therefore required. */
+ * node(s) may have disallowd a previos inode delete. Re-processing
+ * is therefore required. */
ocfs2_queue_recovery_completion(osb->journal, osb->slot_num, NULL,
NULL);
bail:
mutex_lock(&osb->recovery_lock);
- if (!status &&
- !ocfs2_node_map_is_empty(osb, &osb->recovery_map)) {
+ if (!status && !ocfs2_recovery_completed(osb)) {
mutex_unlock(&osb->recovery_lock);
goto restart;
}
/* People waiting on recovery will wait on
* the recovery map to empty. */
- if (!ocfs2_recovery_map_set(osb, node_num))
- mlog(0, "node %d already be in recovery.\n", node_num);
+ if (ocfs2_recovery_map_set(osb, node_num))
+ mlog(0, "node %d already in recovery map.\n", node_num);
mlog(0, "starting recovery thread...\n");
goto out;
osb->recovery_thread_task = kthread_run(__ocfs2_recovery_thread, osb,
- "ocfs2rec-%d", osb->osb_id);
+ "ocfs2rec");
if (IS_ERR(osb->recovery_thread_task)) {
mlog_errno((int)PTR_ERR(osb->recovery_thread_task));
osb->recovery_thread_task = NULL;
mlog_exit_void();
}
+static int ocfs2_read_journal_inode(struct ocfs2_super *osb,
+ int slot_num,
+ struct buffer_head **bh,
+ struct inode **ret_inode)
+{
+ int status = -EACCES;
+ struct inode *inode = NULL;
+
+ BUG_ON(slot_num >= osb->max_slots);
+
+ inode = ocfs2_get_system_file_inode(osb, JOURNAL_SYSTEM_INODE,
+ slot_num);
+ if (!inode || is_bad_inode(inode)) {
+ mlog_errno(status);
+ goto bail;
+ }
+ SET_INODE_JOURNAL(inode);
+
+ status = ocfs2_read_blocks(inode, OCFS2_I(inode)->ip_blkno, 1, bh, 0);
+ if (status < 0) {
+ mlog_errno(status);
+ goto bail;
+ }
+
+ status = 0;
+
+bail:
+ if (inode) {
+ if (status || !ret_inode)
+ iput(inode);
+ else
+ *ret_inode = inode;
+ }
+ 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,
struct ocfs2_dinode *fe;
journal_t *journal = NULL;
struct buffer_head *bh = NULL;
+ u32 slot_reco_gen;
- inode = ocfs2_get_system_file_inode(osb, JOURNAL_SYSTEM_INODE,
- slot_num);
- if (inode == NULL) {
- status = -EACCES;
+ status = ocfs2_read_journal_inode(osb, slot_num, &bh, &inode);
+ if (status) {
mlog_errno(status);
goto done;
}
- if (is_bad_inode(inode)) {
- status = -EACCES;
- iput(inode);
- inode = NULL;
- mlog_errno(status);
+
+ fe = (struct ocfs2_dinode *)bh->b_data;
+ slot_reco_gen = ocfs2_get_recovery_generation(fe);
+ brelse(bh);
+ bh = NULL;
+
+ /*
+ * 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_recovery_generations[slot_num] != slot_reco_gen) {
+ mlog(0, "Slot %u already recovered (old/new=%u/%u)\n", slot_num,
+ osb->slot_recovery_generations[slot_num], slot_reco_gen);
+ osb->slot_recovery_generations[slot_num] = slot_reco_gen;
+ status = -EBUSY;
goto done;
}
- SET_INODE_JOURNAL(inode);
- status = ocfs2_meta_lock_full(inode, NULL, &bh, 1,
- OCFS2_META_LOCK_RECOVERY);
+ /* 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) {
- mlog(0, "status returned from ocfs2_meta_lock=%d\n", status);
+ mlog(0, "status returned from ocfs2_inode_lock=%d\n", status);
if (status != -ERESTARTSYS)
mlog(ML_ERROR, "Could not lock journal!\n");
goto done;
fe = (struct ocfs2_dinode *) bh->b_data;
flags = le32_to_cpu(fe->id1.journal1.ij_flags);
+ slot_reco_gen = ocfs2_get_recovery_generation(fe);
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_recovery_generations[slot_num] = slot_reco_gen;
goto done;
}
}
mlog(0, "calling journal_init_inode\n");
- journal = journal_init_inode(inode);
+ journal = jbd2_journal_init_inode(inode);
if (journal == NULL) {
mlog(ML_ERROR, "Linux journal layer error\n");
status = -EIO;
goto done;
}
- status = journal_load(journal);
+ status = jbd2_journal_load(journal);
if (status < 0) {
mlog_errno(status);
if (!igrab(inode))
BUG();
- journal_destroy(journal);
+ jbd2_journal_destroy(journal);
goto done;
}
/* wipe the journal */
mlog(0, "flushing the journal.\n");
- journal_lock_updates(journal);
- status = journal_flush(journal);
- journal_unlock_updates(journal);
+ jbd2_journal_lock_updates(journal);
+ status = jbd2_journal_flush(journal);
+ jbd2_journal_unlock_updates(journal);
if (status < 0)
mlog_errno(status);
flags &= ~OCFS2_JOURNAL_DIRTY_FL;
fe->id1.journal1.ij_flags = cpu_to_le32(flags);
+ /* Increment recovery generation to indicate successful recovery */
+ ocfs2_bump_recovery_generation(fe);
+ osb->slot_recovery_generations[slot_num] =
+ ocfs2_get_recovery_generation(fe);
+
status = ocfs2_write_block(osb, bh, inode);
if (status < 0)
mlog_errno(status);
if (!igrab(inode))
BUG();
- journal_destroy(journal);
+ jbd2_journal_destroy(journal);
done:
/* drop the lock on this nodes journal */
if (got_lock)
- ocfs2_meta_unlock(inode, 1);
+ ocfs2_inode_unlock(inode, 1);
if (inode)
iput(inode);
- if (bh)
- brelse(bh);
+ brelse(bh);
mlog_exit(status);
return status;
{
int status = 0;
int slot_num;
- struct ocfs2_slot_info *si = osb->slot_info;
struct ocfs2_dinode *la_copy = NULL;
struct ocfs2_dinode *tl_copy = NULL;
* case we should've called ocfs2_journal_load instead. */
BUG_ON(osb->node_num == node_num);
- slot_num = ocfs2_node_num_to_slot(si, node_num);
- if (slot_num == OCFS2_INVALID_SLOT) {
+ slot_num = ocfs2_node_num_to_slot(osb, node_num);
+ if (slot_num == -ENOENT) {
status = 0;
mlog(0, "no slot for this node, so no recovery required.\n");
goto done;
status = ocfs2_replay_journal(osb, node_num, slot_num);
if (status < 0) {
+ if (status == -EBUSY) {
+ mlog(0, "Skipping recovery for slot %u (node %u) "
+ "as another node has recovered it\n", slot_num,
+ node_num);
+ status = 0;
+ goto done;
+ }
mlog_errno(status);
goto done;
}
/* Likewise, this would be a strange but ultimately not so
* harmful place to get an error... */
- ocfs2_clear_slot(si, slot_num);
- status = ocfs2_update_disk_slots(osb, si);
+ status = ocfs2_clear_slot(osb, slot_num);
if (status < 0)
mlog_errno(status);
SET_INODE_JOURNAL(inode);
flags = OCFS2_META_LOCK_RECOVERY | OCFS2_META_LOCK_NOQUEUE;
- status = ocfs2_meta_lock_full(inode, NULL, NULL, 1, flags);
+ status = ocfs2_inode_lock_full(inode, NULL, 1, flags);
if (status < 0) {
if (status != -EAGAIN)
mlog_errno(status);
goto bail;
}
- ocfs2_meta_unlock(inode, 1);
+ ocfs2_inode_unlock(inode, 1);
bail:
if (inode)
iput(inode);
* slot info struct has been updated from disk. */
int ocfs2_mark_dead_nodes(struct ocfs2_super *osb)
{
- int status, i, node_num;
- struct ocfs2_slot_info *si = osb->slot_info;
+ unsigned int node_num;
+ int status, i;
+ u32 gen;
+ struct buffer_head *bh = NULL;
+ struct ocfs2_dinode *di;
/* 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++) {
- if (i == osb->slot_num)
+ for (i = 0; i < osb->max_slots; i++) {
+ /* Read journal inode to get the recovery generation */
+ status = ocfs2_read_journal_inode(osb, i, &bh, NULL);
+ if (status) {
+ mlog_errno(status);
+ goto bail;
+ }
+ di = (struct ocfs2_dinode *)bh->b_data;
+ gen = ocfs2_get_recovery_generation(di);
+ brelse(bh);
+ bh = NULL;
+
+ spin_lock(&osb->osb_lock);
+ osb->slot_recovery_generations[i] = gen;
+
+ mlog(0, "Slot %u recovery generation is %u\n", i,
+ osb->slot_recovery_generations[i]);
+
+ if (i == osb->slot_num) {
+ spin_unlock(&osb->osb_lock);
continue;
- if (ocfs2_is_empty_slot(si, i))
+ }
+
+ status = ocfs2_slot_to_node_num_locked(osb, i, &node_num);
+ if (status == -ENOENT) {
+ spin_unlock(&osb->osb_lock);
continue;
+ }
- node_num = si->si_global_node_nums[i];
- if (ocfs2_node_map_test_bit(osb, &osb->recovery_map, node_num))
+ if (__ocfs2_recovery_map_test(osb, node_num)) {
+ spin_unlock(&osb->osb_lock);
continue;
- spin_unlock(&si->si_lock);
+ }
+ spin_unlock(&osb->osb_lock);
/* Ok, we have a slot occupied by another node which
* is not in the recovery map. We trylock his journal
mlog_errno(status);
goto bail;
}
-
- spin_lock(&si->si_lock);
}
- spin_unlock(&si->si_lock);
status = 0;
bail:
return status;
}
-static int ocfs2_recover_orphans(struct ocfs2_super *osb,
- int slot)
+struct ocfs2_orphan_filldir_priv {
+ struct inode *head;
+ struct ocfs2_super *osb;
+};
+
+static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,
+ loff_t pos, u64 ino, unsigned type)
{
- int status = 0;
- int have_disk_lock = 0;
- struct inode *inode = NULL;
+ struct ocfs2_orphan_filldir_priv *p = priv;
struct inode *iter;
+
+ if (name_len == 1 && !strncmp(".", name, 1))
+ return 0;
+ if (name_len == 2 && !strncmp("..", name, 2))
+ return 0;
+
+ /* Skip bad inodes so that recovery can continue */
+ iter = ocfs2_iget(p->osb, ino,
+ OCFS2_FI_FLAG_ORPHAN_RECOVERY, 0);
+ if (IS_ERR(iter))
+ return 0;
+
+ mlog(0, "queue orphan %llu\n",
+ (unsigned long long)OCFS2_I(iter)->ip_blkno);
+ /* No locking is required for the next_orphan queue as there
+ * is only ever a single process doing orphan recovery. */
+ OCFS2_I(iter)->ip_next_orphan = p->head;
+ p->head = iter;
+
+ return 0;
+}
+
+static int ocfs2_queue_orphans(struct ocfs2_super *osb,
+ int slot,
+ struct inode **head)
+{
+ int status;
struct inode *orphan_dir_inode = NULL;
- unsigned long offset, blk, local;
- struct buffer_head *bh = NULL;
- struct ocfs2_dir_entry *de;
- struct super_block *sb = osb->sb;
- struct ocfs2_inode_info *oi;
+ struct ocfs2_orphan_filldir_priv priv;
+ loff_t pos = 0;
- mlog(0, "Recover inodes from orphan dir in slot %d\n", slot);
+ priv.osb = osb;
+ priv.head = *head;
orphan_dir_inode = ocfs2_get_system_file_inode(osb,
ORPHAN_DIR_SYSTEM_INODE,
if (!orphan_dir_inode) {
status = -ENOENT;
mlog_errno(status);
- goto out;
- }
+ return status;
+ }
mutex_lock(&orphan_dir_inode->i_mutex);
- status = ocfs2_meta_lock(orphan_dir_inode, NULL, NULL, 0);
+ status = ocfs2_inode_lock(orphan_dir_inode, NULL, 0);
if (status < 0) {
- mutex_unlock(&orphan_dir_inode->i_mutex);
mlog_errno(status);
goto out;
}
- have_disk_lock = 1;
- offset = 0;
- iter = NULL;
- while(offset < i_size_read(orphan_dir_inode)) {
- blk = offset >> sb->s_blocksize_bits;
+ status = ocfs2_dir_foreach(orphan_dir_inode, &pos, &priv,
+ ocfs2_orphan_filldir);
+ if (status) {
+ mlog_errno(status);
+ goto out_cluster;
+ }
- bh = ocfs2_bread(orphan_dir_inode, blk, &status, 0);
- if (!bh)
- status = -EINVAL;
- if (status < 0) {
- mutex_unlock(&orphan_dir_inode->i_mutex);
- if (bh)
- brelse(bh);
- mlog_errno(status);
- goto out;
- }
+ *head = priv.head;
- local = 0;
- while(offset < i_size_read(orphan_dir_inode)
- && local < sb->s_blocksize) {
- de = (struct ocfs2_dir_entry *) (bh->b_data + local);
-
- if (!ocfs2_check_dir_entry(orphan_dir_inode,
- de, bh, local)) {
- mutex_unlock(&orphan_dir_inode->i_mutex);
- status = -EINVAL;
- mlog_errno(status);
- brelse(bh);
- goto out;
- }
-
- local += le16_to_cpu(de->rec_len);
- offset += le16_to_cpu(de->rec_len);
-
- /* I guess we silently fail on no inode? */
- if (!le64_to_cpu(de->inode))
- continue;
- if (de->file_type > OCFS2_FT_MAX) {
- mlog(ML_ERROR,
- "block %llu contains invalid de: "
- "inode = %"MLFu64", rec_len = %u, "
- "name_len = %u, file_type = %u, "
- "name='%.*s'\n",
- (unsigned long long)bh->b_blocknr,
- le64_to_cpu(de->inode),
- le16_to_cpu(de->rec_len),
- de->name_len,
- de->file_type,
- de->name_len,
- de->name);
- continue;
- }
- if (de->name_len == 1 && !strncmp(".", de->name, 1))
- continue;
- if (de->name_len == 2 && !strncmp("..", de->name, 2))
- continue;
-
- iter = ocfs2_iget(osb, le64_to_cpu(de->inode));
- if (IS_ERR(iter))
- continue;
-
- mlog(0, "queue orphan %"MLFu64"\n",
- OCFS2_I(iter)->ip_blkno);
- OCFS2_I(iter)->ip_next_orphan = inode;
- inode = iter;
- }
- brelse(bh);
- }
+out_cluster:
+ ocfs2_inode_unlock(orphan_dir_inode, 0);
+out:
mutex_unlock(&orphan_dir_inode->i_mutex);
+ iput(orphan_dir_inode);
+ return status;
+}
- ocfs2_meta_unlock(orphan_dir_inode, 0);
- have_disk_lock = 0;
+static int ocfs2_orphan_recovery_can_continue(struct ocfs2_super *osb,
+ int slot)
+{
+ int ret;
- iput(orphan_dir_inode);
- orphan_dir_inode = NULL;
+ spin_lock(&osb->osb_lock);
+ ret = !osb->osb_orphan_wipes[slot];
+ spin_unlock(&osb->osb_lock);
+ return ret;
+}
+
+static void ocfs2_mark_recovering_orphan_dir(struct ocfs2_super *osb,
+ int slot)
+{
+ spin_lock(&osb->osb_lock);
+ /* Mark ourselves such that new processes in delete_inode()
+ * know to quit early. */
+ ocfs2_node_map_set_bit(osb, &osb->osb_recovering_orphan_dirs, slot);
+ while (osb->osb_orphan_wipes[slot]) {
+ /* If any processes are already in the middle of an
+ * orphan wipe on this dir, then we need to wait for
+ * them. */
+ spin_unlock(&osb->osb_lock);
+ wait_event_interruptible(osb->osb_wipe_event,
+ ocfs2_orphan_recovery_can_continue(osb, slot));
+ spin_lock(&osb->osb_lock);
+ }
+ spin_unlock(&osb->osb_lock);
+}
+
+static void ocfs2_clear_recovering_orphan_dir(struct ocfs2_super *osb,
+ int slot)
+{
+ ocfs2_node_map_clear_bit(osb, &osb->osb_recovering_orphan_dirs, slot);
+}
+
+/*
+ * Orphan recovery. Each mounted node has it's own orphan dir which we
+ * must run during recovery. Our strategy here is to build a list of
+ * the inodes in the orphan dir and iget/iput them. The VFS does
+ * (most) of the rest of the work.
+ *
+ * Orphan recovery can happen at any time, not just mount so we have a
+ * couple of extra considerations.
+ *
+ * - We grab as many inodes as we can under the orphan dir lock -
+ * doing iget() outside the orphan dir risks getting a reference on
+ * an invalid inode.
+ * - We must be sure not to deadlock with other processes on the
+ * system wanting to run delete_inode(). This can happen when they go
+ * to lock the orphan dir and the orphan recovery process attempts to
+ * iget() inside the orphan dir lock. This can be avoided by
+ * advertising our state to ocfs2_delete_inode().
+ */
+static int ocfs2_recover_orphans(struct ocfs2_super *osb,
+ int slot)
+{
+ int ret = 0;
+ struct inode *inode = NULL;
+ struct inode *iter;
+ struct ocfs2_inode_info *oi;
+
+ mlog(0, "Recover inodes from orphan dir in slot %d\n", slot);
+
+ ocfs2_mark_recovering_orphan_dir(osb, slot);
+ ret = ocfs2_queue_orphans(osb, slot, &inode);
+ ocfs2_clear_recovering_orphan_dir(osb, slot);
+
+ /* Error here should be noted, but we want to continue with as
+ * many queued inodes as we've got. */
+ if (ret)
+ mlog_errno(ret);
while (inode) {
oi = OCFS2_I(inode);
- mlog(0, "iput orphan %"MLFu64"\n", oi->ip_blkno);
+ mlog(0, "iput orphan %llu\n", (unsigned long long)oi->ip_blkno);
iter = oi->ip_next_orphan;
spin_lock(&oi->ip_lock);
- /* Delete voting may have set these on the assumption
- * that the other node would wipe them successfully.
- * If they are still in the node's orphan dir, we need
- * to reset that state. */
+ /* The remote delete code may have set these on the
+ * assumption that the other node would wipe them
+ * successfully. If they are still in the node's
+ * orphan dir, we need to reset that state. */
oi->ip_flags &= ~(OCFS2_INODE_DELETED|OCFS2_INODE_SKIP_DELETE);
/* Set the proper information to get us going into
* ocfs2_delete_inode. */
oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
- oi->ip_orphaned_slot = slot;
spin_unlock(&oi->ip_lock);
iput(inode);
inode = iter;
}
-out:
- if (have_disk_lock)
- ocfs2_meta_unlock(orphan_dir_inode, 0);
-
- if (orphan_dir_inode)
- iput(orphan_dir_inode);
-
- return status;
+ return ret;
}
static int ocfs2_wait_on_mount(struct ocfs2_super *osb)
while (!(kthread_should_stop() &&
atomic_read(&journal->j_num_trans) == 0)) {
- wait_event_interruptible_timeout(osb->checkpoint_event,
- atomic_read(&journal->j_num_trans)
- || kthread_should_stop(),
- OCFS2_CHECKPOINT_INTERVAL);
+ wait_event_interruptible(osb->checkpoint_event,
+ atomic_read(&journal->j_num_trans)
+ || kthread_should_stop());
status = ocfs2_commit_cache(osb);
if (status < 0)
return 0;
}
-/* Look for a dirty journal without taking any cluster locks. Used for
- * hard readonly access to determine whether the file system journals
- * require recovery. */
+/* Reads all the journal inodes without taking any cluster locks. Used
+ * for hard readonly access to determine whether any journal requires
+ * recovery. Also used to refresh the recovery generation numbers after
+ * a journal has been recovered by another node.
+ */
int ocfs2_check_journals_nolocks(struct ocfs2_super *osb)
{
int ret = 0;
unsigned int slot;
- struct buffer_head *di_bh;
+ struct buffer_head *di_bh = NULL;
struct ocfs2_dinode *di;
- struct inode *journal = NULL;
+ int journal_dirty = 0;
for(slot = 0; slot < osb->max_slots; slot++) {
- journal = ocfs2_get_system_file_inode(osb,
- JOURNAL_SYSTEM_INODE,
- slot);
- if (!journal || is_bad_inode(journal)) {
- ret = -EACCES;
- mlog_errno(ret);
- goto out;
- }
-
- di_bh = NULL;
- ret = ocfs2_read_block(osb, OCFS2_I(journal)->ip_blkno, &di_bh,
- 0, journal);
- if (ret < 0) {
+ ret = ocfs2_read_journal_inode(osb, slot, &di_bh, NULL);
+ if (ret) {
mlog_errno(ret);
goto out;
}
di = (struct ocfs2_dinode *) di_bh->b_data;
+ osb->slot_recovery_generations[slot] =
+ ocfs2_get_recovery_generation(di);
+
if (le32_to_cpu(di->id1.journal1.ij_flags) &
OCFS2_JOURNAL_DIRTY_FL)
- ret = -EROFS;
+ journal_dirty = 1;
brelse(di_bh);
- if (ret)
- break;
+ di_bh = NULL;
}
out:
- if (journal)
- iput(journal);
-
+ if (journal_dirty)
+ ret = -EROFS;
return ret;
}