dlmglue.c

来自「ocfs1.2.7 源码」· C语言 代码 · 共 2,504 行 · 第 1/5 页

				ocfs2_log_dlm_error("dlmlock", status,
						    lockres);
				ret = -EINVAL;
			}
			ocfs2_recover_from_dlm_error(lockres, 1);
			goto out;
		}

		mlog(0, "lock %s, successfull return from dlmlock\n",
		     lockres->l_name);

		/* At this point we've gone inside the dlm and need to
		 * complete our work regardless. */
		catch_signals = 0;

		/* wait for busy to clear and carry on */
		goto again;
	}

	/* Ok, if we get here then we're good to go. */
	ocfs2_inc_holders(lockres, level);

	ret = 0;
unlock:
	spin_unlock(&lockres->l_lock);
out:
	/* Non-async callers will always wait here for dlm operations
	 * to complete. We must be careful to re-initialize the
	 * completion before looping back. */
	if (ret == -EIOCBRETRY && sync) {
		ret = ocfs2_wait_for_status_completion(&sc);
		if (ret == 0)
			goto again;
		mlog_errno(ret);
	}

	/* free the async fcb on error or if it is not queued. */
	if ((ret && ret != -EIOCBRETRY && !sync) || (!ret && !sync)) {
		mlog_bug_on_msg(!list_empty(&fcb->fc_lockres_item),
				"Lockres %s, freeing flag callback in use\n",
				lockres->l_name);
		kfree(fcb);
	}

	mlog_exit(ret);
	return ret;
}

static void ocfs2_cluster_unlock(struct ocfs2_super *osb,
				 struct ocfs2_lock_res *lockres,
				 int level)
{
	mlog_entry_void();
	spin_lock(&lockres->l_lock);
	ocfs2_dec_holders(lockres, level);
	ocfs2_vote_on_unlock(osb, lockres);
	spin_unlock(&lockres->l_lock);
	mlog_exit_void();
}

/* Grants us an EX lock on the data and metadata resources, skipping
 * the normal cluster directory lookup. Use this ONLY on newly created
 * inodes which other nodes can't possibly see, and which haven't been
 * hashed in the inode hash yet. This can give us a good performance
 * increase as it'll skip the network broadcast normally associated
 * with creating a new lock resource. */
int ocfs2_create_new_inode_locks(struct inode *inode)
{
	int status;
	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
	struct ocfs2_lock_res *lockres;

	BUG_ON(!inode);
	BUG_ON(!ocfs2_inode_is_new(inode));

	mlog_entry_void();

	mlog(0, "Inode %"MLFu64"\n", OCFS2_I(inode)->ip_blkno);

	/* NOTE: That we don't increment any of the holder counts, nor
	 * do we add anything to a journal handle. Since this is
	 * supposed to be a new inode which the cluster doesn't know
	 * about yet, there is no need to.  As far as the LVB handling
	 * is concerned, this is basically like acquiring an EX lock
	 * on a resource which has an invalid one -- we'll set it
	 * valid when we release the EX. */

	lockres = &OCFS2_I(inode)->ip_meta_lockres;

	spin_lock(&lockres->l_lock);
	BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);
	lockres_or_flags(lockres, OCFS2_LOCK_LOCAL);
	spin_unlock(&lockres->l_lock);

	status = ocfs2_lock_create(osb, lockres, LKM_EXMODE, LKM_LOCAL);
	if (status < 0) {
		mlog_errno(status);
		goto bail;
	}

	lockres = &OCFS2_I(inode)->ip_data_lockres;

	spin_lock(&lockres->l_lock);
	BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);
	lockres_or_flags(lockres, OCFS2_LOCK_LOCAL);
	spin_unlock(&lockres->l_lock);

	status = ocfs2_lock_create(osb, lockres, LKM_EXMODE, LKM_LOCAL);
	if (status < 0) {
		mlog_errno(status);
		goto bail;
	}

	status = 0;
bail:
	mlog_exit(status);
	return status;
}

