common/os/msg.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */
/*
 * Copyright 2007 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

/*	Copyright (c) 1984, 1986, 1987, 1988, 1989 AT&T	*/
/*	  All Rights Reserved  	*/


#pragma ident	"@(#)msg.c	1.71	07/12/26 SMI"

/*
 * Inter-Process Communication Message Facility.
 *
 * See os/ipc.c for a description of common IPC functionality.
 *
 * Resource controls
 * -----------------
 *
 * Control:      zone.max-msg-ids (rc_zone_msgmni)
 * Description:  Maximum number of message queue ids allowed a zone.
 *
 *   When msgget() is used to allocate a message queue, one id is
 *   allocated.  If the id allocation doesn't succeed, msgget() fails
 *   and errno is set to ENOSPC.  Upon successful msgctl(, IPC_RMID)
 *   the id is deallocated.
 *
 * Control:      project.max-msg-ids (rc_project_msgmni)
 * Description:  Maximum number of message queue ids allowed a project.
 *
 *   When msgget() is used to allocate a message queue, one id is
 *   allocated.  If the id allocation doesn't succeed, msgget() fails
 *   and errno is set to ENOSPC.  Upon successful msgctl(, IPC_RMID)
 *   the id is deallocated.
 *
 * Control:      process.max-msg-qbytes (rc_process_msgmnb)
 * Description:  Maximum number of bytes of messages on a message queue.
 *
 *   When msgget() successfully allocates a message queue, the minimum
 *   enforced value of this limit is used to initialize msg_qbytes.
 *
 * Control:      process.max-msg-messages (rc_process_msgtql)
 * Description:  Maximum number of messages on a message queue.
 *
 *   When msgget() successfully allocates a message queue, the minimum
 *   enforced value of this limit is used to initialize a per-queue
 *   limit on the number of messages.
 */

#include <sys/types.h>
#include <sys/t_lock.h>
#include <sys/param.h>
#include <sys/cred.h>
#include <sys/user.h>
#include <sys/proc.h>
#include <sys/time.h>
#include <sys/ipc.h>
#include <sys/ipc_impl.h>
#include <sys/msg.h>
#include <sys/msg_impl.h>
#include <sys/list.h>
#include <sys/systm.h>
#include <sys/sysmacros.h>
#include <sys/cpuvar.h>
#include <sys/kmem.h>
#include <sys/ddi.h>
#include <sys/errno.h>
#include <sys/cmn_err.h>
#include <sys/debug.h>
#include <sys/project.h>
#include <sys/modctl.h>
#include <sys/syscall.h>
#include <sys/policy.h>
#include <sys/zone.h>

#include <c2/audit.h>

/*
 * The following tunables are obsolete.  Though for compatibility we
 * still read and interpret msginfo_msgmnb, msginfo_msgmni, and
 * msginfo_msgtql (see os/project.c and os/rctl_proc.c), the preferred
 * mechanism for administrating the IPC Message facility is through the
 * resource controls described at the top of this file.
 */
size_t	msginfo_msgmax = 2048;	/* (obsolete) */
size_t	msginfo_msgmnb = 4096;	/* (obsolete) */
int	msginfo_msgmni = 50;	/* (obsolete) */
int	msginfo_msgtql = 40;	/* (obsolete) */
int	msginfo_msgssz = 8;	/* (obsolete) */
int	msginfo_msgmap = 0;	/* (obsolete) */
ushort_t msginfo_msgseg = 1024;	/* (obsolete) */

extern rctl_hndl_t rc_zone_msgmni;
extern rctl_hndl_t rc_project_msgmni;
extern rctl_hndl_t rc_process_msgmnb;
extern rctl_hndl_t rc_process_msgtql;
static ipc_service_t *msq_svc;
static zone_key_t msg_zone_key;

static void msg_dtor(kipc_perm_t *);
static void msg_rmid(kipc_perm_t *);
static void msg_remove_zone(zoneid_t, void *);

/*
 * Module linkage information for the kernel.
 */
static ssize_t msgsys(int opcode, uintptr_t a0, uintptr_t a1, uintptr_t a2,
	uintptr_t a4, uintptr_t a5);

static struct sysent ipcmsg_sysent = {
	6,
#ifdef	_LP64
	SE_ARGC | SE_NOUNLOAD | SE_64RVAL,
#else
	SE_ARGC | SE_NOUNLOAD | SE_32RVAL1,
#endif
	(int (*)())msgsys
};

#ifdef	_SYSCALL32_IMPL
static ssize32_t msgsys32(int opcode, uint32_t a0, uint32_t a1, uint32_t a2,
	uint32_t a4, uint32_t a5);

static struct sysent ipcmsg_sysent32 = {
	6,
	SE_ARGC | SE_NOUNLOAD | SE_32RVAL1,
	(int (*)())msgsys32
};
#endif	/* _SYSCALL32_IMPL */

