copy messages directly from sender into a receiver-supplied buffer

Require the receiver to pre-allocate a buffer and register it with
HELLO. The sender will copy its data directly into the revceiver's
buffer.

The RECV call will now only return a pointer to the next message
in the buffer. The buffer needs to be free()d with the FREE ioctl.

FIXME: currently missing is a proper allocator to manage the
receiver's buffer, in the current state the allocator will only clear
the buffer if all messages are free()d. This is obviously only useful
for testing.

15 files changed, 905 insertions, 701 deletions

diff --git a/Makefile b/Makefile
index c522067ab01..28ea15118f9 100644
--- a/Makefile
+++ b/Makefile
@@ -1,6 +1,7 @@
 kdbus-y	:= \
 	bus.o \
 	connection.o \
+	buffer.o \
 	ep.o \
 	main.o \
 	match.o \
diff --git a/TODO b/TODO
index 0e68bea2921..68b9ed2e04d 100644
--- a/TODO
+++ b/TODO
@@ -1,7 +1,14 @@
 External API:
-  - check that enum values of 0 make sense, otherwise insert "*_NONE,"
-  - always zero-out, put \0 in the padding bytes?
-  - merge dst_id and src_ id into a union?
+
+Internal:
+  - chunk get_user_pages() while writing to the internal buffer api
+  - update/rethink kdbus_conn_accounting_sub_size() use, we don't allocate
+    kernel memory anymore
+  - add real allocator to kdbus_conn_buffer_alloc()
+  - BUG: connections are currently listed in the endpoint not the bus, so
+    broadcasting will never see the other endpoints on the same bus
+  - BUG: the sender needs to ref the receiver when looking it up to protect
+    against the receiver going away while its ressources are accessed
 
 Kernel Core:
   - remove our own task_cgroup_path_from_hierarchy() as soon as it's available:
diff --git a/buffer.c b/buffer.c
new file mode 100644
index 00000000000..e97fb77ae72
--- /dev/null
+++ b/buffer.c
@@ -0,0 +1,182 @@
+/*
+ * Copyright (C) 2013 Kay Sievers
+ * Copyright (C) 2013 Greg Kroah-Hartman <gregkh@linuxfoundation.org>
+ * Copyright (C) 2013 Linux Foundation
+ * Copyright (C) 2013 Daniel Mack <daniel@zonque.org>
+ *
+ * kdbus is free software; you can redistribute it and/or modify it under
+ * the terms of the GNU Lesser General Public License as published by the
+ * Free Software Foundation; either version 2.1 of the License, or (at
+ * your option) any later version.
+ */
+
+#define pr_fmt(fmt)	KBUILD_MODNAME ": " fmt
+
+#include <linux/module.h>
+#include <linux/fs.h>
+#include <linux/slab.h>
+#include <linux/sizes.h>
+#include <linux/sched.h>
+#include <linux/mutex.h>
+#include <linux/init.h>
+#include <linux/mm.h>
+#include <linux/highmem.h>
+
+#include "message.h"
+
+/*
+ * At KDBUS_CMD_MSG_SEND, messages are placed direcly into the buffer the
+ * receiver has registered with KDBUS_HELLO_BUFFER.
+ *
+ * To receive a message, KDBUS_CMD_MSG_RECV is called, which returns a pointer
+ * into the buffer.
+ *
+ * The internally allocated memory needs to be freed by the receiver with
+ * KDBUS_CMD_MSG_FREE.
+ */
+struct kdbus_buffer {
+	void __user *buf;	/* receiver-supplied buffer */
+	size_t size;		/* size of buffer  */
+	size_t pos;		/* current wrire position */
+	unsigned int users;
+};
+
+/* allocate a slot on the given size in the receiver's buffer */
+struct kdbus_msg __user *
+kdbus_buffer_alloc(struct kdbus_buffer *buf, size_t len)
+{
+	size_t pos;
+
+	pos = KDBUS_ALIGN8(buf->pos);
+	if (pos + len > buf->size)
+		return NULL;
+
+	buf->pos = pos + len;
+	buf->users++;
+
+	return buf->buf + pos;
+}
+
+/* free the allocated slot */
+void kdbus_buffer_free(struct kdbus_buffer *buf, struct kdbus_msg __user *msg)
+{
+	if (!msg)
+		return;
+
+	BUG_ON(buf->users == 0);
+
+	/* FIXME: dumbest possible version of an allocator: just reset the buffer
+	 * when it is empty; replace with rbtree/slice/list allocator */
+	if (--buf->users == 0)
+		buf->pos = 0;
+}
+
+/*
+ * Temporarily map a range of the receiver's buffer to write chunks of data from
+ * the sender into it.
+ */
+struct kdbus_buffer_map {
+	struct page **pages;	/* array of pages representign the buffer */
+	unsigned int n;		/* number pf pages in the array */
+	unsigned long cur;	/* current page we write to */
+	unsigned long pos;	/* position in current page we write to*/
+};
+
+/* unpin the receiver's pages */
+void kdbus_buffer_map_close(struct kdbus_buffer_map *buf)
+{
+	unsigned int i;
+
+	for (i = 0; i < buf->n; i++)
+		put_page(buf->pages[i]);
+	kfree(buf->pages);
+}
+
+/* pin the receiver's memory range/pages */
+int kdbus_buffer_map_open(struct kdbus_buffer_map *map,
+			  struct task_struct *task,
+			  void __user *to, size_t len)
+{
+	unsigned int n;
+	int have;
+	unsigned long base;
+	unsigned long addr;
+	struct mm_struct *mm;
+
+	memset(map, 0, sizeof(struct kdbus_buffer));
+
+	/* calculate the number of pages involved in the range */
+	addr = (unsigned long)to;
+	n = (addr + len - 1) / PAGE_SIZE - addr / PAGE_SIZE + 1;
+
+	map->pages = kmalloc(n * sizeof(struct page *), GFP_KERNEL);
+	if (!map->pages)
+		return -ENOMEM;
+
+	/* start address in our first page */
+	base = addr & PAGE_MASK;
+	map->pos = addr - base;
+
+	/* pin the receiver's buffer page(s); the task
+	 * is pinned as long as the connection is open */
+	mm = get_task_mm(task);
+	if (!mm) {
+		kdbus_buffer_map_close(map);
+		return -ESHUTDOWN;
+	}
+	down_read(&mm->mmap_sem);
+	have = get_user_pages(task, mm, base, n,
+			      true, false, map->pages, NULL);
+	up_read(&mm->mmap_sem);
+	mmput(mm);
+
+	if (have < 0) {
+		kdbus_buffer_map_close(map);
+		return have;
+	}
+
+	map->n = have;
+
+	/* fewer pages than requested */
+	if (map->n < n) {
+		kdbus_buffer_map_close(map);
+		return -EFAULT;
+	}
+
+	return 0;
+}
+
+/* copy a memory range from the current user process page by
+ * page into the pinned receiver's buffer */
+int kdbus_buffer_map_write(struct kdbus_buffer_map *map,
+			   void __user *from, size_t len)
+{
+	int ret = 0;
+
+	while (len > 0) {
+		void *addr;
+		size_t bytes;
+
+		/* bytes copy to remaining space of current page */
+		bytes = min(PAGE_SIZE - map->pos, len);
+
+		/* map, fill, unmap current page */
+		addr = kmap(map->pages[map->cur]) + map->pos;
+		if (copy_from_user(addr, from, bytes))
+			ret = -EFAULT;
+		kunmap(map->pages[map->cur]);
+		if (ret < 0)
+			break;
+
+		/* add to pos, or move to next page */
+		map->pos += bytes;
+		if (map->pos == PAGE_SIZE) {
+			map->pos = 0;
+			map->cur++;
+		}
+
+		len -= bytes;
+	}
+
+	return ret;
+}
diff --git a/buffer.h b/buffer.h
new file mode 100644
index 00000000000..0cdb065d870
--- /dev/null
+++ b/buffer.h
@@ -0,0 +1,52 @@
+/*
+ * Copyright (C) 2013 Kay Sievers
+ * Copyright (C) 2013 Greg Kroah-Hartman <gregkh@linuxfoundation.org>
+ * Copyright (C) 2013 Linux Foundation
+ * Copyright (C) 2013 Daniel Mack <daniel@zonque.org>
+ *
+ * kdbus is free software; you can redistribute it and/or modify it under
+ * the terms of the GNU Lesser General Public License as published by the
+ * Free Software Foundation; either version 2.1 of the License, or (at
+ * your option) any later version.
+ */
+
+#ifndef __KDBUS_BUFFER_H
+#define __KDBUS_BUFFER_H
+
+/*
+ * At KDBUS_CMD_MSG_SEND, messages are placed direcly into the buffer the
+ * receiver has registered with KDBUS_HELLO_BUFFER.
+ *
+ * To receive a message, KDBUS_CMD_MSG_RECV is called, which returns a pointer
+ * into the buffer.
+ *
+ * The internally allocated memory needs to be freed by the receiver with
+ * KDBUS_CMD_MSG_FREE.
+ */
+struct kdbus_buffer {
+	void __user *buf;	/* receiver-supplied buffer */
+	size_t size;		/* size of buffer  */
+	size_t pos;		/* current write position */
+	unsigned int users;
+};
+
+/*
+ * Structure to keep the state of a mapped range on the buffer while
+ * writing chunks of data to it from the sender.
+ */
+struct kdbus_buffer_map {
+	struct page **pages;	/* array of pages representign the buffer */
+	unsigned int n;		/* number pf pages in the array */
+	unsigned long cur;	/* current page we write to */
+	unsigned long pos;	/* position in current page we write to */
+};
+
+struct kdbus_msg __user *kdbus_buffer_alloc(struct kdbus_buffer *buf, size_t len);
+void kdbus_buffer_free(struct kdbus_buffer *buf, struct kdbus_msg __user *msg);
+void kdbus_buffer_map_close(struct kdbus_buffer *buf);
+int kdbus_buffer_map_open(struct kdbus_buffer *buf,
+			  struct task_struct *task,
+			  void __user *to, size_t len);
+int kdbus_buffer_map_write(struct kdbus_buffer *buf,
+			   void __user *from, size_t len);
+#endif
diff --git a/connection.c b/connection.c
index bd9b6cde914..c5f528e3f05 100644
--- a/connection.c
+++ b/connection.c
@@ -22,12 +22,16 @@
 #include <linux/mutex.h>
 #include <linux/init.h>
 #include <linux/poll.h>
+#include <linux/file.h>
 #include <linux/hashtable.h>
 #include <linux/audit.h>
 #include <linux/security.h>
+#include <linux/mm.h>
+#include <linux/highmem.h>
 #include <uapi/linux/major.h>
 
 #include "connection.h"
+#include "buffer.h"
 #include "message.h"
 #include "notify.h"
 #include "ns.h"
@@ -37,7 +41,309 @@
 #include "names.h"
 #include "policy.h"
 