int ocfs2_data_lock(struct inode *inode,
		    int write)
{
	int status = 0, level;
	struct ocfs2_lock_res *lockres;
	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);

	BUG_ON(!inode);

	mlog_entry_void();

	mlog(0, "inode %"MLFu64" take %s DATA lock\n",
	     OCFS2_I(inode)->ip_blkno,
	     write ? "EXMODE" : "PRMODE");

	/* We'll allow faking a readonly data lock for
	 * rodevices. */
	if (ocfs2_is_hard_readonly(OCFS2_SB(inode->i_sb))) {
		if (write) {
			status = -EROFS;
			mlog_errno(status);
		}
		goto out;
	}

	if (ocfs2_mount_local(osb))
		goto out;

	lockres = &OCFS2_I(inode)->ip_data_lockres;

	level = write ? LKM_EXMODE : LKM_PRMODE;

	status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres, level, 0,
				    NULL, 0);
	if (status < 0)
		mlog_errno(status);

out:
	mlog_exit(status);
	return status;
}

static void ocfs2_vote_on_unlock(struct ocfs2_super *osb,
				 struct ocfs2_lock_res *lockres)
{
	int kick = 0;

	mlog_entry_void();

	/* If we know that another node is waiting on our lock, kick
	 * the vote thread * pre-emptively when we reach a release
	 * condition. */
	if (lockres->l_flags & OCFS2_LOCK_BLOCKED) {
		switch(lockres->l_blocking) {
		case LKM_EXMODE:
			if (!lockres->l_ex_holders && !lockres->l_ro_holders)
				kick = 1;
			break;
		case LKM_PRMODE:
			if (!lockres->l_ex_holders)
				kick = 1;
			break;
		default:
			BUG();
		}
	}

	if (kick)
		ocfs2_kick_vote_thread(osb);

	mlog_exit_void();
}

void ocfs2_data_unlock(struct inode *inode,
		       int write)
{
	int level = write ? LKM_EXMODE : LKM_PRMODE;
	struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_data_lockres;
	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);

	mlog_entry_void();

	mlog(0, "inode %"MLFu64" drop %s DATA lock\n",
	     OCFS2_I(inode)->ip_blkno,
	     write ? "EXMODE" : "PRMODE");

	if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb))
		ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres, level);

	mlog_exit_void();
}

#define OCFS2_SEC_BITS   34
#define OCFS2_SEC_SHIFT  (64 - 34)
#define OCFS2_NSEC_MASK  ((1ULL << OCFS2_SEC_SHIFT) - 1)

/* LVB only has room for 64 bits of time here so we pack it for
 * now. */
static u64 ocfs2_pack_timespec(struct timespec *spec)
{
	u64 res;
	u64 sec = spec->tv_sec;
	u32 nsec = spec->tv_nsec;

	res = (sec << OCFS2_SEC_SHIFT) | (nsec & OCFS2_NSEC_MASK);

	return res;
}

/* Call this with the lockres locked. I am reasonably sure we don't
 * need ip_lock in this function as anyone who would be changing those
 * values is supposed to be blocked in ocfs2_meta_lock right now. */
static void __ocfs2_stuff_meta_lvb(struct inode *inode)
{
	struct ocfs2_inode_info *oi = OCFS2_I(inode);
	struct ocfs2_lock_res *lockres = &oi->ip_meta_lockres;
	struct ocfs2_meta_lvb *lvb;

	mlog_entry_void();

	lvb = (struct ocfs2_meta_lvb *) lockres->l_lksb.lvb;

	/* Setting this to zero will ensure that old versions of the
	 * LVB code don't trust our information. */
	lvb->lvb_old_seq   = cpu_to_be32(0);
	lvb->lvb_version   = cpu_to_be32(OCFS2_LVB_VERSION);

	lvb->lvb_isize     = cpu_to_be64(i_size_read(inode));
	lvb->lvb_iclusters = cpu_to_be32(oi->ip_clusters);
	lvb->lvb_iuid      = cpu_to_be32(inode->i_uid);
	lvb->lvb_igid      = cpu_to_be32(inode->i_gid);
	lvb->lvb_imode     = cpu_to_be16(inode->i_mode);
	lvb->lvb_inlink    = cpu_to_be16(inode->i_nlink);
	lvb->lvb_iatime_packed  =
		cpu_to_be64(ocfs2_pack_timespec(&inode->i_atime));
	lvb->lvb_ictime_packed =
		cpu_to_be64(ocfs2_pack_timespec(&inode->i_ctime));
	lvb->lvb_imtime_packed =
		cpu_to_be64(ocfs2_pack_timespec(&inode->i_mtime));

	mlog_meta_lvb(0, lockres);

	mlog_exit_void();
}