static struct modlsys modlsys = {
	&mod_syscallops, "System V message facility", &ipcmsg_sysent
};

#ifdef _SYSCALL32_IMPL
static struct modlsys modlsys32 = {
	&mod_syscallops32, "32-bit System V message facility", &ipcmsg_sysent32
};
#endif

/*
 *      Big Theory statement for message queue correctness
 *
 * The msgrcv and msgsnd functions no longer uses cv_broadcast to wake up
 * receivers who are waiting for an event.  Using the cv_broadcast method
 * resulted in negative scaling when the number of waiting receivers are large
 * (the thundering herd problem).  Instead, the receivers waiting to receive a
 * message are now linked in a queue-like fashion and awaken one at a time in
 * a controlled manner.
 *
 * Receivers can block on two different classes of waiting list:
 *    1) "sendwait" list, which is the more complex list of the two.  The
 *	  receiver will be awakened by a sender posting a new message.  There
 *	  are two types of "sendwait" list used:
 *		a) msg_wait_snd: handles all receivers who are looking for
 *		   a message type >= 0, but was unable to locate a match.
 *
 *		   slot 0: reserved for receivers that have designated they
 *			   will take any message type.
 *		   rest:   consist of receivers requesting a specific type
 *			   but the type was not present.  The entries are
 *			   hashed into a bucket in an attempt to keep
 *			   any list search relatively short.
 * 		b) msg_wait_snd_ngt: handles all receivers that have designated
 *		   a negative message type. Unlike msg_wait_snd, the hash bucket
 *		   serves a range of negative message types (-1 to -5, -6 to -10
 *		   and so forth), where the last bucket is reserved for all the
 *		   negative message types that hash outside of MSG_MAX_QNUM - 1.
 *		   This is done this way to simplify the operation of locating a
 *		   negative message type.
 *
 *    2) "copyout" list, where the receiver is awakened by another
 *	 receiver after a message is copied out.  This is a linked list
 *	 of waiters that are awakened one at a time.  Although the solution is
 *	 not optimal, the complexity that would be added in for waking
 *	 up the right entry far exceeds any potential pay back (too many
 *	 correctness and corner case issues).
 *
 * The lists are doubly linked.  In the case of the "sendwait"
 * list, this allows the thread to remove itself from the list without having
 * to traverse the list.  In the case of the "copyout" list it simply allows
 * us to use common functions with the "sendwait" list.
 *
 * To make sure receivers are not hung out to dry, we must guarantee:
 *    1. If any queued message matches any receiver, then at least one
 *       matching receiver must be processing the request.
 *    2. Blocking on the copyout queue is only temporary while messages
 *	 are being copied out.  The process is guaranted to wakeup
 *	 when it gets to front of the queue (copyout is a FIFO).
 *
 * Rules for blocking and waking up:
 *   1. A receiver entering msgrcv must examine all messages for a match
 *      before blocking on a sendwait queue.
 *   2. If the receiver blocks because the message it chose is already
 *	being copied out, then when it wakes up needs to start start
 *	checking the messages from the beginning.
 *   3) When ever a process returns from msgrcv for any reason, if it
 *	had attempted to copy a message or blocked waiting for a copy
 *	to complete it needs to wakeup the next receiver blocked on
 *	a copy out.
 *   4) When a message is sent, the sender selects a process waiting
 *	for that type of message.  This selection process rotates between
 *	receivers types of 0, negative and positive to prevent starvation of
 *	any one particular receiver type.
 *   5) The following are the scenarios for processes that are awakened
 *	by a msgsnd:
 *		a) The process finds the message and is able to copy
 *		   it out.  Once complete, the process returns.
 *		b) The message that was sent that triggered the wakeup is no
 *		   longer available (another process found the message first).
 *		   We issue a wakeup on copy queue and then go back to
 *		   sleep waiting for another matching message to be sent.
 *		c) The message that was supposed to be processed was
 *		   already serviced by another process.  However a different
 *		   message is present which we can service.  The message
 *		   is copied and the process returns.
 *		d) The message is found, but some sort of error occurs that
 *		   prevents the message from being copied.  The receiver
 *		   wakes up the next sender that can service this message
 *		   type and returns an error to the caller.
 *		e) The message is found, but it is marked as being copied
 *		   out.  The receiver then goes to sleep on the copyout
 *		   queue where it will be awakened again sometime in the future.
 *
 *
 *   6) Whenever a message is found that matches the message type designated,
 * 	but is being copied out we have to block on the copyout queue.
 *	After process copying finishes the copy out, it  must wakeup (either
 *	directly or indirectly) all receivers who blocked on its copyout,
 *	so they are guaranteed a chance to examine the remaining messages.
 *	This is implemented via a chain of wakeups: Y wakes X, who wakes Z,
 *	and so on.  The chain cannot be broken.  This leads to the following
 *	cases:
 *		a) A receiver is finished copying the message (or encountered)
 *		   an error), the first entry on the copyout queue is woken
 *		   up.
 *		b) When the receiver is woken up, it attempts to locate
 *		   a message type match.
 *		c) If a message type is found and
 *			-- MSG_RCVCOPY flag is not set, the message is
 *			   marked for copying out.  Regardless of the copyout
 *			   success the next entry on the copyout queue is
 *			   awakened and the operation is completed.
 *			-- MSG_RCVCOPY is set, we simply go back to sleep again
 *			   on the copyout queue.
 *		d) If the message type is not found then we wakeup the next
 *		   process on the copyout queue.
 */