-int kdbus_conn_add_size_allocation(struct kdbus_conn *conn, size_t size)
+static void kdbus_conn_fds_unref(struct kdbus_conn_queue *queue)
+{
+	unsigned int i;
+
+	if (!queue->fds_fp)
+		return;
+
+	for (i = 0; i < queue->fds_count; i++) {
+		if (!queue->fds_fp[i])
+			break;
+
+		fput(queue->fds_fp[i]);
+	}
+
+	kfree(queue->fds_fp);
+	queue->fds_fp = NULL;
+}
+
+/* grab references of passed-in FDS for in-flight messages */
+static int kdbus_conn_fds_ref(struct kdbus_conn_queue *queue,
+			 const int *fds, unsigned int fds_count)
+{
+	unsigned int i;
+
+	queue->fds_fp = kmalloc(fds_count * sizeof(struct file *), GFP_KERNEL);
+	if (!queue->fds_fp)
+		return -ENOMEM;
+
+	for (i = 0; i < fds_count; i++) {
+		queue->fds_fp[i] = fget(fds[i]);
+		if (!queue->fds_fp[i]) {
+			kdbus_conn_fds_unref(queue);
+			return -EBADF;
+		}
+	}
+
+	return 0;
+}
+
+void kdbus_conn_queue_cleanup(struct kdbus_conn_queue *queue)
+{
+	kdbus_conn_fds_unref(queue);
+	kfree(queue);
+}
+
+/* enqueue a message into the receiver's connection */
+int kdbus_conn_queue_insert(struct kdbus_conn *conn,
+			    struct kdbus_kmsg *kmsg,
+			    u64 deadline_ns)
+{
+	struct kdbus_conn_queue *queue;
+	const struct kdbus_item *item;
+	void __user *buf;
+	u64 msg_size;
+	size_t vec = 0;
+	size_t fds = 0;
+	size_t meta = 0;
+	size_t vec_data;
+	struct kdbus_buffer buffer;
+	int ret = 0;
+
+	if (!conn->active)
+		return -ENOTCONN;
+
+	if (kmsg->fds && !(conn->flags & KDBUS_HELLO_ACCEPT_FD))
+		return -ECOMM;
+
+	queue = kzalloc(sizeof(struct kdbus_conn_queue), GFP_KERNEL);
+	if (!queue)
+		return -ENOMEM;
+
+	INIT_LIST_HEAD(&queue->entry);
+
+	/* copy message properies we need for the queue management */
+	queue->deadline_ns = deadline_ns;
+	queue->src_id = kmsg->msg.src_id;
+	queue->cookie = kmsg->msg.cookie;
+	if (kmsg->msg.flags & KDBUS_MSG_FLAGS_EXPECT_REPLY)
+		queue->expect_reply = true;
+
+	/* space for message header */
+	msg_size = KDBUS_MSG_HEADER_SIZE;
+
+	/* space for PAYLOAD_VEC item */
+	if (kmsg->vecs_size > 0) {
+		vec = msg_size;
+		msg_size += KDBUS_ITEM_SIZE(sizeof(struct kdbus_vec));
+	}
+
+	/* space for FDS item */
+	if (kmsg->fds_count > 0) {
+		fds = msg_size;
+		msg_size += KDBUS_ITEM_SIZE(kmsg->fds_count * sizeof(int));
+	}
+
+	/* space for metadata/credential items */
+	if (kmsg->meta_size > 0) {
+		meta = msg_size;
+		msg_size += kmsg->meta_size;
+	}
+
+	/* data starts after the message */
+	vec_data = KDBUS_ALIGN8(msg_size);
+
+	/* allocate the buffer from the receiver */
+	mutex_lock(&conn->lock);
+	if (conn->msg_count > KDBUS_CONN_MAX_MSGS) {
+		ret = -EXFULL;
+		goto exit_unlock;
+	}
+
+	buf = kdbus_buffer_alloc(&conn->buffer, vec_data + kmsg->vecs_size);
+	if (!buf) {
+		ret = -EXFULL;
+		goto exit_unlock;
+	}
+	mutex_unlock(&conn->lock);
+
+	/* update and copy the message header */
+	if (copy_to_user(buf, &kmsg->msg, KDBUS_MSG_HEADER_SIZE)) {
+		ret = -EFAULT;
+		goto exit;
+	}
+
+	/* update the size */
+	if (kdbus_size_set_user(msg_size, buf, struct kdbus_msg)) {
+		ret = -EFAULT;
+		goto exit;
+	}
+
+	/* add a PAYLOAD_VEC item */
+	if (kmsg->vecs_size > 0) {
+		size_t size = KDBUS_ITEM_HEADER_SIZE + sizeof(struct kdbus_vec);
+		char tmp[size];
+		struct kdbus_item *it = (struct kdbus_item *)tmp;
+
+		it->type = KDBUS_MSG_PAYLOAD_VEC;
+		it->size = size;
+		it->vec.address = (u64)buf + vec_data;
+		it->vec.size = kmsg->vecs_size;
+		if (copy_to_user(buf + vec, it, size)) {
+			ret = -EFAULT;
+			goto exit;
+		}
+	}
+
+	/* add a FDS item; the array content will be updated at RECV time */
+	if (kmsg->fds_count > 0) {
+		size_t size = KDBUS_ITEM_HEADER_SIZE;
+		char tmp[size];
+		struct kdbus_item *it = (struct kdbus_item *)tmp;
+
+		it->type = KDBUS_MSG_FDS;
+		it->size = size + (kmsg->fds_count * sizeof(int));
+		if (copy_to_user(buf + fds, it, size)) {
+			ret = -EFAULT;
+			goto exit;
+		}
+
+		queue->fds = it->fds;
+
+		ret = kdbus_conn_fds_ref(queue, kmsg->fds, kmsg->fds_count);
+		if (ret < 0)
+			goto exit;
+	}
+
+	/* append message metadata/credential items */
+	if (kmsg->meta_size > 0) {
+		if (copy_to_user(buf + meta, kmsg->meta, kmsg->meta_size)) {
+			ret = -EFAULT;
+			goto exit;
+		}
+	}
+
+	/* copy all payload data */
+	if (kmsg->vecs_size > 0) {
+		kdbus_buffer_map_open(&buffer, conn->task,
+				       buf + vec_data, kmsg->vecs_size);
+		KDBUS_ITEM_FOREACH(item, &kmsg->msg) {
+			if (item->type != KDBUS_MSG_PAYLOAD_VEC)
+				continue;
+
+			ret = kdbus_buffer_map_write(&buffer,
+					(void *)(uintptr_t)item->vec.address,
+					item->vec.size);
+			if (ret < 0)
+				break;
+
+			vec_data += item->vec.size;
+		}
+		kdbus_buffer_map_close(&buffer);
+		if (ret < 0)
+			goto exit;
+	}
+
+	/* remember the pointer to the message */
+	queue->msg = buf;
+
+	/* link the message into the receiver's queue */
+	mutex_lock(&conn->lock);
+	list_add_tail(&queue->entry, &conn->msg_list);
+	conn->msg_count++;
+	mutex_unlock(&conn->lock);
+
+	/* wake up poll() */
+	wake_up_interruptible(&conn->ep->wait);
+	return 0;
+
+exit_unlock:
+	mutex_unlock(&conn->lock);
+exit:
+	kdbus_conn_queue_cleanup(queue);
+	return ret;
+}
+
+static int kdbus_conn_fds_install(struct kdbus_conn_queue *queue)
+{
+	unsigned int i;
+	int *fds;
+	size_t size;
+	int ret = 0;
+
+	/* get array of file descriptors */
+	size = queue->fds_count * sizeof(int);
+	fds = kmalloc(size, GFP_KERNEL);
+	if (!fds)
+		return -ENOMEM;
+
+	/* allocate new file descriptors in the receiver's process */
+	for (i = 0; i < queue->fds_count; i++) {
+		fds[i] = get_unused_fd();
+		if (fds[i] < 0) {
+			ret = fds[i];
+			goto remove_unused;
+		}
+	}
+
+	/* copy the array into the message item */
+	if (copy_to_user(queue->fds, fds, size)) {
+		ret = -EFAULT;
+		goto remove_unused;
+	}
+
+	/* install files in the receiver's process */
+	for (i = 0; i < queue->fds_count; i++)
+		fd_install(fds[i], get_file(queue->fds_fp[i]));
+
+	kfree(fds);
+	return 0;
+
+remove_unused:
+	for (i = 0; i < queue->fds_count; i++) {
+		if (fds[i] < 0)
+			break;
+		put_unused_fd(fds[i]);
+	}
+
+	kfree(fds);
+	return ret;
+}
+
+static int
+kdbus_conn_recv_msg(struct kdbus_conn *conn, struct kdbus_msg __user **msg_ptr)
+{
+	struct kdbus_conn_queue *queue;
+	int ret;
+
+	if (!KDBUS_IS_ALIGNED8((unsigned long)msg_ptr))
+		return -EFAULT;
+
+	mutex_lock(&conn->lock);
+	if (conn->msg_count == 0) {
+		ret = -EAGAIN;
+		goto exit_unlock;
+	}
+
+	/* return the address of the next message in the buffer */
+	queue = list_first_entry(&conn->msg_list,
+				 struct kdbus_conn_queue, entry);
+	if (put_user(queue->msg, msg_ptr)) {
+		ret = -EFAULT;
+		goto exit_unlock;
+	}
+
+	if (queue->fds_count) {
+		ret = kdbus_conn_fds_install(queue);
+		if (ret < 0)
+			goto exit_unlock;
+	}
+
+	conn->msg_count--;
+	list_del(&queue->entry);
+	mutex_unlock(&conn->lock);
+
+	kdbus_conn_queue_cleanup(queue);
+	return 0;
+
+exit_unlock:
+	mutex_unlock(&conn->lock);
+	return ret;
+}
+
+int kdbus_conn_accounting_add_size(struct kdbus_conn *conn, size_t size)
 {
 	int ret = 0;
 
@@ -54,7 +360,7 @@ int kdbus_conn_add_size_allocation(struct kdbus_conn *conn, size_t size)
 	return ret;
 }
 
