dlmconvert.c

linux 内核源代码

		mlog(0, "bailing out early since res is RECOVERING "
		     "on secondary queue\n");
		/* __dlm_print_one_lock_resource(res); */
		status = DLM_RECOVERING;
		goto bail;
	}
	/* will exit this call with spinlock held */
	__dlm_wait_on_lockres(res);

	if (lock->ml.convert_type != LKM_IVMODE) {
		__dlm_print_one_lock_resource(res);
		mlog(ML_ERROR, "converting a remote lock that is already "
		     "converting! (cookie=%u:%llu, conv=%d)\n",
		     dlm_get_lock_cookie_node(be64_to_cpu(lock->ml.cookie)),
		     dlm_get_lock_cookie_seq(be64_to_cpu(lock->ml.cookie)),
		     lock->ml.convert_type);
		status = DLM_DENIED;
		goto bail;
	}
	res->state |= DLM_LOCK_RES_IN_PROGRESS;
	/* move lock to local convert queue */
	/* do not alter lock refcount.  switching lists. */
	list_move_tail(&lock->list, &res->converting);
	lock->convert_pending = 1;
	lock->ml.convert_type = type;

	if (flags & LKM_VALBLK) {
		if (lock->ml.type == LKM_EXMODE) {
			flags |= LKM_PUT_LVB;
			lock->lksb->flags |= DLM_LKSB_PUT_LVB;
		} else {
			if (lock->ml.convert_type == LKM_NLMODE)
				flags &= ~LKM_VALBLK;
			else {
				flags |= LKM_GET_LVB;
				lock->lksb->flags |= DLM_LKSB_GET_LVB;
			}
		}
	}
	spin_unlock(&res->spinlock);

	/* no locks held here.
	 * need to wait for a reply as to whether it got queued or not. */
	status = dlm_send_remote_convert_request(dlm, res, lock, flags, type);

	spin_lock(&res->spinlock);
	res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
	lock->convert_pending = 0;
	/* if it failed, move it back to granted queue */
	if (status != DLM_NORMAL) {
		if (status != DLM_NOTQUEUED)
			dlm_error(status);
		dlm_revert_pending_convert(res, lock);
	}
bail:
	spin_unlock(&res->spinlock);

	/* TODO: should this be a wake_one? */
	/* wake up any IN_PROGRESS waiters */
	wake_up(&res->wq);

	return status;
}

/* sends DLM_CONVERT_LOCK_MSG to master site
 * locking:
 *   caller needs:  none
 *   taken:         none
 *   held on exit:  none
 * returns: DLM_NOLOCKMGR, status from remote node
 */
static enum dlm_status dlm_send_remote_convert_request(struct dlm_ctxt *dlm,
					   struct dlm_lock_resource *res,
					   struct dlm_lock *lock, int flags, int type)
{
	struct dlm_convert_lock convert;
	int tmpret;
	enum dlm_status ret;
	int status = 0;
	struct kvec vec[2];
	size_t veclen = 1;

	mlog_entry("%.*s\n", res->lockname.len, res->lockname.name);

	memset(&convert, 0, sizeof(struct dlm_convert_lock));
	convert.node_idx = dlm->node_num;
	convert.requested_type = type;
	convert.cookie = lock->ml.cookie;
	convert.namelen = res->lockname.len;
	convert.flags = cpu_to_be32(flags);
	memcpy(convert.name, res->lockname.name, convert.namelen);

	vec[0].iov_len = sizeof(struct dlm_convert_lock);
	vec[0].iov_base = &convert;

	if (flags & LKM_PUT_LVB) {
		/* extra data to send if we are updating lvb */
		vec[1].iov_len = DLM_LVB_LEN;
		vec[1].iov_base = lock->lksb->lvb;
		veclen++;
	}

	tmpret = o2net_send_message_vec(DLM_CONVERT_LOCK_MSG, dlm->key,
					vec, veclen, res->owner, &status);
	if (tmpret >= 0) {
		// successfully sent and received
		ret = status;  // this is already a dlm_status
		if (ret == DLM_RECOVERING) {
			mlog(0, "node %u returned DLM_RECOVERING from convert "
			     "message!\n", res->owner);
		} else if (ret == DLM_MIGRATING) {
			mlog(0, "node %u returned DLM_MIGRATING from convert "
			     "message!\n", res->owner);
		} else if (ret == DLM_FORWARD) {
			mlog(0, "node %u returned DLM_FORWARD from convert "
			     "message!\n", res->owner);
		} else if (ret != DLM_NORMAL && ret != DLM_NOTQUEUED)
			dlm_error(ret);
	} else {
		mlog_errno(tmpret);
		if (dlm_is_host_down(tmpret)) {
			/* instead of logging the same network error over
			 * and over, sleep here and wait for the heartbeat
			 * to notice the node is dead.  times out after 5s. */
			dlm_wait_for_node_death(dlm, res->owner, 
						DLM_NODE_DEATH_WAIT_MAX);
			ret = DLM_RECOVERING;
			mlog(0, "node %u died so returning DLM_RECOVERING "
			     "from convert message!\n", res->owner);
		} else {
			ret = dlm_err_to_dlm_status(tmpret);
		}
	}

