xpc_channel.c

来自「底层驱动开发」· C语言 代码 · 共 2,298 行 · 第 1/4 页

{
	enum xpc_retval ret;


	DBUG_ON(!spin_is_locked(&ch->lock));

	if (!(ch->flags & XPC_C_OPENREQUEST) ||
				!(ch->flags & XPC_C_ROPENREQUEST)) {
		/* nothing more to do for now */
		return;
	}
	DBUG_ON(!(ch->flags & XPC_C_CONNECTING));

	if (!(ch->flags & XPC_C_SETUP)) {
		spin_unlock_irqrestore(&ch->lock, *irq_flags);
		ret = xpc_allocate_msgqueues(ch);
		spin_lock_irqsave(&ch->lock, *irq_flags);

		if (ret != xpcSuccess) {
			XPC_DISCONNECT_CHANNEL(ch, ret, irq_flags);
		}
		if (ch->flags & (XPC_C_CONNECTED | XPC_C_DISCONNECTING)) {
			return;
		}

		DBUG_ON(!(ch->flags & XPC_C_SETUP));
		DBUG_ON(ch->local_msgqueue == NULL);
		DBUG_ON(ch->remote_msgqueue == NULL);
	}

	if (!(ch->flags & XPC_C_OPENREPLY)) {
		ch->flags |= XPC_C_OPENREPLY;
		xpc_IPI_send_openreply(ch, irq_flags);
	}

	if (!(ch->flags & XPC_C_ROPENREPLY)) {
		return;
	}

	DBUG_ON(ch->remote_msgqueue_pa == 0);

	ch->flags = (XPC_C_CONNECTED | XPC_C_SETUP);	/* clear all else */

	dev_info(xpc_chan, "channel %d to partition %d connected\n",
		ch->number, ch->partid);

	spin_unlock_irqrestore(&ch->lock, *irq_flags);
	xpc_create_kthreads(ch, 1);
	spin_lock_irqsave(&ch->lock, *irq_flags);
}


/*
 * Free up message queues and other stuff that were allocated for the specified
 * channel.
 *
 * Note: ch->reason and ch->reason_line are left set for debugging purposes,
 * they're cleared when XPC_C_DISCONNECTED is cleared.
 */
static void
xpc_free_msgqueues(struct xpc_channel *ch)
{
	DBUG_ON(!spin_is_locked(&ch->lock));
	DBUG_ON(atomic_read(&ch->n_to_notify) != 0);

	ch->remote_msgqueue_pa = 0;
	ch->func = NULL;
	ch->key = NULL;
	ch->msg_size = 0;
	ch->local_nentries = 0;
	ch->remote_nentries = 0;
	ch->kthreads_assigned_limit = 0;
	ch->kthreads_idle_limit = 0;

	ch->local_GP->get = 0;
	ch->local_GP->put = 0;
	ch->remote_GP.get = 0;
	ch->remote_GP.put = 0;
	ch->w_local_GP.get = 0;
	ch->w_local_GP.put = 0;
	ch->w_remote_GP.get = 0;
	ch->w_remote_GP.put = 0;
	ch->next_msg_to_pull = 0;

	if (ch->flags & XPC_C_SETUP) {
		ch->flags &= ~XPC_C_SETUP;

		dev_dbg(xpc_chan, "ch->flags=0x%x, partid=%d, channel=%d\n",
			ch->flags, ch->partid, ch->number);

		kfree(ch->local_msgqueue_base);
		ch->local_msgqueue = NULL;
		kfree(ch->remote_msgqueue_base);
		ch->remote_msgqueue = NULL;
		kfree(ch->notify_queue);
		ch->notify_queue = NULL;

		/* in case someone is waiting for the teardown to complete */
		up(&ch->teardown_sema);
	}
}


/*
 * spin_lock_irqsave() is expected to be held on entry.
 */
static void
xpc_process_disconnect(struct xpc_channel *ch, unsigned long *irq_flags)
{
	struct xpc_partition *part = &xpc_partitions[ch->partid];
	u32 ch_flags = ch->flags;


	DBUG_ON(!spin_is_locked(&ch->lock));

	if (!(ch->flags & XPC_C_DISCONNECTING)) {
		return;
	}

	DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));

	/* make sure all activity has settled down first */

	if (atomic_read(&ch->references) > 0) {
		return;
	}
	DBUG_ON(atomic_read(&ch->kthreads_assigned) != 0);

	/* it's now safe to free the channel's message queues */

	xpc_free_msgqueues(ch);
	DBUG_ON(ch->flags & XPC_C_SETUP);

	if (part->act_state != XPC_P_DEACTIVATING) {

		/* as long as the other side is up do the full protocol */

		if (!(ch->flags & XPC_C_RCLOSEREQUEST)) {
			return;
		}

		if (!(ch->flags & XPC_C_CLOSEREPLY)) {
			ch->flags |= XPC_C_CLOSEREPLY;
			xpc_IPI_send_closereply(ch, irq_flags);
		}

		if (!(ch->flags & XPC_C_RCLOSEREPLY)) {
			return;
		}
	}

	/* both sides are disconnected now */

	ch->flags = XPC_C_DISCONNECTED;	/* clear all flags, but this one */

	atomic_dec(&part->nchannels_active);

	if (ch_flags & XPC_C_WASCONNECTED) {
		dev_info(xpc_chan, "channel %d to partition %d disconnected, "
			"reason=%d\n", ch->number, ch->partid, ch->reason);
	}
}


