V1 1/2] ipc: let message queues use SIGEV_THREAD_ID with mq_notify

From: Steven Stewart-Gallus <[email protected]>

Currently the only thread-safe way of using mq_notify with message queues is to use the SIGEV_THREAD option. Unfortunately, existing wrappers around such functionality spawn a thread each time a notification happens instead of caching threads which is slow and inconvenient for debugging. This change lets message queues use SIGEV_THREAD_ID with mq_notify and so target notifications to only one thread at a time which is thread-safe.

---

Hello! This is a simple patch I made more for the purpose of learning how the kernel works then for any serious purpose. Currently, in my own program I now simply use a thread pool and poll for asynchronously running message sends and receives but a thread-safe mq_notify would have been useful for me and possibly saved me some effort. So, I won't be seriously bothered if this patch is rejected but I think other people might find it useful nonetheless. I also think this behaviour seems quite natural and appropriate to support.

I am sending this patch for review but I am pretty sure it is mistaken in some ways. In particular, I believe this patch is wrong because I don't believe it filters out thread ids that belong to other processes and I am not sure how to do that.

Thank you,
Steven Stewart-Gallus

index a71aac4..41a1b96 100644
@@ -95,6 +95,16 @@ as if it were the start function of a new thread.
 See
 .BR sigevent (7)
 for details.
+.TP
+.BR SIGEV_THREAD_ID " (Linux-specific)"
+As for
+.BR SIGEV_SIGNAL ,
+but the signal is targeted at the thread whose ID is given in
+.IR sigev_notify_thread_id ,
+which must be a thread in the same process as the caller.
+See
+.BR sigevent (7)
+for general details.
 .PP
 Only one process can be registered to receive notification
 from a message queue.