	return ret;
}

/* handler for DLM_CONVERT_LOCK_MSG on master site
 * locking:
 *   caller needs:  none
 *   taken:         takes and drop res->spinlock
 *   held on exit:  none
 * returns: DLM_NORMAL, DLM_IVLOCKID, DLM_BADARGS,
 *          status from __dlmconvert_master
 */
int dlm_convert_lock_handler(struct o2net_msg *msg, u32 len, void *data,
			     void **ret_data)
{
	struct dlm_ctxt *dlm = data;
	struct dlm_convert_lock *cnv = (struct dlm_convert_lock *)msg->buf;
	struct dlm_lock_resource *res = NULL;
	struct list_head *iter;
	struct dlm_lock *lock = NULL;
	struct dlm_lockstatus *lksb;
	enum dlm_status status = DLM_NORMAL;
	u32 flags;
	int call_ast = 0, kick_thread = 0, ast_reserved = 0, wake = 0;

	if (!dlm_grab(dlm)) {
		dlm_error(DLM_REJECTED);
		return DLM_REJECTED;
	}

	mlog_bug_on_msg(!dlm_domain_fully_joined(dlm),
			"Domain %s not fully joined!\n", dlm->name);

	if (cnv->namelen > DLM_LOCKID_NAME_MAX) {
		status = DLM_IVBUFLEN;
		dlm_error(status);
		goto leave;
	}

	flags = be32_to_cpu(cnv->flags);

	if ((flags & (LKM_PUT_LVB|LKM_GET_LVB)) ==
	     (LKM_PUT_LVB|LKM_GET_LVB)) {
		mlog(ML_ERROR, "both PUT and GET lvb specified\n");
		status = DLM_BADARGS;
		goto leave;
	}

	mlog(0, "lvb: %s\n", flags & LKM_PUT_LVB ? "put lvb" :
	     (flags & LKM_GET_LVB ? "get lvb" : "none"));

	status = DLM_IVLOCKID;
	res = dlm_lookup_lockres(dlm, cnv->name, cnv->namelen);
	if (!res) {
		dlm_error(status);
		goto leave;
	}

	spin_lock(&res->spinlock);
	status = __dlm_lockres_state_to_status(res);
	if (status != DLM_NORMAL) {
		spin_unlock(&res->spinlock);
		dlm_error(status);
		goto leave;
	}
	list_for_each(iter, &res->granted) {
		lock = list_entry(iter, struct dlm_lock, list);
		if (lock->ml.cookie == cnv->cookie &&
		    lock->ml.node == cnv->node_idx) {
			dlm_lock_get(lock);
			break;
		}
		lock = NULL;
	}
	spin_unlock(&res->spinlock);
	if (!lock) {
		status = DLM_IVLOCKID;
		mlog(ML_ERROR, "did not find lock to convert on grant queue! "
			       "cookie=%u:%llu\n",
		     dlm_get_lock_cookie_node(be64_to_cpu(cnv->cookie)),
		     dlm_get_lock_cookie_seq(be64_to_cpu(cnv->cookie)));
		__dlm_print_one_lock_resource(res);
		goto leave;
	}

	/* found the lock */
	lksb = lock->lksb;

	/* see if caller needed to get/put lvb */
	if (flags & LKM_PUT_LVB) {
		BUG_ON(lksb->flags & (DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB));
		lksb->flags |= DLM_LKSB_PUT_LVB;
		memcpy(&lksb->lvb[0], &cnv->lvb[0], DLM_LVB_LEN);
	} else if (flags & LKM_GET_LVB) {
		BUG_ON(lksb->flags & (DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB));
		lksb->flags |= DLM_LKSB_GET_LVB;
	}

	spin_lock(&res->spinlock);
	status = __dlm_lockres_state_to_status(res);
	if (status == DLM_NORMAL) {
		__dlm_lockres_reserve_ast(res);
		ast_reserved = 1;
		res->state |= DLM_LOCK_RES_IN_PROGRESS;
		status = __dlmconvert_master(dlm, res, lock, flags,
					     cnv->requested_type,
					     &call_ast, &kick_thread);
		res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
		wake = 1;
	}
	spin_unlock(&res->spinlock);
	if (wake)
		wake_up(&res->wq);

	if (status != DLM_NORMAL) {
		if (status != DLM_NOTQUEUED)
			dlm_error(status);
		lksb->flags &= ~(DLM_LKSB_GET_LVB|DLM_LKSB_PUT_LVB);
	}

leave:
	if (lock)
		dlm_lock_put(lock);

	/* either queue the ast or release it, if reserved */
	if (call_ast)
		dlm_queue_ast(dlm, lock);
	else if (ast_reserved)
		dlm_lockres_release_ast(dlm, res);

	if (kick_thread)
		dlm_kick_thread(dlm, res);

	if (res)
		dlm_lockres_put(res);

	dlm_put(dlm);

	return status;
}