vote.c

ocfs1.2.7 源码

		goto respond;
	case OCFS2_VOTE_REQ_MOUNT:
		ocfs2_process_mount_request(osb, node_num);
		goto respond;
	default:
		/* avoids a gcc warning */
		break;
	}

	/* We cannot process the remaining message types before we're
	 * fully mounted. It's perfectly safe however to send a 'yes'
	 * response as we can't possibly have any of the state they're
	 * asking us to modify yet. */
	if (atomic_read(&osb->vol_state) == VOLUME_INIT)
		goto respond;

	/* If we get here, then the request is against an inode. */
	inode = ocfs2_ilookup_for_vote(osb, blkno,
				       request == OCFS2_VOTE_REQ_DELETE);

	/* Not finding the inode is perfectly valid - it means we're
	 * not interested in what the other node is about to do to it
	 * so in those cases we automatically respond with an
	 * affirmative. Cluster locking ensures that we won't race
	 * interest in the inode with this vote request. */
	if (!inode)
		goto respond;

	/* Check generation values. It's possible for us to get a
	 * request against a stale inode. If so then we proceed as if
	 * we had not found an inode in the first place. */
	if (inode->i_generation != generation) {
		mlog(0, "generation passed %u != inode generation = %u, "
		     "ip_flags = %x, ip_blkno = %"MLFu64", msg %"MLFu64", "
		     "i_count = %u, message type = %u\n",
		     generation, inode->i_generation, OCFS2_I(inode)->ip_flags,
		     OCFS2_I(inode)->ip_blkno, blkno,
		     atomic_read(&inode->i_count), request);
		iput(inode);
		inode = NULL;
		goto respond;
	}

	switch (request) {
	case OCFS2_VOTE_REQ_DELETE:
		vote_response = ocfs2_process_delete_request(inode,
					     &orphaned_slot, deleting_node);
		break;
	case OCFS2_VOTE_REQ_RENAME:
		rename = 1;
		/* fall through */
	case OCFS2_VOTE_REQ_UNLINK:
		parent_blkno = be64_to_cpu(msg->v_unlink_parent);
		namelen = be32_to_cpu(msg->v_unlink_namelen);
		/* new_nlink will be ignored in case of a rename vote */
		new_nlink = be32_to_cpu(msg->md1.v_nlink);
		ocfs2_process_dentry_request(inode, rename, new_nlink,
					     parent_blkno, namelen,
					     msg->v_unlink_dirent);
		break;
	default:
		mlog(ML_ERROR, "node %u, invalid request: %u\n",
		     node_num, request);
		vote_response = OCFS2_RESPONSE_BAD_MSG;
	}

respond:
	/* Response struture is small so we just put it on the stack
	 * and stuff it inline. */
	memset(&response, 0, sizeof(struct ocfs2_response_msg));
	response.r_hdr.h_response_id = hdr->h_response_id;
	response.r_hdr.h_blkno = hdr->h_blkno;
	response.r_hdr.h_generation = hdr->h_generation;
	response.r_hdr.h_node_num = cpu_to_be32(osb->node_num);
	response.r_response = cpu_to_be32(vote_response);
	response.r_orphaned_slot = cpu_to_be32(orphaned_slot);

	net_status = o2net_send_message(OCFS2_MESSAGE_TYPE_RESPONSE,
					osb->net_key,
					&response,
					sizeof(struct ocfs2_response_msg),
					node_num,
					NULL);
	/* We still want to error print for ENOPROTOOPT here. The
	 * sending node shouldn't have unregistered his net handler
	 * without sending an unmount vote 1st */
	if (net_status < 0
	    && net_status != -ETIMEDOUT
	    && net_status != -ENOTCONN)
		mlog(ML_ERROR, "message to node %u fails with error %d!\n",
		     node_num, net_status);

	if (inode)
		iput(inode);
}

