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

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

[Steven Stewart-Gallus]

From: Steven Stewart-Gallus <sstewartgallus00@mylangara.bc.ca>

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.

Signed-off-by: Steven Stewart-Gallus <sstewartgallus00@mylangara.bc.ca>
---

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

diff --git a/man3/mq_notify.3 b/man3/mq_notify.3
index a71aac4..41a1b96 100644
--- a/man3/mq_notify.3
+++ b/man3/mq_notify.3
@@ -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.
diff --git a/man7/sigevent.7 b/man7/sigevent.7
index 9cec77e..15da9fe 100644
--- a/man7/sigevent.7
+++ b/man7/sigevent.7
@@ -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

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

[Steven Stewart-Gallus]

diff --git a/ipc/mqueue.c b/ipc/mqueue.c
index 4fcf39a..8ca70a2 100644
--- a/ipc/mqueue.c
+++ b/ipc/mqueue.c
@@ -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;
 	}

Re: [PATCH] V1 2/2] ipc: let message queues use SIGEV_THREAD_ID with mq_notify

[Steven Stewart-Gallus]

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;
}

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

[Steven Stewart-Gallus]

> Could you explain what you mean with "only thread-safe way"?
> I'm a bit relunctant to extend mq_notify() without understanding 
> the reason.
> 
> What about:
> - use sigprocmask()
> - create one worker thread
> - then in a loop in that worker thread: use sigwaitinfo() or 
> signalfd() 
> to collect the signals.

Those all modify global state and are unusable by library
code. Hogging a signal for your own library is bad.

> And one point I don't like: Within timer_create():

> >  SIGEV_THREAD_ID (Linux-specific)
> >  [...]
> > The sigev_notify_thread_id field specifies a kernel thread ID,
> > that is, the value returned by clone(2) or gettid(2).
>
> Does that mean that SIGEV_THREAD_ID is guaranteed to remain
> Linux-specific, it's implicitely linked to the Linux
> clone()/gettid() threading model?
>
> > This flag is intended only for use by threading libraries.

Yes this is low-level Linux specific stuff.

Maybe GLibc could create a more abstract and portable interface on top
of SIGEV_THREAD_ID that other OS's and libraries could implement? This
is pretty much not relevant to Linux kernel code though except that
more features that use SIGEV_THREAD_ID would motivate GLibc developers
to create a wrapper over this and that maybe the Linux kernel could be
modified to make it easier to create a more abstract and portable
wrapper.

If you feel that the whole SIGEV_THREAD_ID line of functionality was a
mistake in the first place and that people should use signals less and
use functionality like epoll more that's also okay.

If you also feel that nobody would use the feature in the first place
that's okay too.

Personally, it just bugged me that SIGEV_THREAD_ID was only used by
timers and not in a few other places where it seems like an obvious
fit.

Thank you,
Steven Stewart-Gallus

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

[Manfred Spraul]

Hi Steven,

You wrote:
> Currently the only thread-safe way of using mq_notify with message 
> queues is to use the SIGEV_THREAD option.
Could you explain what you mean with "only thread-safe way"?
I'm a bit relunctant to extend mq_notify() without understanding the reason.

What about:
- use sigprocmask()
- create one worker thread
- then in a loop in that worker thread: use sigwaitinfo() or signalfd() 
to collect the signals.


And one point I don't like: Within timer_create():
>  SIGEV_THREAD_ID (Linux-specific)
>  [...]
> The sigev_notify_thread_id field specifies a kernel thread ID, that 
> is, the value returned by clone(2) or gettid(2).
Does that mean that SIGEV_THREAD_ID is guaranteed to remain 
Linux-specific, it's implicitely linked to the Linux clone()/gettid() 
threading model?

> This flag is intended only for use by threading libraries.


--
     Manfred