6545064 Posix message queue should fail reliably when a client process crashes

1 files changed, 74 insertions, 16 deletions

diff --git a/usr/src/lib/libc/port/rt/mqueue.c b/usr/src/lib/libc/port/rt/mqueue.c
index 1f9d6bc770..59ff825210 100644
--- a/usr/src/lib/libc/port/rt/mqueue.c
+++ b/usr/src/lib/libc/port/rt/mqueue.c
@@ -82,6 +82,7 @@ typedef struct mq_des {
 	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;
 
 /*
@@ -186,9 +187,10 @@ mq_init(mqhdr_t *mqhp, size_t msgsize, ssize_t maxmsg)
 	/*
 	 * We only need to initialize the non-zero fields.  The use of
 	 * ftruncate() on the message queue file assures that the
-	 * pages will be zfod.
+	 * pages will be zero-filled.
 	 */
-	(void) mutex_init(&mqhp->mq_exclusive, USYNC_PROCESS, NULL);
+	(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);
@@ -335,12 +337,47 @@ mq_putmsg(mqhdr_t *mqhp, const char *msgp, ssize_t len, uint_t prio)
 	}
 }
 
+/*
+ * 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;
-	struct mq_attr	*attr;
+	mode_t		mode = 0;
+	struct mq_attr	*attr = NULL;
 	int		fd;
 	int		err;
 	int		cr_flag = 0;
@@ -498,6 +535,7 @@ mq_open(const char *path, int oflag, /* mode_t mode, mq_attr *attr */ ...)
 	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) {
 		lmutex_lock(&mq_list_lock);
 		mqdp->mqd_next = mq_list;
@@ -538,7 +576,8 @@ mq_close_cleanup(mqdes_t *mqdp)
 
 	/* invalidate the descriptor before freeing it */
 	mqdp->mqd_magic = 0;
-	(void) mutex_unlock(&mqhp->mq_exclusive);
+	if (!mqdp->mqd_ownerdead)
+		(void) mutex_unlock(&mqhp->mq_exclusive);
 
 	lmutex_lock(&mq_list_lock);
 	if (mqdp->mqd_next)
@@ -560,6 +599,7 @@ 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;
@@ -567,7 +607,12 @@ mq_close(mqd_t mqdes)
 	}
 
 	mqhp = mqdp->mqd_mq;
-	(void) mutex_lock(&mqhp->mq_exclusive);
+	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()) {
@@ -670,7 +715,10 @@ __mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len,
 	 * By the time we're here, we know that we've got the capacity
 	 * to add to the queue...now acquire the exclusive lock.
 	 */
-	(void) mutex_lock(&mqhp->mq_exclusive);
+	if ((err = mutex_lock(&mqhp->mq_exclusive)) != 0) {
+		owner_dead(mqdp, err);
+		return (-1);
+	}
 
 	/*
 	 * Now determine if we want to kick the notification.  POSIX
@@ -703,12 +751,7 @@ __mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len,
 
 	if (notify) {
 		/* notify and also delete the registration */
-		(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;
+		do_notify(mqhp);
 	}
 
 	MQ_ASSERT_SEMVAL_LEQ(&mqhp->mq_notempty, ((int)mqhp->mq_maxmsg));
@@ -820,7 +863,10 @@ __mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
 		}
 	}
 
-	(void) mutex_lock(&mqhp->mq_exclusive);
+	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));
@@ -898,6 +944,7 @@ mq_notify(mqd_t mqdes, const struct sigevent *sigevp)
 	int rval = -1;
 	int ntype;
 	int port;
+	int error;
 
 	if (!mq_is_valid(mqdp)) {
 		errno = EBADF;
@@ -906,7 +953,13 @@ mq_notify(mqd_t mqdes, const struct sigevent *sigevp)
 
 	mqhp = mqdp->mqd_mq;
 
-	(void) mutex_lock(&mqhp->mq_exclusive);
+	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 &&
@@ -1012,7 +1065,12 @@ mq_notify(mqd_t mqdes, const struct sigevent *sigevp)
 
 	rval = 0;	/* success */
 bad:
-	(void) mutex_unlock(&mqhp->mq_exclusive);
+	if (error == 0) {
+		(void) mutex_unlock(&mqhp->mq_exclusive);
+	} else {
+		errno = EBADMSG;
+		rval = -1;
+	}
 	return (rval);
 }