index 9cec77e..15da9fe 100644
@@ -125,8 +125,10 @@ structure that defines attributes for the new thread (see
 .TP
 .BR SIGEV_THREAD_ID " (Linux-specific)"
 .\" | SIGEV_SIGNAL vs not?
-Currently used only by POSIX timers; see
-.BR timer_create (2).
+Currently used only by POSIX timers and message queues; see
+.BR timer_create (2)
+and
+.BR mq_notify (2).
 .SH CONFORMING TO
 POSIX.1-2001.
 .SH SEE ALSO

index 4fcf39a..8ca70a2 100644
@@ -88,7 +88,7 @@ struct mqueue_inode_info {
 static const struct inode_operations mqueue_dir_inode_operations;
 static const struct file_operations mqueue_file_operations;
 static const struct super_operations mqueue_super_ops;
-static void remove_notification(struct mqueue_inode_info *info);
+static bool maybe_remove_notification(struct mqueue_inode_info *info);
 
 static struct kmem_cache *mqueue_inode_cachep;
 

@@ -495,7 +495,8 @@ static ssize_t mqueue_read_file(struct file *filp, char

__user *u_data,
info->qsize,
info->notify_owner ? info->notify.sigev_notify : 0, (info->notify_owner &&
- info->notify.sigev_notify == SIGEV_SIGNAL) ?
+ (info->notify.sigev_notify == SIGEV_SIGNAL ||
+ info->notify.sigev_notify == SIGEV_THREAD_ID)) ?
info->notify.sigev_signo : 0,
pid_vnr(info->notify_owner));
spin_unlock(&info->lock);
@@ -515,8 +516,8 @@ static int mqueue_flush_file(struct file *filp, fl_owner_t id)
struct mqueue_inode_info *info = MQUEUE_I(file_inode(filp));

spin_lock(&info->lock);
- if (task_tgid(current) == info->notify_owner)
- remove_notification(info);
+
+ maybe_remove_notification(info);

spin_unlock(&info->lock);
return 0;
@@ -642,6 +643,7 @@ static void __do_notify(struct mqueue_inode_info *info)
case SIGEV_NONE:
break;
case SIGEV_SIGNAL:
+ case SIGEV_THREAD_ID:
/* sends signal */

sig_i.si_signo = info->notify.sigev_signo;
@@ -684,8 +686,24 @@ static int prepare_timeout(const struct timespec __user *u_abs_timeout,
return 0;
}

-static void remove_notification(struct mqueue_inode_info *info) +static bool maybe_remove_notification(struct mqueue_inode_info *info)
{
+ struct pid *pid;
+
+ switch (info->notify.sigev_notify) {
+ case SIGEV_THREAD_ID:
+ pid = task_pid(current);
+ break;
+
+ default:
+ pid = task_tgid(current);
+ break;
+ }
+
+ if (pid != info->notify_owner) {
+ return false;
+ }
+
if (info->notify_owner != NULL &&
info->notify.sigev_notify == SIGEV_THREAD) {
set_cookie(info->notify_cookie, NOTIFY_REMOVED);
@@ -695,6 +713,8 @@ static void remove_notification(struct mqueue_inode_info *info)
put_user_ns(info->notify_user_ns);
info->notify_owner = NULL;
info->notify_user_ns = NULL;
+
+ return true;
}

static int mq_attr_ok(struct ipc_namespace *ipc_ns, struct mq_attr *attr)
@@ -1203,12 +1223,18 @@ SYSCALL_DEFINE2(mq_notify, mqd_t, mqdes,
if (u_notification != NULL) {
if (unlikely(notification.sigev_notify != SIGEV_NONE &&
notification.sigev_notify != SIGEV_SIGNAL &&

-			     notification.sigev_notify != SIGEV_THREAD))
+			     notification.sigev_notify != SIGEV_THREAD &&
+			     notification.sigev_notify != SIGEV_THREAD_ID))

return -EINVAL;
- if (notification.sigev_notify == SIGEV_SIGNAL &&
+ if ((notification.sigev_notify == SIGEV_SIGNAL ||
+ notification.sigev_notify == SIGEV_THREAD_ID) &&
!valid_signal(notification.sigev_signo)) { return -EINVAL;
}
+ if (notification.sigev_notify == SIGEV_THREAD_ID &&
+ notification.sigev_notify_thread_id <= 0) {
+ return -EINVAL;
+ }
if (notification.sigev_notify == SIGEV_THREAD) {
long timeo;

@@ -1270,8 +1296,7 @@ retry:
ret = 0;
spin_lock(&info->lock);
if (u_notification == NULL) {
- if (info->notify_owner == task_tgid(current)) {
- remove_notification(info);
+ if (maybe_remove_notification(info)) {
inode->i_atime = inode->i_ctime = CURRENT_TIME;
}
} else if (info->notify_owner != NULL) {
@@ -1286,16 +1311,39 @@ retry:
info->notify_cookie = nc;
sock = NULL;
nc = NULL;
+ info->notify_owner = get_pid(task_tgid(current));
info->notify.sigev_notify = SIGEV_THREAD; break;
case SIGEV_SIGNAL:
info->notify.sigev_signo = notification.sigev_signo; info->notify.sigev_value = notification.sigev_value;
+ info->notify_owner = get_pid(task_tgid(current));
info->notify.sigev_notify = SIGEV_SIGNAL; break;
+ case SIGEV_THREAD_ID:
+ {
+ struct pid *tid;
+ pid_t tid_nr;
+
+ tid_nr = notification.sigev_notify_thread_id;
+
+ info->notify.sigev_signo = notification.sigev_signo;
+ info->notify.sigev_value = notification.sigev_value;
+
+ tid = find_get_pid(tid_nr);
+ if (NULL == tid) {
+ ret = -ESRCH;
+ goto out_fput;
+ }
+
+ info->notify.sigev_notify_thread_id = tid_nr;
+
+ info->notify_owner = tid;
+ info->notify.sigev_notify = SIGEV_THREAD_ID;
+ break;
+ }
}

- info->notify_owner = get_pid(task_tgid(current));
info->notify_user_ns = get_user_ns(current_user_ns()); inode->i_atime = inode->i_ctime = CURRENT_TIME;
}

Any one who wants a quick way to test the changes can use the following hacky program that works as an init program.

Thank you,
Steven Stewart-Gallus

#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <mqueue.h>
#include <pthread.h>
#include <signal.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/reboot.h>
#include <sys/syscall.h>
#include <sys/wait.h>
#include <unistd.h>

/* GLibc does not expose sigev_notify_thread_id so we hack it in manually */

#ifndef __ARCH_SIGEV_PREAMBLE_SIZE
#define __ARCH_SIGEV_PREAMBLE_SIZE (sizeof(int) * 2 + sizeof(sigval_t)) #endif

#define SIGEV_MAX_SIZE 64

#define SIGEV_PAD_SIZE	((SIGEV_MAX_SIZE - __ARCH_SIGEV_PREAMBLE_SIZE)	\
			 / sizeof(int))

struct my_sigevent {
	sigval_t sigev_value;
	int sigev_signo;
	int sigev_notify;
	union {
		int _pad[SIGEV_PAD_SIZE];
		int _tid;

		struct {
			void (*_function)(sigval_t);
			void *_attribute;	/* really pthread_attr_t */
		} _sigev_thread;
	} _sigev_un;
};

#define sigev_notify_function _sigev_un._sigev_thread._function
#define sigev_notify_attributes _sigev_un._sigev_thread._attribute
#define sigev_notify_thread_id _sigev_un._tid

struct message {
	int number;
};

static mqd_t mq;

static pid_t gettid(void)

{
	return syscall(__NR_gettid);
}

static void * start_routine(void * arg)

{
	{
		struct message message = {
			.number = 4
		};
		if (-1 == mq_send(mq, (char*)&message, sizeof message, 0)) {
			perror("mq_send");
			_Exit(EXIT_FAILURE);
		}
	}

	{
		struct message message = {
			.number = 5
		};
		if (-1 == mq_send(mq, (char*)&message, sizeof message, 0)) {
			perror("mq_send");
			_Exit(EXIT_FAILURE);
		}
	}

	{
		struct message message = {
			.number = 0
		};
		if (-1 == mq_send(mq, (char*)&message, sizeof message, 0)) {
			perror("mq_send");
			_Exit(EXIT_FAILURE);
		}
	}

	return NULL;
}

int main(int argc, char *argv[])

{
	int errnum;

	/* Fork to allow signals to be received */
	{
		pid_t child = fork();
		if (-1 == child) {
			perror("fork");
			return EXIT_FAILURE;
		}

		if (child != 0) {
			siginfo_t info;
			do {
				errnum = -1 == waitid(P_PID, child, &info, WEXITED) ? errno : 0;
			} while (EINTR == errnum);
			if (errnum != 0) {
				assert(errnum != EINVAL);
				assert(errnum != ECHILD);
				assert(false);
			}
			reboot(RB_POWER_OFF);
		}
	}

	{
		struct mq_attr attr = { 0 };
		attr.mq_maxmsg = 8U;
		attr.mq_msgsize = sizeof (struct message);
		mq = mq_open("/foo", O_RDWR | O_CREAT | O_EXCL | O_NONBLOCK, 0, &attr);
	}
	if (-1 == mq) {
		perror("mq_open");
		return EXIT_FAILURE;
	}


	sigset_t rtset;
	sigemptyset(&rtset);
	sigaddset(&rtset, SIGRTMIN);
	pthread_sigmask(SIG_BLOCK, &rtset, NULL);

	pthread_t worker;
	if ((errnum = pthread_create(&worker, NULL, start_routine, NULL)) != 0) {
		errno = errnum;
		perror("pthread_create");
		return EXIT_FAILURE;
	}

	for (;;) {
		{
			struct my_sigevent sev = { 0 };

			sev.sigev_notify = SIGEV_THREAD_ID;
			sev.sigev_signo = SIGRTMIN;

			sev.sigev_notify_thread_id = gettid();

			if (-1 == mq_notify(mq,(struct sigevent*) &sev)) {
				perror("mq_notify");
				return EXIT_FAILURE;
			}
		}

		for (;;) {
			struct message message;
			if (-1 == mq_receive(mq, (char*)&message, sizeof message, 0)) {
				errnum = errno;
			} else {
				errnum = 0;
			}

			if (EAGAIN == errnum) {
				break;
			}

			if (errnum != 0) {
				errno = errnum;
				perror("mq_receive");
				return EXIT_FAILURE;
			}

			printf("received: %u\n", message.number);

			if (0 == message.number) {
				goto exit_loop;
			}
		}

		{
			int xx;
			if (-1 == sigwait(&rtset, &xx)) {
				perror("sigwait");
				return EXIT_FAILURE;
			}
		}

		/* Flush pending signals */
		pthread_sigmask(SIG_UNBLOCK, &rtset, NULL);
		pthread_sigmask(SIG_BLOCK, &rtset, NULL);
	}
exit_loop:
	return EXIT_SUCCESS;
}