static void ocfs2_vote_thread_do_work(struct ocfs2_super *osb)
{
	unsigned long processed;
	struct ocfs2_lock_res *lockres;
	struct ocfs2_vote_work *work;

	mlog_entry_void();

	spin_lock(&osb->vote_task_lock);
	/* grab this early so we know to try again if a state change and
	 * wake happens part-way through our work  */
	osb->vote_work_sequence = osb->vote_wake_sequence;

	processed = osb->blocked_lock_count;
	while (processed) {
		BUG_ON(list_empty(&osb->blocked_lock_list));

		lockres = list_entry(osb->blocked_lock_list.next,
				     struct ocfs2_lock_res, l_blocked_list);
		list_del_init(&lockres->l_blocked_list);
		osb->blocked_lock_count--;
		spin_unlock(&osb->vote_task_lock);

		BUG_ON(!processed);
		processed--;

		ocfs2_process_blocked_lock(osb, lockres);

		spin_lock(&osb->vote_task_lock);
	}

	while (osb->vote_count) {
		BUG_ON(list_empty(&osb->vote_list));
		work = list_entry(osb->vote_list.next,
				  struct ocfs2_vote_work, w_list);
		list_del(&work->w_list);
		osb->vote_count--;
		spin_unlock(&osb->vote_task_lock);

		ocfs2_process_vote(osb, &work->w_msg);
		kfree(work);

		spin_lock(&osb->vote_task_lock);
	}
	spin_unlock(&osb->vote_task_lock);

	mlog_exit_void();
}

static int ocfs2_vote_thread_lists_empty(struct ocfs2_super *osb)
{
	int empty = 0;

	spin_lock(&osb->vote_task_lock);
	if (list_empty(&osb->blocked_lock_list) &&
	    list_empty(&osb->vote_list))
		empty = 1;

	spin_unlock(&osb->vote_task_lock);
	return empty;
}

static int ocfs2_vote_thread_should_wake(struct ocfs2_super *osb)
{
	int should_wake = 0;

	spin_lock(&osb->vote_task_lock);
	if (osb->vote_work_sequence != osb->vote_wake_sequence)
		should_wake = 1;
	spin_unlock(&osb->vote_task_lock);

	return should_wake;
}

int ocfs2_vote_thread(void *arg)
{
	int status = 0;
	struct ocfs2_super *osb = arg;

	/* only quit once we've been asked to stop and there is no more
	 * work available */
	while (!(kthread_should_stop() &&
		 ocfs2_vote_thread_lists_empty(osb))) {

		wait_event_interruptible(osb->vote_event,
					 ocfs2_vote_thread_should_wake(osb) ||
					 kthread_should_stop());

		mlog(0, "vote_thread: awoken\n");

		ocfs2_vote_thread_do_work(osb);
	}

	osb->vote_task = NULL;
	return status;
}

static struct ocfs2_net_wait_ctxt *ocfs2_new_net_wait_ctxt(unsigned int response_id)
{
	struct ocfs2_net_wait_ctxt *w;

	w = kcalloc(1, sizeof(*w), GFP_NOFS);
	if (!w) {
		mlog_errno(-ENOMEM);
		goto bail;
	}

	INIT_LIST_HEAD(&w->n_list);
	init_waitqueue_head(&w->n_event);
	ocfs2_node_map_init(&w->n_node_map);
	w->n_response_id = response_id;
	w->n_callback = NULL;
bail:
	return w;
}

static unsigned int ocfs2_new_response_id(struct ocfs2_super *osb)
{
	unsigned int ret;

	spin_lock(&osb->net_response_lock);
	ret = ++osb->net_response_ids;
	spin_unlock(&osb->net_response_lock);

	return ret;
}

static void ocfs2_dequeue_net_wait_ctxt(struct ocfs2_super *osb,
					struct ocfs2_net_wait_ctxt *w)
{
	spin_lock(&osb->net_response_lock);
	list_del(&w->n_list);
	spin_unlock(&osb->net_response_lock);
}

static void ocfs2_queue_net_wait_ctxt(struct ocfs2_super *osb,
				      struct ocfs2_net_wait_ctxt *w)
{
	spin_lock(&osb->net_response_lock);
	list_add_tail(&w->n_list,
		      &osb->net_response_list);
	spin_unlock(&osb->net_response_lock);
}

static void __ocfs2_mark_node_responded(struct ocfs2_super *osb,
					struct ocfs2_net_wait_ctxt *w,
					int node_num)
{
	assert_spin_locked(&osb->net_response_lock);

	ocfs2_node_map_clear_bit(osb, &w->n_node_map, node_num);
	if (ocfs2_node_map_is_empty(osb, &w->n_node_map))
		wake_up(&w->n_event);
}

/* Intended to be called from the node down callback, we fake remove
 * the node from all our response contexts */