static ulong_t msg_type_hash(long);
static int msgq_check_err(kmsqid_t *qp, int cvres);
static int msg_rcvq_sleep(list_t *, msgq_wakeup_t *, kmutex_t **,
    kmsqid_t *);
static int msg_copyout(kmsqid_t *, long, kmutex_t **, size_t *, size_t,
    struct msg *, struct ipcmsgbuf *, int);
static void msg_rcvq_wakeup_all(list_t *);
static void msg_wakeup_rdr(kmsqid_t *, msg_select_t **, long);
static msgq_wakeup_t *msg_fnd_any_snd(kmsqid_t *, int, long);
static msgq_wakeup_t *msg_fnd_any_rdr(kmsqid_t *, int, long);
static msgq_wakeup_t *msg_fnd_neg_snd(kmsqid_t *, int, long);
static msgq_wakeup_t *msg_fnd_spc_snd(kmsqid_t *, int, long);
static struct msg *msgrcv_lookup(kmsqid_t *, long);

msg_select_t msg_fnd_sndr[] = {
	{ msg_fnd_any_snd, &msg_fnd_sndr[1] },
	{ msg_fnd_spc_snd, &msg_fnd_sndr[2] },
	{ msg_fnd_neg_snd, &msg_fnd_sndr[0] }
};

msg_select_t msg_fnd_rdr[1] = {
	{ msg_fnd_any_rdr, &msg_fnd_rdr[0] },
};

static struct modlinkage modlinkage = {
	MODREV_1,
	&modlsys,
#ifdef _SYSCALL32_IMPL
	&modlsys32,
#endif
	NULL
};


int
_init(void)
{
	int result;

	msq_svc = ipcs_create("msqids", rc_project_msgmni, rc_zone_msgmni,
	    sizeof (kmsqid_t), msg_dtor, msg_rmid, AT_IPC_MSG,
	    offsetof(ipc_rqty_t, ipcq_msgmni));
	zone_key_create(&msg_zone_key, NULL, msg_remove_zone, NULL);

	if ((result = mod_install(&modlinkage)) == 0)
		return (0);

	(void) zone_key_delete(msg_zone_key);
	ipcs_destroy(msq_svc);

	return (result);
}

int
_fini(void)
{
	return (EBUSY);
}

int
_info(struct modinfo *modinfop)
{
	return (mod_info(&modlinkage, modinfop));
}

static void
msg_dtor(kipc_perm_t *perm)
{
	kmsqid_t *qp = (kmsqid_t *)perm;
	int		ii;

	for (ii = 0; ii <= MSG_MAX_QNUM; ii++) {
		ASSERT(list_is_empty(&qp->msg_wait_snd[ii]));
		ASSERT(list_is_empty(&qp->msg_wait_snd_ngt[ii]));
		list_destroy(&qp->msg_wait_snd[ii]);
		list_destroy(&qp->msg_wait_snd_ngt[ii]);
	}
	ASSERT(list_is_empty(&qp->msg_cpy_block));
	list_destroy(&qp->msg_cpy_block);
	ASSERT(qp->msg_snd_cnt == 0);
	ASSERT(qp->msg_cbytes == 0);
	list_destroy(&qp->msg_list);
}


#define	msg_hold(mp)	(mp)->msg_copycnt++

/*
 * msg_rele - decrement the reference count on the message.  When count
 * reaches zero, free message header and contents.
 */
static void
msg_rele(struct msg *mp)
{
	ASSERT(mp->msg_copycnt > 0);
	if (mp->msg_copycnt-- == 1) {
		if (mp->msg_addr)
			kmem_free(mp->msg_addr, mp->msg_size);
		kmem_free(mp, sizeof (struct msg));
	}
}

/*
 * msgunlink - Unlink msg from queue, decrement byte count and wake up anyone
 * waiting for free bytes on queue.
 *
 * Called with queue locked.
 */
