[RFC PATCH net-next 06/12] io_uring/opaque: allow rings to share a store

[Alex Markuze <amarkuze@redhat.com>]

A multi-threaded server often uses one ring per thread. Allow an object
received or created through one ring to be used through another ring
without copying its payload.

Add EXPORT and IMPORT controls to IORING_REGISTER_OPAQUE_STORE. EXPORT
returns a close-on-exec anonymous-inode descriptor; IMPORT registers its
store under a new ring-local ID. The descriptor cannot read, map or poll
payload. Handles and stream tokens belong to the store and are valid
through every ring sharing it.

Each registered ring entry and export file holds one user reference and
one lifetime reference. The last user stops the store; the last lifetime
reference releases its memory and accounting. An export file can keep a
store alive after the creating ring exits. Import preserves the store's
memory-cgroup ownership and original ring user/mm accounting and adds no
backing charge.

Export reserves the descriptor and creates the file, copies the number
to userspace, and installs it only after that copy succeeds. Failure
unwinds the references and descriptor reservation. Import erases the
new ring-local ID if copying its result to userspace fails.

Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
 include/uapi/linux/io_uring/opaque.h |  3 +
 io_uring/opaque_register.c           | 84 ++++++++++++++++++++++++----
 2 files changed, 76 insertions(+), 11 deletions(-)

diff --git a/include/uapi/linux/io_uring/opaque.h b/include/uapi/linux/io_uring/opaque.h
index 692daa7d7e32..9a129b5da5af 100644
--- a/include/uapi/linux/io_uring/opaque.h
+++ b/include/uapi/linux/io_uring/opaque.h
@@ -16,6 +16,9 @@ struct io_uring_opaque_register {
 	__u64 __resv[2];
 };
 
+#define IORING_OPAQUE_REG_IMPORT	(1U << 0)
+#define IORING_OPAQUE_REG_EXPORT	(1U << 1)
+
 /* config/config_size select a size-versioned configuration on creation. */
 struct io_uring_opaque_config {
 	__u64 hard_limit;
diff --git a/io_uring/opaque_register.c b/io_uring/opaque_register.c
index 9e39d1ee5072..472c739a17de 100644
--- a/io_uring/opaque_register.c
+++ b/io_uring/opaque_register.c
@@ -15,6 +15,50 @@ static void io_opaque_unregister_user(struct io_opaque_store *store)
 	io_opaque_put(store);
 }
 
+static int io_opaque_box_release(struct inode *inode, struct file *file)
+{
+	io_opaque_unregister_user(file->private_data);
+	return 0;
+}
+
+static const struct file_operations io_opaque_box_fops = {
+	.owner = THIS_MODULE,
+	.release = io_opaque_box_release,
+};
+
+static int io_opaque_export(struct io_ring_ctx *ctx, void __user *arg,
+			    struct io_uring_opaque_register *reg, u32 size)
+{
+	struct io_opaque_store *store = xa_load(&ctx->opaque_stores, reg->store_id);
+	struct file *file;
+	int fd, ret;
+
+	if (!store)
+		return -ENXIO;
+	fd = get_unused_fd_flags(O_CLOEXEC);
+	if (fd < 0)
+		return fd;
+	io_opaque_get(store);
+	file = anon_inode_create_getfile("[io_uring-opaque]", &io_opaque_box_fops,
+					 store, O_CLOEXEC, NULL);
+	if (IS_ERR(file)) {
+		ret = PTR_ERR(file);
+		io_opaque_unregister_user(store);
+		goto put_fd;
+	}
+	reg->fd = fd;
+	ret = copy_struct_to_user(arg, size, reg, sizeof(*reg), NULL);
+	if (ret) {
+		fput(file);
+		goto put_fd;
+	}
+	fd_install(fd, file);
+	return 0;
+put_fd:
+	put_unused_fd(fd);
+	return ret;
+}
+
 int io_register_opaque(struct io_ring_ctx *ctx, void __user *arg)
 {
 	struct io_uring_opaque_register reg;
@@ -38,19 +82,37 @@ int io_register_opaque(struct io_ring_ctx *ctx, void __user *arg)
 	    !mem_is_zero(reg.__resv, sizeof(reg.__resv)))
 		return -EINVAL;
 	reg.size = sizeof(reg);
-	if (reg.store_id || reg.flags)
+	if (reg.flags == IORING_OPAQUE_REG_EXPORT) {
+		if (reg.fd != -1 || reg.config || reg.config_size)
+			return -EINVAL;
+		return io_opaque_export(ctx, arg, &reg, size);
+	}
+	if (reg.store_id)
 		return -EINVAL;
-	if (reg.fd != -1 || !reg.config ||
-	    reg.config_size < offsetof(struct io_uring_opaque_config, __resv0) ||
-	    reg.config_size > PAGE_SIZE)
+	if (reg.flags == IORING_OPAQUE_REG_IMPORT) {
+		CLASS(fd, f)(reg.fd);
+
+		if (reg.config || reg.config_size)
+			return -EINVAL;
+		if (fd_empty(f) || fd_file(f)->f_op != &io_opaque_box_fops)
+			return -EBADF;
+		store = fd_file(f)->private_data;
+		io_opaque_get(store);
+	} else if (!reg.flags) {
+		if (reg.fd != -1 || !reg.config ||
+		    reg.config_size < offsetof(struct io_uring_opaque_config, __resv0) ||
+		    reg.config_size > PAGE_SIZE)
+			return -EINVAL;
+		ret = copy_struct_from_user(&config, sizeof(config),
+					    u64_to_user_ptr(reg.config), reg.config_size);
+		if (ret)
+			return ret;
+		store = io_opaque_alloc(ctx, &config);
+		if (IS_ERR(store))
+			return PTR_ERR(store);
+	} else {
 		return -EINVAL;
-	ret = copy_struct_from_user(&config, sizeof(config),
-				    u64_to_user_ptr(reg.config), reg.config_size);
-	if (ret)
-		return ret;
-	store = io_opaque_alloc(ctx, &config);
-	if (IS_ERR(store))
-		return PTR_ERR(store);
+	}
 	ret = xa_alloc(&ctx->opaque_stores, &id, store, xa_limit_31b, GFP_KERNEL);
 	if (ret)
 		goto put_store;
-- 
2.34.1