[PATCH v5 00/13] io_uring: buffer registration enhancements

[PATCH v5 00/13] io_uring: buffer registration enhancements

[Bijan Mottahedeh]

v5:

- call io_get_fixed_rsrc_ref for buffers
- make percpu_ref_release names consistent
- rebase on for-5.12/io_uring

v4:

- address v3 comments (TBD REGISTER_BUFFERS)
- rebase

v3:

- batch file->rsrc renames into a signle patch when possible
- fix other review changes from v2
- fix checkpatch warnings

v2:

- drop readv/writev with fixed buffers patch
- handle ref_nodes both both files/buffers with a single ref_list
- make file/buffer handling more unified

This patchset implements a set of enhancements to buffer registration
consistent with existing file registration functionality:

- buffer registration updates		IORING_REGISTER_BUFFERS_UPDATE
					IORING_OP_BUFFERS_UPDATE

- buffer registration sharing		IORING_SETUP_SHARE_BUF
					IORING_SETUP_ATTACH_BUF

Patch 1-5 generalize fixed_file functionality to fixed_rsrc.

Patch 6 applies fixed_rsrc functionality for fixed buffers support.

Patch 7-8 generalize files_update functionality to rsrc_update.

Patch 9 implements buffer registration update, and introduces
IORING_REGISTER_BUFFERS_UPDATE and IORING_OP_BUFFERS_UPDATE, consistent
with file registration update.

Patch 10 generalizes fixed resource allocation 

Patch 11 renames percpu release routines for consistency

Patch 12 calls io_get_fixed_rsrc_ref() for buffers as well as files

Patch 13 implements buffer sharing among multiple rings; it works as follows:

- A new ring, A,  is setup. Since no buffers have been registered, the
  registered buffer state is an empty set, Z. That's different from the
  NULL state in current implementation.

- Ring B is setup, attaching to Ring A. It's also attaching to it's
  buffer registrations, now we have two references to the same empty
  set, Z.

- Ring A registers buffers into set Z, which is no longer empty.

- Ring B sees this immediately, since it's already sharing that set.

Testing

I have used liburing file-{register,update} tests as models for
buffer-{register,update,share}, tests and they run ok.

TBD

- Need a patch from Pavel to address a race between fixed IO from async
context and buffer unregister, or force buffer registration ops to do
full quiesce.

- Need to still address Pavel's comments about unkillable deadlocks. It
seems that we should no long hange unkillably in io_rsrc_ref_quiesce()
because of Pavel's changes.

- I tried to use a single opcode for files/buffers but ran into an
issue since work_flags is different for files/buffers.  This should
be ok for the most part since req->work.flags is ultimately examined;
however, there are place where io_op_defs[opcode].work_flags is examined
directly, and I wasn't sure what would the best way to handle that.

Bijan Mottahedeh (13):
  io_uring: rename file related variables to rsrc
  io_uring: generalize io_queue_rsrc_removal
  io_uring: separate ref_list from fixed_rsrc_data
  io_uring: split alloc_fixed_file_ref_node
  io_uring: add rsrc_ref locking routines
  io_uring: implement fixed buffers registration similar to fixed files
  io_uring: create common fixed_rsrc_ref_node handling routines
  io_uring: generalize files_update functionlity to rsrc_update
  io_uring: support buffer registration updates
  io_uring: create common fixed_rsrc_data allocation routines
  io_uring: make percpu_ref_release names consistent
  io_uring: call io_get_fixed_rsrc_ref for buffers
  io_uring: support buffer registration sharing

 fs/io_uring.c                 | 803 ++++++++++++++++++++++++++++++++----------
 include/uapi/linux/io_uring.h |  13 +-
 2 files changed, 626 insertions(+), 190 deletions(-)

-- 
1.8.3.1

[PATCH v5 01/13] io_uring: rename file related variables to rsrc

[Bijan Mottahedeh]

This is a prep rename patch for subsequent patches to generalize file
registration.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c                 | 226 ++++++++++++++++++++++--------------------
 include/uapi/linux/io_uring.h |   3 +-
 2 files changed, 118 insertions(+), 111 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index ad925b2..195ba12 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -195,24 +195,29 @@ struct io_mapped_ubuf {
 	unsigned long	acct_pages;
 };
 