/*
 * Process a change in the channel's remote connection state.
 */
static void
xpc_process_openclose_IPI(struct xpc_partition *part, int ch_number,
				u8 IPI_flags)
{
	unsigned long irq_flags;
	struct xpc_openclose_args *args =
				&part->remote_openclose_args[ch_number];
	struct xpc_channel *ch = &part->channels[ch_number];
	enum xpc_retval reason;



	spin_lock_irqsave(&ch->lock, irq_flags);


	if (IPI_flags & XPC_IPI_CLOSEREQUEST) {

		dev_dbg(xpc_chan, "XPC_IPI_CLOSEREQUEST (reason=%d) received "
			"from partid=%d, channel=%d\n", args->reason,
			ch->partid, ch->number);

		/*
		 * If RCLOSEREQUEST is set, we're probably waiting for
		 * RCLOSEREPLY. We should find it and a ROPENREQUEST packed
		 * with this RCLOSEQREUQEST in the IPI_flags.
		 */

		if (ch->flags & XPC_C_RCLOSEREQUEST) {
			DBUG_ON(!(ch->flags & XPC_C_DISCONNECTING));
			DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
			DBUG_ON(!(ch->flags & XPC_C_CLOSEREPLY));
			DBUG_ON(ch->flags & XPC_C_RCLOSEREPLY);

			DBUG_ON(!(IPI_flags & XPC_IPI_CLOSEREPLY));
			IPI_flags &= ~XPC_IPI_CLOSEREPLY;
			ch->flags |= XPC_C_RCLOSEREPLY;

			/* both sides have finished disconnecting */
			xpc_process_disconnect(ch, &irq_flags);
		}

		if (ch->flags & XPC_C_DISCONNECTED) {
			// >>> explain this section

			if (!(IPI_flags & XPC_IPI_OPENREQUEST)) {
				DBUG_ON(part->act_state !=
							XPC_P_DEACTIVATING);
				spin_unlock_irqrestore(&ch->lock, irq_flags);
				return;
			}

			XPC_SET_REASON(ch, 0, 0);
			ch->flags &= ~XPC_C_DISCONNECTED;

			atomic_inc(&part->nchannels_active);
			ch->flags |= (XPC_C_CONNECTING | XPC_C_ROPENREQUEST);
		}

		IPI_flags &= ~(XPC_IPI_OPENREQUEST | XPC_IPI_OPENREPLY);

		/*
		 * The meaningful CLOSEREQUEST connection state fields are:
		 *      reason = reason connection is to be closed
		 */

		ch->flags |= XPC_C_RCLOSEREQUEST;

		if (!(ch->flags & XPC_C_DISCONNECTING)) {
			reason = args->reason;
			if (reason <= xpcSuccess || reason > xpcUnknownReason) {
				reason = xpcUnknownReason;
			} else if (reason == xpcUnregistering) {
				reason = xpcOtherUnregistering;
			}

			XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
		} else {
			xpc_process_disconnect(ch, &irq_flags);
		}
	}


	if (IPI_flags & XPC_IPI_CLOSEREPLY) {

		dev_dbg(xpc_chan, "XPC_IPI_CLOSEREPLY received from partid=%d,"
			" channel=%d\n", ch->partid, ch->number);

		if (ch->flags & XPC_C_DISCONNECTED) {
			DBUG_ON(part->act_state != XPC_P_DEACTIVATING);
			spin_unlock_irqrestore(&ch->lock, irq_flags);
			return;
		}

		DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
		DBUG_ON(!(ch->flags & XPC_C_RCLOSEREQUEST));

		ch->flags |= XPC_C_RCLOSEREPLY;

		if (ch->flags & XPC_C_CLOSEREPLY) {
			/* both sides have finished disconnecting */
			xpc_process_disconnect(ch, &irq_flags);
		}
	}


	if (IPI_flags & XPC_IPI_OPENREQUEST) {

		dev_dbg(xpc_chan, "XPC_IPI_OPENREQUEST (msg_size=%d, "
			"local_nentries=%d) received from partid=%d, "
			"channel=%d\n", args->msg_size, args->local_nentries,
			ch->partid, ch->number);

		if ((ch->flags & XPC_C_DISCONNECTING) ||
					part->act_state == XPC_P_DEACTIVATING) {
			spin_unlock_irqrestore(&ch->lock, irq_flags);
			return;
		}
		DBUG_ON(!(ch->flags & (XPC_C_DISCONNECTED |
							XPC_C_OPENREQUEST)));
		DBUG_ON(ch->flags & (XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
					XPC_C_OPENREPLY | XPC_C_CONNECTED));

		/*
		 * The meaningful OPENREQUEST connection state fields are:
		 *      msg_size = size of channel's messages in bytes
		 *      local_nentries = remote partition's local_nentries
		 */
		DBUG_ON(args->msg_size == 0);
		DBUG_ON(args->local_nentries == 0);

		ch->flags |= (XPC_C_ROPENREQUEST | XPC_C_CONNECTING);
		ch->remote_nentries = args->local_nentries;


		if (ch->flags & XPC_C_OPENREQUEST) {
			if (args->msg_size != ch->msg_size) {
				XPC_DISCONNECT_CHANNEL(ch, xpcUnequalMsgSizes,
								&irq_flags);
				spin_unlock_irqrestore(&ch->lock, irq_flags);
				return;
			}
		} else {
			ch->msg_size = args->msg_size;

			XPC_SET_REASON(ch, 0, 0);
			ch->flags &= ~XPC_C_DISCONNECTED;

			atomic_inc(&part->nchannels_active);
		}

		xpc_process_connect(ch, &irq_flags);
	}


	if (IPI_flags & XPC_IPI_OPENREPLY) {

		dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY (local_msgqueue_pa=0x%lx, "
			"local_nentries=%d, remote_nentries=%d) received from "
			"partid=%d, channel=%d\n", args->local_msgqueue_pa,
			args->local_nentries, args->remote_nentries,
			ch->partid, ch->number);

		if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED)) {
			spin_unlock_irqrestore(&ch->lock, irq_flags);
			return;
		}
		DBUG_ON(!(ch->flags & XPC_C_OPENREQUEST));
		DBUG_ON(!(ch->flags & XPC_C_ROPENREQUEST));
		DBUG_ON(ch->flags & XPC_C_CONNECTED);

		/*
		 * The meaningful OPENREPLY connection state fields are:
		 *      local_msgqueue_pa = physical address of remote
		 *			    partition's local_msgqueue
		 *      local_nentries = remote partition's local_nentries
		 *      remote_nentries = remote partition's remote_nentries
		 */
		DBUG_ON(args->local_msgqueue_pa == 0);
		DBUG_ON(args->local_nentries == 0);
		DBUG_ON(args->remote_nentries == 0);

		ch->flags |= XPC_C_ROPENREPLY;
		ch->remote_msgqueue_pa = args->local_msgqueue_pa;

		if (args->local_nentries < ch->remote_nentries) {
			dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY: new "
				"remote_nentries=%d, old remote_nentries=%d, "
				"partid=%d, channel=%d\n",
				args->local_nentries, ch->remote_nentries,
				ch->partid, ch->number);

			ch->remote_nentries = args->local_nentries;
		}
		if (args->remote_nentries < ch->local_nentries) {
			dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY: new "
				"local_nentries=%d, old local_nentries=%d, "
				"partid=%d, channel=%d\n",
				args->remote_nentries, ch->local_nentries,
				ch->partid, ch->number);

			ch->local_nentries = args->remote_nentries;
		}

		xpc_process_connect(ch, &irq_flags);
	}

	spin_unlock_irqrestore(&ch->lock, irq_flags);
}