static void
msgunlink(kmsqid_t *qp, struct msg *mp)
{
	list_remove(&qp->msg_list, mp);
	qp->msg_qnum--;
	qp->msg_cbytes -= mp->msg_size;
	msg_rele(mp);

	/* Wake up waiting writers */
	if (qp->msg_snd_cnt)
		cv_broadcast(&qp->msg_snd_cv);
}

static void
msg_rmid(kipc_perm_t *perm)
{
	kmsqid_t *qp = (kmsqid_t *)perm;
	struct msg *mp;
	int		ii;


	while ((mp = list_head(&qp->msg_list)) != NULL)
		msgunlink(qp, mp);
	ASSERT(qp->msg_cbytes == 0);

	/*
	 * Wake up everyone who is in a wait state of some sort
	 * for this message queue.
	 */
	for (ii = 0; ii <= MSG_MAX_QNUM; ii++) {
		msg_rcvq_wakeup_all(&qp->msg_wait_snd[ii]);
		msg_rcvq_wakeup_all(&qp->msg_wait_snd_ngt[ii]);
	}
	msg_rcvq_wakeup_all(&qp->msg_cpy_block);
	if (qp->msg_snd_cnt)
		cv_broadcast(&qp->msg_snd_cv);
}

/*
 * msgctl system call.
 *
 * gets q lock (via ipc_lookup), releases before return.
 * may call users of msg_lock
 */
static int
msgctl(int msgid, int cmd, void *arg)
{
	STRUCT_DECL(msqid_ds, ds);		/* SVR4 queue work area */
	kmsqid_t		*qp;		/* ptr to associated q */
	int			error;
	struct	cred		*cr;
	model_t	mdl = get_udatamodel();
	struct msqid_ds64	ds64;
	kmutex_t		*lock;
	proc_t			*pp = curproc;

	STRUCT_INIT(ds, mdl);
	cr = CRED();

	/*
	 * Perform pre- or non-lookup actions (e.g. copyins, RMID).
	 */
	switch (cmd) {
	case IPC_SET:
		if (copyin(arg, STRUCT_BUF(ds), STRUCT_SIZE(ds)))
			return (set_errno(EFAULT));
		break;

	case IPC_SET64:
		if (copyin(arg, &ds64, sizeof (struct msqid_ds64)))
			return (set_errno(EFAULT));
		break;

	case IPC_RMID:
		if (error = ipc_rmid(msq_svc, msgid, cr))
			return (set_errno(error));
		return (0);
	}

	/*
	 * get msqid_ds for this msgid
	 */
	if ((lock = ipc_lookup(msq_svc, msgid, (kipc_perm_t **)&qp)) == NULL)
		return (set_errno(EINVAL));

	switch (cmd) {
	case IPC_SET:
		if (STRUCT_FGET(ds, msg_qbytes) > qp->msg_qbytes &&
		    secpolicy_ipc_config(cr) != 0) {
			mutex_exit(lock);
			return (set_errno(EPERM));
		}
		if (error = ipcperm_set(msq_svc, cr, &qp->msg_perm,
		    &STRUCT_BUF(ds)->msg_perm, mdl)) {
			mutex_exit(lock);
			return (set_errno(error));
		}
		qp->msg_qbytes = STRUCT_FGET(ds, msg_qbytes);
		qp->msg_ctime = gethrestime_sec();
		break;

	case IPC_STAT:
		if (error = ipcperm_access(&qp->msg_perm, MSG_R, cr)) {
			mutex_exit(lock);
			return (set_errno(error));
		}

		if (qp->msg_rcv_cnt)
			qp->msg_perm.ipc_mode |= MSG_RWAIT;
		if (qp->msg_snd_cnt)
			qp->msg_perm.ipc_mode |= MSG_WWAIT;
		ipcperm_stat(&STRUCT_BUF(ds)->msg_perm, &qp->msg_perm, mdl);
		qp->msg_perm.ipc_mode &= ~(MSG_RWAIT|MSG_WWAIT);
		STRUCT_FSETP(ds, msg_first, NULL); 	/* kernel addr */
		STRUCT_FSETP(ds, msg_last, NULL);
		STRUCT_FSET(ds, msg_cbytes, qp->msg_cbytes);
		STRUCT_FSET(ds, msg_qnum, qp->msg_qnum);
		STRUCT_FSET(ds, msg_qbytes, qp->msg_qbytes);
		STRUCT_FSET(ds, msg_lspid, qp->msg_lspid);
		STRUCT_FSET(ds, msg_lrpid, qp->msg_lrpid);
		STRUCT_FSET(ds, msg_stime, qp->msg_stime);
		STRUCT_FSET(ds, msg_rtime, qp->msg_rtime);
		STRUCT_FSET(ds, msg_ctime, qp->msg_ctime);
		break;

	case IPC_SET64:
		mutex_enter(&pp->p_lock);
		if ((ds64.msgx_qbytes > qp->msg_qbytes) &&
		    secpolicy_ipc_config(cr) != 0 &&
		    rctl_test(rc_process_msgmnb, pp->p_rctls, pp,
		    ds64.msgx_qbytes, RCA_SAFE) & RCT_DENY) {
			mutex_exit(&pp->p_lock);
			mutex_exit(lock);
			return (set_errno(EPERM));
		}
		mutex_exit(&pp->p_lock);
		if (error = ipcperm_set64(msq_svc, cr, &qp->msg_perm,
		    &ds64.msgx_perm)) {
			mutex_exit(lock);
			return (set_errno(error));
		}
		qp->msg_qbytes = ds64.msgx_qbytes;
		qp->msg_ctime = gethrestime_sec();
		break;

	case IPC_STAT64:
		if (qp->msg_rcv_cnt)
			qp->msg_perm.ipc_mode |= MSG_RWAIT;
		if (qp->msg_snd_cnt)
			qp->msg_perm.ipc_mode |= MSG_WWAIT;
		ipcperm_stat64(&ds64.msgx_perm, &qp->msg_perm);
		qp->msg_perm.ipc_mode &= ~(MSG_RWAIT|MSG_WWAIT);
		ds64.msgx_cbytes = qp->msg_cbytes;
		ds64.msgx_qnum = qp->msg_qnum;
		ds64.msgx_qbytes = qp->msg_qbytes;
		ds64.msgx_lspid = qp->msg_lspid;
		ds64.msgx_lrpid = qp->msg_lrpid;
		ds64.msgx_stime = qp->msg_stime;
		ds64.msgx_rtime = qp->msg_rtime;
		ds64.msgx_ctime = qp->msg_ctime;
		break;

	default:
		mutex_exit(lock);
		return (set_errno(EINVAL));
	}

	mutex_exit(lock);

	/*
	 * Do copyout last (after releasing mutex).
	 */
	switch (cmd) {
	case IPC_STAT:
		if (copyout(STRUCT_BUF(ds), arg, STRUCT_SIZE(ds)))
			return (set_errno(EFAULT));
		break;

	case IPC_STAT64:
		if (copyout(&ds64, arg, sizeof (struct msqid_ds64)))
			return (set_errno(EFAULT));
		break;
	}

	return (0);
}