-struct fixed_file_table {
+struct io_rsrc_put {
+	struct list_head list;
+	struct file *file;
+};
+
+struct fixed_rsrc_table {
 	struct file		**files;
 };
 
-struct fixed_file_ref_node {
+struct fixed_rsrc_ref_node {
 	struct percpu_ref		refs;
 	struct list_head		node;
-	struct list_head		file_list;
-	struct fixed_file_data		*file_data;
+	struct list_head		rsrc_list;
+	struct fixed_rsrc_data		*rsrc_data;
 	struct llist_node		llist;
 	bool				done;
 };
 
-struct fixed_file_data {
-	struct fixed_file_table		*table;
+struct fixed_rsrc_data {
+	struct fixed_rsrc_table		*table;
 	struct io_ring_ctx		*ctx;
 
-	struct fixed_file_ref_node	*node;
+	struct fixed_rsrc_ref_node	*node;
 	struct percpu_ref		refs;
 	struct completion		done;
 	struct list_head		ref_list;
@@ -318,7 +323,7 @@ struct io_ring_ctx {
 	 * readers must ensure that ->refs is alive as long as the file* is
 	 * used. Only updated through io_uring_register(2).
 	 */
-	struct fixed_file_data	*file_data;
+	struct fixed_rsrc_data	*file_data;
 	unsigned		nr_user_files;
 
 	/* if used, fixed mapped user buffers */
@@ -382,8 +387,8 @@ struct io_ring_ctx {
 		struct list_head	inflight_list;
 	} ____cacheline_aligned_in_smp;
 
-	struct delayed_work		file_put_work;
-	struct llist_head		file_put_llist;
+	struct delayed_work		rsrc_put_work;
+	struct llist_head		rsrc_put_llist;
 
 	struct work_struct		exit_work;
 	struct io_restriction		restrictions;
@@ -492,7 +497,7 @@ struct io_open {
 	unsigned long			nofile;
 };
 
-struct io_files_update {
+struct io_rsrc_update {
 	struct file			*file;
 	u64				arg;
 	u32				nr_args;
@@ -686,7 +691,7 @@ struct io_kiocb {
 		struct io_sr_msg	sr_msg;
 		struct io_open		open;
 		struct io_close		close;
-		struct io_files_update	files_update;
+		struct io_rsrc_update	rsrc_update;
 		struct io_fadvise	fadvise;
 		struct io_madvise	madvise;
 		struct io_epoll		epoll;
@@ -716,7 +721,7 @@ struct io_kiocb {
 	u64				user_data;
 
 	struct io_kiocb			*link;
-	struct percpu_ref		*fixed_file_refs;
+	struct percpu_ref		*fixed_rsrc_refs;
 
 	/*
 	 * 1. used with ctx->iopoll_list with reads/writes
@@ -991,8 +996,8 @@ enum io_mem_account {
 	ACCT_PINNED,
 };
 
-static void destroy_fixed_file_ref_node(struct fixed_file_ref_node *ref_node);
-static struct fixed_file_ref_node *alloc_fixed_file_ref_node(
+static void destroy_fixed_rsrc_ref_node(struct fixed_rsrc_ref_node *ref_node);
+static struct fixed_rsrc_ref_node *alloc_fixed_file_ref_node(
 			struct io_ring_ctx *ctx);
 
 static void __io_complete_rw(struct io_kiocb *req, long res, long res2,
@@ -1005,13 +1010,13 @@ static void __io_complete_rw(struct io_kiocb *req, long res, long res2,
 static void __io_queue_linked_timeout(struct io_kiocb *req);
 static void io_queue_linked_timeout(struct io_kiocb *req);
 static int __io_sqe_files_update(struct io_ring_ctx *ctx,
-				 struct io_uring_files_update *ip,
+				 struct io_uring_rsrc_update *ip,
 				 unsigned nr_args);
 static void __io_clean_op(struct io_kiocb *req);
 static struct file *io_file_get(struct io_submit_state *state,
 				struct io_kiocb *req, int fd, bool fixed);
 static void __io_queue_sqe(struct io_kiocb *req, struct io_comp_state *cs);
-static void io_file_put_work(struct work_struct *work);
+static void io_rsrc_put_work(struct work_struct *work);
 
 static ssize_t io_import_iovec(int rw, struct io_kiocb *req,
 			       struct iovec **iovec, struct iov_iter *iter,
@@ -1051,9 +1056,9 @@ static inline void io_set_resource_node(struct io_kiocb *req)
 {
 	struct io_ring_ctx *ctx = req->ctx;
 
-	if (!req->fixed_file_refs) {
-		req->fixed_file_refs = &ctx->file_data->node->refs;
-		percpu_ref_get(req->fixed_file_refs);
+	if (!req->fixed_rsrc_refs) {
+		req->fixed_rsrc_refs = &ctx->file_data->node->refs;
+		percpu_ref_get(req->fixed_rsrc_refs);
 	}
 }
 
@@ -1312,8 +1317,8 @@ static struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
 	INIT_LIST_HEAD(&ctx->timeout_list);
 	spin_lock_init(&ctx->inflight_lock);
 	INIT_LIST_HEAD(&ctx->inflight_list);
-	INIT_DELAYED_WORK(&ctx->file_put_work, io_file_put_work);
-	init_llist_head(&ctx->file_put_llist);
+	INIT_DELAYED_WORK(&ctx->rsrc_put_work, io_rsrc_put_work);
+	init_llist_head(&ctx->rsrc_put_llist);
 	return ctx;
 err:
 	if (ctx->fallback_req)
@@ -1941,8 +1946,8 @@ static void io_dismantle_req(struct io_kiocb *req)
 		kfree(req->async_data);
 	if (req->file)
 		io_put_file(req, req->file, (req->flags & REQ_F_FIXED_FILE));
-	if (req->fixed_file_refs)
-		percpu_ref_put(req->fixed_file_refs);
+	if (req->fixed_rsrc_refs)
+		percpu_ref_put(req->fixed_rsrc_refs);
 	io_req_clean_work(req);
 }
 
@@ -5916,7 +5921,7 @@ static int io_async_cancel(struct io_kiocb *req)
 	return 0;
 }
 
-static int io_files_update_prep(struct io_kiocb *req,
+static int io_rsrc_update_prep(struct io_kiocb *req,
 				const struct io_uring_sqe *sqe)
 {
 	if (unlikely(req->ctx->flags & IORING_SETUP_SQPOLL))
@@ -5926,11 +5931,11 @@ static int io_files_update_prep(struct io_kiocb *req,
 	if (sqe->ioprio || sqe->rw_flags)
 		return -EINVAL;
 
-	req->files_update.offset = READ_ONCE(sqe->off);
-	req->files_update.nr_args = READ_ONCE(sqe->len);
-	if (!req->files_update.nr_args)
+	req->rsrc_update.offset = READ_ONCE(sqe->off);
+	req->rsrc_update.nr_args = READ_ONCE(sqe->len);
+	if (!req->rsrc_update.nr_args)
 		return -EINVAL;
-	req->files_update.arg = READ_ONCE(sqe->addr);
+	req->rsrc_update.arg = READ_ONCE(sqe->addr);
 	return 0;
 }
 
@@ -5938,17 +5943,17 @@ static int io_files_update(struct io_kiocb *req, bool force_nonblock,
 			   struct io_comp_state *cs)
 {
 	struct io_ring_ctx *ctx = req->ctx;
-	struct io_uring_files_update up;
+	struct io_uring_rsrc_update up;
 	int ret;
 
 	if (force_nonblock)
 		return -EAGAIN;
 
-	up.offset = req->files_update.offset;
-	up.fds = req->files_update.arg;
+	up.offset = req->rsrc_update.offset;
+	up.fds = req->rsrc_update.arg;
 
 	mutex_lock(&ctx->uring_lock);
-	ret = __io_sqe_files_update(ctx, &up, req->files_update.nr_args);
+	ret = __io_sqe_files_update(ctx, &up, req->rsrc_update.nr_args);
 	mutex_unlock(&ctx->uring_lock);
 
 	if (ret < 0)
@@ -6003,7 +6008,7 @@ static int io_req_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
 	case IORING_OP_CLOSE:
 		return io_close_prep(req, sqe);
 	case IORING_OP_FILES_UPDATE:
-		return io_files_update_prep(req, sqe);
+		return io_rsrc_update_prep(req, sqe);
 	case IORING_OP_STATX:
 		return io_statx_prep(req, sqe);
 	case IORING_OP_FADVISE:
@@ -6373,7 +6378,7 @@ static struct io_wq_work *io_wq_submit_work(struct io_wq_work *work)
 static inline struct file *io_file_from_index(struct io_ring_ctx *ctx,
 					      int index)
 {
-	struct fixed_file_table *table;
+	struct fixed_rsrc_table *table;
 
 	table = &ctx->file_data->table[index >> IORING_FILE_TABLE_SHIFT];
 	return table->files[index & IORING_FILE_TABLE_MASK];
@@ -6759,7 +6764,7 @@ static int io_init_req(struct io_ring_ctx *ctx, struct io_kiocb *req,
 	req->ctx = ctx;
 	req->flags = 0;
 	req->link = NULL;
-	req->fixed_file_refs = NULL;
+	req->fixed_rsrc_refs = NULL;
 	/* one is dropped after submission, the other at completion */
 	refcount_set(&req->refs, 2);
 	req->task = current;
@@ -7234,28 +7239,28 @@ static void __io_sqe_files_unregister(struct io_ring_ctx *ctx)
 #endif
 }
 
-static void io_file_ref_kill(struct percpu_ref *ref)
+static void io_rsrc_ref_kill(struct percpu_ref *ref)
 {
-	struct fixed_file_data *data;
+	struct fixed_rsrc_data *data;
 
-	data = container_of(ref, struct fixed_file_data, refs);
+	data = container_of(ref, struct fixed_rsrc_data, refs);
 	complete(&data->done);
 }
 
-static void io_sqe_files_set_node(struct fixed_file_data *file_data,
-				  struct fixed_file_ref_node *ref_node)
+static void io_sqe_rsrc_set_node(struct fixed_rsrc_data *rsrc_data,
+				 struct fixed_rsrc_ref_node *ref_node)
 {
-	spin_lock_bh(&file_data->lock);
-	file_data->node = ref_node;
-	list_add_tail(&ref_node->node, &file_data->ref_list);
-	spin_unlock_bh(&file_data->lock);
-	percpu_ref_get(&file_data->refs);
+	spin_lock_bh(&rsrc_data->lock);
+	rsrc_data->node = ref_node;
+	list_add_tail(&ref_node->node, &rsrc_data->ref_list);
+	spin_unlock_bh(&rsrc_data->lock);
+	percpu_ref_get(&rsrc_data->refs);
 }
 
 static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 {
-	struct fixed_file_data *data = ctx->file_data;
-	struct fixed_file_ref_node *backup_node, *ref_node = NULL;
+	struct fixed_rsrc_data *data = ctx->file_data;
+	struct fixed_rsrc_ref_node *backup_node, *ref_node = NULL;
 	unsigned nr_tables, i;
 	int ret;
 
@@ -7274,7 +7279,7 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 	percpu_ref_kill(&data->refs);
 
 	/* wait for all refs nodes to complete */
-	flush_delayed_work(&ctx->file_put_work);
+	flush_delayed_work(&ctx->rsrc_put_work);
 	do {
 		ret = wait_for_completion_interruptible(&data->done);
 		if (!ret)
@@ -7283,7 +7288,7 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 		if (ret < 0) {
 			percpu_ref_resurrect(&data->refs);
 			reinit_completion(&data->done);
-			io_sqe_files_set_node(data, backup_node);
+			io_sqe_rsrc_set_node(data, backup_node);
 			return ret;
 		}
 	} while (1);
@@ -7297,7 +7302,7 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 	kfree(data);
 	ctx->file_data = NULL;
 	ctx->nr_user_files = 0;
-	destroy_fixed_file_ref_node(backup_node);
+	destroy_fixed_rsrc_ref_node(backup_node);
 	return 0;
 }
 
@@ -7520,13 +7525,13 @@ static int io_sqe_files_scm(struct io_ring_ctx *ctx)
 }
 #endif
 
-static int io_sqe_alloc_file_tables(struct fixed_file_data *file_data,
+static int io_sqe_alloc_file_tables(struct fixed_rsrc_data *file_data,
 				    unsigned nr_tables, unsigned nr_files)
 {
 	int i;
 
 	for (i = 0; i < nr_tables; i++) {
-		struct fixed_file_table *table = &file_data->table[i];
+		struct fixed_rsrc_table *table = &file_data->table[i];
 		unsigned this_files;
 
 		this_files = min(nr_files, IORING_MAX_FILES_TABLE);
@@ -7541,7 +7546,7 @@ static int io_sqe_alloc_file_tables(struct fixed_file_data *file_data,
 		return 0;
 
 	for (i = 0; i < nr_tables; i++) {
-		struct fixed_file_table *table = &file_data->table[i];
+		struct fixed_rsrc_table *table = &file_data->table[i];
 		kfree(table->files);
 	}
 	return 1;
@@ -7609,56 +7614,51 @@ static void io_ring_file_put(struct io_ring_ctx *ctx, struct file *file)
 #endif
 }
 
-struct io_file_put {
-	struct list_head list;
-	struct file *file;
-};
-
-static void __io_file_put_work(struct fixed_file_ref_node *ref_node)
+static void __io_rsrc_put_work(struct fixed_rsrc_ref_node *ref_node)
 {
-	struct fixed_file_data *file_data = ref_node->file_data;
-	struct io_ring_ctx *ctx = file_data->ctx;
-	struct io_file_put *pfile, *tmp;
+	struct fixed_rsrc_data *rsrc_data = ref_node->rsrc_data;
+	struct io_ring_ctx *ctx = rsrc_data->ctx;
+	struct io_rsrc_put *prsrc, *tmp;
 
-	list_for_each_entry_safe(pfile, tmp, &ref_node->file_list, list) {
-		list_del(&pfile->list);
-		io_ring_file_put(ctx, pfile->file);
-		kfree(pfile);
+	list_for_each_entry_safe(prsrc, tmp, &ref_node->rsrc_list, list) {
+		list_del(&prsrc->list);
+		io_ring_file_put(ctx, prsrc->file);
+		kfree(prsrc);
 	}
 
 	percpu_ref_exit(&ref_node->refs);
 	kfree(ref_node);
-	percpu_ref_put(&file_data->refs);
+	percpu_ref_put(&rsrc_data->refs);
 }
 
-static void io_file_put_work(struct work_struct *work)
+static void io_rsrc_put_work(struct work_struct *work)
 {
 	struct io_ring_ctx *ctx;
 	struct llist_node *node;
 
-	ctx = container_of(work, struct io_ring_ctx, file_put_work.work);
-	node = llist_del_all(&ctx->file_put_llist);
+	ctx = container_of(work, struct io_ring_ctx, rsrc_put_work.work);
+	node = llist_del_all(&ctx->rsrc_put_llist);
 
 	while (node) {
-		struct fixed_file_ref_node *ref_node;
+		struct fixed_rsrc_ref_node *ref_node;
 		struct llist_node *next = node->next;
 
-		ref_node = llist_entry(node, struct fixed_file_ref_node, llist);
-		__io_file_put_work(ref_node);
+		ref_node = llist_entry(node, struct fixed_rsrc_ref_node, llist);
+		__io_rsrc_put_work(ref_node);
 		node = next;
 	}
 }
 
-static void io_file_data_ref_zero(struct percpu_ref *ref)
+static void io_rsrc_data_ref_zero(struct percpu_ref *ref)
 {
-	struct fixed_file_ref_node *ref_node;
-	struct fixed_file_data *data;
+	struct fixed_rsrc_ref_node *ref_node;
+	struct fixed_rsrc_data *data;
 	struct io_ring_ctx *ctx;
 	bool first_add = false;
 	int delay = HZ;
 
-	ref_node = container_of(ref, struct fixed_file_ref_node, refs);
-	data = ref_node->file_data;
+	ref_node = container_of(ref, struct fixed_rsrc_ref_node, refs);
+	data = ref_node->rsrc_data;
 	ctx = data->ctx;
 
 	spin_lock_bh(&data->lock);
@@ -7666,12 +7666,12 @@ static void io_file_data_ref_zero(struct percpu_ref *ref)
 
 	while (!list_empty(&data->ref_list)) {
 		ref_node = list_first_entry(&data->ref_list,
-					struct fixed_file_ref_node, node);
+					struct fixed_rsrc_ref_node, node);
 		/* recycle ref nodes in order */
 		if (!ref_node->done)
 			break;
 		list_del(&ref_node->node);
-		first_add |= llist_add(&ref_node->llist, &ctx->file_put_llist);
+		first_add |= llist_add(&ref_node->llist, &ctx->rsrc_put_llist);
 	}
 	spin_unlock_bh(&data->lock);
 
@@ -7679,33 +7679,33 @@ static void io_file_data_ref_zero(struct percpu_ref *ref)
 		delay = 0;
 
 	if (!delay)
-		mod_delayed_work(system_wq, &ctx->file_put_work, 0);
+		mod_delayed_work(system_wq, &ctx->rsrc_put_work, 0);
 	else if (first_add)
-		queue_delayed_work(system_wq, &ctx->file_put_work, delay);
+		queue_delayed_work(system_wq, &ctx->rsrc_put_work, delay);
 }
 
-static struct fixed_file_ref_node *alloc_fixed_file_ref_node(
+static struct fixed_rsrc_ref_node *alloc_fixed_file_ref_node(
 			struct io_ring_ctx *ctx)
 {
-	struct fixed_file_ref_node *ref_node;
+	struct fixed_rsrc_ref_node *ref_node;
 
 	ref_node = kzalloc(sizeof(*ref_node), GFP_KERNEL);
 	if (!ref_node)
 		return ERR_PTR(-ENOMEM);
 
-	if (percpu_ref_init(&ref_node->refs, io_file_data_ref_zero,
+	if (percpu_ref_init(&ref_node->refs, io_rsrc_data_ref_zero,
 			    0, GFP_KERNEL)) {
 		kfree(ref_node);
 		return ERR_PTR(-ENOMEM);
 	}
 	INIT_LIST_HEAD(&ref_node->node);
-	INIT_LIST_HEAD(&ref_node->file_list);
-	ref_node->file_data = ctx->file_data;
+	INIT_LIST_HEAD(&ref_node->rsrc_list);
+	ref_node->rsrc_data = ctx->file_data;
 	ref_node->done = false;
 	return ref_node;
 }
 
-static void destroy_fixed_file_ref_node(struct fixed_file_ref_node *ref_node)
+static void destroy_fixed_rsrc_ref_node(struct fixed_rsrc_ref_node *ref_node)
 {
 	percpu_ref_exit(&ref_node->refs);
 	kfree(ref_node);
@@ -7718,8 +7718,8 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 	unsigned nr_tables, i;
 	struct file *file;
 	int fd, ret = -ENOMEM;
-	struct fixed_file_ref_node *ref_node;
-	struct fixed_file_data *file_data;
+	struct fixed_rsrc_ref_node *ref_node;
+	struct fixed_rsrc_data *file_data;
 
 	if (ctx->file_data)
 		return -EBUSY;
@@ -7742,7 +7742,7 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 	if (!file_data->table)
 		goto out_free;
 
-	if (percpu_ref_init(&file_data->refs, io_file_ref_kill,
+	if (percpu_ref_init(&file_data->refs, io_rsrc_ref_kill,
 				PERCPU_REF_ALLOW_REINIT, GFP_KERNEL))
 		goto out_free;
 
@@ -7751,7 +7751,7 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 	ctx->file_data = file_data;
 
 	for (i = 0; i < nr_args; i++, ctx->nr_user_files++) {
-		struct fixed_file_table *table;
+		struct fixed_rsrc_table *table;
 		unsigned index;
 
 		if (copy_from_user(&fd, &fds[i], sizeof(fd))) {
@@ -7795,7 +7795,7 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 		return PTR_ERR(ref_node);
 	}
 
-	io_sqe_files_set_node(file_data, ref_node);
+	io_sqe_rsrc_set_node(file_data, ref_node);
 	return ret;
 out_fput:
 	for (i = 0; i < ctx->nr_user_files; i++) {
@@ -7858,28 +7858,34 @@ static int io_sqe_file_register(struct io_ring_ctx *ctx, struct file *file,
 #endif
 }
 
-static int io_queue_file_removal(struct fixed_file_data *data,
-				 struct file *file)
+static int io_queue_rsrc_removal(struct fixed_rsrc_data *data,
+				 struct file *rsrc)
 {
-	struct io_file_put *pfile;
-	struct fixed_file_ref_node *ref_node = data->node;
+	struct io_rsrc_put *prsrc;
+	struct fixed_rsrc_ref_node *ref_node = data->node;
 
-	pfile = kzalloc(sizeof(*pfile), GFP_KERNEL);
-	if (!pfile)
+	prsrc = kzalloc(sizeof(*prsrc), GFP_KERNEL);
+	if (!prsrc)
 		return -ENOMEM;
 
-	pfile->file = file;
-	list_add(&pfile->list, &ref_node->file_list);
+	prsrc->file = rsrc;
+	list_add(&prsrc->list, &ref_node->rsrc_list);
 
 	return 0;
 }
 
+static inline int io_queue_file_removal(struct fixed_rsrc_data *data,
+					struct file *file)
+{
+	return io_queue_rsrc_removal(data, file);
+}
+
 static int __io_sqe_files_update(struct io_ring_ctx *ctx,
-				 struct io_uring_files_update *up,
+				 struct io_uring_rsrc_update *up,
 				 unsigned nr_args)
 {
-	struct fixed_file_data *data = ctx->file_data;
-	struct fixed_file_ref_node *ref_node;
+	struct fixed_rsrc_data *data = ctx->file_data;
+	struct fixed_rsrc_ref_node *ref_node;
 	struct file *file;
 	__s32 __user *fds;
 	int fd, i, err;
@@ -7898,7 +7904,7 @@ static int __io_sqe_files_update(struct io_ring_ctx *ctx,
 	done = 0;
 	fds = u64_to_user_ptr(up->fds);
 	while (nr_args) {
-		struct fixed_file_table *table;
+		struct fixed_rsrc_table *table;
 		unsigned index;
 
 		err = 0;
@@ -7951,9 +7957,9 @@ static int __io_sqe_files_update(struct io_ring_ctx *ctx,
 
 	if (needs_switch) {
 		percpu_ref_kill(&data->node->refs);
-		io_sqe_files_set_node(data, ref_node);
+		io_sqe_rsrc_set_node(data, ref_node);
 	} else
-		destroy_fixed_file_ref_node(ref_node);
+		destroy_fixed_rsrc_ref_node(ref_node);
 
 	return done ? done : err;
 }
@@ -7961,7 +7967,7 @@ static int __io_sqe_files_update(struct io_ring_ctx *ctx,
 static int io_sqe_files_update(struct io_ring_ctx *ctx, void __user *arg,
 			       unsigned nr_args)
 {
-	struct io_uring_files_update up;
+	struct io_uring_rsrc_update up;
 
 	if (!ctx->file_data)
 		return -ENXIO;
@@ -9332,7 +9338,7 @@ static void __io_uring_show_fdinfo(struct io_ring_ctx *ctx, struct seq_file *m)
 	seq_printf(m, "SqThreadCpu:\t%d\n", sq ? task_cpu(sq->thread) : -1);
 	seq_printf(m, "UserFiles:\t%u\n", ctx->nr_user_files);
 	for (i = 0; has_lock && i < ctx->nr_user_files; i++) {
-		struct fixed_file_table *table;
+		struct fixed_rsrc_table *table;
 		struct file *f;
 
 		table = &ctx->file_data->table[i >> IORING_FILE_TABLE_SHIFT];
diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
index d31a2a1..77de7c0 100644
--- a/include/uapi/linux/io_uring.h
+++ b/include/uapi/linux/io_uring.h
@@ -285,12 +285,13 @@ enum {
 	IORING_REGISTER_LAST
 };
 
-struct io_uring_files_update {
+struct io_uring_rsrc_update {
 	__u32 offset;
 	__u32 resv;
 	__aligned_u64 /* __s32 * */ fds;
 };
 
+#define io_uring_files_update	io_uring_rsrc_update
 #define IO_URING_OP_SUPPORTED	(1U << 0)
 
 struct io_uring_probe_op {
-- 
1.8.3.1

[PATCH v5 02/13] io_uring: generalize io_queue_rsrc_removal

[Bijan Mottahedeh]

Generalize io_queue_rsrc_removal to handle both files and buffers.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 28 ++++++++++++++++++++--------
 1 file changed, 20 insertions(+), 8 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 195ba12..12eede8 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -195,13 +195,22 @@ struct io_mapped_ubuf {
 	unsigned long	acct_pages;
 };
 
+struct io_ring_ctx;
+
 struct io_rsrc_put {
 	struct list_head list;
-	struct file *file;
+	union {
+		void *rsrc;
+		struct file *file;
+		struct io_mapped_ubuf *buf;
+	};
 };
 
 struct fixed_rsrc_table {
-	struct file		**files;
+	union {
+		struct file		**files;
+		struct io_mapped_ubuf	*bufs;
+	};
 };
 
 struct fixed_rsrc_ref_node {
@@ -209,6 +218,8 @@ struct fixed_rsrc_ref_node {
 	struct list_head		node;
 	struct list_head		rsrc_list;
 	struct fixed_rsrc_data		*rsrc_data;
+	void				(*rsrc_put)(struct io_ring_ctx *ctx,
+						    struct io_rsrc_put *prsrc);
 	struct llist_node		llist;
 	bool				done;
 };
@@ -7552,8 +7563,9 @@ static int io_sqe_alloc_file_tables(struct fixed_rsrc_data *file_data,
 	return 1;
 }
 
-static void io_ring_file_put(struct io_ring_ctx *ctx, struct file *file)
+static void io_ring_file_put(struct io_ring_ctx *ctx, struct io_rsrc_put *prsrc)
 {
+	struct file *file = prsrc->file;
 #if defined(CONFIG_UNIX)
 	struct sock *sock = ctx->ring_sock->sk;
 	struct sk_buff_head list, *head = &sock->sk_receive_queue;
@@ -7622,7 +7634,7 @@ static void __io_rsrc_put_work(struct fixed_rsrc_ref_node *ref_node)
 
 	list_for_each_entry_safe(prsrc, tmp, &ref_node->rsrc_list, list) {
 		list_del(&prsrc->list);
-		io_ring_file_put(ctx, prsrc->file);
+		ref_node->rsrc_put(ctx, prsrc);
 		kfree(prsrc);
 	}
 
@@ -7701,6 +7713,7 @@ static struct fixed_rsrc_ref_node *alloc_fixed_file_ref_node(
 	INIT_LIST_HEAD(&ref_node->node);
 	INIT_LIST_HEAD(&ref_node->rsrc_list);
 	ref_node->rsrc_data = ctx->file_data;
+	ref_node->rsrc_put = io_ring_file_put;
 	ref_node->done = false;
 	return ref_node;
 }
@@ -7858,8 +7871,7 @@ static int io_sqe_file_register(struct io_ring_ctx *ctx, struct file *file,
 #endif
 }
 
-static int io_queue_rsrc_removal(struct fixed_rsrc_data *data,
-				 struct file *rsrc)
+static int io_queue_rsrc_removal(struct fixed_rsrc_data *data, void *rsrc)
 {
 	struct io_rsrc_put *prsrc;
 	struct fixed_rsrc_ref_node *ref_node = data->node;
@@ -7868,7 +7880,7 @@ static int io_queue_rsrc_removal(struct fixed_rsrc_data *data,
 	if (!prsrc)
 		return -ENOMEM;
 
-	prsrc->file = rsrc;
+	prsrc->rsrc = rsrc;
 	list_add(&prsrc->list, &ref_node->rsrc_list);
 
 	return 0;
@@ -7877,7 +7889,7 @@ static int io_queue_rsrc_removal(struct fixed_rsrc_data *data,
 static inline int io_queue_file_removal(struct fixed_rsrc_data *data,
 					struct file *file)
 {
-	return io_queue_rsrc_removal(data, file);
+	return io_queue_rsrc_removal(data, (void *)file);
 }
 
 static int __io_sqe_files_update(struct io_ring_ctx *ctx,
-- 
1.8.3.1

[PATCH v5 03/13] io_uring: separate ref_list from fixed_rsrc_data

[Bijan Mottahedeh]

Uplevel ref_list and make it common to all resources.  This is to
allow one common ref_list to be used for both files, and buffers
in upcoming patches.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 35 ++++++++++++++++++-----------------
 1 file changed, 18 insertions(+), 17 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 12eede8..158b53f 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -231,8 +231,6 @@ struct fixed_rsrc_data {
 	struct fixed_rsrc_ref_node	*node;
 	struct percpu_ref		refs;
 	struct completion		done;
-	struct list_head		ref_list;
-	spinlock_t			lock;
 };
 
 struct io_buffer {
@@ -400,6 +398,8 @@ struct io_ring_ctx {
 
 	struct delayed_work		rsrc_put_work;
 	struct llist_head		rsrc_put_llist;
+	struct list_head		rsrc_ref_list;
+	spinlock_t			rsrc_ref_lock;
 
 	struct work_struct		exit_work;
 	struct io_restriction		restrictions;
@@ -1328,6 +1328,8 @@ static struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
 	INIT_LIST_HEAD(&ctx->timeout_list);
 	spin_lock_init(&ctx->inflight_lock);
 	INIT_LIST_HEAD(&ctx->inflight_list);
+	spin_lock_init(&ctx->rsrc_ref_lock);
+	INIT_LIST_HEAD(&ctx->rsrc_ref_list);
 	INIT_DELAYED_WORK(&ctx->rsrc_put_work, io_rsrc_put_work);
 	init_llist_head(&ctx->rsrc_put_llist);
 	return ctx;
@@ -7258,13 +7260,14 @@ static void io_rsrc_ref_kill(struct percpu_ref *ref)
 	complete(&data->done);
 }
 
-static void io_sqe_rsrc_set_node(struct fixed_rsrc_data *rsrc_data,
+static void io_sqe_rsrc_set_node(struct io_ring_ctx *ctx,
+				 struct fixed_rsrc_data *rsrc_data,
 				 struct fixed_rsrc_ref_node *ref_node)
 {
-	spin_lock_bh(&rsrc_data->lock);
+	spin_lock_bh(&ctx->rsrc_ref_lock);
 	rsrc_data->node = ref_node;
-	list_add_tail(&ref_node->node, &rsrc_data->ref_list);
-	spin_unlock_bh(&rsrc_data->lock);
+	list_add_tail(&ref_node->node, &ctx->rsrc_ref_list);
+	spin_unlock_bh(&ctx->rsrc_ref_lock);
 	percpu_ref_get(&rsrc_data->refs);
 }
 
@@ -7281,9 +7284,9 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 	if (!backup_node)
 		return -ENOMEM;
 
-	spin_lock_bh(&data->lock);
+	spin_lock_bh(&ctx->rsrc_ref_lock);
 	ref_node = data->node;
-	spin_unlock_bh(&data->lock);
+	spin_unlock_bh(&ctx->rsrc_ref_lock);
 	if (ref_node)
 		percpu_ref_kill(&ref_node->refs);
 
@@ -7299,7 +7302,7 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 		if (ret < 0) {
 			percpu_ref_resurrect(&data->refs);
 			reinit_completion(&data->done);
-			io_sqe_rsrc_set_node(data, backup_node);
+			io_sqe_rsrc_set_node(ctx, data, backup_node);
 			return ret;
 		}
 	} while (1);
@@ -7673,11 +7676,11 @@ static void io_rsrc_data_ref_zero(struct percpu_ref *ref)
 	data = ref_node->rsrc_data;
 	ctx = data->ctx;
 
-	spin_lock_bh(&data->lock);
+	spin_lock_bh(&ctx->rsrc_ref_lock);
 	ref_node->done = true;
 
-	while (!list_empty(&data->ref_list)) {
-		ref_node = list_first_entry(&data->ref_list,
+	while (!list_empty(&ctx->rsrc_ref_list)) {
+		ref_node = list_first_entry(&ctx->rsrc_ref_list,
 					struct fixed_rsrc_ref_node, node);
 		/* recycle ref nodes in order */
 		if (!ref_node->done)
@@ -7685,7 +7688,7 @@ static void io_rsrc_data_ref_zero(struct percpu_ref *ref)
 		list_del(&ref_node->node);
 		first_add |= llist_add(&ref_node->llist, &ctx->rsrc_put_llist);
 	}
-	spin_unlock_bh(&data->lock);
+	spin_unlock_bh(&ctx->rsrc_ref_lock);
 
 	if (percpu_ref_is_dying(&data->refs))
 		delay = 0;
@@ -7746,8 +7749,6 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 		return -ENOMEM;
 	file_data->ctx = ctx;
 	init_completion(&file_data->done);
-	INIT_LIST_HEAD(&file_data->ref_list);
-	spin_lock_init(&file_data->lock);
 
 	nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_FILES_TABLE);
 	file_data->table = kcalloc(nr_tables, sizeof(*file_data->table),
@@ -7808,7 +7809,7 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 		return PTR_ERR(ref_node);
 	}
 
-	io_sqe_rsrc_set_node(file_data, ref_node);
+	io_sqe_rsrc_set_node(ctx, file_data, ref_node);
 	return ret;
 out_fput:
 	for (i = 0; i < ctx->nr_user_files; i++) {
@@ -7969,7 +7970,7 @@ static int __io_sqe_files_update(struct io_ring_ctx *ctx,
 
 	if (needs_switch) {
 		percpu_ref_kill(&data->node->refs);
-		io_sqe_rsrc_set_node(data, ref_node);
+		io_sqe_rsrc_set_node(ctx, data, ref_node);
 	} else
 		destroy_fixed_rsrc_ref_node(ref_node);
 
-- 
1.8.3.1

[PATCH v5 04/13] io_uring: split alloc_fixed_file_ref_node

[Bijan Mottahedeh]

Split alloc_fixed_file_ref_node into resource generic/specific parts,
to be leveraged for fixed buffers.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 158b53f..d0bc4c8 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -7699,7 +7699,7 @@ static void io_rsrc_data_ref_zero(struct percpu_ref *ref)
 		queue_delayed_work(system_wq, &ctx->rsrc_put_work, delay);
 }
 
-static struct fixed_rsrc_ref_node *alloc_fixed_file_ref_node(
+static struct fixed_rsrc_ref_node *alloc_fixed_rsrc_ref_node(
 			struct io_ring_ctx *ctx)
 {
 	struct fixed_rsrc_ref_node *ref_node;
@@ -7715,9 +7715,21 @@ static struct fixed_rsrc_ref_node *alloc_fixed_file_ref_node(
 	}
 	INIT_LIST_HEAD(&ref_node->node);
 	INIT_LIST_HEAD(&ref_node->rsrc_list);
+	ref_node->done = false;
+	return ref_node;
+}
+
+static struct fixed_rsrc_ref_node *alloc_fixed_file_ref_node(
+			struct io_ring_ctx *ctx)
+{
+	struct fixed_rsrc_ref_node *ref_node;
+
+	ref_node = alloc_fixed_rsrc_ref_node(ctx);
+	if (!ref_node)
+		return NULL;
+
 	ref_node->rsrc_data = ctx->file_data;
 	ref_node->rsrc_put = io_ring_file_put;
-	ref_node->done = false;
 	return ref_node;
 }
 
-- 
1.8.3.1

[PATCH v5 05/13] io_uring: add rsrc_ref locking routines

[Bijan Mottahedeh]

Encapsulate resource reference locking into separate routines.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 22 ++++++++++++++++------
 1 file changed, 16 insertions(+), 6 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index d0bc4c8..c5e3269 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -7260,14 +7260,24 @@ static void io_rsrc_ref_kill(struct percpu_ref *ref)
 	complete(&data->done);
 }
 
+static inline void io_rsrc_ref_lock(struct io_ring_ctx *ctx)
+{
+	spin_lock_bh(&ctx->rsrc_ref_lock);
+}
+
+static inline void io_rsrc_ref_unlock(struct io_ring_ctx *ctx)
+{
+	spin_unlock_bh(&ctx->rsrc_ref_lock);
+}
+
 static void io_sqe_rsrc_set_node(struct io_ring_ctx *ctx,
 				 struct fixed_rsrc_data *rsrc_data,
 				 struct fixed_rsrc_ref_node *ref_node)
 {
-	spin_lock_bh(&ctx->rsrc_ref_lock);
+	io_rsrc_ref_lock(ctx);
 	rsrc_data->node = ref_node;
 	list_add_tail(&ref_node->node, &ctx->rsrc_ref_list);
-	spin_unlock_bh(&ctx->rsrc_ref_lock);
+	io_rsrc_ref_unlock(ctx);
 	percpu_ref_get(&rsrc_data->refs);
 }
 
@@ -7284,9 +7294,9 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 	if (!backup_node)
 		return -ENOMEM;
 
-	spin_lock_bh(&ctx->rsrc_ref_lock);
+	io_rsrc_ref_lock(ctx);
 	ref_node = data->node;
-	spin_unlock_bh(&ctx->rsrc_ref_lock);
+	io_rsrc_ref_unlock(ctx);
 	if (ref_node)
 		percpu_ref_kill(&ref_node->refs);
 
@@ -7676,7 +7686,7 @@ static void io_rsrc_data_ref_zero(struct percpu_ref *ref)
 	data = ref_node->rsrc_data;
 	ctx = data->ctx;
 
-	spin_lock_bh(&ctx->rsrc_ref_lock);
+	io_rsrc_ref_lock(ctx);
 	ref_node->done = true;
 
 	while (!list_empty(&ctx->rsrc_ref_list)) {
@@ -7688,7 +7698,7 @@ static void io_rsrc_data_ref_zero(struct percpu_ref *ref)
 		list_del(&ref_node->node);
 		first_add |= llist_add(&ref_node->llist, &ctx->rsrc_put_llist);
 	}
-	spin_unlock_bh(&ctx->rsrc_ref_lock);
+	io_rsrc_ref_unlock(ctx);
 
 	if (percpu_ref_is_dying(&data->refs))
 		delay = 0;
-- 
1.8.3.1

[PATCH v5 06/13] io_uring: implement fixed buffers registration similar to fixed files

[Bijan Mottahedeh]

Apply fixed_rsrc functionality for fixed buffers support.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 234 +++++++++++++++++++++++++++++++++++++++++++++++++---------
 1 file changed, 198 insertions(+), 36 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index c5e3269..1477015 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -104,6 +104,14 @@
 #define IORING_MAX_RESTRICTIONS	(IORING_RESTRICTION_LAST + \
 				 IORING_REGISTER_LAST + IORING_OP_LAST)
 
+/*
+ * Shift of 7 is 128 entries, or exactly one page on 64-bit archs
+ */
+#define IORING_BUF_TABLE_SHIFT	7	/* struct io_mapped_ubuf */
+#define IORING_MAX_BUFS_TABLE	(1U << IORING_BUF_TABLE_SHIFT)
+#define IORING_BUF_TABLE_MASK	(IORING_MAX_BUFS_TABLE - 1)
+#define IORING_MAX_FIXED_BUFS	UIO_MAXIOV
+
 struct io_uring {
 	u32 head ____cacheline_aligned_in_smp;
 	u32 tail ____cacheline_aligned_in_smp;
@@ -336,8 +344,8 @@ struct io_ring_ctx {
 	unsigned		nr_user_files;
 
 	/* if used, fixed mapped user buffers */
+	struct fixed_rsrc_data	*buf_data;
 	unsigned		nr_user_bufs;
-	struct io_mapped_ubuf	*user_bufs;
 
 	struct user_struct	*user;
 
@@ -2951,6 +2959,15 @@ static void kiocb_done(struct kiocb *kiocb, ssize_t ret,
 		io_rw_done(kiocb, ret);
 }
 
+static inline struct io_mapped_ubuf *io_buf_from_index(struct io_ring_ctx *ctx,
+						       int index)
+{
+	struct fixed_rsrc_table *table;
+
+	table = &ctx->buf_data->table[index >> IORING_BUF_TABLE_SHIFT];
+	return &table->bufs[index & IORING_BUF_TABLE_MASK];
+}
+
 static ssize_t io_import_fixed(struct io_kiocb *req, int rw,
 			       struct iov_iter *iter)
 {
@@ -2964,7 +2981,7 @@ static ssize_t io_import_fixed(struct io_kiocb *req, int rw,
 	if (unlikely(buf_index >= ctx->nr_user_bufs))
 		return -EFAULT;
 	index = array_index_nospec(buf_index, ctx->nr_user_bufs);
-	imu = &ctx->user_bufs[index];
+	imu = io_buf_from_index(ctx, index);
 	buf_addr = req->rw.addr;
 
 	/* overflow */
@@ -8317,28 +8334,71 @@ static unsigned long ring_pages(unsigned sq_entries, unsigned cq_entries)
 	return pages;
 }
 
-static int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
+static void io_buffer_unmap(struct io_ring_ctx *ctx, struct io_mapped_ubuf *imu)
 {
-	int i, j;
+	unsigned int i;
 
-	if (!ctx->user_bufs)
-		return -ENXIO;
+	for (i = 0; i < imu->nr_bvecs; i++)
+		unpin_user_page(imu->bvec[i].bv_page);
 
-	for (i = 0; i < ctx->nr_user_bufs; i++) {
-		struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
+	if (imu->acct_pages)
+		io_unaccount_mem(ctx, imu->nr_bvecs, ACCT_PINNED);
+	kvfree(imu->bvec);
+	imu->nr_bvecs = 0;
+}
 
-		for (j = 0; j < imu->nr_bvecs; j++)
-			unpin_user_page(imu->bvec[j].bv_page);
+static void io_buffers_unmap(struct io_ring_ctx *ctx)
+{
+	unsigned int i;
+	struct io_mapped_ubuf *imu;
 
-		if (imu->acct_pages)
-			io_unaccount_mem(ctx, imu->acct_pages, ACCT_PINNED);
-		kvfree(imu->bvec);
-		imu->nr_bvecs = 0;
+	for (i = 0; i < ctx->nr_user_bufs; i++) {
+		imu = io_buf_from_index(ctx, i);
+		io_buffer_unmap(ctx, imu);
 	}
+}
+
+static void io_buffers_map_free(struct io_ring_ctx *ctx)
+{
+	struct fixed_rsrc_data *data = ctx->buf_data;
+	unsigned int nr_tables, i;
 
-	kfree(ctx->user_bufs);
-	ctx->user_bufs = NULL;
+	if (!data)
+		return;
+
+	nr_tables = DIV_ROUND_UP(ctx->nr_user_bufs, IORING_MAX_BUFS_TABLE);
+	for (i = 0; i < nr_tables; i++)
+		kfree(data->table[i].bufs);
+	kfree(data->table);
+	percpu_ref_exit(&data->refs);
+	kfree(data);
+	ctx->buf_data = NULL;
 	ctx->nr_user_bufs = 0;
+}
+
+static int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
+{
+	struct fixed_rsrc_data *data = ctx->buf_data;
+	struct fixed_rsrc_ref_node *ref_node = NULL;
+
+	if (!data)
+		return -ENXIO;
+
+	io_rsrc_ref_lock(ctx);
+	ref_node = data->node;
+	io_rsrc_ref_unlock(ctx);
+	if (ref_node)
+		percpu_ref_kill(&ref_node->refs);
+
+	percpu_ref_kill(&data->refs);
+
+	/* wait for all refs nodes to complete */
+	flush_delayed_work(&ctx->rsrc_put_work);
+	wait_for_completion(&data->done);
+
+	io_buffers_unmap(ctx);
+	io_buffers_map_free(ctx);
+
 	return 0;
 }
 
@@ -8391,7 +8451,9 @@ static bool headpage_already_acct(struct io_ring_ctx *ctx, struct page **pages,
 
 	/* check previously registered pages */
 	for (i = 0; i < ctx->nr_user_bufs; i++) {
-		struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
+		struct io_mapped_ubuf *imu;
+
+		imu = io_buf_from_index(ctx, i);
 
 		for (j = 0; j < imu->nr_bvecs; j++) {
 			if (!PageCompound(imu->bvec[j].bv_page))
@@ -8526,19 +8588,79 @@ static int io_sqe_buffer_register(struct io_ring_ctx *ctx, struct iovec *iov,
 	return ret;
 }
 
-static int io_buffers_map_alloc(struct io_ring_ctx *ctx, unsigned int nr_args)
+static void io_free_buf_tables(struct fixed_rsrc_data *buf_data,
+			       unsigned int nr_tables)
 {
-	if (ctx->user_bufs)
-		return -EBUSY;
-	if (!nr_args || nr_args > UIO_MAXIOV)
-		return -EINVAL;
+	int i;
 
-	ctx->user_bufs = kcalloc(nr_args, sizeof(struct io_mapped_ubuf),
-					GFP_KERNEL);
-	if (!ctx->user_bufs)
-		return -ENOMEM;
+	for (i = 0; i < nr_tables; i++) {
+		struct fixed_rsrc_table *table = &buf_data->table[i];
 
-	return 0;
+		kfree(table->bufs);
+	}
+}
+
+static int io_alloc_buf_tables(struct fixed_rsrc_data *buf_data,
+			       unsigned int nr_tables, unsigned int nr_bufs)
+{
+	int i;
+
+	for (i = 0; i < nr_tables; i++) {
+		struct fixed_rsrc_table *table = &buf_data->table[i];
+		unsigned int this_bufs;
+
+		this_bufs = min(nr_bufs, IORING_MAX_BUFS_TABLE);
+		table->bufs = kcalloc(this_bufs, sizeof(struct io_mapped_ubuf),
+				      GFP_KERNEL);
+		if (!table->bufs)
+			break;
+		nr_bufs -= this_bufs;
+	}
+
+	if (i == nr_tables)
+		return 0;
+
+	io_free_buf_tables(buf_data, i);
+	return 1;
+}
+
+static struct fixed_rsrc_data *io_buffers_map_alloc(struct io_ring_ctx *ctx,
+						    unsigned int nr_args)
+{
+	unsigned int nr_tables;
+	struct fixed_rsrc_data *buf_data;
+	int ret = -ENOMEM;
+
+	if (!nr_args || nr_args > IORING_MAX_FIXED_BUFS)
+		return ERR_PTR(-EINVAL);
+
+	buf_data = kzalloc(sizeof(*ctx->buf_data), GFP_KERNEL);
+	if (!buf_data)
+		return ERR_PTR(-ENOMEM);
+	buf_data->ctx = ctx;
+	init_completion(&buf_data->done);
+
+	nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_BUFS_TABLE);
+	buf_data->table = kcalloc(nr_tables, sizeof(*buf_data->table),
+				  GFP_KERNEL);
+	if (!buf_data->table)
+		goto out_free;
+
+	if (percpu_ref_init(&buf_data->refs, io_rsrc_ref_kill,
+			    PERCPU_REF_ALLOW_REINIT, GFP_KERNEL))
+		goto out_free;
+
+	if (io_alloc_buf_tables(buf_data, nr_tables, nr_args))
+		goto out_ref;
+
+	return buf_data;
+
+out_ref:
+	percpu_ref_exit(&buf_data->refs);
+out_free:
+	kfree(buf_data->table);
+	kfree(buf_data);
+	return ERR_PTR(ret);
 }
 
 static int io_buffer_validate(struct iovec *iov)
@@ -8558,39 +8680,78 @@ static int io_buffer_validate(struct iovec *iov)
 	return 0;
 }
 
+static void io_ring_buf_put(struct io_ring_ctx *ctx, struct io_rsrc_put *prsrc)
+{
+	io_buffer_unmap(ctx, prsrc->buf);
+}
+
+static struct fixed_rsrc_ref_node *alloc_fixed_buf_ref_node(
+			struct io_ring_ctx *ctx)
+{
+	struct fixed_rsrc_ref_node *ref_node;
+
+	ref_node = alloc_fixed_rsrc_ref_node(ctx);
+	ref_node->rsrc_data = ctx->buf_data;
+	ref_node->rsrc_put = io_ring_buf_put;
+	return ref_node;
+}
+
 static int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
 				   unsigned int nr_args)
 {
 	int i, ret;
 	struct iovec iov;
 	struct page *last_hpage = NULL;
+	struct fixed_rsrc_ref_node *ref_node;
+	struct fixed_rsrc_data *buf_data;
 
-	ret = io_buffers_map_alloc(ctx, nr_args);
-	if (ret)
-		return ret;
+	if (ctx->nr_user_bufs)
+		return -EBUSY;
 
-	for (i = 0; i < nr_args; i++) {
-		struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
+	buf_data = io_buffers_map_alloc(ctx, nr_args);
+	if (IS_ERR(buf_data))
+		return PTR_ERR(buf_data);
+
+	for (i = 0; i < nr_args; i++, ctx->nr_user_bufs++) {
+		struct io_mapped_ubuf *imu;
 
 		ret = io_copy_iov(ctx, &iov, arg, i);
 		if (ret)
 			break;
 
+		/* allow sparse sets */
+		if (!iov.iov_base && !iov.iov_len)
+			continue;
+
 		ret = io_buffer_validate(&iov);
 		if (ret)
 			break;
 
+		imu = io_buf_from_index(ctx, i);
+
 		ret = io_sqe_buffer_register(ctx, &iov, imu, &last_hpage);
 		if (ret)
 			break;
+	}
 
-		ctx->nr_user_bufs++;
+	ctx->buf_data = buf_data;
+	if (ret) {
+		io_sqe_buffers_unregister(ctx);
+		return ret;
 	}
 
-	if (ret)
+	ref_node = alloc_fixed_buf_ref_node(ctx);
+	if (IS_ERR(ref_node)) {
 		io_sqe_buffers_unregister(ctx);
+		return PTR_ERR(ref_node);
+	}
 
-	return ret;
+	buf_data->node = ref_node;
+	io_rsrc_ref_lock(ctx);
+	list_add(&ref_node->node, &ctx->rsrc_ref_list);
+	io_rsrc_ref_unlock(ctx);
+	percpu_ref_get(&buf_data->refs);
+	return 0;
 }
 
 static int io_eventfd_register(struct io_ring_ctx *ctx, void __user *arg)
@@ -9385,7 +9546,7 @@ static void __io_uring_show_fdinfo(struct io_ring_ctx *ctx, struct seq_file *m)
 	}
 	seq_printf(m, "UserBufs:\t%u\n", ctx->nr_user_bufs);
 	for (i = 0; has_lock && i < ctx->nr_user_bufs; i++) {
-		struct io_mapped_ubuf *buf = &ctx->user_bufs[i];
+		struct io_mapped_ubuf *buf = io_buf_from_index(ctx, i);
 
 		seq_printf(m, "%5u: 0x%llx/%u\n", i, buf->ubuf,
 						(unsigned int) buf->len);
@@ -9900,6 +10061,7 @@ static bool io_register_op_must_quiesce(int op)
 	switch (op) {
 	case IORING_UNREGISTER_FILES:
 	case IORING_REGISTER_FILES_UPDATE:
+	case IORING_UNREGISTER_BUFFERS:
 	case IORING_REGISTER_PROBE:
 	case IORING_REGISTER_PERSONALITY:
 	case IORING_UNREGISTER_PERSONALITY:
-- 
1.8.3.1

[PATCH v5 07/13] io_uring: create common fixed_rsrc_ref_node handling routines

[Bijan Mottahedeh]

Create common routines to be used for both files/buffers registration.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 53 +++++++++++++++++++++++++++++------------------------
 1 file changed, 29 insertions(+), 24 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 1477015..6ebfe1f 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -1018,6 +1018,8 @@ enum io_mem_account {
 static void destroy_fixed_rsrc_ref_node(struct fixed_rsrc_ref_node *ref_node);
 static struct fixed_rsrc_ref_node *alloc_fixed_file_ref_node(
 			struct io_ring_ctx *ctx);
+static struct fixed_rsrc_ref_node *alloc_fixed_buf_ref_node(
+			struct io_ring_ctx *ctx);
 
 static void __io_complete_rw(struct io_kiocb *req, long res, long res2,
 			     struct io_comp_state *cs);
@@ -7298,16 +7300,15 @@ static void io_sqe_rsrc_set_node(struct io_ring_ctx *ctx,
 	percpu_ref_get(&rsrc_data->refs);
 }
 
-static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
+static int io_rsrc_ref_quiesce(struct fixed_rsrc_data *data,
+			       struct io_ring_ctx *ctx,
+			       struct fixed_rsrc_ref_node *(*alloc)(
+			       struct io_ring_ctx *ctx))
 {
-	struct fixed_rsrc_data *data = ctx->file_data;
 	struct fixed_rsrc_ref_node *backup_node, *ref_node = NULL;
-	unsigned nr_tables, i;
 	int ret;
 
-	if (!data)
-		return -ENXIO;
-	backup_node = alloc_fixed_file_ref_node(ctx);
+	backup_node = alloc(ctx);
 	if (!backup_node)
 		return -ENOMEM;
 
@@ -7334,6 +7335,23 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 		}
 	} while (1);
 
+	destroy_fixed_rsrc_ref_node(backup_node);
+	return 0;
+}
+
+static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
+{
+	struct fixed_rsrc_data *data = ctx->file_data;
+	unsigned int nr_tables, i;
+	int ret;
+
+	if (!data)
+		return -ENXIO;
+
+	ret = io_rsrc_ref_quiesce(data, ctx, alloc_fixed_file_ref_node);
+	if (ret)
+		return ret;
+
 	__io_sqe_files_unregister(ctx);
 	nr_tables = DIV_ROUND_UP(ctx->nr_user_files, IORING_MAX_FILES_TABLE);
 	for (i = 0; i < nr_tables; i++)
@@ -7343,7 +7361,6 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 	kfree(data);
 	ctx->file_data = NULL;
 	ctx->nr_user_files = 0;
-	destroy_fixed_rsrc_ref_node(backup_node);
 	return 0;
 }
 
@@ -8379,22 +8396,14 @@ static void io_buffers_map_free(struct io_ring_ctx *ctx)
 static int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
 {
 	struct fixed_rsrc_data *data = ctx->buf_data;
-	struct fixed_rsrc_ref_node *ref_node = NULL;
+	int ret;
 
 	if (!data)
 		return -ENXIO;
 
-	io_rsrc_ref_lock(ctx);
-	ref_node = data->node;
-	io_rsrc_ref_unlock(ctx);
-	if (ref_node)
-		percpu_ref_kill(&ref_node->refs);
-
-	percpu_ref_kill(&data->refs);
-
-	/* wait for all refs nodes to complete */
-	flush_delayed_work(&ctx->rsrc_put_work);
-	wait_for_completion(&data->done);
+	ret = io_rsrc_ref_quiesce(data, ctx, alloc_fixed_buf_ref_node);
+	if (ret)
+		return ret;
 
 	io_buffers_unmap(ctx);
 	io_buffers_map_free(ctx);
@@ -8746,11 +8755,7 @@ static int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
 		return PTR_ERR(ref_node);
 	}
 
-	buf_data->node = ref_node;
-	io_rsrc_ref_lock(ctx);
-	list_add(&ref_node->node, &ctx->rsrc_ref_list);
-	io_rsrc_ref_unlock(ctx);
-	percpu_ref_get(&buf_data->refs);
+	io_sqe_rsrc_set_node(ctx, buf_data, ref_node);
 	return 0;
 }
 
-- 
1.8.3.1

[PATCH v5 08/13] io_uring: generalize files_update functionlity to rsrc_update

[Bijan Mottahedeh]

Generalize files_update functionality to rsrc_update in order to
leverage it for buffers updates.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c                 | 19 ++++++++++++++-----
 include/uapi/linux/io_uring.h |  6 +++++-
 2 files changed, 19 insertions(+), 6 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 6ebfe1f..f9f458c 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -5954,7 +5954,7 @@ static int io_async_cancel(struct io_kiocb *req)
 }
 
 static int io_rsrc_update_prep(struct io_kiocb *req,
-				const struct io_uring_sqe *sqe)
+			       const struct io_uring_sqe *sqe)
 {
 	if (unlikely(req->ctx->flags & IORING_SETUP_SQPOLL))
 		return -EINVAL;
@@ -5971,8 +5971,11 @@ static int io_rsrc_update_prep(struct io_kiocb *req,
 	return 0;
 }
 
-static int io_files_update(struct io_kiocb *req, bool force_nonblock,
-			   struct io_comp_state *cs)
+static int io_rsrc_update(struct io_kiocb *req, bool force_nonblock,
+			  struct io_comp_state *cs,
+			  int (*update)(struct io_ring_ctx *ctx,
+					struct io_uring_rsrc_update *up,
+					unsigned int nr_args))
 {
 	struct io_ring_ctx *ctx = req->ctx;
 	struct io_uring_rsrc_update up;
@@ -5982,10 +5985,10 @@ static int io_files_update(struct io_kiocb *req, bool force_nonblock,
 		return -EAGAIN;
 
 	up.offset = req->rsrc_update.offset;
-	up.fds = req->rsrc_update.arg;
+	up.rsrc = req->rsrc_update.arg;
 
 	mutex_lock(&ctx->uring_lock);
-	ret = __io_sqe_files_update(ctx, &up, req->rsrc_update.nr_args);
+	ret = (*update)(ctx, &up, req->rsrc_update.nr_args);
 	mutex_unlock(&ctx->uring_lock);
 
 	if (ret < 0)
@@ -5994,6 +5997,12 @@ static int io_files_update(struct io_kiocb *req, bool force_nonblock,
 	return 0;
 }
 
+static int io_files_update(struct io_kiocb *req, bool force_nonblock,
+			   struct io_comp_state *cs)
+{
+	return io_rsrc_update(req, force_nonblock, cs, __io_sqe_files_update);
+}
+
 static int io_req_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
 {
 	switch (req->opcode) {
diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
index 77de7c0..f51190b 100644
--- a/include/uapi/linux/io_uring.h
+++ b/include/uapi/linux/io_uring.h
@@ -288,7 +288,11 @@ enum {
 struct io_uring_rsrc_update {
 	__u32 offset;
 	__u32 resv;
-	__aligned_u64 /* __s32 * */ fds;
+	union {
+		__aligned_u64 /* __s32 * */ fds;
+		__aligned_u64 /* __s32 * */ iovs;
+		__aligned_u64 /* __s32 * */ rsrc;
+	};
 };
 
 #define io_uring_files_update	io_uring_rsrc_update
-- 
1.8.3.1

[PATCH v5 09/13] io_uring: support buffer registration updates

[Bijan Mottahedeh]

Introduce IORING_REGISTER_BUFFERS_UPDATE and IORING_OP_BUFFERS_UPDATE,
consistent with file registration update.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c                 | 128 +++++++++++++++++++++++++++++++++++++++++-
 include/uapi/linux/io_uring.h |   2 +
 2 files changed, 128 insertions(+), 2 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index f9f458c..d3a6185 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -1008,6 +1008,9 @@ struct io_op_def {
 		.work_flags		= IO_WQ_WORK_MM | IO_WQ_WORK_FILES |
 						IO_WQ_WORK_FS | IO_WQ_WORK_BLKCG,
 	},
+	[IORING_OP_BUFFERS_UPDATE] = {
+		.work_flags		= IO_WQ_WORK_MM,
+	},
 };
 
 enum io_mem_account {
@@ -1033,6 +1036,9 @@ static void __io_complete_rw(struct io_kiocb *req, long res, long res2,
 static int __io_sqe_files_update(struct io_ring_ctx *ctx,
 				 struct io_uring_rsrc_update *ip,
 				 unsigned nr_args);
+static int __io_sqe_buffers_update(struct io_ring_ctx *ctx,
+				   struct io_uring_rsrc_update *up,
+				   unsigned int nr_args);
 static void __io_clean_op(struct io_kiocb *req);
 static struct file *io_file_get(struct io_submit_state *state,
 				struct io_kiocb *req, int fd, bool fixed);
@@ -6003,6 +6009,12 @@ static int io_files_update(struct io_kiocb *req, bool force_nonblock,
 	return io_rsrc_update(req, force_nonblock, cs, __io_sqe_files_update);
 }
 
+static int io_buffers_update(struct io_kiocb *req, bool force_nonblock,
+			     struct io_comp_state *cs)
+{
+	return io_rsrc_update(req, force_nonblock, cs, __io_sqe_buffers_update);
+}
+
 static int io_req_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
 {
 	switch (req->opcode) {
@@ -6074,11 +6086,13 @@ static int io_req_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
 		return io_renameat_prep(req, sqe);
 	case IORING_OP_UNLINKAT:
 		return io_unlinkat_prep(req, sqe);
+	case IORING_OP_BUFFERS_UPDATE:
+		return io_rsrc_update_prep(req, sqe);
 	}
 
 	printk_once(KERN_WARNING "io_uring: unhandled opcode %d\n",
 			req->opcode);
-	return-EINVAL;
+	return -EINVAL;
 }
 
 static int io_req_defer_prep(struct io_kiocb *req,
@@ -6333,6 +6347,9 @@ static int io_issue_sqe(struct io_kiocb *req, bool force_nonblock,
 	case IORING_OP_UNLINKAT:
 		ret = io_unlinkat(req, force_nonblock);
 		break;
+	case IORING_OP_BUFFERS_UPDATE:
+		ret = io_buffers_update(req, force_nonblock, cs);
+		break;
 	default:
 		ret = -EINVAL;
 		break;
@@ -8036,8 +8053,9 @@ static int __io_sqe_files_update(struct io_ring_ctx *ctx,
 	if (needs_switch) {
 		percpu_ref_kill(&data->node->refs);
 		io_sqe_rsrc_set_node(ctx, data, ref_node);
-	} else
+	} else {
 		destroy_fixed_rsrc_ref_node(ref_node);
+	}
 
 	return done ? done : err;
 }
@@ -8370,6 +8388,7 @@ static void io_buffer_unmap(struct io_ring_ctx *ctx, struct io_mapped_ubuf *imu)
 	if (imu->acct_pages)
 		io_unaccount_mem(ctx, imu->nr_bvecs, ACCT_PINNED);
 	kvfree(imu->bvec);
+	imu->bvec = NULL;
 	imu->nr_bvecs = 0;
 }
 
@@ -8573,6 +8592,7 @@ static int io_sqe_buffer_register(struct io_ring_ctx *ctx, struct iovec *iov,
 		if (pret > 0)
 			unpin_user_pages(pages, pret);
 		kvfree(imu->bvec);
+		imu->bvec = NULL;
 		goto done;
 	}
 
@@ -8701,6 +8721,8 @@ static int io_buffer_validate(struct iovec *iov)
 static void io_ring_buf_put(struct io_ring_ctx *ctx, struct io_rsrc_put *prsrc)
 {
 	io_buffer_unmap(ctx, prsrc->buf);
+	kvfree(prsrc->buf);
+	prsrc->buf = NULL;
 }
 
 static struct fixed_rsrc_ref_node *alloc_fixed_buf_ref_node(
@@ -8768,6 +8790,104 @@ static int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
 	return 0;
 }
 
+static inline int io_queue_buffer_removal(struct fixed_rsrc_data *data,
+					  struct io_mapped_ubuf *imu)
+{
+	return io_queue_rsrc_removal(data, (void *)imu);
+}
+
+static int __io_sqe_buffers_update(struct io_ring_ctx *ctx,
+				   struct io_uring_rsrc_update *up,
+				   unsigned int nr_args)
+{
+	struct fixed_rsrc_data *data = ctx->buf_data;
+	struct fixed_rsrc_ref_node *ref_node;
+	struct io_mapped_ubuf *imu;
+	struct iovec iov;
+	struct iovec __user *iovs;
+	struct page *last_hpage = NULL;
+	__u32 done;
+	int i, err;
+	bool needs_switch = false;
+
+	if (check_add_overflow(up->offset, nr_args, &done))
+		return -EOVERFLOW;
+	if (done > ctx->nr_user_bufs)
+		return -EINVAL;
+
+	ref_node = alloc_fixed_buf_ref_node(ctx);
+	if (IS_ERR(ref_node))
+		return PTR_ERR(ref_node);
+
+	done = 0;
+	iovs = u64_to_user_ptr(up->iovs);
+	while (nr_args) {
+		struct fixed_rsrc_table *table;
+		unsigned int index;
+
+		err = 0;
+		if (copy_from_user(&iov, &iovs[done], sizeof(iov))) {
+			err = -EFAULT;
+			break;
+		}
+		i = array_index_nospec(up->offset, ctx->nr_user_bufs);
+		table = &ctx->buf_data->table[i >> IORING_BUF_TABLE_SHIFT];
+		index = i & IORING_BUF_TABLE_MASK;
+		imu = &table->bufs[index];
+		if (table->bufs[index].ubuf) {
+			struct io_mapped_ubuf *dup;
+
+			dup = kmemdup(imu, sizeof(*imu), GFP_KERNEL);
+			if (!dup) {
+				err = -ENOMEM;
+				break;
+			}
+			err = io_queue_buffer_removal(data, dup);
+			if (err)
+				break;
+			memset(imu, 0, sizeof(*imu));
+			needs_switch = true;
+		}
+		if (!io_buffer_validate(&iov)) {
+			err = io_sqe_buffer_register(ctx, &iov, imu,
+						     &last_hpage);
+			if (err) {
+				memset(imu, 0, sizeof(*imu));
+				break;
+			}
+		}
+		nr_args--;
+		done++;
+		up->offset++;
+	}
+
+	if (needs_switch) {
+		percpu_ref_kill(&data->node->refs);
+		io_sqe_rsrc_set_node(ctx, data, ref_node);
+	} else {
+		destroy_fixed_rsrc_ref_node(ref_node);
+	}
+
+	return done ? done : err;
+}
+
+static int io_sqe_buffers_update(struct io_ring_ctx *ctx, void __user *arg,
+				 unsigned int nr_args)
+{
+	struct io_uring_rsrc_update up;
+
+	if (!ctx->buf_data)
+		return -ENXIO;
+	if (!nr_args)
+		return -EINVAL;
+	if (copy_from_user(&up, arg, sizeof(up)))
+		return -EFAULT;
+	if (up.resv)
+		return -EINVAL;
+
+	return __io_sqe_buffers_update(ctx, &up, nr_args);
+}
+
 static int io_eventfd_register(struct io_ring_ctx *ctx, void __user *arg)
 {
 	__s32 __user *fds = arg;
@@ -10076,6 +10196,7 @@ static bool io_register_op_must_quiesce(int op)
 	case IORING_UNREGISTER_FILES:
 	case IORING_REGISTER_FILES_UPDATE:
 	case IORING_UNREGISTER_BUFFERS:
+	case IORING_REGISTER_BUFFERS_UPDATE:
 	case IORING_REGISTER_PROBE:
 	case IORING_REGISTER_PERSONALITY:
 	case IORING_UNREGISTER_PERSONALITY:
@@ -10151,6 +10272,9 @@ static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
 			break;
 		ret = io_sqe_buffers_unregister(ctx);
 		break;
+	case IORING_REGISTER_BUFFERS_UPDATE:
+		ret = io_sqe_buffers_update(ctx, arg, nr_args);
+		break;
 	case IORING_REGISTER_FILES:
 		ret = io_sqe_files_register(ctx, arg, nr_args);
 		break;
diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
index f51190b..b289ef8 100644
--- a/include/uapi/linux/io_uring.h
+++ b/include/uapi/linux/io_uring.h
@@ -137,6 +137,7 @@ enum {
 	IORING_OP_SHUTDOWN,
 	IORING_OP_RENAMEAT,
 	IORING_OP_UNLINKAT,
+	IORING_OP_BUFFERS_UPDATE,
 
 	/* this goes last, obviously */
 	IORING_OP_LAST,
@@ -280,6 +281,7 @@ enum {
 	IORING_UNREGISTER_PERSONALITY		= 10,
 	IORING_REGISTER_RESTRICTIONS		= 11,
 	IORING_REGISTER_ENABLE_RINGS		= 12,
+	IORING_REGISTER_BUFFERS_UPDATE		= 13,
 
 	/* this goes last */
 	IORING_REGISTER_LAST
-- 
1.8.3.1

[PATCH v5 10/13] io_uring: create common fixed_rsrc_data allocation routines

[Bijan Mottahedeh]

Create common alloc/free fixed_rsrc_data routines for both files and
buffers.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 75 ++++++++++++++++++++++++++++++++---------------------------
 1 file changed, 41 insertions(+), 34 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index d3a6185..96f760e 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -7365,6 +7365,33 @@ static int io_rsrc_ref_quiesce(struct fixed_rsrc_data *data,
 	return 0;
 }
 
+static struct fixed_rsrc_data *alloc_fixed_rsrc_data(struct io_ring_ctx *ctx)
+{
+	struct fixed_rsrc_data *data;
+
+	data = kzalloc(sizeof(*data), GFP_KERNEL);
+	if (!data)
+		return ERR_PTR(-ENOMEM);
+
+	data->ctx = ctx;
+	init_completion(&data->done);
+
+	if (percpu_ref_init(&data->refs, io_rsrc_ref_kill,
+			    PERCPU_REF_ALLOW_REINIT, GFP_KERNEL)) {
+		kfree(data);
+		return ERR_PTR(-ENOMEM);
+	}
+
+	return data;
+}
+
+static void free_fixed_rsrc_data(struct fixed_rsrc_data *data)
+{
+	percpu_ref_exit(&data->refs);
+	kfree(data->table);
+	kfree(data);
+}
+
 static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 {
 	struct fixed_rsrc_data *data = ctx->file_data;
@@ -7382,9 +7409,7 @@ static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
 	nr_tables = DIV_ROUND_UP(ctx->nr_user_files, IORING_MAX_FILES_TABLE);
 	for (i = 0; i < nr_tables; i++)
 		kfree(data->table[i].files);
-	kfree(data->table);
-	percpu_ref_exit(&data->refs);
-	kfree(data);
+	free_fixed_rsrc_data(ctx->file_data);
 	ctx->file_data = NULL;
 	ctx->nr_user_files = 0;
 	return 0;
@@ -7826,11 +7851,9 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 	if (nr_args > IORING_MAX_FIXED_FILES)
 		return -EMFILE;
 
-	file_data = kzalloc(sizeof(*ctx->file_data), GFP_KERNEL);
-	if (!file_data)
-		return -ENOMEM;
-	file_data->ctx = ctx;
-	init_completion(&file_data->done);
+	file_data = alloc_fixed_rsrc_data(ctx);
+	if (IS_ERR(file_data))
+		return PTR_ERR(ref_node);
 
 	nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_FILES_TABLE);
 	file_data->table = kcalloc(nr_tables, sizeof(*file_data->table),
@@ -7838,12 +7861,8 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 	if (!file_data->table)
 		goto out_free;
 
-	if (percpu_ref_init(&file_data->refs, io_rsrc_ref_kill,
-				PERCPU_REF_ALLOW_REINIT, GFP_KERNEL))
-		goto out_free;
-
 	if (io_sqe_alloc_file_tables(file_data, nr_tables, nr_args))
-		goto out_ref;
+		goto out_free;
 	ctx->file_data = file_data;
 
 	for (i = 0; i < nr_args; i++, ctx->nr_user_files++) {
@@ -7902,11 +7921,8 @@ static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
 	for (i = 0; i < nr_tables; i++)
 		kfree(file_data->table[i].files);
 	ctx->nr_user_files = 0;
-out_ref:
-	percpu_ref_exit(&file_data->refs);
 out_free:
-	kfree(file_data->table);
-	kfree(file_data);
+	free_fixed_rsrc_data(ctx->file_data);
 	ctx->file_data = NULL;
 	return ret;
 }
@@ -8672,32 +8688,23 @@ static struct fixed_rsrc_data *io_buffers_map_alloc(struct io_ring_ctx *ctx,
 	if (!nr_args || nr_args > IORING_MAX_FIXED_BUFS)
 		return ERR_PTR(-EINVAL);
 
-	buf_data = kzalloc(sizeof(*ctx->buf_data), GFP_KERNEL);
-	if (!buf_data)
-		return ERR_PTR(-ENOMEM);
-	buf_data->ctx = ctx;
-	init_completion(&buf_data->done);
+	buf_data = alloc_fixed_rsrc_data(ctx);
+	if (IS_ERR(buf_data))
+		return buf_data;
 
 	nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_BUFS_TABLE);
 	buf_data->table = kcalloc(nr_tables, sizeof(*buf_data->table),
 				  GFP_KERNEL);
 	if (!buf_data->table)
-		goto out_free;
-
-	if (percpu_ref_init(&buf_data->refs, io_rsrc_ref_kill,
-			    PERCPU_REF_ALLOW_REINIT, GFP_KERNEL))
-		goto out_free;
+		goto out;
 
 	if (io_alloc_buf_tables(buf_data, nr_tables, nr_args))
-		goto out_ref;
+		goto out;
 
 	return buf_data;
-
-out_ref:
-	percpu_ref_exit(&buf_data->refs);
-out_free:
-	kfree(buf_data->table);
-	kfree(buf_data);
+out:
+	free_fixed_rsrc_data(ctx->buf_data);
+	ctx->buf_data = NULL;
 	return ERR_PTR(ret);
 }
 
-- 
1.8.3.1

[PATCH v5 11/13] io_uring: make percpu_ref_release names consistent

[Bijan Mottahedeh]

Make the percpu ref release function names consistent between rsrc data
and nodes.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 96f760e..5ab0ff1 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -7297,7 +7297,7 @@ static void __io_sqe_files_unregister(struct io_ring_ctx *ctx)
 #endif
 }
 
-static void io_rsrc_ref_kill(struct percpu_ref *ref)
+static void io_rsrc_data_ref_zero(struct percpu_ref *ref)
 {
 	struct fixed_rsrc_data *data;
 
@@ -7376,7 +7376,7 @@ static struct fixed_rsrc_data *alloc_fixed_rsrc_data(struct io_ring_ctx *ctx)
 	data->ctx = ctx;
 	init_completion(&data->done);
 
-	if (percpu_ref_init(&data->refs, io_rsrc_ref_kill,
+	if (percpu_ref_init(&data->refs, io_rsrc_data_ref_zero,
 			    PERCPU_REF_ALLOW_REINIT, GFP_KERNEL)) {
 		kfree(data);
 		return ERR_PTR(-ENOMEM);
@@ -7759,7 +7759,7 @@ static void io_rsrc_put_work(struct work_struct *work)
 	}
 }
 
-static void io_rsrc_data_ref_zero(struct percpu_ref *ref)
+static void io_rsrc_node_ref_zero(struct percpu_ref *ref)
 {
 	struct fixed_rsrc_ref_node *ref_node;
 	struct fixed_rsrc_data *data;
@@ -7803,7 +7803,7 @@ static struct fixed_rsrc_ref_node *alloc_fixed_rsrc_ref_node(
 	if (!ref_node)
 		return ERR_PTR(-ENOMEM);
 
-	if (percpu_ref_init(&ref_node->refs, io_rsrc_data_ref_zero,
+	if (percpu_ref_init(&ref_node->refs, io_rsrc_node_ref_zero,
 			    0, GFP_KERNEL)) {
 		kfree(ref_node);
 		return ERR_PTR(-ENOMEM);
-- 
1.8.3.1

[PATCH v5 12/13] io_uring: call io_get_fixed_rsrc_ref for buffers

[Bijan Mottahedeh]

io_get_fixed_rsrc_ref() must be called for both buffers and files.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
---
 fs/io_uring.c | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 5ab0ff1..37639b9 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -1079,12 +1079,11 @@ static inline void io_clean_op(struct io_kiocb *req)
 		__io_clean_op(req);
 }
 
-static inline void io_set_resource_node(struct io_kiocb *req)
+static inline void io_get_fixed_rsrc_ref(struct io_kiocb *req,
+					 struct fixed_rsrc_data *rsrc_data)
 {
-	struct io_ring_ctx *ctx = req->ctx;
-
 	if (!req->fixed_rsrc_refs) {
-		req->fixed_rsrc_refs = &ctx->file_data->node->refs;
+		req->fixed_rsrc_refs = &rsrc_data->node->refs;
 		percpu_ref_get(req->fixed_rsrc_refs);
 	}
 }
@@ -2921,6 +2920,9 @@ static int io_prep_rw(struct io_kiocb *req, const struct io_uring_sqe *sqe)
 	req->rw.addr = READ_ONCE(sqe->addr);
 	req->rw.len = READ_ONCE(sqe->len);
 	req->buf_index = READ_ONCE(sqe->buf_index);
+	if (req->opcode == IORING_OP_READ_FIXED ||
+	    req->opcode == IORING_OP_WRITE_FIXED)
+		io_get_fixed_rsrc_ref(req, ctx->buf_data);
 	return 0;
 }
 
@@ -6453,7 +6455,7 @@ static struct file *io_file_get(struct io_submit_state *state,
 			return NULL;
 		fd = array_index_nospec(fd, ctx->nr_user_files);
 		file = io_file_from_index(ctx, fd);
-		io_set_resource_node(req);
+		io_get_fixed_rsrc_ref(req, ctx->file_data);
 	} else {
 		trace_io_uring_file_get(ctx, fd);
 		file = __io_file_get(state, fd);
-- 
1.8.3.1

[PATCH v5 13/13] io_uring: support buffer registration sharing

[Bijan Mottahedeh]

Implement buffer sharing among multiple rings.

A ring shares its (future) buffer registrations at setup time with
IORING_SETUP_SHARE_BUF. A ring attaches to another ring's buffer
registration at setup time with IORING_SETUP_ATTACH_BUF, after
authenticating with the buffer registration owner's fd. Any updates to
the owner's buffer registrations become immediately available to the
attached rings.

Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>

Conflicts:
	fs/io_uring.c
---
 fs/io_uring.c                 | 85 +++++++++++++++++++++++++++++++++++++++++--
 include/uapi/linux/io_uring.h |  2 +
 2 files changed, 83 insertions(+), 4 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 37639b9..856a570b 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -8439,6 +8439,13 @@ static void io_buffers_map_free(struct io_ring_ctx *ctx)
 	ctx->nr_user_bufs = 0;
 }
 
+static void io_detach_buf_data(struct io_ring_ctx *ctx)
+{
+	percpu_ref_put(&ctx->buf_data->refs);
+	ctx->buf_data = NULL;
+	ctx->nr_user_bufs = 0;
+}
+
 static int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
 {
 	struct fixed_rsrc_data *data = ctx->buf_data;
@@ -8447,6 +8454,11 @@ static int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
 	if (!data)
 		return -ENXIO;
 
+	if (ctx->flags & IORING_SETUP_ATTACH_BUF) {
+		io_detach_buf_data(ctx);
+		return 0;
+	}
+
 	ret = io_rsrc_ref_quiesce(data, ctx, alloc_fixed_buf_ref_node);
 	if (ret)
 		return ret;
@@ -8690,9 +8702,13 @@ static struct fixed_rsrc_data *io_buffers_map_alloc(struct io_ring_ctx *ctx,
 	if (!nr_args || nr_args > IORING_MAX_FIXED_BUFS)
 		return ERR_PTR(-EINVAL);
 
-	buf_data = alloc_fixed_rsrc_data(ctx);
-	if (IS_ERR(buf_data))
-		return buf_data;
+	if (ctx->buf_data) {
+		buf_data = ctx->buf_data;
+	} else {
+		buf_data = alloc_fixed_rsrc_data(ctx);
+		if (IS_ERR(buf_data))
+			return buf_data;
+	}
 
 	nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_BUFS_TABLE);
 	buf_data->table = kcalloc(nr_tables, sizeof(*buf_data->table),
@@ -8757,9 +8773,17 @@ static int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
 	if (ctx->nr_user_bufs)
 		return -EBUSY;
 
+	if (ctx->flags & IORING_SETUP_ATTACH_BUF) {
+		if (!ctx->buf_data)
+			return -EFAULT;
+		ctx->nr_user_bufs = ctx->buf_data->ctx->nr_user_bufs;
+		return 0;
+	}
+
 	buf_data = io_buffers_map_alloc(ctx, nr_args);
 	if (IS_ERR(buf_data))
 		return PTR_ERR(buf_data);
+	ctx->buf_data = buf_data;
 
 	for (i = 0; i < nr_args; i++, ctx->nr_user_bufs++) {
 		struct io_mapped_ubuf *imu;
@@ -8783,7 +8807,6 @@ static int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
 			break;
 	}
 
-	ctx->buf_data = buf_data;
 	if (ret) {
 		io_sqe_buffers_unregister(ctx);
 		return ret;
@@ -9831,6 +9854,55 @@ static struct file *io_uring_get_file(struct io_ring_ctx *ctx)
 	return file;
 }
 
+static int io_attach_buf_data(struct io_ring_ctx *ctx,
+			      struct io_uring_params *p)
+{
+	struct io_ring_ctx *ctx_attach;
+	struct fd f;
+
+	f = fdget(p->wq_fd);
+	if (!f.file)
+		return -EBADF;
+	if (f.file->f_op != &io_uring_fops) {
+		fdput(f);
+		return -EINVAL;
+	}
+
+	ctx_attach = f.file->private_data;
+	if (!ctx_attach->buf_data) {
+		fdput(f);
+		return -EINVAL;
+	}
+	ctx->buf_data = ctx_attach->buf_data;
+
+	percpu_ref_get(&ctx->buf_data->refs);
+	fdput(f);
+	return 0;
+}
+
+static int io_init_buf_data(struct io_ring_ctx *ctx, struct io_uring_params *p)
+{
+	if ((p->flags & (IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF)) ==
+	    (IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF))
+		return -EINVAL;
+
+	if (p->flags & IORING_SETUP_SHARE_BUF) {
+		struct fixed_rsrc_data *buf_data;
+
+		buf_data = alloc_fixed_rsrc_data(ctx);
+		if (IS_ERR(buf_data))
+			return PTR_ERR(buf_data);
+
+		ctx->buf_data = buf_data;
+		return 0;
+	}
+
+	if (p->flags & IORING_SETUP_ATTACH_BUF)
+		return io_attach_buf_data(ctx, p);
+
+	return 0;
+}
+
 static int io_uring_create(unsigned entries, struct io_uring_params *p,
 			   struct io_uring_params __user *params)
 {
@@ -9948,6 +10020,10 @@ static int io_uring_create(unsigned entries, struct io_uring_params *p,
 	if (ret)
 		goto err;
 
+	ret = io_init_buf_data(ctx, p);
+	if (ret)
+		goto err;
+
 	ret = io_sq_offload_create(ctx, p);
 	if (ret)
 		goto err;
@@ -10028,6 +10104,7 @@ static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
 	if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
 			IORING_SETUP_SQ_AFF | IORING_SETUP_CQSIZE |
 			IORING_SETUP_CLAMP | IORING_SETUP_ATTACH_WQ |
+			IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF |
 			IORING_SETUP_R_DISABLED))
 		return -EINVAL;
 
diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
index b289ef8..3ad786a 100644
--- a/include/uapi/linux/io_uring.h
+++ b/include/uapi/linux/io_uring.h
@@ -98,6 +98,8 @@ enum {
 #define IORING_SETUP_CLAMP	(1U << 4)	/* clamp SQ/CQ ring sizes */
 #define IORING_SETUP_ATTACH_WQ	(1U << 5)	/* attach to existing wq */
 #define IORING_SETUP_R_DISABLED	(1U << 6)	/* start with ring disabled */
+#define IORING_SETUP_SHARE_BUF	(1U << 7)	/* share buffer registration */
+#define IORING_SETUP_ATTACH_BUF	(1U << 8)	/* attach buffer registration */
 
 enum {
 	IORING_OP_NOP,
-- 
1.8.3.1

Re: [PATCH v5 13/13] io_uring: support buffer registration sharing

[Bijan Mottahedeh]

On 1/12/2021 1:33 PM, Bijan Mottahedeh wrote:
> Implement buffer sharing among multiple rings.
> 
> A ring shares its (future) buffer registrations at setup time with
> IORING_SETUP_SHARE_BUF. A ring attaches to another ring's buffer
> registration at setup time with IORING_SETUP_ATTACH_BUF, after
> authenticating with the buffer registration owner's fd. Any updates to
> the owner's buffer registrations become immediately available to the
> attached rings.
> 
> Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
> 
> Conflicts:
> 	fs/io_uring.c
> ---
>   fs/io_uring.c                 | 85 +++++++++++++++++++++++++++++++++++++++++--
>   include/uapi/linux/io_uring.h |  2 +
>   2 files changed, 83 insertions(+), 4 deletions(-)
> 
> diff --git a/fs/io_uring.c b/fs/io_uring.c
> index 37639b9..856a570b 100644
> --- a/fs/io_uring.c
> +++ b/fs/io_uring.c
> @@ -8439,6 +8439,13 @@ static void io_buffers_map_free(struct io_ring_ctx *ctx)
>   	ctx->nr_user_bufs = 0;
>   }
>   
> +static void io_detach_buf_data(struct io_ring_ctx *ctx)
> +{
> +	percpu_ref_put(&ctx->buf_data->refs);
> +	ctx->buf_data = NULL;
> +	ctx->nr_user_bufs = 0;
> +}
> +
>   static int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
>   {
>   	struct fixed_rsrc_data *data = ctx->buf_data;
> @@ -8447,6 +8454,11 @@ static int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
>   	if (!data)
>   		return -ENXIO;
>   
> +	if (ctx->flags & IORING_SETUP_ATTACH_BUF) {
> +		io_detach_buf_data(ctx);
> +		return 0;
> +	}
> +
>   	ret = io_rsrc_ref_quiesce(data, ctx, alloc_fixed_buf_ref_node);
>   	if (ret)
>   		return ret;
> @@ -8690,9 +8702,13 @@ static struct fixed_rsrc_data *io_buffers_map_alloc(struct io_ring_ctx *ctx,
>   	if (!nr_args || nr_args > IORING_MAX_FIXED_BUFS)
>   		return ERR_PTR(-EINVAL);
>   
> -	buf_data = alloc_fixed_rsrc_data(ctx);
> -	if (IS_ERR(buf_data))
> -		return buf_data;
> +	if (ctx->buf_data) {
> +		buf_data = ctx->buf_data;
> +	} else {
> +		buf_data = alloc_fixed_rsrc_data(ctx);
> +		if (IS_ERR(buf_data))
> +			return buf_data;
> +	}
>   
>   	nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_BUFS_TABLE);
>   	buf_data->table = kcalloc(nr_tables, sizeof(*buf_data->table),
> @@ -8757,9 +8773,17 @@ static int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
>   	if (ctx->nr_user_bufs)
>   		return -EBUSY;
>   
> +	if (ctx->flags & IORING_SETUP_ATTACH_BUF) {
> +		if (!ctx->buf_data)
> +			return -EFAULT;
> +		ctx->nr_user_bufs = ctx->buf_data->ctx->nr_user_bufs;
> +		return 0;
> +	}
> +
>   	buf_data = io_buffers_map_alloc(ctx, nr_args);
>   	if (IS_ERR(buf_data))
>   		return PTR_ERR(buf_data);
> +	ctx->buf_data = buf_data;
>   
>   	for (i = 0; i < nr_args; i++, ctx->nr_user_bufs++) {
>   		struct io_mapped_ubuf *imu;
> @@ -8783,7 +8807,6 @@ static int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
>   			break;
>   	}
>   
> -	ctx->buf_data = buf_data;
>   	if (ret) {
>   		io_sqe_buffers_unregister(ctx);
>   		return ret;
> @@ -9831,6 +9854,55 @@ static struct file *io_uring_get_file(struct io_ring_ctx *ctx)
>   	return file;
>   }
>   
> +static int io_attach_buf_data(struct io_ring_ctx *ctx,
> +			      struct io_uring_params *p)
> +{
> +	struct io_ring_ctx *ctx_attach;
> +	struct fd f;
> +
> +	f = fdget(p->wq_fd);
> +	if (!f.file)
> +		return -EBADF;
> +	if (f.file->f_op != &io_uring_fops) {
> +		fdput(f);
> +		return -EINVAL;
> +	}
> +
> +	ctx_attach = f.file->private_data;
> +	if (!ctx_attach->buf_data) {
> +		fdput(f);
> +		return -EINVAL;
> +	}
> +	ctx->buf_data = ctx_attach->buf_data;
> +
> +	percpu_ref_get(&ctx->buf_data->refs);
> +	fdput(f);
> +	return 0;
> +}
> +
> +static int io_init_buf_data(struct io_ring_ctx *ctx, struct io_uring_params *p)
> +{
> +	if ((p->flags & (IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF)) ==
> +	    (IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF))
> +		return -EINVAL;
> +
> +	if (p->flags & IORING_SETUP_SHARE_BUF) {
> +		struct fixed_rsrc_data *buf_data;
> +
> +		buf_data = alloc_fixed_rsrc_data(ctx);
> +		if (IS_ERR(buf_data))
> +			return PTR_ERR(buf_data);
> +
> +		ctx->buf_data = buf_data;
> +		return 0;
> +	}
> +
> +	if (p->flags & IORING_SETUP_ATTACH_BUF)
> +		return io_attach_buf_data(ctx, p);
> +
> +	return 0;
> +}
> +
>   static int io_uring_create(unsigned entries, struct io_uring_params *p,
>   			   struct io_uring_params __user *params)
>   {
> @@ -9948,6 +10020,10 @@ static int io_uring_create(unsigned entries, struct io_uring_params *p,
>   	if (ret)
>   		goto err;
>   
> +	ret = io_init_buf_data(ctx, p);
> +	if (ret)
> +		goto err;
> +
>   	ret = io_sq_offload_create(ctx, p);
>   	if (ret)
>   		goto err;
> @@ -10028,6 +10104,7 @@ static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
>   	if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
>   			IORING_SETUP_SQ_AFF | IORING_SETUP_CQSIZE |
>   			IORING_SETUP_CLAMP | IORING_SETUP_ATTACH_WQ |
> +			IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF |
>   			IORING_SETUP_R_DISABLED))
>   		return -EINVAL;
>   
> diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
> index b289ef8..3ad786a 100644
> --- a/include/uapi/linux/io_uring.h
> +++ b/include/uapi/linux/io_uring.h
> @@ -98,6 +98,8 @@ enum {
>   #define IORING_SETUP_CLAMP	(1U << 4)	/* clamp SQ/CQ ring sizes */
>   #define IORING_SETUP_ATTACH_WQ	(1U << 5)	/* attach to existing wq */
>   #define IORING_SETUP_R_DISABLED	(1U << 6)	/* start with ring disabled */
> +#define IORING_SETUP_SHARE_BUF	(1U << 7)	/* share buffer registration */
> +#define IORING_SETUP_ATTACH_BUF	(1U << 8)	/* attach buffer registration */
>   
>   enum {
>   	IORING_OP_NOP,
> 

I recreated the deadlock scenario you had raised:

 > Ok, now the original io_uring instance will wait until the attached
 > once get rid of their references. That's a versatile ground to have
 > in kernel deadlocks.
 >
 > task1: uring1 = create()
 > task2: uring2 = create()
 > task1: uring3 = create(share=uring2);
 > task2: uring4 = create(share=uring1);
 >
 > task1: io_sqe_buffers_unregister(uring1)
 > task2: io_sqe_buffers_unregister(uring2)
 >
 > If I skimmed through the code right, that should hang unkillably.

with the following test:

+static int test_deadlock(void)
+{
+       int i, pid, ret;
+       struct io_uring rings[4];
+       struct io_uring_params p = {};
+
+       p.flags = IORING_SETUP_SHARE_BUF;
+
+       for (i = 0; i < 2; i++) {
+               ret = io_uring_queue_init_params(1, &rings[i], &p);
+               if (ret) {
+                       verror("queue_init share");
+                       return ret;
+               }
+       }
+
+       p.flags = IORING_SETUP_ATTACH_BUF;
+
+       pid = fork();
+       if (pid) {
+               p.wq_fd = rings[1].ring_fd;
+               ret = io_uring_queue_init_params(1, &rings[3], &p);
+       } else {
+               p.wq_fd = rings[0].ring_fd;
+               ret = io_uring_queue_init_params(1, &rings[4], &p);
+       }
+
+       if (ret) {
+               verror("queue_init attach");
+               return ret;
+       }
+
+
+       vinfo(V1, "unregister\n");
+
+       if (pid) {
+               close(rings[1].ring_fd);
+               ret = io_uring_unregister_buffers(&rings[0]);
+       } else {
+               close(rings[0].ring_fd);
+               ret = io_uring_unregister_buffers(&rings[1]);
+       }
+
+       vinfo(V1, "unregister done\n");
+
+       if (ret)
+               verror("unregister");
+
+       return ret;
+}


The two processe in the test hang but can be interrupted.

I checked that

ret = wait_for_completion_interruptible(&data->done);

in io_rsrc_ref_quiesce() returns -ERESTARTSYS (-512) after hitting ^C 
and that

ret = io_run_task_work_sig();

returns -EINTR (-4)

Finally in

io_ring_ctx_free()
-> io_sqe_buffers_unregister()
    -> io_rsrc_ref_quiesce()

ret = wait_for_completion_interruptible(&data->done);

returns 0.

So it looks like the unkillable hang is not there.

However, when I take out the io_uring_unregister_buffers() calls from 
the test, one of the processes gets a segfault with the following trace 
and I'm not sure what the cause is.

buffer-share[2791]: segfault at 7f2ca196b040 ip ]
Code: 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00

#0  io_uring_release (inode=0xffff88810665a5a8, file=0xffff88811b5eb540)
     at fs/io_uring.c:9116
#1  0xffffffff812eb208 in __fput (file=0xffff88811b5eb540)
     at fs/file_table.c:280
#2  0xffffffff810aa857 in task_work_run () at kernel/task_work.c:140
#3  0xffffffff8108b37d in exit_task_work (task=0xffff8881321b8040)
     at ./include/linux/task_work.h:30
#4  do_exit (code=code@entry=139) at kernel/exit.c:825
#5  0xffffffff8108bbe7 in do_group_exit (exit_code=139) at kernel/exit.c:922
#6  0xffffffff81099697 in get_signal (ksig=ksig@entry=0xffffc900023d3ea8)
     at kernel/signal.c:2770
#7  0xffffffff81031819 in arch_do_signal_or_restart 
(regs=0xffffc900023d3f58,
     has_signal=<optimized out>) at ./arch/x86/include/asm/current.h:15
#8  0xffffffff8111dd72 in handle_signal_work (ti_work=<optimized out>,
     regs=0xffffc900023d3f58) at kernel/entry/common.c:147
#9  exit_to_user_mode_loop (ti_work=<optimized out>, regs=<optimized out>)
     at kernel/entry/common.c:171
#10 exit_to_user_mode_prepare (regs=0xffffc900023d3f58)
     at kernel/entry/common.c:201
#11 0xffffffff819c5645 in irqentry_exit_to_user_mode (regs=<optimized out>)
     at kernel/entry/common.c:315
#12 0xffffffff81a00ade in asm_exc_page_fault ()
     at ./arch/x86/include/asm/idtentry.h:580
#13 0x0000000000000000 in ?? ()

Re: [PATCH v5 13/13] io_uring: support buffer registration sharing

[Bijan Mottahedeh]

On 1/13/2021 6:01 PM, Bijan Mottahedeh wrote:
> On 1/12/2021 1:33 PM, Bijan Mottahedeh wrote:
>> Implement buffer sharing among multiple rings.
>>
>> A ring shares its (future) buffer registrations at setup time with
>> IORING_SETUP_SHARE_BUF. A ring attaches to another ring's buffer
>> registration at setup time with IORING_SETUP_ATTACH_BUF, after
>> authenticating with the buffer registration owner's fd. Any updates to
>> the owner's buffer registrations become immediately available to the
>> attached rings.
>>
>> Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
>>
>> Conflicts:
>>     fs/io_uring.c
>> ---
>>   fs/io_uring.c                 | 85 
>> +++++++++++++++++++++++++++++++++++++++++--
>>   include/uapi/linux/io_uring.h |  2 +
>>   2 files changed, 83 insertions(+), 4 deletions(-)
>>
>> diff --git a/fs/io_uring.c b/fs/io_uring.c
>> index 37639b9..856a570b 100644
>> --- a/fs/io_uring.c
>> +++ b/fs/io_uring.c
>> @@ -8439,6 +8439,13 @@ static void io_buffers_map_free(struct 
>> io_ring_ctx *ctx)
>>       ctx->nr_user_bufs = 0;
>>   }
>> +static void io_detach_buf_data(struct io_ring_ctx *ctx)
>> +{
>> +    percpu_ref_put(&ctx->buf_data->refs);
>> +    ctx->buf_data = NULL;
>> +    ctx->nr_user_bufs = 0;
>> +}
>> +
>>   static int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
>>   {
>>       struct fixed_rsrc_data *data = ctx->buf_data;
>> @@ -8447,6 +8454,11 @@ static int io_sqe_buffers_unregister(struct 
>> io_ring_ctx *ctx)
>>       if (!data)
>>           return -ENXIO;
>> +    if (ctx->flags & IORING_SETUP_ATTACH_BUF) {
>> +        io_detach_buf_data(ctx);
>> +        return 0;
>> +    }
>> +
>>       ret = io_rsrc_ref_quiesce(data, ctx, alloc_fixed_buf_ref_node);
>>       if (ret)
>>           return ret;
>> @@ -8690,9 +8702,13 @@ static struct fixed_rsrc_data 
>> *io_buffers_map_alloc(struct io_ring_ctx *ctx,
>>       if (!nr_args || nr_args > IORING_MAX_FIXED_BUFS)
>>           return ERR_PTR(-EINVAL);
>> -    buf_data = alloc_fixed_rsrc_data(ctx);
>> -    if (IS_ERR(buf_data))
>> -        return buf_data;
>> +    if (ctx->buf_data) {
>> +        buf_data = ctx->buf_data;
>> +    } else {
>> +        buf_data = alloc_fixed_rsrc_data(ctx);
>> +        if (IS_ERR(buf_data))
>> +            return buf_data;
>> +    }
>>       nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_BUFS_TABLE);
>>       buf_data->table = kcalloc(nr_tables, sizeof(*buf_data->table),
>> @@ -8757,9 +8773,17 @@ static int io_sqe_buffers_register(struct 
>> io_ring_ctx *ctx, void __user *arg,
>>       if (ctx->nr_user_bufs)
>>           return -EBUSY;
>> +    if (ctx->flags & IORING_SETUP_ATTACH_BUF) {
>> +        if (!ctx->buf_data)
>> +            return -EFAULT;
>> +        ctx->nr_user_bufs = ctx->buf_data->ctx->nr_user_bufs;
>> +        return 0;
>> +    }
>> +
>>       buf_data = io_buffers_map_alloc(ctx, nr_args);
>>       if (IS_ERR(buf_data))
>>           return PTR_ERR(buf_data);
>> +    ctx->buf_data = buf_data;
>>       for (i = 0; i < nr_args; i++, ctx->nr_user_bufs++) {
>>           struct io_mapped_ubuf *imu;
>> @@ -8783,7 +8807,6 @@ static int io_sqe_buffers_register(struct 
>> io_ring_ctx *ctx, void __user *arg,
>>               break;
>>       }
>> -    ctx->buf_data = buf_data;
>>       if (ret) {
>>           io_sqe_buffers_unregister(ctx);
>>           return ret;
>> @@ -9831,6 +9854,55 @@ static struct file *io_uring_get_file(struct 
>> io_ring_ctx *ctx)
>>       return file;
>>   }
>> +static int io_attach_buf_data(struct io_ring_ctx *ctx,
>> +                  struct io_uring_params *p)
>> +{
>> +    struct io_ring_ctx *ctx_attach;
>> +    struct fd f;
>> +
>> +    f = fdget(p->wq_fd);
>> +    if (!f.file)
>> +        return -EBADF;
>> +    if (f.file->f_op != &io_uring_fops) {
>> +        fdput(f);
>> +        return -EINVAL;
>> +    }
>> +
>> +    ctx_attach = f.file->private_data;
>> +    if (!ctx_attach->buf_data) {
>> +        fdput(f);
>> +        return -EINVAL;
>> +    }
>> +    ctx->buf_data = ctx_attach->buf_data;
>> +
>> +    percpu_ref_get(&ctx->buf_data->refs);
>> +    fdput(f);
>> +    return 0;
>> +}
>> +
>> +static int io_init_buf_data(struct io_ring_ctx *ctx, struct 
>> io_uring_params *p)
>> +{
>> +    if ((p->flags & (IORING_SETUP_SHARE_BUF | 
>> IORING_SETUP_ATTACH_BUF)) ==
>> +        (IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF))
>> +        return -EINVAL;
>> +
>> +    if (p->flags & IORING_SETUP_SHARE_BUF) {
>> +        struct fixed_rsrc_data *buf_data;
>> +
>> +        buf_data = alloc_fixed_rsrc_data(ctx);
>> +        if (IS_ERR(buf_data))
>> +            return PTR_ERR(buf_data);
>> +
>> +        ctx->buf_data = buf_data;
>> +        return 0;
>> +    }
>> +
>> +    if (p->flags & IORING_SETUP_ATTACH_BUF)
>> +        return io_attach_buf_data(ctx, p);
>> +
>> +    return 0;
>> +}
>> +
>>   static int io_uring_create(unsigned entries, struct io_uring_params *p,
>>                  struct io_uring_params __user *params)
>>   {
>> @@ -9948,6 +10020,10 @@ static int io_uring_create(unsigned entries, 
>> struct io_uring_params *p,
>>       if (ret)
>>           goto err;
>> +    ret = io_init_buf_data(ctx, p);
>> +    if (ret)
>> +        goto err;
>> +
>>       ret = io_sq_offload_create(ctx, p);
>>       if (ret)
>>           goto err;
>> @@ -10028,6 +10104,7 @@ static long io_uring_setup(u32 entries, struct 
>> io_uring_params __user *params)
>>       if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
>>               IORING_SETUP_SQ_AFF | IORING_SETUP_CQSIZE |
>>               IORING_SETUP_CLAMP | IORING_SETUP_ATTACH_WQ |
>> +            IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF |
>>               IORING_SETUP_R_DISABLED))
>>           return -EINVAL;
>> diff --git a/include/uapi/linux/io_uring.h 
>> b/include/uapi/linux/io_uring.h
>> index b289ef8..3ad786a 100644
>> --- a/include/uapi/linux/io_uring.h
>> +++ b/include/uapi/linux/io_uring.h
>> @@ -98,6 +98,8 @@ enum {
>>   #define IORING_SETUP_CLAMP    (1U << 4)    /* clamp SQ/CQ ring sizes */
>>   #define IORING_SETUP_ATTACH_WQ    (1U << 5)    /* attach to existing 
>> wq */
>>   #define IORING_SETUP_R_DISABLED    (1U << 6)    /* start with ring 
>> disabled */
>> +#define IORING_SETUP_SHARE_BUF    (1U << 7)    /* share buffer 
>> registration */
>> +#define IORING_SETUP_ATTACH_BUF    (1U << 8)    /* attach buffer 
>> registration */
>>   enum {
>>       IORING_OP_NOP,
>>
> 
> I recreated the deadlock scenario you had raised:
> 
>  > Ok, now the original io_uring instance will wait until the attached
>  > once get rid of their references. That's a versatile ground to have
>  > in kernel deadlocks.
>  >
>  > task1: uring1 = create()
>  > task2: uring2 = create()
>  > task1: uring3 = create(share=uring2);
>  > task2: uring4 = create(share=uring1);
>  >
>  > task1: io_sqe_buffers_unregister(uring1)
>  > task2: io_sqe_buffers_unregister(uring2)
>  >
>  > If I skimmed through the code right, that should hang unkillably.
> 
> with the following test:
> 
> +static int test_deadlock(void)
> +{
> +       int i, pid, ret;
> +       struct io_uring rings[4];
> +       struct io_uring_params p = {};
> +
> +       p.flags = IORING_SETUP_SHARE_BUF;
> +
> +       for (i = 0; i < 2; i++) {
> +               ret = io_uring_queue_init_params(1, &rings[i], &p);
> +               if (ret) {
> +                       verror("queue_init share");
> +                       return ret;
> +               }
> +       }
> +
> +       p.flags = IORING_SETUP_ATTACH_BUF;
> +
> +       pid = fork();
> +       if (pid) {
> +               p.wq_fd = rings[1].ring_fd;
> +               ret = io_uring_queue_init_params(1, &rings[3], &p);
                                                             ^^^
                                                             2
> +       } else {
> +               p.wq_fd = rings[0].ring_fd;
> +               ret = io_uring_queue_init_params(1, &rings[4], &p);
                                                             ^^^
                                                             3
> +       }
> +
> +       if (ret) {
> +               verror("queue_init attach");
> +               return ret;
> +       }
> +
> +
> +       vinfo(V1, "unregister\n");
> +
> +       if (pid) {
> +               close(rings[1].ring_fd);
> +               ret = io_uring_unregister_buffers(&rings[0]);
> +       } else {
> +               close(rings[0].ring_fd);
> +               ret = io_uring_unregister_buffers(&rings[1]);
> +       }
> +
> +       vinfo(V1, "unregister done\n");
> +
> +       if (ret)
> +               verror("unregister");
> +
> +       return ret;
> +}
> 
> 
> The two processe in the test hang but can be interrupted.
> 
> I checked that
> 
> ret = wait_for_completion_interruptible(&data->done);
> 
> in io_rsrc_ref_quiesce() returns -ERESTARTSYS (-512) after hitting ^C 
> and that
> 
> ret = io_run_task_work_sig();
> 
> returns -EINTR (-4)
> 
> Finally in
> 
> io_ring_ctx_free()
> -> io_sqe_buffers_unregister()
>     -> io_rsrc_ref_quiesce()
> 
> ret = wait_for_completion_interruptible(&data->done);
> 
> returns 0.
> 
> So it looks like the unkillable hang is not there.
> 
> However, when I take out the io_uring_unregister_buffers() calls from 
> the test, one of the processes gets a segfault with the following trace 
> and I'm not sure what the cause is.

Bug in test  itself.  Wrong index passed to io_uring_queue_init_params() 
above.

Re: [PATCH v5 00/13] io_uring: buffer registration enhancements

[Pavel Begunkov]

On 12/01/2021 21:33, Bijan Mottahedeh wrote:
> v5:
> 
> - call io_get_fixed_rsrc_ref for buffers
> - make percpu_ref_release names consistent
> - rebase on for-5.12/io_uring

To reduce the burden I'll take the generalisation patches from that,
review and resend to Jens with small changes leaving your "from:".
I hope you don't mind, that should be faster.

I'll remove your signed-off and will need it back by you replying
on this coming resend.

> 
> v4:
> 
> - address v3 comments (TBD REGISTER_BUFFERS)
> - rebase
> 
> v3:
> 
> - batch file->rsrc renames into a signle patch when possible
> - fix other review changes from v2
> - fix checkpatch warnings
> 
> v2:
> 
> - drop readv/writev with fixed buffers patch
> - handle ref_nodes both both files/buffers with a single ref_list
> - make file/buffer handling more unified
> 
> This patchset implements a set of enhancements to buffer registration
> consistent with existing file registration functionality:
> 
> - buffer registration updates		IORING_REGISTER_BUFFERS_UPDATE
> 					IORING_OP_BUFFERS_UPDATE
> 
> - buffer registration sharing		IORING_SETUP_SHARE_BUF
> 					IORING_SETUP_ATTACH_BUF
> 
> Patch 1-5 generalize fixed_file functionality to fixed_rsrc.
> 
> Patch 6 applies fixed_rsrc functionality for fixed buffers support.
> 
> Patch 7-8 generalize files_update functionality to rsrc_update.
> 
> Patch 9 implements buffer registration update, and introduces
> IORING_REGISTER_BUFFERS_UPDATE and IORING_OP_BUFFERS_UPDATE, consistent
> with file registration update.
> 
> Patch 10 generalizes fixed resource allocation 
> 
> Patch 11 renames percpu release routines for consistency
> 
> Patch 12 calls io_get_fixed_rsrc_ref() for buffers as well as files
> 
> Patch 13 implements buffer sharing among multiple rings; it works as follows:
> 
> - A new ring, A,  is setup. Since no buffers have been registered, the
>   registered buffer state is an empty set, Z. That's different from the
>   NULL state in current implementation.
> 
> - Ring B is setup, attaching to Ring A. It's also attaching to it's
>   buffer registrations, now we have two references to the same empty
>   set, Z.
> 
> - Ring A registers buffers into set Z, which is no longer empty.
> 
> - Ring B sees this immediately, since it's already sharing that set.
> 
> Testing
> 
> I have used liburing file-{register,update} tests as models for
> buffer-{register,update,share}, tests and they run ok.
> 
> TBD
> 
> - Need a patch from Pavel to address a race between fixed IO from async
> context and buffer unregister, or force buffer registration ops to do
> full quiesce.
> 
> - Need to still address Pavel's comments about unkillable deadlocks. It
> seems that we should no long hange unkillably in io_rsrc_ref_quiesce()
> because of Pavel's changes.
> 
> - I tried to use a single opcode for files/buffers but ran into an
> issue since work_flags is different for files/buffers.  This should
> be ok for the most part since req->work.flags is ultimately examined;
> however, there are place where io_op_defs[opcode].work_flags is examined
> directly, and I wasn't sure what would the best way to handle that.
> 
> Bijan Mottahedeh (13):
>   io_uring: rename file related variables to rsrc
>   io_uring: generalize io_queue_rsrc_removal
>   io_uring: separate ref_list from fixed_rsrc_data
>   io_uring: split alloc_fixed_file_ref_node
>   io_uring: add rsrc_ref locking routines
>   io_uring: implement fixed buffers registration similar to fixed files
>   io_uring: create common fixed_rsrc_ref_node handling routines
>   io_uring: generalize files_update functionlity to rsrc_update
>   io_uring: support buffer registration updates
>   io_uring: create common fixed_rsrc_data allocation routines
>   io_uring: make percpu_ref_release names consistent
>   io_uring: call io_get_fixed_rsrc_ref for buffers
>   io_uring: support buffer registration sharing
> 
>  fs/io_uring.c                 | 803 ++++++++++++++++++++++++++++++++----------
>  include/uapi/linux/io_uring.h |  13 +-
>  2 files changed, 626 insertions(+), 190 deletions(-)
> 

-- 
Pavel Begunkov

Re: [PATCH v5 00/13] io_uring: buffer registration enhancements

[Bijan Mottahedeh]

On 1/14/2021 1:20 PM, Pavel Begunkov wrote:
> On 12/01/2021 21:33, Bijan Mottahedeh wrote:
>> v5:
>>
>> - call io_get_fixed_rsrc_ref for buffers
>> - make percpu_ref_release names consistent
>> - rebase on for-5.12/io_uring
> 
> To reduce the burden I'll take the generalisation patches from that,
> review and resend to Jens with small changes leaving your "from:".
> I hope you don't mind, that should be faster.
> 
> I'll remove your signed-off and will need it back by you replying
> on this coming resend.

Sure, thanks.

Re: [PATCH v5 00/13] io_uring: buffer registration enhancements

[Bijan Mottahedeh]

On 1/14/2021 2:44 PM, Bijan Mottahedeh wrote:
> On 1/14/2021 1:20 PM, Pavel Begunkov wrote:
>> On 12/01/2021 21:33, Bijan Mottahedeh wrote:
>>> v5:
>>>
>>> - call io_get_fixed_rsrc_ref for buffers
>>> - make percpu_ref_release names consistent
>>> - rebase on for-5.12/io_uring
>>
>> To reduce the burden I'll take the generalisation patches from that,
>> review and resend to Jens with small changes leaving your "from:".
>> I hope you don't mind, that should be faster.
>>
>> I'll remove your signed-off and will need it back by you replying
>> on this coming resend.
> 
> Sure, thanks.

Do you have any other concerns about the buffer sharing patch itself 
that I can address?

Re: [PATCH v5 00/13] io_uring: buffer registration enhancements

[Pavel Begunkov]

On 14/01/2021 22:54, Bijan Mottahedeh wrote:
> On 1/14/2021 2:44 PM, Bijan Mottahedeh wrote:
>> On 1/14/2021 1:20 PM, Pavel Begunkov wrote:
>>> On 12/01/2021 21:33, Bijan Mottahedeh wrote:
>>>> v5:
>>>>
>>>> - call io_get_fixed_rsrc_ref for buffers
>>>> - make percpu_ref_release names consistent
>>>> - rebase on for-5.12/io_uring
>>>
>>> To reduce the burden I'll take the generalisation patches from that,
>>> review and resend to Jens with small changes leaving your "from:".
>>> I hope you don't mind, that should be faster.
>>>
>>> I'll remove your signed-off and will need it back by you replying
>>> on this coming resend.
>>
>> Sure, thanks.
> 
> Do you have any other concerns about the buffer sharing patch itself that I can address?

Not yet, I'll go over it this week

-- 
Pavel Begunkov

Re: [PATCH v5 06/13] io_uring: implement fixed buffers registration similar to fixed files

[Pavel Begunkov]

On 12/01/2021 21:33, Bijan Mottahedeh wrote:
> Apply fixed_rsrc functionality for fixed buffers support.
> 
> Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
> ---
>  fs/io_uring.c | 234 +++++++++++++++++++++++++++++++++++++++++++++++++---------
>  1 file changed, 198 insertions(+), 36 deletions(-)
[...]
> +static struct fixed_rsrc_ref_node *alloc_fixed_buf_ref_node(
> +			struct io_ring_ctx *ctx)
> +{
> +	struct fixed_rsrc_ref_node *ref_node;
> +
> +	ref_node = alloc_fixed_rsrc_ref_node(ctx);

if (!ref_node)
    return NULL;

No need to resend yet, just we need to keep an eye on it.


> +	ref_node->rsrc_data = ctx->buf_data;
> +	ref_node->rsrc_put = io_ring_buf_put;
> +	return ref_node;
> +}


-- 
Pavel Begunkov

Re: [PATCH v5 12/13] io_uring: call io_get_fixed_rsrc_ref for buffers

[Pavel Begunkov]

On 12/01/2021 21:33, Bijan Mottahedeh wrote:
> io_get_fixed_rsrc_ref() must be called for both buffers and files.

This should go before you wire up ref_node'ed buffers stuff, or
merged into it. You know, the intention is for each patch to be
bug-free, it's cleaner and easier to debug/bisect.

[again, no need to resend yet]

-- 
Pavel Begunkov

Re: [PATCH v5 08/13] io_uring: generalize files_update functionlity to rsrc_update

[Pavel Begunkov]

On 12/01/2021 21:33, Bijan Mottahedeh wrote:
> Generalize files_update functionality to rsrc_update in order to
> leverage it for buffers updates.
> 
> Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
> ---
>  fs/io_uring.c                 | 19 ++++++++++++++-----
>  include/uapi/linux/io_uring.h |  6 +++++-
>  2 files changed, 19 insertions(+), 6 deletions(-)
> 
> diff --git a/fs/io_uring.c b/fs/io_uring.c
> index 6ebfe1f..f9f458c 100644
> --- a/fs/io_uring.c
> +++ b/fs/io_uring.c
> @@ -5954,7 +5954,7 @@ static int io_async_cancel(struct io_kiocb *req)
>  }
>  
>  static int io_rsrc_update_prep(struct io_kiocb *req,
> -				const struct io_uring_sqe *sqe)
> +			       const struct io_uring_sqe *sqe)
>  {
>  	if (unlikely(req->ctx->flags & IORING_SETUP_SQPOLL))
>  		return -EINVAL;
> @@ -5971,8 +5971,11 @@ static int io_rsrc_update_prep(struct io_kiocb *req,
>  	return 0;
>  }
>  
> -static int io_files_update(struct io_kiocb *req, bool force_nonblock,
> -			   struct io_comp_state *cs)
> +static int io_rsrc_update(struct io_kiocb *req, bool force_nonblock,
> +			  struct io_comp_state *cs,
> +			  int (*update)(struct io_ring_ctx *ctx,
> +					struct io_uring_rsrc_update *up,
> +					unsigned int nr_args))

I don't like excessive use of higher order functions in C. How about
replacing it with a switch/ifs? e.g.

... /* prepare up */
mutex_lock()
if (req->opcode == FILES_UPDATE)
	update_file(); // the one that was in callback @update arg
else (req->opcode == ...)
	...
mutex_unlock();

>  {
>  	struct io_ring_ctx *ctx = req->ctx;
>  	struct io_uring_rsrc_update up;
> @@ -5982,10 +5985,10 @@ static int io_files_update(struct io_kiocb *req, bool force_nonblock,
>  		return -EAGAIN;
>  
>  	up.offset = req->rsrc_update.offset;
> -	up.fds = req->rsrc_update.arg;
> +	up.rsrc = req->rsrc_update.arg;
>  
>  	mutex_lock(&ctx->uring_lock);
> -	ret = __io_sqe_files_update(ctx, &up, req->rsrc_update.nr_args);
> +	ret = (*update)(ctx, &up, req->rsrc_update.nr_args);
>  	mutex_unlock(&ctx->uring_lock);
>  
>  	if (ret < 0)
> @@ -5994,6 +5997,12 @@ static int io_files_update(struct io_kiocb *req, bool force_nonblock,
>  	return 0;
>  }
>  
> +static int io_files_update(struct io_kiocb *req, bool force_nonblock,
> +			   struct io_comp_state *cs)
> +{
> +	return io_rsrc_update(req, force_nonblock, cs, __io_sqe_files_update);
> +}
> +
>  static int io_req_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
>  {
>  	switch (req->opcode) {
> diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
> index 77de7c0..f51190b 100644
> --- a/include/uapi/linux/io_uring.h
> +++ b/include/uapi/linux/io_uring.h
> @@ -288,7 +288,11 @@ enum {
>  struct io_uring_rsrc_update {
>  	__u32 offset;
>  	__u32 resv;
> -	__aligned_u64 /* __s32 * */ fds;
> +	union {
> +		__aligned_u64 /* __s32 * */ fds;
> +		__aligned_u64 /* __s32 * */ iovs;
> +		__aligned_u64 /* __s32 * */ rsrc;
> +	};
>  };
>  
>  #define io_uring_files_update	io_uring_rsrc_update
> 

-- 
Pavel Begunkov

Re: [PATCH v5 10/13] io_uring: create common fixed_rsrc_data allocation routines

[Pavel Begunkov]

On 12/01/2021 21:33, Bijan Mottahedeh wrote:
> Create common alloc/free fixed_rsrc_data routines for both files and
> buffers.
> 
> Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
> ---
>  fs/io_uring.c | 75 ++++++++++++++++++++++++++++++++---------------------------
>  1 file changed, 41 insertions(+), 34 deletions(-)
> 
[...]
> @@ -8672,32 +8688,23 @@ static struct fixed_rsrc_data *io_buffers_map_alloc(struct io_ring_ctx *ctx,
>  	if (!nr_args || nr_args > IORING_MAX_FIXED_BUFS)
>  		return ERR_PTR(-EINVAL);
>  
> -	buf_data = kzalloc(sizeof(*ctx->buf_data), GFP_KERNEL);
> -	if (!buf_data)
> -		return ERR_PTR(-ENOMEM);
> -	buf_data->ctx = ctx;
> -	init_completion(&buf_data->done);
> +	buf_data = alloc_fixed_rsrc_data(ctx);
> +	if (IS_ERR(buf_data))
> +		return buf_data;

As you remember it's planned to be partially merged first, so when
you'll be rebasing, please use all these helpers from the beginning
(i.e. in 8/13 or so). Just a nit

>  
>  	nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_BUFS_TABLE);
>  	buf_data->table = kcalloc(nr_tables, sizeof(*buf_data->table),
>  				  GFP_KERNEL);
>  	if (!buf_data->table)
> -		goto out_free;
> -
> -	if (percpu_ref_init(&buf_data->refs, io_rsrc_ref_kill,
> -			    PERCPU_REF_ALLOW_REINIT, GFP_KERNEL))
> -		goto out_free;
> +		goto out;
>  
>  	if (io_alloc_buf_tables(buf_data, nr_tables, nr_args))
> -		goto out_ref;
> +		goto out;
>  
>  	return buf_data;
> -
> -out_ref:
> -	percpu_ref_exit(&buf_data->refs);
> -out_free:
> -	kfree(buf_data->table);
> -	kfree(buf_data);
> +out:
> +	free_fixed_rsrc_data(ctx->buf_data);
> +	ctx->buf_data = NULL;
>  	return ERR_PTR(ret);
>  }
>  
> 

-- 
Pavel Begunkov

Re: [PATCH v5 13/13] io_uring: support buffer registration sharing

[Pavel Begunkov]

On 12/01/2021 21:33, Bijan Mottahedeh wrote:
> Implement buffer sharing among multiple rings.
> 
> A ring shares its (future) buffer registrations at setup time with
> IORING_SETUP_SHARE_BUF. A ring attaches to another ring's buffer
> registration at setup time with IORING_SETUP_ATTACH_BUF, after
> authenticating with the buffer registration owner's fd. Any updates to
> the owner's buffer registrations become immediately available to the
> attached rings.

I'm thinking it through, but there is an easy to miss potential bug,
so see below a comment to not forget.

> 
> Signed-off-by: Bijan Mottahedeh <bijan.mottahedeh@oracle.com>
> 
> Conflicts:
> 	fs/io_uring.c
> ---
>  fs/io_uring.c                 | 85 +++++++++++++++++++++++++++++++++++++++++--
>  include/uapi/linux/io_uring.h |  2 +
>  2 files changed, 83 insertions(+), 4 deletions(-)
[...]
> +
> +static int io_init_buf_data(struct io_ring_ctx *ctx, struct io_uring_params *p)
> +{
> +	if ((p->flags & (IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF)) ==
> +	    (IORING_SETUP_SHARE_BUF | IORING_SETUP_ATTACH_BUF))
> +		return -EINVAL;
> +
> +	if (p->flags & IORING_SETUP_SHARE_BUF) {
> +		struct fixed_rsrc_data *buf_data;
> +
> +		buf_data = alloc_fixed_rsrc_data(ctx);
> +		if (IS_ERR(buf_data))
> +			return PTR_ERR(buf_data);

Because of sneaked through 5.11 fixes it'll be

if (!buf_data) return -ENOMEM

> +
> +		ctx->buf_data = buf_data;
> +		return 0;
> +	}
> +
> +	if (p->flags & IORING_SETUP_ATTACH_BUF)
> +		return io_attach_buf_data(ctx, p);
> +
> +	return 0;
> +}
> +


-- 
Pavel Begunkov