/*
 * Attempt to establish a channel connection to a remote partition.
 */
static enum xpc_retval
xpc_connect_channel(struct xpc_channel *ch)
{
	unsigned long irq_flags;
	struct xpc_registration *registration = &xpc_registrations[ch->number];


	if (down_interruptible(&registration->sema) != 0) {
		return xpcInterrupted;
	}

	if (!XPC_CHANNEL_REGISTERED(ch->number)) {
		up(&registration->sema);
		return xpcUnregistered;
	}

	spin_lock_irqsave(&ch->lock, irq_flags);

	DBUG_ON(ch->flags & XPC_C_CONNECTED);
	DBUG_ON(ch->flags & XPC_C_OPENREQUEST);

	if (ch->flags & XPC_C_DISCONNECTING) {
		spin_unlock_irqrestore(&ch->lock, irq_flags);
		up(&registration->sema);
		return ch->reason;
	}


	/* add info from the channel connect registration to the channel */

	ch->kthreads_assigned_limit = registration->assigned_limit;
	ch->kthreads_idle_limit = registration->idle_limit;
	DBUG_ON(atomic_read(&ch->kthreads_assigned) != 0);
	DBUG_ON(atomic_read(&ch->kthreads_idle) != 0);
	DBUG_ON(atomic_read(&ch->kthreads_active) != 0);

	ch->func = registration->func;
	DBUG_ON(registration->func == NULL);
	ch->key = registration->key;

	ch->local_nentries = registration->nentries;

	if (ch->flags & XPC_C_ROPENREQUEST) {
		if (registration->msg_size != ch->msg_size) {
			/* the local and remote sides aren't the same */

			/*
			 * Because XPC_DISCONNECT_CHANNEL() can block we're
			 * forced to up the registration sema before we unlock
			 * the channel lock. But that's okay here because we're
			 * done with the part that required the registration
			 * sema. XPC_DISCONNECT_CHANNEL() requires that the
			 * channel lock be locked and will unlock and relock
			 * the channel lock as needed.
			 */
			up(&registration->sema);
			XPC_DISCONNECT_CHANNEL(ch, xpcUnequalMsgSizes,
								&irq_flags);
			spin_unlock_irqrestore(&ch->lock, irq_flags);
			return xpcUnequalMsgSizes;
		}
	} else {
		ch->msg_size = registration->msg_size;

		XPC_SET_REASON(ch, 0, 0);
		ch->flags &= ~XPC_C_DISCONNECTED;

		atomic_inc(&xpc_partitions[ch->partid].nchannels_active);
	}

	up(&registration->sema);


	/* initiate the connection */

	ch->flags |= (XPC_C_OPENREQUEST | XPC_C_CONNECTING);
	xpc_IPI_send_openrequest(ch, &irq_flags);

	xpc_process_connect(ch, &irq_flags);

	spin_unlock_irqrestore(&ch->lock, irq_flags);

	return xpcSuccess;
}