-void kdbus_conn_sub_size_allocation(struct kdbus_conn *conn, size_t size)
+void kdbus_conn_accounting_sub_size(struct kdbus_conn *conn, size_t size)
 {
 	if (!conn)
 		return;
@@ -66,7 +372,7 @@ void kdbus_conn_sub_size_allocation(struct kdbus_conn *conn, size_t size)
 
 static void kdbus_conn_scan_timeout(struct kdbus_conn *conn)
 {
-	struct kdbus_msg_list_entry *entry, *tmp;
+	struct kdbus_conn_queue *queue, *tmp;
 	u64 deadline = -1;
 	struct timespec ts;
 	uint64_t now;
@@ -74,24 +380,23 @@ static void kdbus_conn_scan_timeout(struct kdbus_conn *conn)
 	ktime_get_ts(&ts);
 	now = timespec_to_ns(&ts);
 
-	mutex_lock(&conn->msg_lock);
-	list_for_each_entry_safe(entry, tmp, &conn->msg_list, entry) {
-		struct kdbus_kmsg *kmsg = entry->kmsg;
-
-		if (kmsg->deadline_ns == 0)
+	mutex_lock(&conn->lock);
+	list_for_each_entry_safe(queue, tmp, &conn->msg_list, entry) {
+		if (queue->deadline_ns == 0)
 			continue;
 
-		if (kmsg->deadline_ns <= now) {
-			if (kmsg->msg.flags & KDBUS_MSG_FLAGS_EXPECT_REPLY)
-				kdbus_notify_reply_timeout(conn->ep, &kmsg->msg);
-			kdbus_kmsg_unref(entry->kmsg);
-			list_del(&entry->entry);
-			kfree(entry);
-		} else if (kmsg->deadline_ns < deadline) {
-			deadline = kmsg->deadline_ns;
+		if (queue->deadline_ns <= now) {
+			if (queue->expect_reply)
+				kdbus_notify_reply_timeout(conn->ep,
+					queue->src_id, queue->cookie);
+			kdbus_buffer_free(&conn->buffer, queue->msg);
+			list_del(&queue->entry);
+			kdbus_conn_queue_cleanup(queue);
+		} else if (queue->deadline_ns < deadline) {
+			deadline = queue->deadline_ns;
 		}
 	}
-	mutex_unlock(&conn->msg_lock);
+	mutex_unlock(&conn->lock);
 
 	if (deadline != -1) {
 		u64 usecs = deadline - now;
@@ -106,7 +411,7 @@ static void kdbus_conn_work(struct work_struct *work)
 	kdbus_conn_scan_timeout(conn);
 }
 
-void kdbus_conn_schedule_timeout_scan(struct kdbus_conn *conn)
+void kdbus_conn_timeout_schedule_scan(struct kdbus_conn *conn)
 {
 	schedule_work(&conn->work);
 }
@@ -114,7 +419,7 @@ void kdbus_conn_schedule_timeout_scan(struct kdbus_conn *conn)
 static void kdbus_conn_timer_func(unsigned long val)
 {
 	struct kdbus_conn *conn = (struct kdbus_conn *) val;
-	kdbus_conn_schedule_timeout_scan(conn);
+	kdbus_conn_timeout_schedule_scan(conn);
 }
 
 #ifdef CONFIG_AUDITSYSCALL
@@ -171,7 +476,6 @@ static int kdbus_conn_open(struct inode *inode, struct file *file)
 	/* control device node */
 	if (MINOR(inode->i_rdev) == 0) {
 		conn->type = KDBUS_CONN_CONTROL;
-		file->private_data = conn;
 		pr_debug("opened control device '%s/control'\n",
 			 conn->ns->devpath);
 		return 0;
@@ -195,7 +499,7 @@ static int kdbus_conn_open(struct inode *inode, struct file *file)
 	/* add this connection to hash table */
 	hash_add(conn->ep->bus->conn_hash, &conn->hentry, conn->id);
 
-	mutex_init(&conn->msg_lock);
+	mutex_init(&conn->lock);
 	mutex_init(&conn->names_lock);
 	mutex_init(&conn->accounting_lock);
 	INIT_LIST_HEAD(&conn->msg_list);
@@ -231,10 +535,14 @@ static int kdbus_conn_open(struct inode *inode, struct file *file)
 		 (unsigned long long)conn->id, conn->ns->devpath,
 		 conn->ep->bus->name);
 
+	//FIXME: cleanup here!
 	ret = kdbus_notify_id_change(conn->ep, KDBUS_MSG_ID_ADD, conn->id, conn->flags);
 	if (ret < 0)
 		return ret;
 
+	/* pin and store the task, so a sender can copy to the receiver */
+	get_task_struct(current);
+	conn->task = current;
 	return 0;
 
 exit_unlock:
@@ -258,43 +566,39 @@ static int kdbus_conn_release(struct inode *inode, struct file *file)
 		break;
 
 	case KDBUS_CONN_EP: {
-		struct kdbus_msg_list_entry *entry, *tmp;
+		struct kdbus_conn_queue *queue, *tmp;
 		struct list_head list;
 
 		INIT_LIST_HEAD(&list);
 
 		hash_del(&conn->hentry);
 		list_del(&conn->connection_entry);
+
 		/* clean up any messages still left on this endpoint */
-		mutex_lock(&conn->msg_lock);
-		list_for_each_entry_safe(entry, tmp, &conn->msg_list, entry) {
-			struct kdbus_kmsg *kmsg = entry->kmsg;
-			struct kdbus_msg *msg = &kmsg->msg;
-
-			list_del(&entry->entry);
-
-			/*
-			 * calling kdbus_notify_reply_dead() with msg_lock held
-			 * causes a lockdep warning, so let's re-link those
-			 * messages into a temporary list and handle it later.
-			 */
-			if (msg->src_id != conn->id &&
-			    msg->flags & KDBUS_MSG_FLAGS_EXPECT_REPLY) {
-				list_add_tail(&entry->entry, &list);
+		mutex_lock(&conn->lock);
+		list_for_each_entry_safe(queue, tmp, &conn->msg_list, entry) {
+			list_del(&queue->entry);
+
+			/* we cannot hold "lock" and enqueue new messages with
+			 * kdbus_notify_reply_dead(); move these messages
+			 * into a temporary list and handle them below */
+			if (queue->src_id != conn->id && queue->expect_reply) {
+				list_add_tail(&queue->entry, &list);
 			} else {
-				kdbus_kmsg_unref(kmsg);
-				kfree(entry);
+				kdbus_buffer_free(&conn->buffer, queue->msg);
+				kdbus_conn_queue_cleanup(queue);
 			}
 		}
-		mutex_unlock(&conn->msg_lock);
-
-		list_for_each_entry_safe(entry, tmp, &list, entry) {
-			struct kdbus_kmsg *kmsg = entry->kmsg;
-			struct kdbus_msg *msg = &kmsg->msg;
-
-			kdbus_notify_reply_dead(conn->ep, msg);
-			kdbus_kmsg_unref(kmsg);
-			kfree(entry);
+		mutex_unlock(&conn->lock);
+
+		list_for_each_entry_safe(queue, tmp, &list, entry) {
+			kdbus_notify_reply_dead(conn->ep, queue->src_id,
+						queue->cookie);
+			mutex_lock(&conn->lock);
+			kdbus_buffer_free(&conn->buffer, queue->msg);
+			mutex_unlock(&conn->lock);
+			kdbus_conn_queue_cleanup(queue);
+			list_del(&queue->entry);
 		}
 
 		del_timer(&conn->timer);
@@ -311,6 +615,7 @@ static int kdbus_conn_release(struct inode *inode, struct file *file)
 		kdbus_match_db_unref(conn->match_db);
 		kdbus_ep_unref(conn->ep);
 
+		put_task_struct(current);
 		break;
 	}
 
@@ -410,8 +715,8 @@ static long kdbus_conn_ioctl_ep(struct file *file, unsigned int cmd,
 {
 	struct kdbus_conn *conn = file->private_data;
 	struct kdbus_cmd_ep_kmake *kmake = NULL;
+	struct kdbus_cmd_hello *hello = NULL;
 	struct kdbus_bus *bus = NULL;
-	struct kdbus_kmsg *kmsg;
 	long ret = 0;
 
 	if (conn && conn->ep)
@@ -444,33 +749,72 @@ static long kdbus_conn_ioctl_ep(struct file *file, unsigned int cmd,
 
 	case KDBUS_CMD_HELLO: {
 		/* turn this fd into a connection. */
-		struct kdbus_cmd_hello hello;
+		const struct kdbus_item *item;
+		size_t size;
+		void *v;
 
 		if (conn->active) {
 			ret = -EISCONN;
 			break;
 		}
 
-		if (copy_from_user(&hello, buf, sizeof(hello))) {
+		if (kdbus_size_get_user(size, buf, struct kdbus_cmd_hello)) {
 			ret = -EFAULT;
 			break;
 		}
 
-		if (!check_flags(hello.conn_flags)) {
+		if (size < sizeof(struct kdbus_cmd_hello) || size > KDBUS_HELLO_MAX_SIZE) {
+			ret = -EMSGSIZE;
+			break;
+		}
+
+		v = memdup_user(buf, size);
+		if (IS_ERR(v)) {
+			ret = PTR_ERR(v);
+			break;
+		}
+		hello = v;
+
+		if (!check_flags(hello->conn_flags)) {
 			ret = -ENOTSUPP;
 			break;
 		}
 
-		conn->flags = hello.conn_flags;
-		hello.bus_flags = bus->bus_flags;
-		hello.bloom_size = bus->bloom_size;
-		hello.id = conn->id;
+		KDBUS_ITEM_FOREACH_VALIDATE(item, hello) {
+			/* empty data records are invalid */
+			if (item->size <= KDBUS_ITEM_HEADER_SIZE) {
+				ret = -EINVAL;
+				break;
+			}
 
-		if (copy_to_user(buf, &hello, sizeof(hello))) {
+			switch (item->type) {
+			case KDBUS_HELLO_BUFFER:
+				/* enforce page alignment and page granularity */
+				if (!KDBUS_IS_ALIGNED_PAGE(item->vec.address) ||
+				    !KDBUS_IS_ALIGNED_PAGE(item->vec.size)) {
+					ret = -EFAULT;
+					break;
+				}
+
+				conn->buffer.buf = (void *)(uintptr_t)item->vec.address;
+				conn->buffer.size = item->vec.size;
+				break;
+
+			default:
+				ret = -ENOTSUPP;
+			}
+		}
+
+		/* return properties of this connection to the caller */
+		hello->bus_flags = bus->bus_flags;
+		hello->bloom_size = bus->bloom_size;
+		hello->id = conn->id;
+		if (copy_to_user(buf, hello, sizeof(struct kdbus_cmd_hello))) {
 			ret = -EFAULT;
 			break;
 		}
 
+		conn->flags = hello->conn_flags;
 		conn->active = true;
 
 		break;
@@ -484,43 +828,36 @@ static long kdbus_conn_ioctl_ep(struct file *file, unsigned int cmd,
 			return -ENOMEM;
 
 		ret = kdbus_cmd_policy_set_from_user(conn->ep->policy_db, buf);
-
 		break;
 
 	case KDBUS_CMD_NAME_ACQUIRE:
 		/* acquire a well-known name */
 		ret = kdbus_cmd_name_acquire(bus->name_registry, conn, buf);
-
 		break;
 
 	case KDBUS_CMD_NAME_RELEASE:
 		/* release a well-known name */
 		ret = kdbus_cmd_name_release(bus->name_registry, conn, buf);
-
 		break;
 
 	case KDBUS_CMD_NAME_LIST:
 		/* return all current well-known names */
 		ret = kdbus_cmd_name_list(bus->name_registry, conn, buf);
-
 		break;
 
 	case KDBUS_CMD_NAME_QUERY:
 		/* return details about a specific well-known name */
 		ret = kdbus_cmd_name_query(bus->name_registry, conn, buf);
-
 		break;
 
 	case KDBUS_CMD_MATCH_ADD:
 		/* subscribe to/filter for broadcast messages */
 		ret = kdbus_cmd_match_db_add(conn, buf);
-
 		break;
 
 	case KDBUS_CMD_MATCH_REMOVE:
 		/* unsubscribe from broadcast messages */
 		ret = kdbus_cmd_match_db_remove(conn->match_db, buf);
-
 		break;
 
 	case KDBUS_CMD_MONITOR: {
@@ -530,34 +867,42 @@ static long kdbus_conn_ioctl_ep(struct file *file, unsigned int cmd,
 			return -EFAULT;
 
 		conn->monitor = !!cmd_monitor.enabled;
-
 		break;
 	}
 
-	case KDBUS_CMD_MSG_SEND:
-		/* send a message */
+	case KDBUS_CMD_MSG_SEND: {
+		struct kdbus_kmsg *kmsg;
+
+		/* submit a message which will be queued in the receiver */
 		ret = kdbus_kmsg_new_from_user(conn, buf, &kmsg);
 		if (ret < 0)
 			break;
 
 		ret = kdbus_kmsg_send(conn->ep, conn, kmsg);
-		kdbus_kmsg_unref(kmsg);
-
+		kdbus_kmsg_free(kmsg);
 		break;
+	}
 
 	case KDBUS_CMD_MSG_RECV:
-		/* receive a message */
-		ret = kdbus_kmsg_recv(conn, buf);
+		/* receive a pointer to a queued message */
+		ret = kdbus_conn_recv_msg(conn, buf);
+		break;
 
+	case KDBUS_CMD_MSG_FREE: {
+		/* cleanup the memory used in the receiver's buffer */
+		mutex_lock(&conn->lock);
+		kdbus_buffer_free(&conn->buffer, buf);
+		mutex_unlock(&conn->lock);
 		break;
+	}
 
 	default:
 		ret = -ENOTTY;
-
 		break;
 	}
 
 	kfree(kmake);
+	kfree(hello);
 
 	return ret;
 }
@@ -592,10 +937,10 @@ static unsigned int kdbus_conn_poll(struct file *file,
 
 	poll_wait(file, &conn->ep->wait, wait);
 
-	mutex_lock(&conn->msg_lock);
+	mutex_lock(&conn->lock);
 	if (!list_empty(&conn->msg_list))
 		mask |= POLLIN | POLLRDNORM;
-	mutex_unlock(&conn->msg_lock);
+	mutex_unlock(&conn->lock);
 
 	return mask;
 }
diff --git a/connection.h b/connection.h
index d2a28287ce3..0df78a145c3 100644
--- a/connection.h
+++ b/connection.h
@@ -14,6 +14,7 @@
 #define __KDBUS_CONNECTION_H
 
 #include "internal.h"
+#include "buffer.h"
 
 /*
  * kdbus connection
@@ -41,7 +42,7 @@ struct kdbus_conn {
 	bool active;	/* did the connection say hello yet? */
 	bool monitor;
 
-	struct mutex msg_lock;
+	struct mutex lock;
 	struct mutex names_lock;
 	struct mutex accounting_lock;
 	struct list_head msg_list;
@@ -67,12 +68,43 @@ struct kdbus_conn {
 	u32 sec_label_len;
 #endif
 
+	/* reference to the taks owning the connection */
+	struct task_struct *task;
+
+	/* connection accounting */
 	unsigned int msg_count;
 	size_t allocated_size;
+
+	/* userspace-supplied buffer to fill with message data */
+	struct kdbus_buffer buffer;
 };
 
-void kdbus_conn_schedule_timeout_scan(struct kdbus_conn *conn);
-int kdbus_conn_add_size_allocation(struct kdbus_conn *conn, size_t size);
-void kdbus_conn_sub_size_allocation(struct kdbus_conn *conn, size_t size);
+struct kdbus_conn_queue {
+	struct list_head entry;
+
+	/* pointer to message placed in the receiver's buffer */
+	struct __user kdbus_msg *msg;
+
+	/* passed file descriptors */
+	int __user *fds;
+	struct file **fds_fp;
+	unsigned int fds_count;
+
+	/* timeout in the queue */
+	u64 deadline_ns;
+	u64 src_id;
+	u64 cookie;
+	bool expect_reply;
+};
+
+struct kdbus_kmsg;
+
+void kdbus_conn_timeout_schedule_scan(struct kdbus_conn *conn);
+
+int kdbus_conn_accounting_add_size(struct kdbus_conn *conn, size_t size);
+void kdbus_conn_accounting_sub_size(struct kdbus_conn *conn, size_t size);
 
+void kdbus_conn_queue_cleanup(struct kdbus_conn_queue *queue);
+int kdbus_conn_queue_insert(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg,
+			    u64 deadline_ns);
 #endif
diff --git a/internal.h b/internal.h
index 61a56885666..4cc3ab22b99 100644
--- a/internal.h
+++ b/internal.h
@@ -20,15 +20,14 @@
 #define KDBUS_MSG_MAX_FDS		256		/* maximum number of passed file descriptors */
 #define KDBUS_MSG_MAX_PAYLOAD_VECS	64		/* maximum number of payload references */
 #define KDBUS_MSG_MAX_PAYLOAD_SIZE	SZ_128M		/* maximum message payload size */
-#define KDBUS_MSG_MAX_INLINE_SIZE	2 * PAGE_SIZE	/* threshold when to use the external buffer */
 
 #define KDBUS_NAME_MAX_LEN		255		/* maximum length of well-known bus name */
 
 #define KDBUS_MAKE_MAX_LEN		63		/* maximum length of bus, ns, ep name */
 #define KDBUS_MAKE_MAX_SIZE		SZ_32K		/* maximum size of make buffer */
 
+#define KDBUS_HELLO_MAX_SIZE		SZ_32K		/* maximum size of hello buffer */
 #define KDBUS_MATCH_MAX_SIZE		SZ_32K		/* maximum size of match buffer */
-
 #define KDBUS_POLICY_MAX_SIZE		SZ_32K		/* maximum size of policy buffer */
 
 #define KDBUS_CONN_MAX_MSGS		64		/* maximum number of queued messages on the bus */
@@ -36,45 +35,45 @@
 
 #define KDBUS_CHAR_MAJOR		222		/* FIXME: move to uapi/linux/major.h */
 
+#define KDBUS_ALIGN8(s) ALIGN((s), 8)
 #define KDBUS_IS_ALIGNED8(s) (IS_ALIGNED(s, 8))
 #define KDBUS_IS_ALIGNED_PAGE(s) (IS_ALIGNED(s, PAGE_SIZE))
-#define KDBUS_ALIGN8(s) ALIGN((s), 8)
-#define KDBUS_ALIGN_PAGE(s) ALIGN((s), PAGE_SIZE)
 
+#define KDBUS_ITEM_HEADER_SIZE offsetof(struct kdbus_item, data)
+#define KDBUS_ITEM_SIZE(s) KDBUS_ALIGN8((s) + KDBUS_ITEM_HEADER_SIZE)
 #define KDBUS_ITEM_NEXT(item) \
 	(typeof(item))(((u8 *)item) + KDBUS_ALIGN8((item)->size))
 #define KDBUS_ITEM_FOREACH(item, head)						\
 	for (item = (head)->items;						\
 	     (u8 *)(item) < (u8 *)(head) + (head)->size;			\
 	     item = KDBUS_ITEM_NEXT(item))
-
 /* same iterator with more consistency checks, to be used with incoming data */
-#define KDBUS_ITEM_HEADER_SIZE offsetof(struct kdbus_item, data)
-#define KDBUS_ITEM_SIZE(s) KDBUS_ALIGN8((s) + KDBUS_ITEM_HEADER_SIZE)
 #define KDBUS_ITEM_FOREACH_VALIDATE(item, head)					\
 	for (item = (head)->items;						\
 	     (u8 *)(item) + KDBUS_ITEM_HEADER_SIZE <= (u8 *)(head) + (head)->size && \
 	     (u8 *)(item) + (item)->size <= (u8 *)(head) + (head)->size; \
 	     item = KDBUS_ITEM_NEXT(item))
 
+#define KDBUS_MSG_HEADER_SIZE offsetof(struct kdbus_msg, items)
+
 /* some architectures miss get_user_8(); copy only the lower 32bit */
 #ifdef CONFIG_64BIT
 #define kdbus_size_get_user(_s, _b, _t) \
 ({ \
-	u64 __user *_sz = _b + offsetof(typeof(_t), size); \
+	u64 __user *_sz = (void *)(_b) + offsetof(typeof(_t), size); \
 	get_user(_s, _sz); \
 })
 #else
 	#ifdef __LITTLE_ENDIAN__
 	#define kdbus_size_get_user(_s, _b, _t) \
 	({ \
-		u32 __user *_sz = _b + offsetof(typeof(_t), size); \
+		u32 __user *_sz = (void *)(_b) + offsetof(typeof(_t), size); \
 		get_user(_s, _sz); \
 	})
 	#else
 	#define kdbus_size_get_user(_s, _b, _t) \
 	({ \
-		u32 __user *_sz = _b + sizeof(u32) + offsetof(typeof(_t), size); \
+		u32 __user *_sz = (void *)(_b) + sizeof(u32) + offsetof(typeof(_t), size); \
 		get_user(_s, _sz); \
 	})
 	#endif
diff --git a/kdbus.h b/kdbus.h
index 049c988a478..6ae8f1b6d95 100644
--- a/kdbus.h
+++ b/kdbus.h
@@ -70,14 +70,15 @@ enum {
 	KDBUS_MSG_NULL,
 
 	/* Filled in by userspace */
-	KDBUS_MSG_PAYLOAD,		/* .data, inline memory */
 	KDBUS_MSG_PAYLOAD_VEC,		/* .data_vec, reference to memory area */
-	KDBUS_MSG_UNIX_FDS,		/* .data_fds of file descriptors */
+	KDBUS_MSG_PAYLOAD_FD,		/* .data_fd, reference to a data file descriptor */
+	KDBUS_MSG_FDS,			/* .data_fds of file descriptors */
 	KDBUS_MSG_BLOOM,		/* for broadcasts, carries bloom filter blob in .data */
 	KDBUS_MSG_DST_NAME,		/* destination's well-known name, in .str */
+	KDBUS_MSG_PRIORITY,		/* queue priority for message */
 
 	/* Filled in by kernelspace */
-	KDBUS_MSG_SRC_NAMES	= 0x200,/* NUL separated string list with well-known names of source */
+	KDBUS_MSG_SRC_NAMES	= 0x400,/* NUL separated string list with well-known names of source */
 	KDBUS_MSG_TIMESTAMP,		/* .timestamp */
 	KDBUS_MSG_SRC_CREDS,		/* .creds */
 	KDBUS_MSG_SRC_PID_COMM,		/* optional, in .str */
@@ -90,7 +91,7 @@ enum {
 	KDBUS_MSG_SRC_AUDIT,		/* .audit */
 
 	/* Special messages from kernel, consisting of one and only one of these data blocks */
-	KDBUS_MSG_NAME_ADD	= 0x400,/* .name_change */
+	KDBUS_MSG_NAME_ADD	= 0x800,/* .name_change */
 	KDBUS_MSG_NAME_REMOVE,		/* .name_change */
 	KDBUS_MSG_NAME_CHANGE,		/* .name_change */
 	KDBUS_MSG_ID_ADD,		/* .id_change */
@@ -99,14 +100,9 @@ enum {
 	KDBUS_MSG_REPLY_DEAD,		/* dito */
 };
 
-enum {
-	KDBUS_VEC_ALIGNED		= 1 <<  0,
-};
-
 struct kdbus_vec {
 	__u64 address;
 	__u64 size;
-	__u64 flags;
 };
 
 /**
@@ -236,6 +232,8 @@ enum {
 /* Items to append to struct kdbus_cmd_hello */
 enum {
 	KDBUS_HELLO_NULL,
+	KDBUS_HELLO_BUFFER,	/* kdbus_vec, userspace supplied buffer to
+				 * place received messages */
 };
 
 struct kdbus_cmd_hello {
@@ -401,7 +399,8 @@ enum kdbus_cmd {
 
 	/* kdbus ep node commands: require connected state */
 	KDBUS_CMD_MSG_SEND =		_IOWR(KDBUS_IOC_MAGIC, 0x40, struct kdbus_msg),
-	KDBUS_CMD_MSG_RECV =		_IOWR(KDBUS_IOC_MAGIC, 0x41, struct kdbus_msg),
+	KDBUS_CMD_MSG_RECV =		_IOWR(KDBUS_IOC_MAGIC, 0x41, struct kdbus_msg *),
+	KDBUS_CMD_MSG_FREE =		_IOWR(KDBUS_IOC_MAGIC, 0x42, struct kdbus_msg),
 
 	KDBUS_CMD_NAME_ACQUIRE =	_IOWR(KDBUS_IOC_MAGIC, 0x50, struct kdbus_cmd_name),
 	KDBUS_CMD_NAME_RELEASE =	_IOWR(KDBUS_IOC_MAGIC, 0x51, struct kdbus_cmd_name),
diff --git a/match.c b/match.c
index 6cbde0e855e..6cbbfbcdd16 100644
--- a/match.c
+++ b/match.c
@@ -149,12 +149,7 @@ bool kdbus_match_db_match_with_src(struct kdbus_match_db *db,
 				   struct kdbus_conn *conn_src,
 				   struct kdbus_kmsg *kmsg)
 {
-	const struct kdbus_item *bloom =
-		kdbus_msg_get_item(&kmsg->msg, KDBUS_MSG_BLOOM, 0);
-	const struct kdbus_item *src_names =
-		kdbus_msg_get_item(&kmsg->msg, KDBUS_MSG_SRC_NAMES, 0);
 	struct kdbus_match_db_entry *e;
-	size_t bloom_size = conn_src->ep->bus->bloom_size / sizeof(u64);
 	bool matched = false;
 
 	mutex_lock(&db->entries_lock);
@@ -168,19 +163,22 @@ bool kdbus_match_db_match_with_src(struct kdbus_match_db *db,
 		matched = true;
 
 		list_for_each_entry(ei, &e->items_list, list_entry) {
-			if (bloom && ei->type == KDBUS_MATCH_BLOOM)
+			if (kmsg->bloom) {
+				size_t bloom_size =
+					conn_src->ep->bus->bloom_size / sizeof(u64);
+
 				if (!kdbus_match_db_test_bloom(ei->bloom,
-							       bloom->data64,
+							       kmsg->bloom,
 							       bloom_size)) {
 					matched = false;
 					break;
 				}
+			}
 
-			if (src_names && ei->type == KDBUS_MATCH_SRC_NAME) {
-				size_t nlen = src_names->size - KDBUS_ITEM_HEADER_SIZE;
-
-				if (!kdbus_match_db_test_src_names(src_names->str,
-								   nlen, ei->name)) {
+			if (kmsg->src_names) {
+				if (!kdbus_match_db_test_src_names(kmsg->src_names,
+								   kmsg->src_names_len,
+								   ei->name)) {
 					matched = false;
 					break;
 				}
diff --git a/message.c b/message.c
index e87362a0b3c..616146bedbe 100644
--- a/message.c
+++ b/message.c
@@ -15,13 +15,12 @@
 #include <linux/module.h>
 #include <linux/device.h>
 #include <linux/idr.h>
-#include <linux/fs.h>
 #include <linux/slab.h>
+#include <linux/file.h>
 #include <linux/sched.h>
 #include <linux/mutex.h>
 #include <linux/init.h>
 #include <linux/poll.h>
-#include <linux/file.h>
 #include <linux/cgroup.h>
 #include <linux/cred.h>
 #include <linux/capability.h>
@@ -35,61 +34,32 @@
 #include "names.h"
 #include "match.h"
 
-#define KDBUS_MSG_HEADER_SIZE offsetof(struct kdbus_msg, items)
 #define KDBUS_KMSG_HEADER_SIZE offsetof(struct kdbus_kmsg, msg)
 
-static void kdbus_msg_dump(const struct kdbus_msg *msg);
-
-static void kdbus_kmsg_free(struct kdbus_kmsg *kmsg)
-{
-	size_t size = 0;
-
-	if (kmsg->fds_fp) {
-		unsigned int i;
-
-		for (i = 0; i < kmsg->fds_count; i++)
-			fput(kmsg->fds_fp[i]);
-		size += kmsg->fds_count * sizeof(struct file *);
-		kfree(kmsg->fds_fp);
-	}
-
-	if (kmsg->fds) {
-		size += KDBUS_ITEM_HEADER_SIZE + (kmsg->fds_count * sizeof(int));
-		kfree(kmsg->fds);
-	}
-
-	if (kmsg->meta) {
-		size += kmsg->meta_allocated_size;
-		kfree(kmsg->meta);
-	}
-
-	if (kmsg->vecs) {
-		size += kmsg->vecs_size;
-		kfree(kmsg->vecs);
-	}
-
-	size += kmsg->msg.size + KDBUS_KMSG_HEADER_SIZE;
-	kdbus_conn_sub_size_allocation(kmsg->conn_src, size);
-
-	kfree(kmsg);
-}
-
-static void __kdbus_kmsg_free(struct kref *kref)
+static void __maybe_unused kdbus_msg_dump(const struct kdbus_msg *msg)
 {
-	struct kdbus_kmsg *kmsg = container_of(kref, struct kdbus_kmsg, kref);
+	const struct kdbus_item *item;
 
-	return kdbus_kmsg_free(kmsg);
-}
+	pr_debug("msg size=%llu, flags=0x%llx, dst_id=%llu, src_id=%llu, "
+		 "cookie=0x%llx payload_type=0x%llx, timeout=%llu\n",
+		 (unsigned long long) msg->size,
+		 (unsigned long long) msg->flags,
+		 (unsigned long long) msg->dst_id,
+		 (unsigned long long) msg->src_id,
+		 (unsigned long long) msg->cookie,
+		 (unsigned long long) msg->payload_type,
+		 (unsigned long long) msg->timeout_ns);
 
-void kdbus_kmsg_unref(struct kdbus_kmsg *kmsg)
-{
-	kref_put(&kmsg->kref, __kdbus_kmsg_free);
+	KDBUS_ITEM_FOREACH(item, msg)
+		pr_debug("`- msg_item size=%llu, type=0x%llx\n",
+			 item->size, item->type);
 }
 
-static struct kdbus_kmsg *kdbus_kmsg_ref(struct kdbus_kmsg *kmsg)
+void kdbus_kmsg_free(struct kdbus_kmsg *kmsg)
 {
-	kref_get(&kmsg->kref);
-	return kmsg;
+	if (kmsg->meta)
+		kfree(kmsg->meta);
+	kfree(kmsg);
 }
 
 int kdbus_kmsg_new(size_t extra_size, struct kdbus_kmsg **m)
@@ -116,9 +86,8 @@ static int kdbus_msg_scan_items(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg
 	const struct kdbus_item *item;
 	unsigned int num_items = 0;
 	unsigned int num_vecs = 0;
-	unsigned int num_fds = 0;
+	unsigned int fds_count = 0;
 	size_t vecs_size = 0;
-	bool needs_vec = false;
 	bool has_fds = false;
 	bool has_name = false;
 	bool has_bloom = false;
@@ -132,9 +101,6 @@ static int kdbus_msg_scan_items(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg
 			return -E2BIG;
 
 		switch (item->type) {
-		case KDBUS_MSG_PAYLOAD:
-			break;
-
 		case KDBUS_MSG_PAYLOAD_VEC:
 			if (item->size != KDBUS_ITEM_HEADER_SIZE + sizeof(struct kdbus_vec))
 				return -EINVAL;
@@ -142,22 +108,12 @@ static int kdbus_msg_scan_items(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg
 			if (++num_vecs > KDBUS_MSG_MAX_PAYLOAD_VECS)
 				return -E2BIG;
 
-			if (item->vec.flags & KDBUS_VEC_ALIGNED) {
-				/* enforce page alignment and page granularity */
-				if (!KDBUS_IS_ALIGNED_PAGE(item->vec.address) ||
-				    !KDBUS_IS_ALIGNED_PAGE(item->vec.size))
-					return -EFAULT;
-
-				/* we always deliver aligned data as PAYLOAD_VEC */
-				needs_vec = true;
-			}
-
-			vecs_size += KDBUS_ALIGN8(item->vec.size);
+			vecs_size += item->vec.size;
 			if (vecs_size > KDBUS_MSG_MAX_PAYLOAD_SIZE)
 				return -EMSGSIZE;
 			break;
 
-		case KDBUS_MSG_UNIX_FDS:
+		case KDBUS_MSG_FDS:
 			/* do not allow multiple fd arrays */
 			if (has_fds)
 				return -EEXIST;
@@ -167,9 +123,12 @@ static int kdbus_msg_scan_items(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg
 			if (msg->dst_id == KDBUS_DST_ID_BROADCAST)
 				return -ENOTUNIQ;
 
-			num_fds = (item->size - KDBUS_ITEM_HEADER_SIZE) / sizeof(int);
-			if (num_fds > KDBUS_MSG_MAX_FDS)
+			fds_count = (item->size - KDBUS_ITEM_HEADER_SIZE) / sizeof(int);
+			if (fds_count > KDBUS_MSG_MAX_FDS)
 				return -EMFILE;
+
+			kmsg->fds = item->fds;
+			kmsg->fds_count = fds_count;
 			break;
 
 		case KDBUS_MSG_BLOOM:
@@ -189,6 +148,8 @@ static int kdbus_msg_scan_items(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg
 			/* do not allow mismatching bloom filter sizes */
 			if (item->size - KDBUS_ITEM_HEADER_SIZE != conn->ep->bus->bloom_size)
 				return -EDOM;
+
+			kmsg->bloom = item->data64;
 			break;
 
 		case KDBUS_MSG_DST_NAME:
@@ -203,6 +164,8 @@ static int kdbus_msg_scan_items(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg
 
 			if (!kdbus_name_is_valid(item->str))
 				return -EINVAL;
+
+			kmsg->dst_name = item->str;
 			break;
 
 		default:
@@ -214,7 +177,7 @@ static int kdbus_msg_scan_items(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg
 	if ((char *)item - ((char *)msg + msg->size) >= 8)
 		return -EINVAL;
 
-	/* name is needed for broadcast */
+	/* name is needed if no ID is given */
 	if (msg->dst_id == KDBUS_DST_ID_WELL_KNOWN_NAME && !has_name)
 		return -EDESTADDRREQ;
 
@@ -223,128 +186,36 @@ static int kdbus_msg_scan_items(struct kdbus_conn *conn, struct kdbus_kmsg *kmsg
 	    msg->dst_id < KDBUS_DST_ID_BROADCAST && has_name)
 		return -EBADMSG;
 
-	/* broadcast messages require a bloom filter */
-	if (msg->dst_id == KDBUS_DST_ID_BROADCAST && !has_bloom)
-		return -EBADMSG;
+	if (msg->dst_id == KDBUS_DST_ID_BROADCAST) {
+		/* broadcast messages require a bloom filter */
+		if (!has_bloom)
+			return -EBADMSG;
+
+		/* timeouts are not allowed for broadcasts */
+		if (msg->timeout_ns)
+			return -ENOTUNIQ;
+	}
 
 	/* bloom filters are for undirected messages only */
 	if (has_name && has_bloom)
 		return -EBADMSG;
 
-	/* allocate array for file descriptors */
-	if (has_fds) {
-		size_t size;
-		unsigned int i;
-		int ret;
-
-		size = num_fds * sizeof(struct file *);
-		ret = kdbus_conn_add_size_allocation(conn, size);
-		if (ret < 0)
-			return ret;
-
-		kmsg->fds_fp = kzalloc(size, GFP_KERNEL);
-		if (!kmsg->fds_fp)
-			return -ENOMEM;
-
-		size = KDBUS_ITEM_HEADER_SIZE + (num_fds * sizeof(int));
-		ret = kdbus_conn_add_size_allocation(conn, size);
-		if (ret < 0)
-			return ret;
-
-		kmsg->fds = kmalloc(size, GFP_KERNEL);
-		if (!kmsg->fds)
-			return -ENOMEM;
-		for (i = 0; i < num_fds; i++)
-			kmsg->fds->fds[i] = -1;
-	}
-
-	/* if we have only very small PAYLOAD_VECs, they get inlined */
-	if (num_vecs > 0 && !needs_vec &&
-	    msg->size + vecs_size < KDBUS_MSG_MAX_INLINE_SIZE) {
-		size_t size;
-		int ret;
-
-		size = (num_vecs * KDBUS_ITEM_HEADER_SIZE) + vecs_size;
-		ret = kdbus_conn_add_size_allocation(conn, size);
-		if (ret < 0)
-			return ret;
-
-		kmsg->vecs = kzalloc(size, GFP_KERNEL);
-		if (!kmsg->vecs)
-			return -ENOMEM;
-		kmsg->vecs_size = size;
-	}
-
-	return 0;
-}
-
-static int kdbus_inline_user_vec(struct kdbus_kmsg *kmsg,
-				 struct kdbus_item *next,
-				 const struct kdbus_item *item)
-{
-	void __user *user_addr;
-
-	user_addr = (void __user *)(uintptr_t)item->vec.address;
-	if (copy_from_user(next->data, user_addr, item->vec.size))
-		return -EFAULT;
-
-	next->type = KDBUS_MSG_PAYLOAD;
-	next->size = KDBUS_ITEM_HEADER_SIZE + item->vec.size;
-
+	kmsg->vecs_size = vecs_size;
 	return 0;
 }
 
-/*
- * Grab and keep references to passed files descriptors, to install
- * them in the receiving process at message delivery.
- */
-static int kdbus_copy_user_fds(struct kdbus_kmsg *kmsg,
-			       const struct kdbus_item *item)
-{
-	unsigned int i;
-	unsigned int count;
-
-	count = (item->size - KDBUS_ITEM_HEADER_SIZE) / sizeof(int);
-	for (i = 0; i < count; i++) {
-		struct file *fp;
-
-		fp = fget(item->fds[i]);
-		if (!fp)
-			goto unwind;
-
-		kmsg->fds_fp[kmsg->fds_count++] = fp;
-	}
-
-	return 0;
-
-unwind:
-	while (i >= 0) {
-		fput(kmsg->fds_fp[i]);
-		kmsg->fds_fp[i] = NULL;
-		i--;
-	}
-
-	kmsg->fds_count = 0;
-	return -EBADF;
-}
-
-/*
- * Check the validity of a message. The general layout of the received message
- * is not altered before it is delivered.
- */
-int kdbus_kmsg_new_from_user(struct kdbus_conn *conn, void __user *buf,
+int kdbus_kmsg_new_from_user(struct kdbus_conn *conn,
+			     struct kdbus_msg __user *msg,
 			     struct kdbus_kmsg **m)
 {
 	struct kdbus_kmsg *kmsg;
-	const struct kdbus_item *item;
-	struct kdbus_item *vecs_next;
 	u64 size, alloc_size;
 	int ret;
 
-	if (!KDBUS_IS_ALIGNED8((unsigned long)buf))
+	if (!KDBUS_IS_ALIGNED8((unsigned long)msg))
 		return -EFAULT;
 
-	if (kdbus_size_get_user(size, buf, struct kdbus_msg))
+	if (kdbus_size_get_user(size, msg, struct kdbus_msg))
 		return -EFAULT;
 
 	if (size < sizeof(struct kdbus_msg) || size > KDBUS_MSG_MAX_SIZE)
@@ -357,61 +228,20 @@ int kdbus_kmsg_new_from_user(struct kdbus_conn *conn, void __user *buf,
 		return -ENOMEM;
 	memset(kmsg, 0, KDBUS_KMSG_HEADER_SIZE);
 
-	ret = kdbus_conn_add_size_allocation(conn, alloc_size);
-	if (ret < 0) {
-		kfree(kmsg);
-		return ret;
-	}
-
-	if (copy_from_user(&kmsg->msg, buf, size)) {
+	if (copy_from_user(&kmsg->msg, msg, size)) {
 		ret = -EFAULT;
 		goto exit_free;
 	}
 
-	/* check validity and prepare handling of data records */
+	/* check validity and gather some values for processing */
 	ret = kdbus_msg_scan_items(conn, kmsg);
 	if (ret < 0)
 		goto exit_free;
 
-	/* fill in sender ID */
+	/* patch-in the source of this message */
 	kmsg->msg.src_id = conn->id;
 
-	/* keep a reference to the source connection, for accounting */
-	kmsg->conn_src = conn;
-
-	/* Iterate over the items, resolve external references to data
-	 * we need to pass to the receiver; ignore all items used by
-	 * the sender only. */
-	vecs_next = kmsg->vecs;
-	KDBUS_ITEM_FOREACH(item, &kmsg->msg) {
-		switch (item->type) {
-		case KDBUS_MSG_PAYLOAD_VEC:
-			if (!kmsg->vecs) {
-				ret = -ENOSYS;
-				goto exit_free;
-			}
-
-			/* convert PAYLOAD_VEC to PAYLOAD */
-			ret = kdbus_inline_user_vec(kmsg, vecs_next, item);
-			if (ret < 0)
-				goto exit_free;
-			vecs_next = KDBUS_ITEM_NEXT(vecs_next);
-			break;
-
-		case KDBUS_MSG_UNIX_FDS:
-			ret = kdbus_copy_user_fds(kmsg, item);
-			if (ret < 0)
-				goto exit_free;
-			break;
-
-		case KDBUS_MSG_BLOOM:
-			//FIXME: store in kmsg
-			break;
-		}
-	}
-
 	kref_init(&kmsg->kref);
-
 	*m = kmsg;
 	return 0;
 
@@ -420,51 +250,15 @@ exit_free:
 	return ret;
 }
 
-const struct kdbus_item *
-kdbus_msg_get_item(const struct kdbus_msg *msg, u64 type, unsigned int index)
-{
-	const struct kdbus_item *item;
-
-	KDBUS_ITEM_FOREACH(item, msg)
-		if (item->type == type && index-- == 0)
-			return item;
-
-	return NULL;
-}
-
-static void __maybe_unused kdbus_msg_dump(const struct kdbus_msg *msg)
-{
-	const struct kdbus_item *item;
-
-	pr_debug("msg size=%llu, flags=0x%llx, dst_id=%llu, src_id=%llu, "
-		 "cookie=0x%llx payload_type=0x%llx, timeout=%llu\n",
-		 (unsigned long long) msg->size,
-		 (unsigned long long) msg->flags,
-		 (unsigned long long) msg->dst_id,
-		 (unsigned long long) msg->src_id,
-		 (unsigned long long) msg->cookie,
-		 (unsigned long long) msg->payload_type,
-		 (unsigned long long) msg->timeout_ns);
-
-	KDBUS_ITEM_FOREACH(item, msg)
-		pr_debug("`- msg_item size=%llu, type=0x%llx\n",
-			 item->size, item->type);
-}
-
 static struct kdbus_item *
 kdbus_kmsg_append(struct kdbus_kmsg *kmsg, size_t extra_size)
 {
 	struct kdbus_item *item;
 	size_t size;
-	int ret;
 
 	/* get new metadata buffer, pre-allocate at least 512 bytes */
 	if (!kmsg->meta) {
 		size = roundup_pow_of_two(256 + KDBUS_ALIGN8(extra_size));
-		ret = kdbus_conn_add_size_allocation(kmsg->conn_src, size);
-		if (ret < 0)
-			return ERR_PTR(ret);
-
 		kmsg->meta = kzalloc(size, GFP_KERNEL);
 		if (!kmsg->meta)
 			return ERR_PTR(-ENOMEM);
@@ -480,11 +274,6 @@ kdbus_kmsg_append(struct kdbus_kmsg *kmsg, size_t extra_size)
 
 		size = roundup_pow_of_two(size);
 		size_diff = size - kmsg->meta_allocated_size;
-
-		ret = kdbus_conn_add_size_allocation(kmsg->conn_src, size_diff);
-		if (ret < 0)
-			return ERR_PTR(ret);
-
 		pr_debug("%s: grow to size=%zu\n", __func__, size);
 		meta = kmalloc(size, GFP_KERNEL);
 		if (!meta)
@@ -589,6 +378,9 @@ static int kdbus_kmsg_append_src_names(struct kdbus_kmsg *kmsg,
 		pos += strlen(name_entry->name) + 1;
 	}
 
+	kmsg->src_names = item->data;
+	kmsg->src_names_len = pos;
+
 exit_unlock:
 	mutex_unlock(&conn->names_lock);
 
@@ -612,40 +404,6 @@ static int kdbus_kmsg_append_cred(struct kdbus_kmsg *kmsg,
 	return 0;
 }
 
-static int kdbus_conn_enqueue_kmsg(struct kdbus_conn *conn,
-				   struct kdbus_kmsg *kmsg)
-{
-	struct kdbus_msg_list_entry *entry;
-	int ret = 0;
-
-	if (!conn->active)
-		return -ENOTCONN;
-
-	if (kmsg->fds && !(conn->flags & KDBUS_HELLO_ACCEPT_FD))
-		return -ECOMM;
-
-	entry = kzalloc(sizeof(*entry), GFP_KERNEL);
-	if (!entry)
-		return -ENOMEM;
-
-	entry->kmsg = kdbus_kmsg_ref(kmsg);
-	INIT_LIST_HEAD(&entry->entry);
-
-	mutex_lock(&conn->msg_lock);
-	if (conn->msg_count > KDBUS_CONN_MAX_MSGS) {
-		ret = -EXFULL;
-	} else {
-		list_add_tail(&entry->entry, &conn->msg_list);
-		conn->msg_count++;
-	}
-	mutex_unlock(&conn->msg_lock);
-
-	if (ret == 0)
-		wake_up_interruptible(&conn->ep->wait);
-
-	return ret;
-}
-
 /*
  * FIXME: dirty and unsafe version of:
  *   http://git.kernel.org/cgit/linux/kernel/git/tj/cgroup.git/commit/?h=review-task_cgroup_path_from_hierarchy
@@ -686,7 +444,7 @@ int task_cgroup_path_from_hierarchy(struct task_struct *task, int hierarchy_id,
 	return ret;
 }
 
-static int kdbus_msg_append_for_dst(struct kdbus_kmsg *kmsg,
+static int kdbus_kmsg_append_for_dst(struct kdbus_kmsg *kmsg,
 				    struct kdbus_conn *conn_src,
 				    struct kdbus_conn *conn_dst)
 {
@@ -855,6 +613,7 @@ int kdbus_kmsg_send(struct kdbus_ep *ep,
 	struct kdbus_conn *conn_dst = NULL;
 	const struct kdbus_msg *msg;
 	u64 now_ns = 0;
+	u64 deadline_ns = 0;
 	int ret;
 
 	/* augment incoming message */
@@ -873,70 +632,9 @@ int kdbus_kmsg_send(struct kdbus_ep *ep,
 	}
 
 	msg = &kmsg->msg;
-//	kdbus_msg_dump(msg);
-
-	if (msg->dst_id == KDBUS_DST_ID_WELL_KNOWN_NAME) {
-		const struct kdbus_item *name_item;
-		const struct kdbus_name_entry *name_entry;
-
-		name_item = kdbus_msg_get_item(msg, KDBUS_MSG_DST_NAME, 0);
-		if (!name_item)
-			return -EDESTADDRREQ;
-
-		/* lookup and determine conn_dst ... */
-		name_entry = kdbus_name_lookup(ep->bus->name_registry,
-					       name_item->data);
-		if (!name_entry)
-			return -ESRCH;
-
-		conn_dst = name_entry->conn;
-
-		if ((msg->flags & KDBUS_MSG_FLAGS_NO_AUTO_START) &&
-		    (conn_dst->flags & KDBUS_HELLO_STARTER))
-			return -EADDRNOTAVAIL;
-
-	} else if (msg->dst_id != KDBUS_DST_ID_BROADCAST) {
-		/* direct message */
-		conn_dst = kdbus_bus_find_conn_by_id(ep->bus, msg->dst_id);
-		if (!conn_dst)
-			return -ENXIO;
-	}
-
-	if (conn_dst) {
-		/* direct message */
-
-		if (msg->timeout_ns)
-			kmsg->deadline_ns = now_ns + msg->timeout_ns;
-
-		/* check policy */
-		if (ep->policy_db && conn_src) {
-			ret = kdbus_policy_db_check_send_access(ep->policy_db,
-								conn_src,
-								conn_dst,
-								kmsg->deadline_ns);
-			if (ret < 0)
-				return ret;
-		}
-
-		/* direct message */
-		if (conn_src) {
-			ret = kdbus_msg_append_for_dst(kmsg, conn_src, conn_dst);
-			if (ret < 0)
-				return ret;
-		}
-
-		ret = kdbus_conn_enqueue_kmsg(conn_dst, kmsg);
-
-		if (msg->timeout_ns)
-			kdbus_conn_schedule_timeout_scan(conn_dst);
-	} else {
-		/* broadcast */
-		/* timeouts are not allowed for broadcasts */
-		if (msg->timeout_ns)
-			return -ENOTUNIQ;
-
-		ret = 0;
 
+	/* broadcast message */
+	if (msg->dst_id == KDBUS_DST_ID_BROADCAST) {
 		list_for_each_entry(conn_dst, &ep->connection_list,
 				    connection_entry) {
 			if (conn_dst->type != KDBUS_CONN_EP)
@@ -954,170 +652,56 @@ int kdbus_kmsg_send(struct kdbus_ep *ep,
 						       kmsg))
 				continue;
 
-			ret = kdbus_conn_enqueue_kmsg(conn_dst, kmsg);
+			ret = kdbus_conn_queue_insert(conn_dst, kmsg, 0);
 			if (ret < 0)
 				break;
 		}
-	}
-
-	return ret;
-}
-
-int kdbus_kmsg_recv(struct kdbus_conn *conn, void __user *buf)
-{
-	struct kdbus_msg_list_entry *entry;
-	const struct kdbus_kmsg *kmsg = NULL;
-	const struct kdbus_msg *msg;
-	const struct kdbus_item *item;
-	const struct kdbus_item *vecs_next;
-	u64 size, pos, max_size;
-	int ret;
 
-	if (!KDBUS_IS_ALIGNED8((unsigned long)buf))
-		return -EFAULT;
-
-	if (kdbus_size_get_user(size, buf, struct kdbus_msg))
-		return -EFAULT;
-
-	mutex_lock(&conn->msg_lock);
-	if (conn->msg_count == 0) {
-		ret = -EAGAIN;
-		goto exit_unlock;
-	}
-
-	entry = list_first_entry(&conn->msg_list, struct kdbus_msg_list_entry, entry);
-	kmsg = entry->kmsg;
-	msg = &kmsg->msg;
-
-	max_size = msg->size;
-
-	if (kmsg->meta)
-		max_size += kmsg->meta->size;
-
-	if (kmsg->vecs)
-		max_size += kmsg->vecs_size;
-
-	/* reuturn needed buffer size to the receiver */
-	if (size < max_size) {
-		kdbus_size_set_user(max_size, buf, struct kdbus_msg);
-		ret = -ENOBUFS;
-		goto exit_unlock;
-	}
-
-	/* copy the message header */
-	if (copy_to_user(buf, msg, KDBUS_MSG_HEADER_SIZE)) {
-		ret = -EFAULT;
-		goto exit_unlock;
+		return ret;
 	}
 
-	pos = KDBUS_MSG_HEADER_SIZE;
-
-	/* The order and sequence of PAYLOAD and PAYLOAD_VEC is always
-	 * preserved, it might have meaning in the sender/receiver contract;
-	 * one type is freely concerted to the other though, depending
-	 * on the actual copying strategy */
-	vecs_next = kmsg->vecs;
-	KDBUS_ITEM_FOREACH(item, msg) {
-		switch (item->type) {
-		case KDBUS_MSG_PAYLOAD:
-			if (copy_to_user(buf + pos, item, item->size)) {
-				ret = -EFAULT;
-				goto exit_unlock;
-			}
-
-			pos += KDBUS_ALIGN8(item->size);
-			break;
+	/* direct message */
+	if (msg->dst_id == KDBUS_DST_ID_WELL_KNOWN_NAME) {
+		const struct kdbus_name_entry *name_entry;
 
-		case KDBUS_MSG_PAYLOAD_VEC:
-			if (!kmsg->vecs) {
-				/* copy PAYLOAD_VEC from the sender to the receiver */
-				ret = -ENOSYS;
-				goto exit_unlock;
-				break;
-			}
+		name_entry = kdbus_name_lookup(ep->bus->name_registry,
+					       kmsg->dst_name);
+		if (!name_entry)
+			return -ESRCH;
+		conn_dst = name_entry->conn;
 
-			/* copy PAYLOAD_VEC we converted to PAYLOAD */
-			if (copy_to_user(buf + pos, vecs_next, vecs_next->size)) {
-				ret = -EFAULT;
-				goto exit_unlock;
-			}
+		if ((msg->flags & KDBUS_MSG_FLAGS_NO_AUTO_START) &&
+		    (conn_dst->flags & KDBUS_HELLO_STARTER))
+			return -EADDRNOTAVAIL;
 
-			pos += KDBUS_ALIGN8(vecs_next->size);
-			vecs_next = KDBUS_ITEM_NEXT(vecs_next);
-			break;
-		}
+	} else {
+		conn_dst = kdbus_bus_find_conn_by_id(ep->bus, msg->dst_id);
+		if (!conn_dst)
+			return -ENXIO;
 	}
 
-	/* install file descriptors */
-	if (kmsg->fds) {
-		unsigned int i;
-		size_t size;
-
-		for (i = 0; i < kmsg->fds_count; i++) {
-			int fd;
-
-			fd = get_unused_fd();
-			if (fd < 0) {
-				ret = fd;
-				goto exit_unlock_fds;
-			}
-
-			fd_install(fd, get_file(kmsg->fds_fp[i]));
-			kmsg->fds->fds[i] = fd;
-		}
+	if (msg->timeout_ns)
+		deadline_ns = now_ns + msg->timeout_ns;
 
-		size = KDBUS_ITEM_HEADER_SIZE + (sizeof(int) * kmsg->fds_count);
-		kmsg->fds->size = size;
-		kmsg->fds->type = KDBUS_MSG_UNIX_FDS;
-
-		if (copy_to_user(buf + pos, kmsg->fds, size)) {
-			ret = -EFAULT;
-			goto exit_unlock_fds;
-		}
-
-		pos += KDBUS_ALIGN8(size);
+	if (ep->policy_db && conn_src) {
+		ret = kdbus_policy_db_check_send_access(ep->policy_db,
+							conn_src,
+							conn_dst,
+							deadline_ns);
+		if (ret < 0)
+			return ret;
 	}
 
-	/* append metadata records */
-	if (kmsg->meta) {
-		if (copy_to_user(buf + pos, kmsg->meta, kmsg->meta_size)) {
-			ret = -EFAULT;
-			goto exit_unlock_fds;
-		}
-
-		pos += KDBUS_ALIGN8(kmsg->meta_size);
+	if (conn_src) {
+		ret = kdbus_kmsg_append_for_dst(kmsg, conn_src, conn_dst);
+		if (ret < 0)
+			return ret;
 	}
 
-	/* update the returned data size in the message header */
-	ret = kdbus_size_set_user(pos, buf, struct kdbus_msg);
-	if (ret)
-		goto exit_unlock_fds;
-
-	conn->msg_count--;
-	list_del(&entry->entry);
-	kdbus_kmsg_unref(entry->kmsg);
-	kfree(entry);
-	mutex_unlock(&conn->msg_lock);
-
-	return 0;
+	ret = kdbus_conn_queue_insert(conn_dst, kmsg, deadline_ns);
 
-exit_unlock_fds:
-	/* cleanup installed file descriptors */
-	if (kmsg->fds) {
-		unsigned int i;
-
-		for (i = 0; i < kmsg->fds_count; i++) {
-			if (kmsg->fds->fds[i] < 0)
-				continue;
-
-			fput(kmsg->fds_fp[i]);
-			put_unused_fd(kmsg->fds->fds[i]);
-			kmsg->fds->fds[i] = -1;
-		}
-	}
-
-exit_unlock:
-	mutex_unlock(&conn->msg_lock);
+	if (msg->timeout_ns)
+		kdbus_conn_timeout_schedule_scan(conn_dst);
 
 	return ret;
 }
diff --git a/message.h b/message.h
index 196c130b9d7..57d4c166af3 100644
--- a/message.h
+++ b/message.h
@@ -16,43 +16,33 @@
 
 struct kdbus_kmsg {
 	struct kref kref;
-	u64 deadline_ns;
 
-	/* short-hand for faster match db lookup. */
+	/* short-cuts for faster lookup */
 	u64 notification_type;
+	const char *dst_name;
+	const char *src_names;
+	size_t src_names_len;
+	const u64 *bloom;
+	unsigned int bloom_size;
+	const int *fds;
+	unsigned int fds_count;
 
 	/* appended SCM-like metadata */
 	struct kdbus_item *meta;
 	size_t meta_size;
 	size_t meta_allocated_size;
 
-	/* inlined PAYLOAD_VECs */
-	struct kdbus_item *vecs;
+	/* size of PAYLOAD_VECs data */
 	size_t vecs_size;
 
-	/* passed file descriptors */
-	struct kdbus_item *fds;
-	struct file **fds_fp;
-	unsigned int fds_count;
-
-	struct kdbus_conn *conn_src;
 	struct kdbus_msg msg;
 };
 
-struct kdbus_msg_list_entry {
-	struct kdbus_kmsg *kmsg;
-	struct list_head entry;
-};
-
 struct kdbus_ep;
 struct kdbus_conn;
 
 int kdbus_kmsg_new(size_t extra_size, struct kdbus_kmsg **m);
-int kdbus_kmsg_new_from_user(struct kdbus_conn *conn, void __user *argp, struct kdbus_kmsg **m);
-const struct kdbus_item *kdbus_msg_get_item(const struct kdbus_msg *msg, u64 type, unsigned int index);
-void kdbus_kmsg_unref(struct kdbus_kmsg *kmsg);
-int kdbus_kmsg_send(struct kdbus_ep *ep,
-		    struct kdbus_conn *conn_src,
-		    struct kdbus_kmsg *kmsg);
-int kdbus_kmsg_recv(struct kdbus_conn *conn, void __user *buf);
+int kdbus_kmsg_new_from_user(struct kdbus_conn *conn, struct kdbus_msg __user *msg, struct kdbus_kmsg **m);
+int kdbus_kmsg_send(struct kdbus_ep *ep, struct kdbus_conn *conn_src, struct kdbus_kmsg *kmsg);
+void kdbus_kmsg_free(struct kdbus_kmsg *kmsg);
 #endif
diff --git a/notify.c b/notify.c
index 207bb14bb30..5147e5bf54d 100644
--- a/notify.c
+++ b/notify.c
@@ -27,17 +27,15 @@
 #include "ep.h"
 #include "message.h"
 
-static int kdbus_notify_reply(struct kdbus_ep *ep,
-			      const struct kdbus_msg *orig_msg,
-			      u64 msg_type)
+static int kdbus_notify_reply(struct kdbus_ep *ep, u64 src_id,
+			      u64 cookie, u64 msg_type)
 {
 	struct kdbus_conn *dst_conn;
 	struct kdbus_kmsg *kmsg;
 	struct kdbus_item *item;
-	u64 dst_id = orig_msg->src_id;
 	int ret;
 
-	dst_conn = kdbus_bus_find_conn_by_id(ep->bus, dst_id);
+	dst_conn = kdbus_bus_find_conn_by_id(ep->bus, src_id);
 	if (!dst_conn)
 		return -ENXIO;
 
@@ -52,30 +50,28 @@ static int kdbus_notify_reply(struct kdbus_ep *ep,
 	 */
 	kmsg->notification_type = msg_type;
 
-	kmsg->msg.dst_id = dst_id;
+	kmsg->msg.dst_id = src_id;
 	kmsg->msg.src_id = KDBUS_SRC_ID_KERNEL;
 	kmsg->msg.payload_type = KDBUS_PAYLOAD_NULL;
-	kmsg->msg.cookie_reply = orig_msg->cookie;
+	kmsg->msg.cookie_reply = cookie;
 
 	item = kmsg->msg.items;
 	item->type = msg_type;
 
 	ret = kdbus_kmsg_send(ep, NULL, kmsg);
-	kdbus_kmsg_unref(kmsg);
+	kdbus_kmsg_free(kmsg);
 
 	return ret;
 }
 
-int kdbus_notify_reply_timeout(struct kdbus_ep *ep,
-			       const struct kdbus_msg *orig_msg)
+int kdbus_notify_reply_timeout(struct kdbus_ep *ep, u64 src_id, u64 cookie)
 {
-	return kdbus_notify_reply(ep, orig_msg, KDBUS_MSG_REPLY_TIMEOUT);
+	return kdbus_notify_reply(ep, src_id, cookie, KDBUS_MSG_REPLY_TIMEOUT);
 }
 
-int kdbus_notify_reply_dead(struct kdbus_ep *ep,
-			    const struct kdbus_msg *orig_msg)
+int kdbus_notify_reply_dead(struct kdbus_ep *ep, u64 src_id, u64 cookie)
 {
-	return kdbus_notify_reply(ep, orig_msg, KDBUS_MSG_REPLY_DEAD);
+	return kdbus_notify_reply(ep, src_id, cookie, KDBUS_MSG_REPLY_DEAD);
 }
 
 int kdbus_notify_name_change(struct kdbus_ep *ep, u64 type,
@@ -109,7 +105,7 @@ int kdbus_notify_name_change(struct kdbus_ep *ep, u64 type,
 	strcpy(name_change->name, name);
 
 	ret = kdbus_kmsg_send(ep, NULL, kmsg);
-	kdbus_kmsg_unref(kmsg);
+	kdbus_kmsg_free(kmsg);
 
 	return ret;
 }
@@ -141,7 +137,7 @@ int kdbus_notify_id_change(struct kdbus_ep *ep, u64 type,
 	id_change->flags = flags;
 
 	ret = kdbus_kmsg_send(ep, NULL, kmsg);
-	kdbus_kmsg_unref(kmsg);
+	kdbus_kmsg_free(kmsg);
 
 	return ret;
 }
diff --git a/notify.h b/notify.h
index ac88cfa4c33..36305fc18a2 100644
--- a/notify.h
+++ b/notify.h
@@ -21,8 +21,6 @@ int kdbus_notify_name_change(struct kdbus_ep *ep, u64 type,
 			     const char *name);
 int kdbus_notify_id_change(struct kdbus_ep *ep, u64 type,
 			   u64 id, u64 flags);
-int kdbus_notify_reply_timeout(struct kdbus_ep *ep,
-			       const struct kdbus_msg *orig_msg);
-int kdbus_notify_reply_dead(struct kdbus_ep *ep,
-			    const struct kdbus_msg *orig_msg);
+int kdbus_notify_reply_timeout(struct kdbus_ep *ep, u64 src_id, u64 cookie);
+int kdbus_notify_reply_dead(struct kdbus_ep *ep, u64 src_id, u64 cookie);
 #endif
diff --git a/test/kdbus-enum.c b/test/kdbus-enum.c
index 9dcd8fd27f7..92060d59936 100644
--- a/test/kdbus-enum.c
+++ b/test/kdbus-enum.c
@@ -55,9 +55,8 @@ LOOKUP(CMD);
 
 TABLE(MSG) = {
 	ENUM(KDBUS_MSG_NULL),
-	ENUM(KDBUS_MSG_PAYLOAD),
 	ENUM(KDBUS_MSG_PAYLOAD_VEC),
-	ENUM(KDBUS_MSG_UNIX_FDS),
+	ENUM(KDBUS_MSG_FDS),
 	ENUM(KDBUS_MSG_BLOOM),
 	ENUM(KDBUS_MSG_DST_NAME),
 	ENUM(KDBUS_MSG_SRC_CREDS),
diff --git a/test/kdbus-util.c b/test/kdbus-util.c
index 8c328a1487e..0cdf73a91e4 100644
--- a/test/kdbus-util.c
+++ b/test/kdbus-util.c
@@ -21,6 +21,7 @@
 #include <assert.h>
 #include <poll.h>
 #include <sys/ioctl.h>
+#include <sys/mman.h>
 
 //#include "include/uapi/kdbus/kdbus.h"
 #include "../kdbus.h"
@@ -31,9 +32,17 @@
 struct conn *connect_to_bus(const char *path)
 {
 	int fd, ret;
-	struct kdbus_cmd_hello hello;
+	void *buf;
+	struct {
+		struct kdbus_cmd_hello hello;
+		uint64_t v_size;
+		uint64_t v_type;
+		struct kdbus_vec vec;
+	} h;
 	struct conn *conn;
 
+	memset(&h, 0, sizeof(h));
+
 	printf("-- opening bus connection %s\n", path);
 	fd = open(path, O_RDWR|O_CLOEXEC);
 	if (fd < 0) {
@@ -41,23 +50,34 @@ struct conn *connect_to_bus(const char *path)
 		return NULL;
 	}
 
-	memset(&hello, 0, sizeof(hello));
-
-	hello.conn_flags = KDBUS_HELLO_ACCEPT_FD |
-			   KDBUS_HELLO_ATTACH_COMM |
-			   KDBUS_HELLO_ATTACH_EXE |
-			   KDBUS_HELLO_ATTACH_CMDLINE |
-			   KDBUS_HELLO_ATTACH_CAPS |
-			   KDBUS_HELLO_ATTACH_CGROUP |
-			   KDBUS_HELLO_ATTACH_SECLABEL |
-			   KDBUS_HELLO_ATTACH_AUDIT;
-
-	ret = ioctl(fd, KDBUS_CMD_HELLO, &hello);
+	buf = mmap(NULL, 128 * 1024 * 1024, PROT_READ|PROT_WRITE, MAP_ANONYMOUS|MAP_PRIVATE, -1, 0);
+	if (buf == MAP_FAILED) {
+		fprintf(stderr, "--- error mmap (%m)\n");
+		return NULL;
+	}
+	h.v_type = KDBUS_HELLO_BUFFER;
+	h.v_size = KDBUS_ITEM_HEADER_SIZE + sizeof(struct kdbus_vec);
+	h.vec.address = (uint64_t)buf;
+	h.vec.size = 128 * 1024 * 1024;
+
+	h.hello.conn_flags = KDBUS_HELLO_ACCEPT_FD |
+			     KDBUS_HELLO_ATTACH_COMM |
+			     KDBUS_HELLO_ATTACH_EXE |
+			     KDBUS_HELLO_ATTACH_CMDLINE |
+			     KDBUS_HELLO_ATTACH_CAPS |
+			     KDBUS_HELLO_ATTACH_CGROUP |
+			     KDBUS_HELLO_ATTACH_SECLABEL |
+			     KDBUS_HELLO_ATTACH_AUDIT;
+
+	h.hello.size = sizeof(struct kdbus_cmd_hello) +
+		       KDBUS_ITEM_HEADER_SIZE + sizeof(struct kdbus_vec);
+
+	ret = ioctl(fd, KDBUS_CMD_HELLO, &h.hello);
 	if (ret) {
 		fprintf(stderr, "--- error when saying hello: %d (%m)\n", ret);
 		return NULL;
 	}
-	printf("-- Our peer ID for %s: %llu\n", path, (unsigned long long)hello.id);
+	printf("-- Our peer ID for %s: %llu\n", path, (unsigned long long)h.hello.id);
 
 	conn = malloc(sizeof(*conn));
 	if (!conn) {
@@ -66,7 +86,7 @@ struct conn *connect_to_bus(const char *path)
 	}
 
 	conn->fd = fd;
-	conn->id = hello.id;
+	conn->id = h.hello.id;
 	return conn;
 }
 
@@ -76,18 +96,19 @@ int msg_send(const struct conn *conn,
 		    uint64_t dst_id)
 {
 	struct kdbus_msg *msg;
-	const char ref[2048] = "REFERENCED";
+	const char ref1[1024 * 1024] = "0123456789_0";
+	const char ref2[] = "0123456789_1";
 	struct kdbus_item *item;
 	uint64_t size;
 	int ret;
 
-	size = sizeof(*msg);
+	size = sizeof(struct kdbus_msg);
 	size += KDBUS_ITEM_HEADER_SIZE + sizeof(struct kdbus_vec);
-	size += KDBUS_ITEM_HEADER_SIZE + sizeof("INLINE1");
-	
+	size += KDBUS_ITEM_HEADER_SIZE + sizeof(struct kdbus_vec);
+
 	if (dst_id == KDBUS_DST_ID_BROADCAST)
 		size += KDBUS_ITEM_HEADER_SIZE + 64;
-	
+
 	if (name)
 		size += KDBUS_ITEM_HEADER_SIZE + strlen(name) + 1;
 
@@ -115,13 +136,14 @@ int msg_send(const struct conn *conn,
 
 	item->type = KDBUS_MSG_PAYLOAD_VEC;
 	item->size = KDBUS_ITEM_HEADER_SIZE + sizeof(struct kdbus_vec);
-	item->vec.address = (uint64_t)&ref;
-	item->vec.size = sizeof(ref);
+	item->vec.address = (uint64_t)&ref1;
+	item->vec.size = sizeof(ref1);
 	item = KDBUS_ITEM_NEXT(item);
 
-	item->type = KDBUS_MSG_PAYLOAD;
-	item->size = KDBUS_ITEM_HEADER_SIZE + sizeof("INLINE1");
-	memcpy(item->data, "INLINE1", sizeof("INLINE1"));
+	item->type = KDBUS_MSG_PAYLOAD_VEC;
+	item->size = KDBUS_ITEM_HEADER_SIZE + sizeof(struct kdbus_vec);
+	item->vec.address = (uint64_t)&ref2;
+	item->vec.size = sizeof(ref2);
 	item = KDBUS_ITEM_NEXT(item);
 
 	if (dst_id == KDBUS_DST_ID_BROADCAST) {
@@ -169,16 +191,13 @@ void msg_dump(struct kdbus_msg *msg)
 		}
 
 		switch (item->type) {
-		case KDBUS_MSG_PAYLOAD:
-			printf("  +%s (%llu bytes) '%s'\n",
-			       enum_MSG(item->type), item->size, item->data);
-			break;
-
 		case KDBUS_MSG_PAYLOAD_VEC:
-			printf("  +%s (%llu bytes) addr=%p size=%llu flags=0x%llx '%s'\n",
+			printf("  +%s (%llu bytes) addr=%p size=%llu\n",
 			       enum_MSG(item->type), item->size, (void *)(uintptr_t)item->vec.address,
-			       (unsigned long long)item->vec.size, (unsigned long long)item->vec.flags,
-			       (char *)(uintptr_t)item->vec.address);
+			       (unsigned long long)item->vec.size);
+			printf("    '%s' ... '%s'\n",
+			       (char *)(uintptr_t)item->vec.address,
+			       (char *)(uintptr_t)item->vec.address + item->vec.size - 12);
 			break;
 
 		case KDBUS_MSG_SRC_CREDS:
@@ -298,20 +317,23 @@ void msg_dump(struct kdbus_msg *msg)
 
 int msg_recv(struct conn *conn)
 {
-	char tmp[0xffff];
-	struct kdbus_msg *msg = (struct kdbus_msg *) tmp;
+	struct kdbus_msg *msg;
 	int ret;
 
-	memset(tmp, 0, sizeof(tmp));
-	msg->size = sizeof(tmp);
-	ret = ioctl(conn->fd, KDBUS_CMD_MSG_RECV, msg);
-	if (ret) {
+	ret = ioctl(conn->fd, KDBUS_CMD_MSG_RECV, &msg);
+	if (ret < 0) {
 		fprintf(stderr, "error receiving message: %d (%m)\n", ret);
 		return EXIT_FAILURE;
 	}
 
 	msg_dump(msg);
 
+	ret = ioctl(conn->fd, KDBUS_CMD_MSG_FREE, msg);
+	if (ret < 0) {
+		fprintf(stderr, "error free message: %d (%m)\n", ret);
+		return EXIT_FAILURE;
+	}
+
 	return 0;
 }