[PATCH v2 0/2] IPC: message queue checkpoint/restore - requested updates

[PATCH v2 0/2] IPC: message queue checkpoint/restore - requested updates

[Stanislav Kinsbursky]

Comparing to previous version:
1) single patch for all changes related to the feature (i.e. this patch is the
replacement for c-r-ipc-message-queue-stealing-feature-introduced.patch).
2) Added self-test for MSG_PEEK_ALL feature.

The following series consists of:

---

Stanislav Kinsbursky (2):
      IPC: message queue stealing feature introduced
      IPC: selftest tor new MSG_PEEK_ALL flag for msgrcv()


 include/linux/msg.h                           |    8 +
 ipc/compat.c                                  |   42 ++++++
 ipc/msg.c                                     |   65 +++++++++-
 tools/testing/selftests/ipc/Makefile          |   31 +++++
 tools/testing/selftests/ipc/msgque_peek_all.c |  170 +++++++++++++++++++++++++
 5 files changed, 313 insertions(+), 3 deletions(-)
 create mode 100644 tools/testing/selftests/ipc/Makefile
 create mode 100644 tools/testing/selftests/ipc/msgque_peek_all.c

[PATCH v2 1/2] IPC: message queue stealing feature introduced

[Stanislav Kinsbursky]

v3:
1) MSG_PEEK_ALL looks familiar because MSG_PEEK for one socket message is used
already.
2) All new checkpoint/restore code parts are now covered with
CONFIG_CHECKPOINT_RESTORE macro. So it would be easy to remove them, in case
the whole project fails.
3) return -ENOSYS, if user called sys_msgrcv() with MSG_PEEK_ALL flag set and
checkpoint/restore code wasn't compiled.

v2:
1) compat functions added.
2) message slot size in array is now aligned by struct msgbuf_a.
3) check for enough free space in buffer before message copying added.
4) if MSG_STEAL flag is set, then do_msgrcv() returns number of bytes written
to buffer.
5) flag MSG_NOERROR is ignored if MSG_STEAL flag is set.

This patch is required for checkpoint/restore in userspace.
IOW, c/r requires some way to get all pending IPC messages without deleting
them for the queue (checkpoint can fail and in this case tasks will be resumed,
so queue have to be valid).
To achive this, new operation flag MSG_STEAL for sys_msgrcv() system call
introduced.
If this flag is set, then passed struct msgbuf pointer will be used for storing
array of structures:

struct msgbuf_a {
	long mtype;         /* type of message */
	int msize;          /* size of message */
	char mtext[0];      /* message text */
};

each of which will be followed by corresponding message data.

Signed-off-by: Stanislav Kinsbursky <skinsbursky@parallels.com>
Signed-off-by: Cyrill Gorcunov <gorcunov@openvz.org>

---
 include/linux/msg.h |    8 ++++++
 ipc/compat.c        |   42 +++++++++++++++++++++++++++++++--
 ipc/msg.c           |   65 ++++++++++++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 112 insertions(+), 3 deletions(-)

diff --git a/include/linux/msg.h b/include/linux/msg.h
index 9411b76..bd8fe66 100644
--- a/include/linux/msg.h
+++ b/include/linux/msg.h
@@ -11,6 +11,7 @@
 /* msgrcv options */
 #define MSG_NOERROR     010000  /* no error if message is too big */
 #define MSG_EXCEPT      020000  /* recv any msg except of specified type.*/