/*
 * Notify those who wanted to be notified upon delivery of their message.
 */
static void
xpc_notify_senders(struct xpc_channel *ch, enum xpc_retval reason, s64 put)
{
	struct xpc_notify *notify;
	u8 notify_type;
	s64 get = ch->w_remote_GP.get - 1;


	while (++get < put && atomic_read(&ch->n_to_notify) > 0) {

		notify = &ch->notify_queue[get % ch->local_nentries];

		/*
		 * See if the notify entry indicates it was associated with
		 * a message who's sender wants to be notified. It is possible
		 * that it is, but someone else is doing or has done the
		 * notification.
		 */
		notify_type = notify->type;
		if (notify_type == 0 ||
				cmpxchg(&notify->type, notify_type, 0) !=
								notify_type) {
			continue;
		}

		DBUG_ON(notify_type != XPC_N_CALL);

		atomic_dec(&ch->n_to_notify);

		if (notify->func != NULL) {
			dev_dbg(xpc_chan, "notify->func() called, notify=0x%p, "
				"msg_number=%ld, partid=%d, channel=%d\n",
				(void *) notify, get, ch->partid, ch->number);

			notify->func(reason, ch->partid, ch->number,
								notify->key);

			dev_dbg(xpc_chan, "notify->func() returned, "
				"notify=0x%p, msg_number=%ld, partid=%d, "
				"channel=%d\n", (void *) notify, get,
				ch->partid, ch->number);
		}
	}
}


/*
 * Clear some of the msg flags in the local message queue.
 */
static inline void
xpc_clear_local_msgqueue_flags(struct xpc_channel *ch)
{
	struct xpc_msg *msg;
	s64 get;


	get = ch->w_remote_GP.get;
	do {
		msg = (struct xpc_msg *) ((u64) ch->local_msgqueue +
				(get % ch->local_nentries) * ch->msg_size);
		msg->flags = 0;
	} while (++get < (volatile s64) ch->remote_GP.get);
}


/*
 * Clear some of the msg flags in the remote message queue.
 */
static inline void
xpc_clear_remote_msgqueue_flags(struct xpc_channel *ch)
{
	struct xpc_msg *msg;
	s64 put;


	put = ch->w_remote_GP.put;
	do {
		msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue +
				(put % ch->remote_nentries) * ch->msg_size);
		msg->flags = 0;
	} while (++put < (volatile s64) ch->remote_GP.put);
}


static void
xpc_process_msg_IPI(struct xpc_partition *part, int ch_number)
{
	struct xpc_channel *ch = &part->channels[ch_number];
	int nmsgs_sent;


	ch->remote_GP = part->remote_GPs[ch_number];


	/* See what, if anything, has changed for each connected channel */

	xpc_msgqueue_ref(ch);

	if (ch->w_remote_GP.get == ch->remote_GP.get &&
				ch->w_remote_GP.put == ch->remote_GP.put) {
		/* nothing changed since GPs were last pulled */
		xpc_msgqueue_deref(ch);
		return;
	}

	if (!(ch->flags & XPC_C_CONNECTED)){