void ocfs2_remove_node_from_vote_queues(struct ocfs2_super *osb,
					int node_num)
{
	struct list_head *p;
	struct ocfs2_net_wait_ctxt *w = NULL;

	spin_lock(&osb->net_response_lock);

	list_for_each(p, &osb->net_response_list) {
		w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);

		__ocfs2_mark_node_responded(osb, w, node_num);
	}

	spin_unlock(&osb->net_response_lock);
}

static int ocfs2_broadcast_vote(struct ocfs2_super *osb,
				struct ocfs2_vote_msg *request,
				unsigned int response_id,
				int *response,
				struct ocfs2_net_response_cb *callback)
{
	int status, i, remote_err;
	struct ocfs2_net_wait_ctxt *w = NULL;
	int dequeued = 0;

	mlog_entry_void();

	w = ocfs2_new_net_wait_ctxt(response_id);
	if (!w) {
		status = -ENOMEM;
		mlog_errno(status);
		goto bail;
	}
	w->n_callback = callback;

	/* we're pretty much ready to go at this point, and this fills
	 * in n_response which we need anyway... */
	ocfs2_queue_net_wait_ctxt(osb, w);

	i = ocfs2_node_map_iterate(osb, &osb->mounted_map, 0);

	while (i != O2NM_INVALID_NODE_NUM) {
		if (i != osb->node_num) {
			mlog(0, "trying to send request to node %i\n", i);
			ocfs2_node_map_set_bit(osb, &w->n_node_map, i);

			remote_err = 0;
			status = o2net_send_message(OCFS2_MESSAGE_TYPE_VOTE,
						    osb->net_key,
						    request,
						    sizeof(*request),
						    i,
						    &remote_err);
			if (status == -ETIMEDOUT) {
				mlog(0, "remote node %d timed out!\n", i);
				status = -EAGAIN;
				goto bail;
			}
			if (remote_err < 0) {
				status = remote_err;
				mlog(0, "remote error %d on node %d!\n",
				     remote_err, i);
				mlog_errno(status);
				goto bail;
			}
			if (status < 0) {
				mlog_errno(status);
				goto bail;
			}
		}
		i++;
		i = ocfs2_node_map_iterate(osb, &osb->mounted_map, i);
		mlog(0, "next is %d, i am %d\n", i, osb->node_num);
	}
	mlog(0, "done sending, now waiting on responses...\n");

	wait_event(w->n_event, ocfs2_node_map_is_empty(osb, &w->n_node_map));

	ocfs2_dequeue_net_wait_ctxt(osb, w);
	dequeued = 1;

	*response = w->n_response;
	status = 0;
bail:
	if (w) {
		if (!dequeued)
			ocfs2_dequeue_net_wait_ctxt(osb, w);
		kfree(w);
	}

	mlog_exit(status);
	return status;
}

static struct ocfs2_vote_msg * ocfs2_new_vote_request(struct ocfs2_super *osb,
						      u64 blkno,
						      unsigned int generation,
						      enum ocfs2_vote_request type,
						      u32 priv)
{
	struct ocfs2_vote_msg *request;
	struct ocfs2_msg_hdr *hdr;

	BUG_ON(!ocfs2_is_valid_vote_request(type));

	request = kcalloc(1, sizeof(*request), GFP_NOFS);
	if (!request) {
		mlog_errno(-ENOMEM);
	} else {
		hdr = &request->v_hdr;
		hdr->h_node_num = cpu_to_be32(osb->node_num);
		hdr->h_request = cpu_to_be32(type);
		hdr->h_blkno = cpu_to_be64(blkno);
		hdr->h_generation = cpu_to_be32(generation);

		request->md1.v_generic1 = cpu_to_be32(priv);
	}

	return request;
}

/* Complete the buildup of a new vote request and process the
 * broadcast return value. */
static int ocfs2_do_request_vote(struct ocfs2_super *osb,
				 struct ocfs2_vote_msg *request,
				 struct ocfs2_net_response_cb *callback)
{
	int status, response = -EBUSY;
	unsigned int response_id;
	struct ocfs2_msg_hdr *hdr;

	response_id = ocfs2_new_response_id(osb);

	hdr = &request->v_hdr;
	hdr->h_response_id = cpu_to_be32(response_id);

	status = ocfs2_broadcast_vote(osb, request, response_id, &response,
				      callback);
	if (status < 0) {
		mlog_errno(status);
		goto bail;
	}

	status = response;
bail:

	return status;
}

static int ocfs2_request_vote(struct inode *inode,
			      struct ocfs2_vote_msg *request,
			      struct ocfs2_net_response_cb *callback)
{
	int status;