journal.c

ocfs1.2.7 源码

/* -*- mode: c; c-basic-offset: 8; -*-
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * journal.c
 *
 * Defines functions of journalling api
 *
 * Copyright (C) 2003, 2004 Oracle.  All rights reserved.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 021110-1307, USA.
 */

#include <linux/fs.h>
#include <linux/types.h>
#include <linux/slab.h>
#include <linux/highmem.h>
#include <linux/kthread.h>

#define MLOG_MASK_PREFIX ML_JOURNAL
#include <cluster/masklog.h>

#include "ocfs2.h"

#include "alloc.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;

static int ocfs2_force_read_journal(struct inode *inode);
static int ocfs2_recover_node(struct ocfs2_super *osb,
			      int node_num);
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);
static int ocfs2_trylock_journal(struct ocfs2_super *osb,
				 int slot_num);
static int ocfs2_recover_orphans(struct ocfs2_super *osb,
				 int slot, int node_num);
static int ocfs2_commit_thread(void *arg);

static int ocfs2_commit_cache(struct ocfs2_super *osb)
{
	int status = 0;
	unsigned int flushed;
	unsigned long old_id;
	struct ocfs2_journal *journal = NULL;

	mlog_entry_void();

	journal = osb->journal;

	/* Flush all pending commits and checkpoint the journal. */
	down_write(&journal->j_trans_barrier);

	if (atomic_read(&journal->j_num_trans) == 0) {
		up_write(&journal->j_trans_barrier);
		mlog(0, "No transactions for me to flush!\n");
		goto finally;
	}

	journal_lock_updates(journal->j_journal);
	status = journal_flush(journal->j_journal);
	journal_unlock_updates(journal->j_journal);
	if (status < 0) {
		up_write(&journal->j_trans_barrier);
		mlog_errno(status);
		goto finally;
	}

	old_id = ocfs2_inc_trans_id(journal);

	flushed = atomic_read(&journal->j_num_trans);
	atomic_set(&journal->j_num_trans, 0);
	up_write(&journal->j_trans_barrier);

	mlog(0, "commit_thread: flushed transaction %lu (%u handles)\n",
	     journal->j_trans_id, flushed);

	ocfs2_kick_vote_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_NOFS);
	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)
{
	int ret;
	journal_t *journal = osb->journal->j_journal;

	mlog_entry("(max_buffs = %d)\n", max_buffs);

	if (!osb || !osb->journal->j_journal)
		BUG();

	if (ocfs2_is_hard_readonly(osb)) {
		ret = -EROFS;
		goto done_free;
	}

	BUG_ON(osb->journal->j_state == OCFS2_JOURNAL_FREE);
	BUG_ON(max_buffs <= 0);

	/* JBD might support this, but our journalling code doesn't yet. */
	if (journal_current_handle()) {
		mlog(ML_ERROR, "Recursive transaction attempted!\n");
		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)) {
		up_read(&osb->journal->j_trans_barrier);

		ret = PTR_ERR(handle->k_handle);
		handle->k_handle = NULL;
		mlog_errno(ret);

		if (is_journal_aborted(journal)) {
			ocfs2_abort(osb->sb, "Detected aborted journal");
			ret = -EROFS;
		}
		goto done_free;
	}

	if (!ocfs2_mount_local(osb))
		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)
{
	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();

	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);

	up_read(&journal->j_trans_barrier);

	kfree(handle);
	mlog_exit_void();
}