/*
 * Remove all message queues associated with a given zone.  Called by
 * zone_shutdown when the zone is halted.
 */
/*ARGSUSED1*/
static void
msg_remove_zone(zoneid_t zoneid, void *arg)
{
	ipc_remove_zone(msq_svc, zoneid);
}

/*
 * msgget system call.
 */
static int
msgget(key_t key, int msgflg)
{
	kmsqid_t	*qp;
	kmutex_t	*lock;
	int		id, error;
	int		ii;
	proc_t		*pp = curproc;

top:
	if (error = ipc_get(msq_svc, key, msgflg, (kipc_perm_t **)&qp, &lock))
		return (set_errno(error));

	if (IPC_FREE(&qp->msg_perm)) {
		mutex_exit(lock);
		mutex_exit(&pp->p_lock);

		list_create(&qp->msg_list, sizeof (struct msg),
		    offsetof(struct msg, msg_node));
		qp->msg_qnum = 0;
		qp->msg_lspid = qp->msg_lrpid = 0;
		qp->msg_stime = qp->msg_rtime = 0;
		qp->msg_ctime = gethrestime_sec();
		qp->msg_ngt_cnt = 0;
		qp->msg_neg_copy = 0;
		for (ii = 0; ii <= MSG_MAX_QNUM; ii++) {
			list_create(&qp->msg_wait_snd[ii],
			    sizeof (msgq_wakeup_t),
			    offsetof(msgq_wakeup_t, msgw_list));
			list_create(&qp->msg_wait_snd_ngt[ii],
			    sizeof (msgq_wakeup_t),
			    offsetof(msgq_wakeup_t, msgw_list));
		}
		/*
		 * The proper initialization of msg_lowest_type is to the
		 * highest possible value.  By doing this we guarantee that
		 * when the first send happens, the lowest type will be set
		 * properly.
		 */
		qp->msg_lowest_type = -1;
		list_create(&qp->msg_cpy_block,
		    sizeof (msgq_wakeup_t),
		    offsetof(msgq_wakeup_t, msgw_list));
		qp->msg_fnd_sndr = &msg_fnd_sndr[0];
		qp->msg_fnd_rdr = &msg_fnd_rdr[0];
		qp->msg_rcv_cnt = 0;
		qp->msg_snd_cnt = 0;

		if (error = ipc_commit_begin(msq_svc, key, msgflg,
		    (kipc_perm_t *)qp)) {
			if (error == EAGAIN)
				goto top;
			return (set_errno(error));
		}
		qp->msg_qbytes = rctl_enforced_value(rc_process_msgmnb,
		    pp->p_rctls, pp);
		qp->msg_qmax = rctl_enforced_value(rc_process_msgtql,
		    pp->p_rctls, pp);
		lock = ipc_commit_end(msq_svc, &qp->msg_perm);
	}
	if (audit_active)
		audit_ipcget(AT_IPC_MSG, (void *)qp);
	id = qp->msg_perm.ipc_id;
	mutex_exit(lock);
	return (id);
}