static void ocfs2_unpack_timespec(struct timespec *spec,
				  u64 packed_time)
{
	spec->tv_sec = packed_time >> OCFS2_SEC_SHIFT;
	spec->tv_nsec = packed_time & OCFS2_NSEC_MASK;
}

static void ocfs2_refresh_inode_from_lvb(struct inode *inode)
{
	struct ocfs2_inode_info *oi = OCFS2_I(inode);
	struct ocfs2_lock_res *lockres = &oi->ip_meta_lockres;
	struct ocfs2_meta_lvb *lvb;

	mlog_entry_void();

	mlog_meta_lvb(0, lockres);

	lvb = (struct ocfs2_meta_lvb *) lockres->l_lksb.lvb;

	/* We're safe here without the lockres lock... */
	spin_lock(&oi->ip_lock);
	oi->ip_clusters = be32_to_cpu(lvb->lvb_iclusters);
	i_size_write(inode, be64_to_cpu(lvb->lvb_isize));

	if (S_ISREG(inode->i_mode))
		oi->ip_mmu_private = i_size_read(inode);

	/* fast-symlinks are a special case */
	if (S_ISLNK(inode->i_mode) && !oi->ip_clusters)
		inode->i_blocks = 0;
	else
		inode->i_blocks =
			ocfs2_align_bytes_to_sectors(i_size_read(inode));

	inode->i_uid     = be32_to_cpu(lvb->lvb_iuid);
	inode->i_gid     = be32_to_cpu(lvb->lvb_igid);
	inode->i_mode    = be16_to_cpu(lvb->lvb_imode);
	inode->i_nlink   = be16_to_cpu(lvb->lvb_inlink);
	ocfs2_unpack_timespec(&inode->i_atime,
			      be64_to_cpu(lvb->lvb_iatime_packed));
	ocfs2_unpack_timespec(&inode->i_mtime,
			      be64_to_cpu(lvb->lvb_imtime_packed));
	ocfs2_unpack_timespec(&inode->i_ctime,
			      be64_to_cpu(lvb->lvb_ictime_packed));
	spin_unlock(&oi->ip_lock);

	mlog_exit_void();
}

static inline int ocfs2_meta_lvb_is_trustable(struct ocfs2_lock_res *lockres)
{
	struct ocfs2_meta_lvb *lvb = (struct ocfs2_meta_lvb *) lockres->l_lksb.lvb;

	/* Old OCFS2 versions stored a "sequence" in the lvb to
	 * determine whether the information could be trusted. We
	 * don't want to use an lvb populated from a node running the
	 * old code, so check that sequence is not set. */
	if (!lvb->lvb_old_seq &&
	    be32_to_cpu(lvb->lvb_version) == OCFS2_LVB_VERSION)
		return 1;
	return 0;
}

/* Determine whether a lock resource needs to be refreshed, and
 * arbitrate who gets to refresh it.
 *
 *   0 means no refresh needed.
 *
 *   > 0 means you need to refresh this and you MUST call
 *   ocfs2_complete_lock_res_refresh afterwards. */
static int ocfs2_should_refresh_lock_res(struct ocfs2_lock_res *lockres)
{

	int status = 0;
	mlog_entry_void();

refresh_check:
	spin_lock(&lockres->l_lock);
	if (!(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH)) {
		spin_unlock(&lockres->l_lock);
		goto bail;
	}

	if (lockres->l_flags & OCFS2_LOCK_REFRESHING) {
		spin_unlock(&lockres->l_lock);

		ocfs2_wait_on_refreshing_lock(lockres);
		goto refresh_check;
	}

	/* Ok, I'll be the one to refresh this lock. */
	lockres_or_flags(lockres, OCFS2_LOCK_REFRESHING);
	spin_unlock(&lockres->l_lock);

	status = 1;
bail:
	mlog_exit(status);
	return status;
}

/* If status is non zero, I'll mark it as not being in refresh
 * anymroe, but i won't clear the needs refresh flag. */
static inline void ocfs2_complete_lock_res_refresh(struct ocfs2_lock_res *lockres,
						   int status)
{
	mlog_entry_void();

	spin_lock(&lockres->l_lock);
	lockres_clear_flags(lockres, OCFS2_LOCK_REFRESHING);
	if (!status)
		lockres_clear_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
	spin_unlock(&lockres->l_lock);

	wake_up(&lockres->l_event);

	mlog_exit_void();
}

