/*
* 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 (c) 1993, 2011, Oracle and/or its affiliates. All rights reserved.
*/
#include "lint.h"
#include "mtlib.h"
#define _KMEMUSER
#include <sys/param.h> /* _MQ_OPEN_MAX, _MQ_PRIO_MAX, _SEM_VALUE_MAX */
#undef _KMEMUSER
#include <mqueue.h>
#include <sys/types.h>
#include <sys/file.h>
#include <sys/mman.h>
#include <errno.h>
#include <stdarg.h>
#include <limits.h>
#include <pthread.h>
#include <assert.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <inttypes.h>
#include "sigev_thread.h"
#include "pos4obj.h"
/*
* Default values per message queue
*/
#define MQ_MAXMSG 128
#define MQ_MAXSIZE 1024
#define MQ_MAGIC 0x4d534751 /* "MSGQ" */
/*
* Message header which is part of messages in link list
*/
typedef struct {
uint64_t msg_next; /* offset of next message in the link */
uint64_t msg_len; /* length of the message */
} msghdr_t;
/*
* message queue description
*/
struct mq_dn {
size_t mqdn_flags; /* open description flags */
};
/*
* message queue descriptor structure
*/
typedef struct mq_des {
struct mq_des *mqd_next; /* list of all open mq descriptors, */
struct mq_des *mqd_prev; /* needed for fork-safety */
int mqd_magic; /* magic # to identify mq_des */
int mqd_flags; /* operation flag per open */
struct mq_header *mqd_mq; /* address pointer of message Q */
struct mq_dn *mqd_mqdn; /* open description */
thread_communication_data_t *mqd_tcd; /* SIGEV_THREAD notification */
int mqd_ownerdead; /* mq_exclusive is inconsistent */
} mqdes_t;
/*
* message queue common header, part of the mmap()ed file.
* Since message queues may be shared between 32- and 64-bit processes,
* care must be taken to make sure that the elements of this structure
* are identical for both _LP64 and _ILP32 cases.
*/
typedef struct mq_header {
/* first field must be mq_totsize, DO NOT insert before this */
int64_t mq_totsize; /* total size of the Queue */
int64_t mq_maxsz; /* max size of each message */
uint32_t mq_maxmsg; /* max messages in the queue */
uint32_t mq_maxprio; /* maximum mqueue priority */
uint32_t mq_curmaxprio; /* current maximum MQ priority */
uint32_t mq_mask; /* priority bitmask */
uint64_t mq_freep; /* free message's head pointer */
uint64_t mq_headpp; /* pointer to head pointers */
uint64_t mq_tailpp; /* pointer to tail pointers */
signotify_id_t mq_sigid; /* notification id (3 int's) */
uint32_t mq_ntype; /* notification type (SIGEV_*) */
uint64_t mq_des; /* pointer to msg Q descriptor */
mutex_t mq_exclusive; /* acquire for exclusive access */
sem_t mq_rblocked; /* number of processes rblocked */
sem_t mq_notfull; /* mq_send()'s block on this */
sem_t mq_notempty; /* mq_receive()'s block on this */
sem_t mq_spawner; /* spawner thread blocks on this */
} mqhdr_t;
/*
* The code assumes that _MQ_OPEN_MAX == -1 or "no fixed implementation limit".
* If this assumption is somehow invalidated, mq_open() needs to be changed
* back to the old version which kept a count and enforced a limit.
* We make sure that this is pointed out to those changing <sys/param.h>
* by checking _MQ_OPEN_MAX at compile time.
*/
#if _MQ_OPEN_MAX != -1
#error "mq_open() no longer enforces _MQ_OPEN_MAX and needs fixing."
#endif
#define MQ_ALIGNSIZE 8 /* 64-bit alignment */
#ifdef DEBUG
#define MQ_ASSERT(x) assert(x);
#define MQ_ASSERT_PTR(_m, _p) \
assert((_p) != NULL && !((uintptr_t)(_p) & (MQ_ALIGNSIZE -1)) && \
!((uintptr_t)_m + (uintptr_t)(_p) >= (uintptr_t)_m + \
_m->mq_totsize));
#define MQ_ASSERT_SEMVAL_LEQ(sem, val) { \
int _val; \
(void) sem_getvalue((sem), &_val); \
assert((_val) <= val); }
#else
#define MQ_ASSERT(x)
#define MQ_ASSERT_PTR(_m, _p)
#define MQ_ASSERT_SEMVAL_LEQ(sem, val)
#endif
#define MQ_PTR(m, n) ((msghdr_t *)((uintptr_t)m + (uintptr_t)n))
#define HEAD_PTR(m, n) ((uint64_t *)((uintptr_t)m + \
(uintptr_t)m->mq_headpp + n * sizeof (uint64_t)))
#define TAIL_PTR(m, n) ((uint64_t *)((uintptr_t)m + \
(uintptr_t)m->mq_tailpp + n * sizeof (uint64_t)))
#define MQ_RESERVED ((mqdes_t *)-1)
#define ABS_TIME 0
#define REL_TIME 1
#define MQDES_HSIZE 256
#define MQDES_HASH(x) (((uintptr_t)(x) >> 3) & (MQDES_HSIZE - 1))
static mutex_t mq_list_lock = DEFAULTMUTEX;
static mqdes_t **mq_list = NULL;
extern int __signotify(int cmd, siginfo_t *sigonfo, signotify_id_t *sn_id);
static int
mqdes_register(mqdes_t *mqdp)
{
int ind;
mqdes_t *tmqdp;
if (mq_list == NULL) {
mqdes_t **tp;
size_t sz = sizeof (mqdes_t *) * MQDES_HSIZE;
if ((tp = malloc(sz)) == NULL)
return (-1);
lmutex_lock(&mq_list_lock);
if (mq_list != NULL) {
lmutex_unlock(&mq_list_lock);
free(tp);
lmutex_lock(&mq_list_lock);
} else {
(void) memset(tp, 0, sz);
mq_list = tp;
}
} else {
lmutex_lock(&mq_list_lock);
}
/*
* link into hash.
*/
ind = MQDES_HASH(mqdp);
tmqdp = mq_list[ind];
mqdp->mqd_next = tmqdp;
mqdp->mqd_prev = NULL;
if (tmqdp != NULL)
tmqdp->mqd_prev = mqdp;
mq_list[ind] = mqdp;
lmutex_unlock(&mq_list_lock);
return (0);
}
static void
mqdes_unregister(mqdes_t *mqdp)
{
int ind;
lmutex_lock(&mq_list_lock);
if (mqdp->mqd_next != NULL)
mqdp->mqd_next->mqd_prev = mqdp->mqd_prev;
if (mqdp->mqd_prev != NULL)
mqdp->mqd_prev->mqd_next = mqdp->mqd_next;
ind = MQDES_HASH(mqdp);
if (mq_list[ind] == mqdp)
mq_list[ind] = mqdp->mqd_next;
lmutex_unlock(&mq_list_lock);
}
static int
mqdes_is_valid(mqdes_t *mqdp)
{
mqdes_t *tp;
lmutex_lock(&mq_list_lock);
if (mq_list == NULL) {
lmutex_unlock(&mq_list_lock);
return (0);
}
for (tp = mq_list[MQDES_HASH(mqdp)]; tp != NULL; tp = tp->mqd_next) {
if (tp == mqdp)
break;
}
lmutex_unlock(&mq_list_lock);
return (tp != NULL);
}
static int
mq_is_valid(mqdes_t *mqdp)
{
/*
* Any use of a message queue after it was closed is
* undefined. But the standard strongly favours EBADF
* returns.
*/
if (mqdp != NULL && mqdp != MQ_RESERVED && mqdes_is_valid(mqdp)) {
return (mqdp->mqd_magic == MQ_MAGIC);
}
return (0);
}
static void
mq_init(mqhdr_t *mqhp, size_t msgsize, ssize_t maxmsg)
{
int i;
uint64_t temp;
uint64_t currentp;
uint64_t nextp;
/*
* We only need to initialize the non-zero fields. The use of
* ftruncate() on the message queue file assures that the
* pages will be zero-filled.
*/
(void) mutex_init(&mqhp->mq_exclusive,
USYNC_PROCESS | LOCK_ROBUST, NULL);
(void) sem_init(&mqhp->mq_rblocked, 1, 0);
(void) sem_init(&mqhp->mq_notempty, 1, 0);
(void) sem_init(&mqhp->mq_spawner, 1, 0);
(void) sem_init(&mqhp->mq_notfull, 1, (uint_t)maxmsg);
mqhp->mq_maxsz = msgsize;
mqhp->mq_maxmsg = maxmsg;
/*
* As of this writing (1997), there are 32 message queue priorities.
* If this is to change, then the size of the mq_mask will
* also have to change. If DEBUG is defined, assert that
* _MQ_PRIO_MAX hasn't changed.
*/
mqhp->mq_maxprio = _MQ_PRIO_MAX;
#if defined(DEBUG)
/* LINTED always true */
MQ_ASSERT(sizeof (mqhp->mq_mask) * 8 >= _MQ_PRIO_MAX);
#endif
/*
* Since the message queue can be mapped into different
* virtual address ranges by different processes, we don't
* keep track of pointers, only offsets into the shared region.
*/
mqhp->mq_headpp = sizeof (mqhdr_t);
mqhp->mq_tailpp = mqhp->mq_headpp +
mqhp->mq_maxprio * sizeof (uint64_t);
mqhp->mq_freep = mqhp->mq_tailpp +
mqhp->mq_maxprio * sizeof (uint64_t);
currentp = mqhp->mq_freep;
MQ_PTR(mqhp, currentp)->msg_next = 0;
temp = (mqhp->mq_maxsz + MQ_ALIGNSIZE - 1) & ~(MQ_ALIGNSIZE - 1);
for (i = 1; i < mqhp->mq_maxmsg; i++) {
nextp = currentp + sizeof (msghdr_t) + temp;
MQ_PTR(mqhp, currentp)->msg_next = nextp;
MQ_PTR(mqhp, nextp)->msg_next = 0;
currentp = nextp;
}
}
static size_t
mq_getmsg(mqhdr_t *mqhp, char *msgp, uint_t *msg_prio)
{
uint64_t currentp;
msghdr_t *curbuf;
uint64_t *headpp;
uint64_t *tailpp;
MQ_ASSERT(MUTEX_HELD(&mqhp->mq_exclusive));
/*
* Get the head and tail pointers for the queue of maximum
* priority. We shouldn't be here unless there is a message for
* us, so it's fair to assert that both the head and tail
* pointers are non-NULL.
*/
headpp = HEAD_PTR(mqhp, mqhp->mq_curmaxprio);
tailpp = TAIL_PTR(mqhp, mqhp->mq_curmaxprio);
if (msg_prio != NULL)
*msg_prio = mqhp->mq_curmaxprio;
currentp = *headpp;
MQ_ASSERT_PTR(mqhp, currentp);
curbuf = MQ_PTR(mqhp, currentp);
if ((*headpp = curbuf->msg_next) == NULL) {
/*
* We just nuked the last message in this priority's queue.
* Twiddle this priority's bit, and then find the next bit
* tipped.
*/
uint_t prio = mqhp->mq_curmaxprio;
mqhp->mq_mask &= ~(1u << prio);
for (; prio != 0; prio--)
if (mqhp->mq_mask & (1u << prio))
break;
mqhp->mq_curmaxprio = prio;
*tailpp = NULL;
}
/*
* Copy the message, and put the buffer back on the free list.
*/
(void) memcpy(msgp, (char *)&curbuf[1], curbuf->msg_len);
curbuf->msg_next = mqhp->mq_freep;
mqhp->mq_freep = currentp;
return (curbuf->msg_len);
}
static void
mq_putmsg(mqhdr_t *mqhp, const char *msgp, ssize_t len, uint_t prio)
{
uint64_t currentp;
msghdr_t *curbuf;
uint64_t *headpp;
uint64_t *tailpp;
MQ_ASSERT(MUTEX_HELD(&mqhp->mq_exclusive));
/*
* Grab a free message block, and link it in. We shouldn't
* be here unless there is room in the queue for us; it's
* fair to assert that the free pointer is non-NULL.
*/
currentp = mqhp->mq_freep;
MQ_ASSERT_PTR(mqhp, currentp);
curbuf = MQ_PTR(mqhp, currentp);
/*
* Remove a message from the free list, and copy in the new contents.
*/
mqhp->mq_freep = curbuf->msg_next;
curbuf->msg_next = NULL;
(void) memcpy((char *)&curbuf[1], msgp, len);
curbuf->msg_len = len;
headpp = HEAD_PTR(mqhp, prio);
tailpp = TAIL_PTR(mqhp, prio);
if (*tailpp == 0) {
/*
* This is the first message on this queue. Set the
* head and tail pointers, and tip the appropriate bit
* in the priority mask.
*/
*headpp = currentp;
*tailpp = currentp;
mqhp->mq_mask |= (1u << prio);
if (prio > mqhp->mq_curmaxprio)
mqhp->mq_curmaxprio = prio;
} else {
MQ_ASSERT_PTR(mqhp, *tailpp);
MQ_PTR(mqhp, *tailpp)->msg_next = currentp;
*tailpp = currentp;
}
}
/*
* Send a notification and also delete the registration.
*/
static void
do_notify(mqhdr_t *mqhp)
{
(void) __signotify(SN_SEND, NULL, &mqhp->mq_sigid);
if (mqhp->mq_ntype == SIGEV_THREAD ||
mqhp->mq_ntype == SIGEV_PORT)
(void) sem_post(&mqhp->mq_spawner);
mqhp->mq_ntype = 0;
mqhp->mq_des = 0;
}
/*
* Called when the mq_exclusive lock draws EOWNERDEAD or ENOTRECOVERABLE.
* Wake up anyone waiting on mq_*send() or mq_*receive() and ensure that
* they fail with errno == EBADMSG. Trigger any registered notification.
*/
static void
owner_dead(mqdes_t *mqdp, int error)
{
mqhdr_t *mqhp = mqdp->mqd_mq;
mqdp->mqd_ownerdead = 1;
(void) sem_post(&mqhp->mq_notfull);
(void) sem_post(&mqhp->mq_notempty);
if (error == EOWNERDEAD) {
if (mqhp->mq_sigid.sn_pid != 0)
do_notify(mqhp);
(void) mutex_unlock(&mqhp->mq_exclusive);
}
errno = EBADMSG;
}
mqd_t
mq_open(const char *path, int oflag, /* mode_t mode, mq_attr *attr */ ...)
{
va_list ap;
mode_t mode = 0;
struct mq_attr *attr = NULL;
int fd;
int err;
int cr_flag = 0;
int locked = 0;
uint64_t total_size;
size_t msgsize;
ssize_t maxmsg;
uint64_t temp;
void *ptr;
mqdes_t *mqdp;
mqhdr_t *mqhp;
struct mq_dn *mqdnp;
if (__pos4obj_check(path) == -1)
return ((mqd_t)-1);
/* acquire MSGQ lock to have atomic operation */
if (__pos4obj_lock(path, MQ_LOCK_TYPE) < 0)
goto out;
locked = 1;
va_start(ap, oflag);
/* filter oflag to have READ/WRITE/CREATE modes only */
oflag = oflag & (O_RDONLY|O_WRONLY|O_RDWR|O_CREAT|O_EXCL|O_NONBLOCK);
if ((oflag & O_CREAT) != 0) {
mode = va_arg(ap, mode_t);
attr = va_arg(ap, struct mq_attr *);
}
va_end(ap);
if ((fd = __pos4obj_open(path, MQ_PERM_TYPE, oflag,
mode, &cr_flag)) < 0)
goto out;
/* closing permission file */
(void) __close_nc(fd);
/* Try to open/create data file */
if (cr_flag) {
cr_flag = PFILE_CREATE;
if (attr == NULL) {
maxmsg = MQ_MAXMSG;
msgsize = MQ_MAXSIZE;
} else if (attr->mq_maxmsg <= 0 || attr->mq_msgsize <= 0) {
errno = EINVAL;
goto out;
} else if (attr->mq_maxmsg > _SEM_VALUE_MAX) {
errno = ENOSPC;
goto out;
} else {
maxmsg = attr->mq_maxmsg;
msgsize = attr->mq_msgsize;
}
/* adjust for message size at word boundary */
temp = (msgsize + MQ_ALIGNSIZE - 1) & ~(MQ_ALIGNSIZE - 1);
total_size = sizeof (mqhdr_t) +
maxmsg * (temp + sizeof (msghdr_t)) +
2 * _MQ_PRIO_MAX * sizeof (uint64_t);
if (total_size > SSIZE_MAX) {
errno = ENOSPC;
goto out;
}
/*
* data file is opened with read/write to those
* who have read or write permission
*/
mode = mode | (mode & 0444) >> 1 | (mode & 0222) << 1;
if ((fd = __pos4obj_open(path, MQ_DATA_TYPE,
(O_RDWR|O_CREAT|O_EXCL), mode, &err)) < 0)
goto out;
cr_flag |= DFILE_CREATE | DFILE_OPEN;
/* force permissions to avoid umask effect */
if (fchmod(fd, mode) < 0)
goto out;
if (ftruncate64(fd, (off64_t)total_size) < 0)
goto out;
} else {
if ((fd = __pos4obj_open(path, MQ_DATA_TYPE,
O_RDWR, 0666, &err)) < 0)
goto out;
cr_flag = DFILE_OPEN;
/* Message queue has not been initialized yet */
if (read(fd, &total_size, sizeof (total_size)) !=
sizeof (total_size) || total_size == 0) {
errno = ENOENT;
goto out;
}
/* Message queue too big for this process to handle */
if (total_size > SSIZE_MAX) {
errno = EFBIG;
goto out;
}
}
if ((mqdp = (mqdes_t *)malloc(sizeof (mqdes_t))) == NULL) {
errno = ENOMEM;
goto out;
}
cr_flag |= ALLOC_MEM;
if ((ptr = mmap64(NULL, total_size, PROT_READ|PROT_WRITE,
MAP_SHARED, fd, (off64_t)0)) == MAP_FAILED)
goto out;
mqhp = ptr;
cr_flag |= DFILE_MMAP;
/* closing data file */
(void) __close_nc(fd);
cr_flag &= ~DFILE_OPEN;
/*
* create, unlink, size, mmap, and close description file
* all for a flag word in anonymous shared memory
*/
if ((fd = __pos4obj_open(path, MQ_DSCN_TYPE, O_RDWR | O_CREAT,
0666, &err)) < 0)
goto out;
cr_flag |= DFILE_OPEN;
(void) __pos4obj_unlink(path, MQ_DSCN_TYPE);
if (ftruncate64(fd, (off64_t)sizeof (struct mq_dn)) < 0)
goto out;
if ((ptr = mmap64(NULL, sizeof (struct mq_dn),
PROT_READ | PROT_WRITE, MAP_SHARED, fd, (off64_t)0)) == MAP_FAILED)
goto out;
mqdnp = ptr;
cr_flag |= MQDNP_MMAP;
(void) __close_nc(fd);
cr_flag &= ~DFILE_OPEN;
if (mqdes_register(mqdp) != 0)
goto out;
/*
* we follow the same strategy as filesystem open() routine,
* where fcntl.h flags are changed to flags defined in file.h.
*/
mqdp->mqd_flags = (oflag - FOPEN) & (FREAD|FWRITE);
mqdnp->mqdn_flags = (oflag - FOPEN) & (FNONBLOCK);
/* new message queue requires initialization */
if ((cr_flag & DFILE_CREATE) != 0) {
/* message queue header has to be initialized */
mq_init(mqhp, msgsize, maxmsg);
mqhp->mq_totsize = total_size;
}
mqdp->mqd_mq = mqhp;
mqdp->mqd_mqdn = mqdnp;
mqdp->mqd_magic = MQ_MAGIC;
mqdp->mqd_tcd = NULL;
mqdp->mqd_ownerdead = 0;
if (__pos4obj_unlock(path, MQ_LOCK_TYPE) == 0)
return ((mqd_t)mqdp);
mqdes_unregister(mqdp);
locked = 0; /* fall into the error case */
out:
err = errno;
if ((cr_flag & DFILE_OPEN) != 0)
(void) __close_nc(fd);
if ((cr_flag & DFILE_CREATE) != 0)
(void) __pos4obj_unlink(path, MQ_DATA_TYPE);
if ((cr_flag & PFILE_CREATE) != 0)
(void) __pos4obj_unlink(path, MQ_PERM_TYPE);
if ((cr_flag & ALLOC_MEM) != 0)
free((void *)mqdp);
if ((cr_flag & DFILE_MMAP) != 0)
(void) munmap((caddr_t)mqhp, (size_t)total_size);
if ((cr_flag & MQDNP_MMAP) != 0)
(void) munmap((caddr_t)mqdnp, sizeof (struct mq_dn));
if (locked)
(void) __pos4obj_unlock(path, MQ_LOCK_TYPE);
errno = err;
return ((mqd_t)-1);
}
static void
mq_close_cleanup(mqdes_t *mqdp)
{
mqhdr_t *mqhp = mqdp->mqd_mq;
struct mq_dn *mqdnp = mqdp->mqd_mqdn;
/* invalidate the descriptor before freeing it */
mqdp->mqd_magic = 0;
if (!mqdp->mqd_ownerdead)
(void) mutex_unlock(&mqhp->mq_exclusive);
mqdes_unregister(mqdp);
free(mqdp);
(void) munmap((caddr_t)mqdnp, sizeof (struct mq_dn));
(void) munmap((caddr_t)mqhp, (size_t)mqhp->mq_totsize);
}
int
mq_close(mqd_t mqdes)
{
mqdes_t *mqdp = (mqdes_t *)mqdes;
mqhdr_t *mqhp;
thread_communication_data_t *tcdp;
int error;
if (!mq_is_valid(mqdp)) {
errno = EBADF;
return (-1);
}
mqhp = mqdp->mqd_mq;
if ((error = mutex_lock(&mqhp->mq_exclusive)) != 0) {
mqdp->mqd_ownerdead = 1;
if (error == EOWNERDEAD)
(void) mutex_unlock(&mqhp->mq_exclusive);
/* carry on regardless, without holding mq_exclusive */
}
if (mqhp->mq_des == (uintptr_t)mqdp &&
mqhp->mq_sigid.sn_pid == getpid()) {
/* notification is set for this descriptor, remove it */
(void) __signotify(SN_CANCEL, NULL, &mqhp->mq_sigid);
mqhp->mq_ntype = 0;
mqhp->mq_des = 0;
}
pthread_cleanup_push(mq_close_cleanup, mqdp);
if ((tcdp = mqdp->mqd_tcd) != NULL) {
mqdp->mqd_tcd = NULL;
del_sigev_mq(tcdp); /* possible cancellation point */
}
pthread_cleanup_pop(1); /* finish in the cleanup handler */
return (0);
}
int
mq_unlink(const char *path)
{
int err;
if (__pos4obj_check(path) < 0)
return (-1);
if (__pos4obj_lock(path, MQ_LOCK_TYPE) < 0) {
return (-1);
}
err = __pos4obj_unlink(path, MQ_PERM_TYPE);
if (err == 0 || (err == -1 && errno == EEXIST)) {
errno = 0;
err = __pos4obj_unlink(path, MQ_DATA_TYPE);
}
if (__pos4obj_unlock(path, MQ_LOCK_TYPE) < 0)
return (-1);
return (err);
}
static int
__mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len,
uint_t msg_prio, const timespec_t *timeout, int abs_rel)
{
mqdes_t *mqdp = (mqdes_t *)mqdes;
mqhdr_t *mqhp;
int err;
int notify = 0;
/*
* sem_*wait() does cancellation, if called.
* pthread_testcancel() ensures that cancellation takes place if
* there is a cancellation pending when mq_*send() is called.
*/
pthread_testcancel();
if (!mq_is_valid(mqdp) || (mqdp->mqd_flags & FWRITE) == 0) {
errno = EBADF;
return (-1);
}
mqhp = mqdp->mqd_mq;
if (msg_prio >= mqhp->mq_maxprio) {
errno = EINVAL;
return (-1);
}
if (msg_len > mqhp->mq_maxsz) {
errno = EMSGSIZE;
return (-1);
}
if (mqdp->mqd_mqdn->mqdn_flags & O_NONBLOCK)
err = sem_trywait(&mqhp->mq_notfull);
else {
/*
* We might get cancelled here...
*/
if (timeout == NULL)
err = sem_wait(&mqhp->mq_notfull);
else if (abs_rel == ABS_TIME)
err = sem_timedwait(&mqhp->mq_notfull, timeout);
else
err = sem_reltimedwait_np(&mqhp->mq_notfull, timeout);
}
if (err == -1) {
/*
* errno has been set to EAGAIN / EINTR / ETIMEDOUT
* by sem_*wait(), so we can just return.
*/
return (-1);
}
/*
* By the time we're here, we know that we've got the capacity
* to add to the queue...now acquire the exclusive lock.
*/
if ((err = mutex_lock(&mqhp->mq_exclusive)) != 0) {
owner_dead(mqdp, err);
return (-1);
}
/*
* Now determine if we want to kick the notification. POSIX
* requires that if a process has registered for notification,
* we must kick it when the queue makes an empty to non-empty
* transition, and there are no blocked receivers. Note that
* this mechanism does _not_ guarantee that the kicked process
* will be able to receive a message without blocking;
* another receiver could intervene in the meantime. Thus,
* the notification mechanism is inherently racy; all we can
* do is hope to minimize the window as much as possible.
* In general, we want to avoid kicking the notification when
* there are clearly receivers blocked. We'll determine if
* we want to kick the notification before the mq_putmsg(),
* but the actual signotify() won't be done until the message
* is on the queue.
*/
if (mqhp->mq_sigid.sn_pid != 0) {
int nmessages, nblocked;
(void) sem_getvalue(&mqhp->mq_notempty, &nmessages);
(void) sem_getvalue(&mqhp->mq_rblocked, &nblocked);
if (nmessages == 0 && nblocked == 0)
notify = 1;
}
mq_putmsg(mqhp, msg_ptr, (ssize_t)msg_len, msg_prio);
(void) sem_post(&mqhp->mq_notempty);
if (notify) {
/* notify and also delete the registration */
do_notify(mqhp);
}
MQ_ASSERT_SEMVAL_LEQ(&mqhp->mq_notempty, ((int)mqhp->mq_maxmsg));
(void) mutex_unlock(&mqhp->mq_exclusive);
return (0);
}
int
mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, uint_t msg_prio)
{
return (__mq_timedsend(mqdes, msg_ptr, msg_len, msg_prio,
NULL, ABS_TIME));
}
int
mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len,
uint_t msg_prio, const timespec_t *abs_timeout)
{
return (__mq_timedsend(mqdes, msg_ptr, msg_len, msg_prio,
abs_timeout, ABS_TIME));
}
int
mq_reltimedsend_np(mqd_t mqdes, const char *msg_ptr, size_t msg_len,
uint_t msg_prio, const timespec_t *rel_timeout)
{
return (__mq_timedsend(mqdes, msg_ptr, msg_len, msg_prio,
rel_timeout, REL_TIME));
}
static void
decrement_rblocked(mqhdr_t *mqhp)
{
int cancel_state;
(void) pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cancel_state);
while (sem_wait(&mqhp->mq_rblocked) == -1)
continue;
(void) pthread_setcancelstate(cancel_state, NULL);
}
static ssize_t
__mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
uint_t *msg_prio, const timespec_t *timeout, int abs_rel)
{
mqdes_t *mqdp = (mqdes_t *)mqdes;
mqhdr_t *mqhp;
ssize_t msg_size;
int err;
/*
* sem_*wait() does cancellation, if called.
* pthread_testcancel() ensures that cancellation takes place if
* there is a cancellation pending when mq_*receive() is called.
*/
pthread_testcancel();
if (!mq_is_valid(mqdp) || (mqdp->mqd_flags & FREAD) == 0) {
errno = EBADF;
return (ssize_t)(-1);
}
mqhp = mqdp->mqd_mq;
if (msg_len < mqhp->mq_maxsz) {
errno = EMSGSIZE;
return (ssize_t)(-1);
}
/*
* The semaphoring scheme for mq_[timed]receive is a little hairy
* thanks to POSIX.1b's arcane notification mechanism. First,
* we try to take the common case and do a sem_trywait().
* If that doesn't work, and O_NONBLOCK hasn't been set,
* then note that we're going to sleep by incrementing the rblocked
* semaphore. We decrement that semaphore after waking up.
*/
if (sem_trywait(&mqhp->mq_notempty) == -1) {
if ((mqdp->mqd_mqdn->mqdn_flags & O_NONBLOCK) != 0) {
/*
* errno has been set to EAGAIN or EINTR by
* sem_trywait(), so we can just return.
*/
return (-1);
}
/*
* If we're here, then we're probably going to block...
* increment the rblocked semaphore. If we get
* cancelled, decrement_rblocked() will decrement it.
*/
(void) sem_post(&mqhp->mq_rblocked);
pthread_cleanup_push(decrement_rblocked, mqhp);
if (timeout == NULL)
err = sem_wait(&mqhp->mq_notempty);
else if (abs_rel == ABS_TIME)
err = sem_timedwait(&mqhp->mq_notempty, timeout);
else
err = sem_reltimedwait_np(&mqhp->mq_notempty, timeout);
pthread_cleanup_pop(1);
if (err == -1) {
/*
* We took a signal or timeout while waiting
* on mq_notempty...
*/
return (-1);
}
}
if ((err = mutex_lock(&mqhp->mq_exclusive)) != 0) {
owner_dead(mqdp, err);
return (-1);
}
msg_size = mq_getmsg(mqhp, msg_ptr, msg_prio);
(void) sem_post(&mqhp->mq_notfull);
MQ_ASSERT_SEMVAL_LEQ(&mqhp->mq_notfull, ((int)mqhp->mq_maxmsg));
(void) mutex_unlock(&mqhp->mq_exclusive);
return (msg_size);
}
ssize_t
mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, uint_t *msg_prio)
{
return (__mq_timedreceive(mqdes, msg_ptr, msg_len, msg_prio,
NULL, ABS_TIME));
}
ssize_t
mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
uint_t *msg_prio, const timespec_t *abs_timeout)
{
return (__mq_timedreceive(mqdes, msg_ptr, msg_len, msg_prio,
abs_timeout, ABS_TIME));
}
ssize_t
mq_reltimedreceive_np(mqd_t mqdes, char *msg_ptr, size_t msg_len,
uint_t *msg_prio, const timespec_t *rel_timeout)
{
return (__mq_timedreceive(mqdes, msg_ptr, msg_len, msg_prio,
rel_timeout, REL_TIME));
}
/*
* Only used below, in mq_notify().
* We already have a spawner thread.
* Verify that the attributes match; cancel it if necessary.
*/
static int
cancel_if_necessary(thread_communication_data_t *tcdp,
const struct sigevent *sigevp)
{
int do_cancel = !pthread_attr_equal(tcdp->tcd_attrp,
sigevp->sigev_notify_attributes);
if (do_cancel) {
/*
* Attributes don't match, cancel the spawner thread.
*/
(void) pthread_cancel(tcdp->tcd_server_id);
} else {
/*
* Reuse the existing spawner thread with possibly
* changed notification function and value.
*/
tcdp->tcd_notif.sigev_notify = SIGEV_THREAD;
tcdp->tcd_notif.sigev_signo = 0;
tcdp->tcd_notif.sigev_value = sigevp->sigev_value;
tcdp->tcd_notif.sigev_notify_function =
sigevp->sigev_notify_function;
}
return (do_cancel);
}
int
mq_notify(mqd_t mqdes, const struct sigevent *sigevp)
{
mqdes_t *mqdp = (mqdes_t *)mqdes;
mqhdr_t *mqhp;
thread_communication_data_t *tcdp;
siginfo_t mq_siginfo;
struct sigevent sigevent;
struct stat64 statb;
port_notify_t *pn;
void *userval;
int rval = -1;
int ntype;
int port;
int error;
if (!mq_is_valid(mqdp)) {
errno = EBADF;
return (-1);
}
mqhp = mqdp->mqd_mq;
if ((error = mutex_lock(&mqhp->mq_exclusive)) != 0) {
mqdp->mqd_ownerdead = 1;
sigevp = NULL;
if (error == EOWNERDEAD)
(void) mutex_unlock(&mqhp->mq_exclusive);
/* carry on regardless, without holding mq_exclusive */
}
if (sigevp == NULL) { /* remove notification */
if (mqhp->mq_des == (uintptr_t)mqdp &&
mqhp->mq_sigid.sn_pid == getpid()) {
/* notification is set for this descriptor, remove it */
(void) __signotify(SN_CANCEL, NULL, &mqhp->mq_sigid);
if ((tcdp = mqdp->mqd_tcd) != NULL) {
sig_mutex_lock(&tcdp->tcd_lock);
if (tcdp->tcd_msg_enabled) {
/* cancel the spawner thread */
tcdp = mqdp->mqd_tcd;
mqdp->mqd_tcd = NULL;
(void) pthread_cancel(
tcdp->tcd_server_id);
}
sig_mutex_unlock(&tcdp->tcd_lock);
}
mqhp->mq_ntype = 0;
mqhp->mq_des = 0;
} else {
/* notification is not set for this descriptor */
errno = EBUSY;
goto bad;
}
} else { /* register notification with this process */
switch (ntype = sigevp->sigev_notify) {
case SIGEV_THREAD:
userval = sigevp->sigev_value.sival_ptr;
port = -1;
break;
case SIGEV_PORT:
pn = sigevp->sigev_value.sival_ptr;
userval = pn->portnfy_user;
port = pn->portnfy_port;
if (fstat64(port, &statb) != 0 ||
!S_ISPORT(statb.st_mode)) {
errno = EBADF;
goto bad;
}
(void) memset(&sigevent, 0, sizeof (sigevent));
sigevent.sigev_notify = SIGEV_PORT;
sigevp = &sigevent;
break;
}
switch (ntype) {
case SIGEV_NONE:
mq_siginfo.si_signo = 0;
mq_siginfo.si_code = SI_MESGQ;
break;
case SIGEV_SIGNAL:
mq_siginfo.si_signo = sigevp->sigev_signo;
mq_siginfo.si_value = sigevp->sigev_value;
mq_siginfo.si_code = SI_MESGQ;
break;
case SIGEV_THREAD:
if ((tcdp = mqdp->mqd_tcd) != NULL &&
cancel_if_necessary(tcdp, sigevp))
mqdp->mqd_tcd = NULL;
/* FALLTHROUGH */
case SIGEV_PORT:
if ((tcdp = mqdp->mqd_tcd) == NULL) {
/* we must create a spawner thread */
tcdp = setup_sigev_handler(sigevp, MQ);
if (tcdp == NULL) {
errno = EBADF;
goto bad;
}
tcdp->tcd_msg_enabled = 0;
tcdp->tcd_msg_closing = 0;
tcdp->tcd_msg_avail = &mqhp->mq_spawner;
if (launch_spawner(tcdp) != 0) {
free_sigev_handler(tcdp);
goto bad;
}
mqdp->mqd_tcd = tcdp;
}
mq_siginfo.si_signo = 0;
mq_siginfo.si_code = SI_MESGQ;
break;
default:
errno = EINVAL;
goto bad;
}
/* register notification */
if (__signotify(SN_PROC, &mq_siginfo, &mqhp->mq_sigid) < 0)
goto bad;
mqhp->mq_ntype = ntype;
mqhp->mq_des = (uintptr_t)mqdp;
switch (ntype) {
case SIGEV_THREAD:
case SIGEV_PORT:
tcdp->tcd_port = port;
tcdp->tcd_msg_object = mqdp;
tcdp->tcd_msg_userval = userval;
sig_mutex_lock(&tcdp->tcd_lock);
tcdp->tcd_msg_enabled = ntype;
sig_mutex_unlock(&tcdp->tcd_lock);
(void) cond_broadcast(&tcdp->tcd_cv);
break;
}
}
rval = 0; /* success */
bad:
if (error == 0) {
(void) mutex_unlock(&mqhp->mq_exclusive);
} else {
errno = EBADMSG;
rval = -1;
}
return (rval);
}
int
mq_setattr(mqd_t mqdes, const struct mq_attr *mqstat, struct mq_attr *omqstat)
{
mqdes_t *mqdp = (mqdes_t *)mqdes;
mqhdr_t *mqhp;
uint_t flag = 0;
if (!mq_is_valid(mqdp)) {
errno = EBADF;
return (-1);
}
/* store current attributes */
if (omqstat != NULL) {
int count;
mqhp = mqdp->mqd_mq;
omqstat->mq_flags = mqdp->mqd_mqdn->mqdn_flags;
omqstat->mq_maxmsg = (long)mqhp->mq_maxmsg;
omqstat->mq_msgsize = (long)mqhp->mq_maxsz;
(void) sem_getvalue(&mqhp->mq_notempty, &count);
omqstat->mq_curmsgs = count;
}
/* set description attributes */
if ((mqstat->mq_flags & O_NONBLOCK) != 0)
flag = FNONBLOCK;
mqdp->mqd_mqdn->mqdn_flags = flag;
return (0);
}
int
mq_getattr(mqd_t mqdes, struct mq_attr *mqstat)
{
mqdes_t *mqdp = (mqdes_t *)mqdes;
mqhdr_t *mqhp;
int count;
if (!mq_is_valid(mqdp)) {
errno = EBADF;
return (-1);
}
mqhp = mqdp->mqd_mq;
mqstat->mq_flags = mqdp->mqd_mqdn->mqdn_flags;
mqstat->mq_maxmsg = (long)mqhp->mq_maxmsg;
mqstat->mq_msgsize = (long)mqhp->mq_maxsz;
(void) sem_getvalue(&mqhp->mq_notempty, &count);
mqstat->mq_curmsgs = count;
return (0);
}
/*
* Cleanup after fork1() in the child process.
*/
void
postfork1_child_sigev_mq(void)
{
thread_communication_data_t *tcdp;
mqdes_t *mqdp;
int i;
if (mq_list == NULL)
return;
for (i = 0; i < MQDES_HSIZE; i++) {
for (mqdp = mq_list[i]; mqdp != NULL; mqdp = mqdp->mqd_next) {
if ((tcdp = mqdp->mqd_tcd) != NULL) {
mqdp->mqd_tcd = NULL;
tcd_teardown(tcdp);
}
}
}
}