static ssize_t
msgrcv(int msqid, struct ipcmsgbuf *msgp, size_t msgsz, long msgtyp, int msgflg)
{
	struct msg	*smp;	/* ptr to best msg on q */
	kmsqid_t	*qp;	/* ptr to associated q */
	kmutex_t	*lock;
	size_t		xtsz;	/* transfer byte count */
	int		error = 0;
	int		cvres;
	ulong_t		msg_hash;
	msgq_wakeup_t	msg_entry;

	CPU_STATS_ADDQ(CPU, sys, msg, 1);	/* bump msg send/rcv count */

	msg_hash = msg_type_hash(msgtyp);
	if ((lock = ipc_lookup(msq_svc, msqid, (kipc_perm_t **)&qp)) == NULL) {
		return ((ssize_t)set_errno(EINVAL));
	}
	ipc_hold(msq_svc, (kipc_perm_t *)qp);

	if (error = ipcperm_access(&qp->msg_perm, MSG_R, CRED())) {
		goto msgrcv_out;
	}

	/*
	 * Various information (including the condvar_t) required for the
	 * process to sleep is provided by it's stack.
	 */
	msg_entry.msgw_thrd = curthread;
	msg_entry.msgw_snd_wake = 0;
	msg_entry.msgw_type = msgtyp;
findmsg:
	smp = msgrcv_lookup(qp, msgtyp);

	if (smp) {
		/*
		 * We found a possible message to copy out.
		 */
		if ((smp->msg_flags & MSG_RCVCOPY) == 0) {
			/*
			 * It is available, attempt to copy it.
			 */
			error = msg_copyout(qp, msgtyp, &lock, &xtsz, msgsz,
			    smp, msgp, msgflg);
			/*
			 * Don't forget to wakeup a sleeper that blocked because
			 * we were copying things out.
			 */
			msg_wakeup_rdr(qp, &qp->msg_fnd_rdr, 0);
			goto msgrcv_out;
		}
		/*
		 * The selected message is being copied out, so block.  We do
		 * not need to wake the next person up on the msg_cpy_block list
		 * due to the fact some one is copying out and they will get
		 * things moving again once the copy is completed.
		 */
		cvres = msg_rcvq_sleep(&qp->msg_cpy_block,
		    &msg_entry, &lock, qp);
		error = msgq_check_err(qp, cvres);
		if (error) {
			goto msgrcv_out;
		}
		goto findmsg;
	}
	/*
	 * There isn't a message to copy out that matches the designated
	 * criteria.
	 */
	if (msgflg & IPC_NOWAIT) {
		error = ENOMSG;
		goto msgrcv_out;
	}
	msg_wakeup_rdr(qp,  &qp->msg_fnd_rdr, 0);

	/*
	 * Wait for new message.  We keep the negative and positive types
	 * separate for performance reasons.
	 */
	msg_entry.msgw_snd_wake = 0;
	if (msgtyp >= 0) {
		cvres = msg_rcvq_sleep(&qp->msg_wait_snd[msg_hash],
		    &msg_entry, &lock, qp);
	} else {
		qp->msg_ngt_cnt++;
		cvres = msg_rcvq_sleep(&qp->msg_wait_snd_ngt[msg_hash],
		    &msg_entry, &lock, qp);
		qp->msg_ngt_cnt--;
	}

	if (!(error = msgq_check_err(qp, cvres))) {
		goto findmsg;
	}

msgrcv_out:
	if (error) {
		msg_wakeup_rdr(qp,  &qp->msg_fnd_rdr, 0);
		if (msg_entry.msgw_snd_wake) {
			msg_wakeup_rdr(qp, &qp->msg_fnd_sndr,
			    msg_entry.msgw_snd_wake);
		}
		ipc_rele(msq_svc, (kipc_perm_t *)qp);
		return ((ssize_t)set_errno(error));
	}
	ipc_rele(msq_svc, (kipc_perm_t *)qp);
	return ((ssize_t)xtsz);
}

static int
msgq_check_err(kmsqid_t *qp, int cvres)
{
	if (IPC_FREE(&qp->msg_perm)) {
		return (EIDRM);
	}

	if (cvres == 0) {
		return (EINTR);
	}

	return (0);
}

