journal.c

linux 内核源代码

/* -*- 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 "dir.h"
#include "dlmglue.h"
#include "extent_map.h"
#include "heartbeat.h"
#include "inode.h"
#include "journal.h"
#include "localalloc.h"
#include "slot_map.h"
#include "super.h"
#include "vote.h"
#include "sysfile.h"

#include "buffer_head_io.h"

DEFINE_SPINLOCK(trans_inc_lock);

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

/* 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. */
handle_t *ocfs2_start_trans(struct ocfs2_super *osb, int max_buffs)
{
	journal_t *journal = osb->journal->j_journal;
	handle_t *handle;

	BUG_ON(!osb || !osb->journal->j_journal);

	if (ocfs2_is_hard_readonly(osb))
		return ERR_PTR(-EROFS);

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

	down_read(&osb->journal->j_trans_barrier);

	handle = journal_start(journal, max_buffs);
	if (IS_ERR(handle)) {
		up_read(&osb->journal->j_trans_barrier);

		mlog_errno(PTR_ERR(handle));

		if (is_journal_aborted(journal)) {
			ocfs2_abort(osb->sb, "Detected aborted journal");
			handle = ERR_PTR(-EROFS);
		}
	} else {
		if (!ocfs2_mount_local(osb))
			atomic_inc(&(osb->journal->j_num_trans));
	}

	return handle;
}

int ocfs2_commit_trans(struct ocfs2_super *osb,
		       handle_t *handle)
{
	int ret;
	struct ocfs2_journal *journal = osb->journal;

	BUG_ON(!handle);

	ret = journal_stop(handle);
	if (ret < 0)
		mlog_errno(ret);

	up_read(&journal->j_trans_barrier);

	return ret;
}

/*
 * '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.
 *
 * This might call 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.
 *
 * 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(handle_t *handle, int nblocks)
{
	int status;

	BUG_ON(!handle);
	BUG_ON(!nblocks);

	mlog_entry_void();

	mlog(0, "Trying to extend transaction by %d blocks\n", nblocks);

#ifdef OCFS2_DEBUG_FS
	status = 1;
#else
	status = 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, nblocks);
		if (status < 0) {
			mlog_errno(status);
			goto bail;
		}
	}

	status = 0;
bail:

	mlog_exit(status);
	return status;
}

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

	BUG_ON(!inode);
	BUG_ON(!handle);
	BUG_ON(!bh);

	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" :
		   "OCFS2_JOURNAL_ACCESS_WRITE",
		   bh->b_size);

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

	mutex_lock(&OCFS2_I(inode)->ip_io_mutex);
	switch (type) {
	case OCFS2_JOURNAL_ACCESS_CREATE:
	case OCFS2_JOURNAL_ACCESS_WRITE:
		status = journal_get_write_access(handle, bh);
		break;

	case OCFS2_JOURNAL_ACCESS_UNDO:
		status = journal_get_undo_access(handle, bh);
		break;

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

	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(handle_t *handle,
			struct buffer_head *bh)
{
	int status;

	mlog_entry("(bh->b_blocknr=%llu)\n",
		   (unsigned long long)bh->b_blocknr);

	status = journal_dirty_metadata(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;
}

int ocfs2_journal_dirty_data(handle_t *handle,
			     struct buffer_head *bh)
{
	int err = journal_dirty_data(handle, bh);
	if (err)
		mlog_errno(err);
	/* TODO: When we can handle it, abort the handle and go RO on
	 * error here. */

	return err;
}

#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, &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;
	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 = %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);
	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, "j_journal->j_maxlen = %u\n", j_journal->j_maxlen);

	*dirty = (le32_to_cpu(di->id1.journal1.ij_flags) &
		  OCFS2_JOURNAL_DIRTY_FL);

	journal->j_journal = j_journal;
	journal->j_inode = inode;
	journal->j_bh = bh;

	ocfs2_set_journal_params(osb);

	journal->j_state = OCFS2_JOURNAL_LOADED;

	status = 0;
done:
	if (status < 0) {
		if (meta_lock)
			ocfs2_meta_unlock(inode, 1);
		if (bh != NULL)
			brelse(bh);
		if (inode) {
			OCFS2_I(inode)->ip_open_count--;
			iput(inode);
		}
	}

	mlog_exit(status);
	return status;
}

static int ocfs2_journal_toggle_dirty(struct ocfs2_super *osb,
				      int dirty)
{
	int status;
	unsigned int flags;
	struct ocfs2_journal *journal = osb->journal;
	struct buffer_head *bh = journal->j_bh;
	struct ocfs2_dinode *fe;

	mlog_entry_void();

	fe = (struct ocfs2_dinode *)bh->b_data;
	if (!OCFS2_IS_VALID_DINODE(fe)) {
		/* 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 %llu  has invalid "
		     "signature: %.*s",
		     (unsigned long long)le64_to_cpu(fe->i_blkno), 7,
		     fe->i_signature);
		status = -EIO;
		goto out;
	}

	flags = le32_to_cpu(fe->id1.journal1.ij_flags);
	if (dirty)
		flags |= OCFS2_JOURNAL_DIRTY_FL;
	else
		flags &= ~OCFS2_JOURNAL_DIRTY_FL;
	fe->id1.journal1.ij_flags = cpu_to_le32(flags);

	status = ocfs2_write_block(osb, bh, journal->j_inode);
	if (status < 0)
		mlog_errno(status);

out:
	mlog_exit(status);
	return status;
}

/*
 * If the journal has been kmalloc'd it needs to be freed after this
 * call.
 */
void ocfs2_journal_shutdown(struct ocfs2_super *osb)
{
	struct ocfs2_journal *journal = NULL;
	int status = 0;
	struct inode *inode = NULL;
	int num_running_trans = 0;

	mlog_entry_void();

	BUG_ON(!osb);

	journal = osb->journal;
	if (!journal)
		goto done;

	inode = journal->j_inode;

	if (journal->j_state != OCFS2_JOURNAL_LOADED)
		goto done;

	/* need to inc inode use count as journal_destroy will iput. */
	if (!igrab(inode))
		BUG();

	num_running_trans = atomic_read(&(osb->journal->j_num_trans));
	if (num_running_trans > 0)