/* may or may not return a bh if it went to disk. */
static int ocfs2_meta_lock_update(struct inode *inode,
				  struct buffer_head **bh)
{
	int status = 0;
	struct ocfs2_inode_info *oi = OCFS2_I(inode);
	struct ocfs2_lock_res *lockres;
	struct ocfs2_dinode *fe;
	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);

	mlog_entry_void();

	if (ocfs2_mount_local(osb))
		goto bail;

	spin_lock(&oi->ip_lock);
	if (oi->ip_flags & OCFS2_INODE_DELETED) {
		mlog(0, "Orphaned inode %"MLFu64" was deleted while we "
		     "were waiting on a lock. ip_flags = 0x%x\n",
		     oi->ip_blkno, oi->ip_flags);
		spin_unlock(&oi->ip_lock);
		status = -ENOENT;
		goto bail;
	}
	spin_unlock(&oi->ip_lock);

	lockres = &oi->ip_meta_lockres;

	if (!ocfs2_should_refresh_lock_res(lockres))
		goto bail;

	/* This will discard any caching information we might have had
	 * for the inode metadata. */
	ocfs2_metadata_cache_purge(inode);

	/* will do nothing for inode types that don't use the extent
	 * map (directories, bitmap files, etc) */
	ocfs2_extent_map_trunc(inode, 0);

	if (ocfs2_meta_lvb_is_trustable(lockres)) {
		mlog(0, "Trusting LVB on inode %"MLFu64"\n",
		     oi->ip_blkno);
		ocfs2_refresh_inode_from_lvb(inode);
	} else {
		/* Boo, we have to go to disk. */
		/* read bh, cast, ocfs2_refresh_inode */
		status = ocfs2_read_block(OCFS2_SB(inode->i_sb), oi->ip_blkno,
					  bh, OCFS2_BH_CACHED, inode);
		if (status < 0) {
			mlog_errno(status);
			goto bail_refresh;
		}
		fe = (struct ocfs2_dinode *) (*bh)->b_data;

		/* This is a good chance to make sure we're not
		 * locking an invalid object.
		 *
		 * We bug on a stale inode here because we checked
		 * above whether it was wiped from disk. The wiping
		 * node provides a guarantee that we receive that
		 * message and can mark the inode before dropping any
		 * locks associated with it. */
		if (!OCFS2_IS_VALID_DINODE(fe)) {
			OCFS2_RO_ON_INVALID_DINODE(inode->i_sb, fe);
			status = -EIO;
			goto bail_refresh;
		}
		mlog_bug_on_msg(inode->i_generation !=
				le32_to_cpu(fe->i_generation),
				"Invalid dinode %"MLFu64" disk generation: %u "
				"inode->i_generation: %u\n",
				oi->ip_blkno, le32_to_cpu(fe->i_generation),
				inode->i_generation);
		mlog_bug_on_msg(le64_to_cpu(fe->i_dtime) ||
				!(fe->i_flags & cpu_to_le32(OCFS2_VALID_FL)),
				"Stale dinode %"MLFu64" dtime: %"MLFu64" "
				"flags: 0x%x\n", oi->ip_blkno,
				le64_to_cpu(fe->i_dtime),
				le32_to_cpu(fe->i_flags));

		ocfs2_refresh_inode(inode, fe);
	}

#ifdef OCFS2_DELETE_INODE_WORKAROUND
	/* We might as well check this here - since the inode is now
	 * locked, an up to date view will indicate whether this was
	 * never actually orphaned -- i_nlink should be zero for an
	 * orphaned inode. */
	spin_lock(&oi->ip_lock);
	if (inode->i_nlink &&
	    oi->ip_flags & OCFS2_INODE_MAYBE_ORPHANED) {
		mlog(0, "Inode %"MLFu64": clearing maybe_orphaned flag\n",
		     oi->ip_blkno);
		oi->ip_flags &= ~OCFS2_INODE_MAYBE_ORPHANED;
	}
	spin_unlock(&oi->ip_lock);
#endif

	status = 0;
bail_refresh:
	ocfs2_complete_lock_res_refresh(lockres, status);
bail:
	mlog_exit(status);
	return status;
}

static int ocfs2_assign_bh(struct inode *inode,
			   struct buffer_head **ret_bh,
			   struct buffer_head *passed_bh)
{
	int status;

	if (passed_bh) {
		/* Ok, the update went to disk for us, use the
		 * returned bh. */
		*ret_bh = passed_bh;
		get_bh(*ret_bh);

		return 0;