static int
msg_copyout(kmsqid_t *qp, long msgtyp, kmutex_t **lock, size_t *xtsz_ret,
    size_t msgsz, struct msg *smp, struct ipcmsgbuf *msgp, int msgflg)
{
	size_t		xtsz;
	STRUCT_HANDLE(ipcmsgbuf, umsgp);
	model_t		mdl = get_udatamodel();
	int		copyerror = 0;

	STRUCT_SET_HANDLE(umsgp, mdl, msgp);
	if (msgsz < smp->msg_size) {
		if ((msgflg & MSG_NOERROR) == 0) {
			return (E2BIG);
		} else {
			xtsz = msgsz;
		}
	} else {
		xtsz = smp->msg_size;
	}
	*xtsz_ret = xtsz;

	/*
	 * To prevent a DOS attack we mark the message as being
	 * copied out and release mutex.  When the copy is completed
	 * we need to acquire the mutex and make the appropriate updates.
	 */
	ASSERT((smp->msg_flags & MSG_RCVCOPY) == 0);
	smp->msg_flags |= MSG_RCVCOPY;
	msg_hold(smp);
	if (msgtyp < 0) {
		ASSERT(qp->msg_neg_copy == 0);
		qp->msg_neg_copy = 1;
	}
	mutex_exit(*lock);

	if (mdl == DATAMODEL_NATIVE) {
		copyerror = copyout(&smp->msg_type, msgp,
		    sizeof (smp->msg_type));
	} else {
		/*
		 * 32-bit callers need an imploded msg type.
		 */
		int32_t	msg_type32 = smp->msg_type;

		copyerror = copyout(&msg_type32, msgp,
		    sizeof (msg_type32));
	}

	if (copyerror == 0 && xtsz) {
		copyerror = copyout(smp->msg_addr,
		    STRUCT_FADDR(umsgp, mtext), xtsz);
	}

	/*
	 * Reclaim the mutex and make sure the message queue still exists.
	 */

	*lock = ipc_lock(msq_svc, qp->msg_perm.ipc_id);
	if (msgtyp < 0) {
		qp->msg_neg_copy = 0;
	}
	ASSERT(smp->msg_flags & MSG_RCVCOPY);
	smp->msg_flags &= ~MSG_RCVCOPY;
	msg_rele(smp);
	if (IPC_FREE(&qp->msg_perm)) {
		return (EIDRM);
	}
	if (copyerror) {
		return (EFAULT);
	}
	qp->msg_lrpid = ttoproc(curthread)->p_pid;
	qp->msg_rtime = gethrestime_sec();
	msgunlink(qp, smp);
	return (0);
}

static struct msg *
msgrcv_lookup(kmsqid_t *qp, long msgtyp)
{
	struct msg 		*smp = NULL;
	int			qp_low;
	struct msg		*mp;	/* ptr to msg on q */
	int			low_msgtype;
	static struct msg	neg_copy_smp;

	mp = list_head(&qp->msg_list);
	if (msgtyp == 0) {
		smp = mp;
	} else {
		qp_low = qp->msg_lowest_type;
		if (msgtyp > 0) {
			/*
			 * If our lowest possible message type is larger than
			 * the message type desired, then we know there is
			 * no entry present.
			 */
			if (qp_low > msgtyp) {
				return (NULL);
			}

			for (; mp; mp = list_next(&qp->msg_list, mp)) {
				if (msgtyp == mp->msg_type) {
					smp = mp;
					break;
				}
			}
		} else {
			/*
			 * We have kept track of the lowest possible message
			 * type on the send queue.  This allows us to terminate
			 * the search early if we find a message type of that
			 * type.  Note, the lowest type may not be the actual
			 * lowest value in the system, it is only guaranteed
			 * that there isn't a value lower than that.
			 */
			low_msgtype = -msgtyp;
			if (low_msgtype++ < qp_low) {
				return (NULL);
			}
			if (qp->msg_neg_copy) {
				neg_copy_smp.msg_flags = MSG_RCVCOPY;
				return (&neg_copy_smp);
			}
			for (; mp; mp = list_next(&qp->msg_list, mp)) {
				if (mp->msg_type < low_msgtype) {
					smp = mp;
					low_msgtype = mp->msg_type;
					if (low_msgtype == qp_low) {
						break;
					}
				}
			}
			if (smp) {
				/*
				 * Update the lowest message type.
				 */
				qp->msg_lowest_type = smp->msg_type;
			}
		}
	}
	return (smp);
}

/*
 * msgids system call.
 */
static int
msgids(int *buf, uint_t nids, uint_t *pnids)
{
	int error;

	if (error = ipc_ids(msq_svc, buf, nids, pnids))
		return (set_errno(error));

	return (0);
}