/*
 * 'nblocks' is what you want to add to the current
 * transaction. extend_trans will either extend the current handle by
 * nblocks, or commit it and start a new one with nblocks credits.
 *
 * 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.
 *
 * WARNING2: Note that we do *not* drop j_trans_barrier here. This is
 * 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 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);
	if (status < 0) {
		mlog_errno(status);
		goto bail;
	}

	if (status > 0) {
		mlog(0, "journal_extend failed, trying journal_restart\n");
		status = journal_restart(handle->k_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:

	mlog_exit(status);
	return status;
}

int ocfs2_journal_access(struct ocfs2_journal_handle *handle,
			 struct inode *inode,
			 struct buffer_head *bh,
			 int type)
{
	int status;

	BUG_ON(!inode);
	BUG_ON(!handle);
	BUG_ON(!bh);
	BUG_ON(!(handle->flags & OCFS2_HANDLE_STARTED));

	/* we can safely remove this assertion after testing. */
	if (!buffer_uptodate(bh)) {
		mlog(ML_ERROR, "giving me a buffer that's not uptodate!\n");
		mlog(ML_ERROR, "b_blocknr=%llu\n",
		     (unsigned long long)bh->b_blocknr);
		BUG();
	}

	/* Set the current transaction information on the inode so
	 * that the locking code knows whether it can drop it's locks
	 * on this inode or not. We're protected from the commit
	 * thread updating the current transaction id until
	 * ocfs2_commit_trans() because ocfs2_start_trans() took
	 * j_trans_barrier for us. */
	ocfs2_set_inode_lock_trans(OCFS2_SB(inode->i_sb)->journal, inode);

	down(&OCFS2_I(inode)->ip_io_sem);
	switch (type) {
	case OCFS2_JOURNAL_ACCESS_CREATE:
	case OCFS2_JOURNAL_ACCESS_WRITE:
		status = kapi_journal_get_write_access(handle->k_handle, bh);
		break;

	case OCFS2_JOURNAL_ACCESS_UNDO:
		status = kapi_journal_get_undo_access(handle->k_handle, bh);
		break;

	default:
		status = -EINVAL;
		mlog(ML_ERROR, "Uknown access type!\n");
	}
	up(&OCFS2_I(inode)->ip_io_sem);

	if (status < 0)
		mlog(ML_ERROR, "Error %d getting %d access to buffer!\n",
		     status, type);

	mlog_exit(status);
	return status;
}

int ocfs2_journal_dirty(struct ocfs2_journal_handle *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);
	if (status < 0)
		mlog(ML_ERROR, "Could not dirty metadata buffer. "
		     "(bh->b_blocknr=%llu)\n",
		     (unsigned long long)bh->b_blocknr);

	mlog_exit(status);
	return status;
}

/* 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)

void ocfs2_set_journal_params(struct ocfs2_super *osb)
{
	journal_t *journal = osb->journal->j_journal;

	spin_lock(&journal->j_state_lock);
	journal->j_commit_interval = OCFS2_DEFAULT_COMMIT_INTERVAL;
	if (osb->s_mount_opt & OCFS2_MOUNT_BARRIER)
		journal->j_flags |= JFS_BARRIER;
	else
		journal->j_flags &= ~JFS_BARRIER;
	spin_unlock(&journal->j_state_lock);
}

int ocfs2_journal_init(struct ocfs2_journal *journal, int *dirty)
{
	int status = -1;
	struct inode *inode = NULL; /* the journal inode */
	journal_t *j_journal = NULL;
	struct ocfs2_dinode *di = NULL;
	struct buffer_head *bh = NULL;
	struct ocfs2_super *osb;
	int meta_lock = 0;

	mlog_entry_void();

	BUG_ON(!journal);

	osb = journal->j_osb;

	/* already have the inode for our journal */
	inode = ocfs2_get_system_file_inode(osb, JOURNAL_SYSTEM_INODE,
					    osb->slot_num);
	if (inode == NULL) {
		status = -EACCES;
		mlog_errno(status);
		goto done;
	}
	if (is_bad_inode(inode)) {
		mlog(ML_ERROR, "access error (bad inode)\n");
		iput(inode);
		inode = NULL;
		status = -EACCES;
		goto done;
	}

	SET_INODE_JOURNAL(inode);
	OCFS2_I(inode)->ip_open_count++;

	/* 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_meta_lock_full(inode, NULL, &bh, 1,
				      OCFS2_META_LOCK_RECOVERY, NULL, 0);
	if (status < 0) {
		if (status != -ERESTARTSYS)
			mlog(ML_ERROR, "Could not get lock on journal!\n");
		goto done;
	}

	meta_lock = 1;
	di = (struct ocfs2_dinode *)bh->b_data;

	if (inode->i_size <  OCFS2_MIN_JOURNAL_SIZE) {
		mlog(ML_ERROR, "Journal file size (%lld) is too small!\n",
		     inode->i_size);
		status = -EINVAL;
		goto done;
	}

	mlog(0, "inode->i_size = %lld\n", inode->i_size);
	mlog(0, "inode->i_blocks = %lu\n", (unsigned 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);
	if (j_journal == NULL) {
		mlog(ML_ERROR, "Linux journal layer error\n");
		status = -EINVAL;
		goto done;
	}