+#define MSG_PEEK_ALL    040000  /* copy (not remove) all queue messages */
 
 /* Obsolete, used only for backwards compatibility and libc5 compiles */
 struct msqid_ds {
@@ -38,6 +39,13 @@ struct msgbuf {
 	char mtext[1];      /* message text */
 };
 
+/* message buffer for msgrcv in case of array calls */
+struct msgbuf_a {
+	long mtype;         /* type of message */
+	int msize;          /* size of message */
+	char mtext[0];      /* message text */
+};
+
 /* buffer for msgctl calls IPC_INFO, MSG_INFO */
 struct msginfo {
 	int msgpool;
diff --git a/ipc/compat.c b/ipc/compat.c
index 38c1ee5..bf31af7 100644
--- a/ipc/compat.c
+++ b/ipc/compat.c
@@ -38,6 +38,12 @@ struct compat_msgbuf {
 	char mtext[1];
 };
 
+struct compat_msgbuf_a {
+	compat_long_t mtype;
+	int msize;
+	char mtext[0];
+};
+
 struct compat_ipc_perm {
 	key_t key;
 	__compat_uid_t uid;
@@ -328,6 +334,33 @@ long compat_sys_msgsnd(int first, int second, int third, void __user *uptr)
 	return do_msgsnd(first, type, up->mtext, second, third);
 }
 
+#ifdef CONFIG_CHECKPOINT_RESTORE
+static long compat_do_msg_peek_all(void __user *dest, struct msg_msg *msg, size_t bufsz)
+{
+	struct compat_msgbuf_a __user *msgp = dest;
+	size_t msgsz;
+
+	msgsz = roundup(sizeof(struct msgbuf_a) + msg->m_ts,
+			__alignof__(struct msgbuf_a));
+
+	if (bufsz < msgsz)
+		return -E2BIG;
+
+	if (put_user(msg->m_type, &msgp->mtype))
+		return -EFAULT;
+	if (put_user(msg->m_ts, &msgp->msize))
+		return -EFAULT;
+	if (store_msg(msgp->mtext, msg, msg->m_ts))
+		return -EFAULT;
+	return msgsz;
+}
+#else
+static long compat_do_msg_peek_all(void __user *dest, struct msg_msg *msg, size_t bufsz)
+{
+	return -EINVAL;
+}
+#endif
+
 long compat_do_msg_fill(void __user *dest, struct msg_msg *msg, size_t bufsz)
 {
 	struct compat_msgbuf __user *msgp;
@@ -349,7 +382,10 @@ long compat_sys_msgrcv(int first, int second, int msgtyp, int third,
 		return -EINVAL;
 	if (second < 0)
 		return -EINVAL;
-
+#ifndef CONFIG_CHECKPOINT_RESTORE
+	if (third & MSG_PEEK_ALL)
+		return -ENOSYS;
+#endif
 	if (!version) {
 		struct compat_ipc_kludge ipck;
 		if (!uptr)
@@ -359,7 +395,9 @@ long compat_sys_msgrcv(int first, int second, int msgtyp, int third,
 		uptr = compat_ptr(ipck.msgp);
 		msgtyp = ipck.msgtyp;
 	}
-	return do_msgrcv(first, uptr, second, msgtyp, third, compat_do_msg_fill);
+	return do_msgrcv(first, uptr, second, msgtyp, third,
+			 (third & MSG_PEEK_ALL) ? compat_do_msg_peek_all
+						: compat_do_msg_fill);
 }
 
 static inline int get_compat_msqid64(struct msqid64_ds *m64,
diff --git a/ipc/msg.c b/ipc/msg.c
index 1d34c11..e7d07c9 100644
--- a/ipc/msg.c
+++ b/ipc/msg.c
@@ -762,6 +762,40 @@ static inline int convert_mode(long *msgtyp, int msgflg)
 	return SEARCH_EQUAL;
 }
 
+#ifdef CONFIG_CHECKPOINT_RESTORE
+static long do_msg_peek_all(void __user *dest, struct msg_msg *msg, size_t bufsz)
+{
+	struct msgbuf_a __user *msgp = dest;
+	size_t msgsz;
+
+	/*
+	 * Message size have to be aligned.
+	 */
+	msgsz = roundup(sizeof(struct msgbuf_a) + msg->m_ts,
+			__alignof__(struct msgbuf_a));
+
+	/*
+	 * No need to support MSG_NOERROR flag because truncated message array
+	 * is useless.
+	 */
+	if (bufsz < msgsz)
+		return -E2BIG;
+
+	if (put_user(msg->m_type, &msgp->mtype))
+		return -EFAULT;
+	if (put_user(msg->m_ts, &msgp->msize))
+		return -EFAULT;
+	if (store_msg(msgp->mtext, msg, msg->m_ts))
+		return -EFAULT;
+	return msgsz;
+}
+#else
+static long do_msg_peek_all(void __user *dest, struct msg_msg *msg, size_t bufsz)
+{
+	return -EINVAL;
+}
+#endif
+
 static long do_msg_fill(void __user *dest, struct msg_msg *msg, size_t bufsz)
 {
 	struct msgbuf __user *msgp = dest;
@@ -784,9 +818,16 @@ long do_msgrcv(int msqid, void __user *buf, size_t bufsz, long msgtyp,
 	struct msg_msg *msg;
 	int mode;
 	struct ipc_namespace *ns;
+#ifdef CONFIG_CHECKPOINT_RESTORE
+	size_t arrsz = bufsz;
+#endif
 
 	if (msqid < 0 || (long) bufsz < 0)
 		return -EINVAL;
+#ifndef CONFIG_CHECKPOINT_RESTORE
+	if (msgflg & MSG_PEEK_ALL)
+		return -ENOSYS;
+#endif
 	mode = convert_mode(&msgtyp, msgflg);
 	ns = current->nsproxy->ipc_ns;
 
@@ -817,6 +858,18 @@ long do_msgrcv(int msqid, void __user *buf, size_t bufsz, long msgtyp,
 						walk_msg->m_type != 1) {
 					msg = walk_msg;
 					msgtyp = walk_msg->m_type - 1;
+#ifdef CONFIG_CHECKPOINT_RESTORE
+				} else if (msgflg & MSG_PEEK_ALL) {
+					long ret;
+
+					ret = msg_fill(buf, msg, arrsz);
+					if (ret < 0) {
+						msg = ERR_PTR(ret);
+						goto out_unlock;
+					}
+					buf += ret;
+					arrsz -= ret;
+#endif
 				} else {
 					msg = walk_msg;
 					break;
@@ -825,6 +878,10 @@ long do_msgrcv(int msqid, void __user *buf, size_t bufsz, long msgtyp,
 			tmp = tmp->next;
 		}
 		if (!IS_ERR(msg)) {
+#ifdef CONFIG_CHECKPOINT_RESTORE
+			if (msgflg & MSG_PEEK_ALL)
+				goto out_unlock;
+#endif
 			/*
 			 * Found a suitable message.
 			 * Unlink it from the queue.
@@ -919,6 +976,11 @@ out_unlock:
 	if (IS_ERR(msg))
 		return PTR_ERR(msg);
 
+#ifdef CONFIG_CHECKPOINT_RESTORE
+	if (msgflg & MSG_PEEK_ALL)
+		return bufsz - arrsz;
+#endif
+
 	bufsz = msg_fill(buf, msg, bufsz);
 	free_msg(msg);
 
@@ -928,7 +990,8 @@ out_unlock:
 SYSCALL_DEFINE5(msgrcv, int, msqid, struct msgbuf __user *, msgp, size_t, msgsz,
 		long, msgtyp, int, msgflg)
 {
-	return do_msgrcv(msqid, msgp, msgsz, msgtyp, msgflg, do_msg_fill);
+	return do_msgrcv(msqid, msgp, msgsz, msgtyp, msgflg,
+			 (msgflg & MSG_PEEK_ALL) ? do_msg_peek_all : do_msg_fill);
 }
 
 #ifdef CONFIG_PROC_FS

[PATCH v2 2/2] IPC: selftest tor new MSG_PEEK_ALL flag for msgrcv()

[Stanislav Kinsbursky]

This test send two messages, then peek them and check, that they are equal to
original messages. Then it receives messages and check once more to make sure,
that messages are not corrupted or lost after peek operation.

---
 tools/testing/selftests/ipc/Makefile          |   31 +++++
 tools/testing/selftests/ipc/msgque_peek_all.c |  170 +++++++++++++++++++++++++
 2 files changed, 201 insertions(+), 0 deletions(-)
 create mode 100644 tools/testing/selftests/ipc/Makefile
 create mode 100644 tools/testing/selftests/ipc/msgque_peek_all.c

diff --git a/tools/testing/selftests/ipc/Makefile b/tools/testing/selftests/ipc/Makefile
new file mode 100644
index 0000000..10e5051
--- /dev/null
+++ b/tools/testing/selftests/ipc/Makefile
@@ -0,0 +1,31 @@
+ifeq ($(strip $(V)),)
+	E = @echo
+	Q = @
+else
+	E = @\#
+	Q =
+endif
+export E Q
+
+uname_M := $(shell uname -m 2>/dev/null || echo not)
+ARCH ?= $(shell echo $(uname_M) | sed -e s/i.86/i386/)
+ifeq ($(ARCH),i386)
+        ARCH := X86
+	CFLAGS := -DCONFIG_X86_32 -D__i386__
+endif
+ifeq ($(ARCH),x86_64)
+	ARCH := X86
+	CFLAGS := -DCONFIG_X86_64 -D__x86_64__
+endif
+
+all:
+ifeq ($(ARCH),X86)
+	$(E) "  CC run_test"
+	$(Q) gcc msgque_peek_all.c -o run_test
+else
+	$(E) "Not an x86 target, can't build kcmp selftest"
+endif
+
+clean:
+	$(E) "  CLEAN"
+	$(Q) rm -fr ./run_test
diff --git a/tools/testing/selftests/ipc/msgque_peek_all.c b/tools/testing/selftests/ipc/msgque_peek_all.c
new file mode 100644
index 0000000..5308999
--- /dev/null
+++ b/tools/testing/selftests/ipc/msgque_peek_all.c
@@ -0,0 +1,170 @@
+#define _GNU_SOURCE
+#include <sched.h>
+
+#include <stdio.h>
+#include <string.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <sys/sem.h>
+#include <sys/ipc.h>
+#include <sys/msg.h>
+#include <signal.h>
+#include <errno.h>
+
+#define __round_mask(x, y)	((__typeof__(x))((y) - 1))
+#define round_up(x, y)		((((x) - 1) | __round_mask(x, y)) + 1)
+
+#ifndef MSG_PEEK_ALL
+#define MSG_PEEK_ALL		040000
+/* message buffer for msgrcv in case of array calls */
+struct msgbuf_a {
+	long mtype;	/* type of message */
+	int msize;	/* size of message */
+	char mtext[0];	/* message text */
+};
+#endif
+
+const char *test_doc="Tests sysv5 msg queues supporting by checkpointing";
+const char *test_author="Stanislav Kinsbursky <skinsbursky@openvz.org>";
+
+#define MAX_MSG_LENGTH		32
+
+struct my_msg {
+	long mtype;
+	char mtext[MAX_MSG_LENGTH];
+};
+
+struct my_msg messages[] = {
+	{ 1, "Test sysv5 msg" },
+	{ 26538, "Yet another test sysv5 msg"},
+	{ 0, "" }
+};
+
+static int receive_messages(int msgq)
+{
+	int i, ret;
+	struct my_msg msgbuf;
+
+	i = 0;
+	while(messages[i].mtype > 0) {
+		ret = msgrcv(msgq, &msgbuf, MAX_MSG_LENGTH,
+				messages[i].mtype, IPC_NOWAIT);
+		if (ret < 0) {
+			printf("Child: msgrcv failed (%m)\n");
+			return -errno;
+		}
+		if (ret != strlen(messages[i].mtext) + 1) {
+			printf("Received message[%i] size is wrong: %d "
+					"(should be %ld)\n", i,
+					ret, strlen(messages[i].mtext) + 1);
+			return -EINVAL;
+		}
+		if (memcmp(msgbuf.mtext, messages[i].mtext, ret)) {
+			printf("Received message content is wrong\n");
+			return -EINVAL;
+		}
+		i++;
+	}
+	return 0;
+}
+
+static int peek_messages(int msgq)
+{
+	void *msg_array, *ptr;
+	int array_size;
+	struct msqid_ds ds;
+	int id, ret, i;
+
+	ret = msgctl(msgq, IPC_STAT, &ds);
+	if (ret < 0) {
+		printf("Failed to get stats for IPC message queue (%m)\n");
+		return -errno;
+	}
+
+	/*
+	 * Here we allocate memory for struct msgbuf_a twice becase messages in
+	 * array will be aligned by struct msgbuf_a.
+	 */
+	array_size = ds.msg_qnum * sizeof(struct msgbuf_a) * 2 + ds.msg_cbytes;
+	msg_array = malloc(array_size);
+	if (msg_array == 0)
+		return -ENOMEM;
+
+	ret = msgrcv(msgq, msg_array, array_size, 0, IPC_NOWAIT | MSG_PEEK_ALL);
+	if (ret < 0) {
+		printf("Failed to receive IPC messages array (%m)");
+		return -errno;
+	}
+
+	i = 0;
+	ptr = msg_array;
+	while (i < ds.msg_qnum) {
+		struct msgbuf_a *msg = ptr;
+
+		if (msg->mtype != messages[i].mtype) {
+			printf("Peeked message type is wrong: %ld (should be %ld)\n",
+				msg->mtype, messages[i].mtype);
+			return -EINVAL;
+		}
+		if (memcmp(msg->mtext, messages[i].mtext, msg->msize)) {
+			printf("Peeked message content is wrong\n");
+			return -EINVAL;
+		}
+		ptr += round_up(msg->msize + sizeof(struct msgbuf_a), sizeof(struct msgbuf_a));
+		i++;
+	}
+	return 0;
+}
+
+static int send_messages(int msgq)
+{
+	int i = 0;
+
+	while(messages[i].mtype > 0) {
+		if (msgsnd(msgq, &messages[i], strlen(messages[i].mtext) + 1, IPC_NOWAIT) != 0) {
+			printf("Parent: msgsnd[%i] failed (%m)", i);
+			return -errno;
+		};
+		i++;
+	}
+	return 0;
+}
+
+int main(int argc, char **argv)
+{
+	key_t key;
+	int msgq, ret;
+
+	key = ftok(argv[0], 822155650);
+	if (key == -1) {
+		printf("Can't make key");
+		return -errno;
+	}
+
+	msgq = msgget(key, IPC_CREAT | IPC_EXCL | 0666);
+	if (msgq == -1) {
+		msgq = msgget(key, 0666);
+		if (msgq == -1) {
+			printf("Can't get queue");
+			return -errno;
+		}
+	}
+
+	ret = send_messages(msgq);
+	if (ret)
+		goto out;
+	ret = peek_messages(msgq);
+	if (ret)
+		goto out;
+	ret = receive_messages(msgq);
+	if (ret)
+		goto out;
+out:
+	if (msgctl(msgq, IPC_RMID, 0)) {
+		printf("Failed to destroy message queue (%m)\n");
+		return -errno;
+	}
+	return ret;
+}

Re: [PATCH v2 2/2] IPC: selftest tor new MSG_PEEK_ALL flag for msgrcv()

[Andrew Morton]

On Mon, 16 Apr 2012 18:06:13 +0400
Stanislav Kinsbursky <skinsbursky@parallels.com> wrote:

> This test send two messages, then peek them and check, that they are equal to
> original messages. Then it receives messages and check once more to make sure,
> that messages are not corrupted or lost after peek operation.
> 
> ---

Missing signed-off-by.  I added your s-o-b to my copy of the changelog.

Re: [PATCH v2 2/2] IPC: selftest tor new MSG_PEEK_ALL flag for msgrcv()

[Stanislav Kinsbursky]

17.04.2012 02:11, Andrew Morton пишет:
> On Mon, 16 Apr 2012 18:06:13 +0400
> Stanislav Kinsbursky<skinsbursky@parallels.com>  wrote:
>
>> This test send two messages, then peek them and check, that they are equal to
>> original messages. Then it receives messages and check once more to make sure,
>> that messages are not corrupted or lost after peek operation.
>>
>> ---
>
> Missing signed-off-by.  I added your s-o-b to my copy of the changelog.

Sorry. Thanks.

-- 
Best regards,
Stanislav Kinsbursky

Re: [CRIU] [PATCH v2 0/2] IPC: message queue checkpoint/restore - requested updates

[Kinsbursky Stanislav]

Hello, Andrew.
I'm very sorry, but a critical flaw has been found in this feature: copying of 
kernel data to user is done under spinlock.
Since message queue size can be increased up to INT_MAX, then copying to 
temporary kernel buffer while passing message queue in not a solution.
So, please, drop both patches (feature implementation and test).
Most probably Ill replace them with MSG_PEEK feature (the same logic, but only 
for one message - like it's done for sockets).

-- 
Best regards,
Stanislav Kinsbursky