#define	RND(x)		roundup((x), sizeof (size_t))
#define	RND32(x)	roundup((x), sizeof (size32_t))

/*
 * msgsnap system call.
 */
static int
msgsnap(int msqid, caddr_t buf, size_t bufsz, long msgtyp)
{
	struct msg	*mp;	/* ptr to msg on q */
	kmsqid_t	*qp;	/* ptr to associated q */
	kmutex_t	*lock;
	size_t		size;
	size_t		nmsg;
	struct msg	**snaplist;
	int		error, i;
	model_t		mdl = get_udatamodel();
	STRUCT_DECL(msgsnap_head, head);
	STRUCT_DECL(msgsnap_mhead, mhead);

	STRUCT_INIT(head, mdl);
	STRUCT_INIT(mhead, mdl);

	if (bufsz < STRUCT_SIZE(head))
		return (set_errno(EINVAL));

	if ((lock = ipc_lookup(msq_svc, msqid, (kipc_perm_t **)&qp)) == NULL)
		return (set_errno(EINVAL));

	if (error = ipcperm_access(&qp->msg_perm, MSG_R, CRED())) {
		mutex_exit(lock);
		return (set_errno(error));
	}
	ipc_hold(msq_svc, (kipc_perm_t *)qp);

	/*
	 * First compute the required buffer size and
	 * the number of messages on the queue.
	 */
	size = nmsg = 0;
	for (mp = list_head(&qp->msg_list); mp;
	    mp = list_next(&qp->msg_list, mp)) {
		if (msgtyp == 0 ||
		    (msgtyp > 0 && msgtyp == mp->msg_type) ||
		    (msgtyp < 0 && mp->msg_type <= -msgtyp)) {
			nmsg++;
			if (mdl == DATAMODEL_NATIVE)
				size += RND(mp->msg_size);
			else
				size += RND32(mp->msg_size);
		}
	}

	size += STRUCT_SIZE(head) + nmsg * STRUCT_SIZE(mhead);
	if (size > bufsz)
		nmsg = 0;

	if (nmsg > 0) {
		/*
		 * Mark the messages as being copied.
		 */
		snaplist = (struct msg **)kmem_alloc(nmsg *
		    sizeof (struct msg *), KM_SLEEP);
		i = 0;
		for (mp = list_head(&qp->msg_list); mp;
		    mp = list_next(&qp->msg_list, mp)) {
			if (msgtyp == 0 ||
			    (msgtyp > 0 && msgtyp == mp->msg_type) ||
			    (msgtyp < 0 && mp->msg_type <= -msgtyp)) {
				msg_hold(mp);
				snaplist[i] = mp;
				i++;
			}
		}
	}
	mutex_exit(lock);

	/*
	 * Copy out the buffer header.
	 */
	STRUCT_FSET(head, msgsnap_size, size);
	STRUCT_FSET(head, msgsnap_nmsg, nmsg);
	if (copyout(STRUCT_BUF(head), buf, STRUCT_SIZE(head)))
		error = EFAULT;

	buf += STRUCT_SIZE(head);

	/*
	 * Now copy out the messages one by one.
	 */
	for (i = 0; i < nmsg; i++) {
		mp = snaplist[i];
		if (error == 0) {
			STRUCT_FSET(mhead, msgsnap_mlen, mp->msg_size);
			STRUCT_FSET(mhead, msgsnap_mtype, mp->msg_type);
			if (copyout(STRUCT_BUF(mhead), buf, STRUCT_SIZE(mhead)))
				error = EFAULT;
			buf += STRUCT_SIZE(mhead);

			if (error == 0 &&
			    mp->msg_size != 0 &&
			    copyout(mp->msg_addr, buf, mp->msg_size))
				error = EFAULT;
			if (mdl == DATAMODEL_NATIVE)
				buf += RND(mp->msg_size);
			else
				buf += RND32(mp->msg_size);
		}
		lock = ipc_lock(msq_svc, qp->msg_perm.ipc_id);
		msg_rele(mp);
		/* Check for msg q deleted or reallocated */
		if (IPC_FREE(&qp->msg_perm))
			error = EIDRM;
		mutex_exit(lock);
	}

	(void) ipc_lock(msq_svc, qp->msg_perm.ipc_id);
	ipc_rele(msq_svc, (kipc_perm_t *)qp);

	if (nmsg > 0)
		kmem_free(snaplist, nmsg * sizeof (struct msg *));

	if (error)
		return (set_errno(error));
	return (0);
}

#define	MSG_PREALLOC_LIMIT 8192

/*
 * msgsnd system call.
 */
static int
msgsnd(int msqid, struct ipcmsgbuf *msgp, size_t msgsz, int msgflg)
{
	kmsqid_t	*qp;
	kmutex_t	*lock = NULL;
	struct msg	*mp = NULL;