[PATCH V9 0/7] io_uring: support group buffer & ublk zc

[PATCH V9 0/7] io_uring: support group buffer & ublk zc

[Ming Lei]

Hello,

The 1st ~ 4th patches reuse `io_mapped_ubuf` for group kernel buffer.

The 5th patch supports group buffer, so far only kernel buffer is
supported, but it is pretty easy to extend for userspace group buffer.

The 6th patch adds uring_cmd interface for driver.

The last patch applies group kernel buffer for supporting ublk zc.

Tests:

1) pass liburing test
- make runtests

2) add ublk loop-zc test code
https://github.com/ming1/liburing/tree/uring_group

V9:
	- reuse io_mapped_ubuf for group kernel buffer(Jens)
	- rename as REQ_F_GROUP_BUF which can be extended for userspace
	group buffer easily
	- rebase on the latest for-6.13/io_uring
	- make sure that group buffer is imported once
	- use group buffer exclusively with buffer node & buffer select
	- misc cleanup

V8:
	- simplify & clean up group request completion, don't reuse 
	SQE_GROUP as state; meantime improve document; now group
	implementation is quite clean
	- handle short read/recv correctly by zeroing out the remained
	  part(Pavel)
	- fix one group leader reference(Uday Shankar)
	- only allow ublk provide buffer command in case of zc(Uday Shankar)

V7:
	- remove dead code in sqe group support(Pavel)
	- fail single group request(Pavel)
	- remove IORING_PROVIDE_GROUP_KBUF(Pavel)
	- remove REQ_F_SQE_GROUP_DEP(Pavel)
	- rename as leasing buffer
	- improve commit log
	- map group member's IOSQE_IO_DRAIN to GROUP_KBUF, which
	aligns with buffer select use, and it means that io_uring starts
	to support leased kbuf from other subsystem for group member
	requests only

V6:
	- follow Pavel's suggestion to disallow IOSQE_CQE_SKIP_SUCCESS &
	  LINK_TIMEOUT
	- kill __io_complete_group_member() (Pavel)
	- simplify link failure handling (Pavel)
	- move members' queuing out of completion lock (Pavel)
	- cleanup group io complete handler
	- add more comment
	- add ublk zc into liburing test for covering
	  IOSQE_SQE_GROUP & IORING_PROVIDE_GROUP_KBUF 

V5:
	- follow Pavel's suggestion to minimize change on io_uring fast code
	  path: sqe group code is called in by single 'if (unlikely())' from
	  both issue & completion code path

	- simplify & re-write group request completion
		avoid to touch io-wq code by completing group leader via tw
		directly, just like ->task_complete

		re-write group member & leader completion handling, one
		simplification is always to free leader via the last member

		simplify queueing group members, not support issuing leader
		and members in parallel

	- fail the whole group if IO_*LINK & IO_DRAIN is set on group
	  members, and test code to cover this change

	- misc cleanup

V4:
	- address most comments from Pavel
	- fix request double free
	- don't use io_req_commit_cqe() in io_req_complete_defer()
	- make members' REQ_F_INFLIGHT discoverable
	- use common assembling check in submission code path
	- drop patch 3 and don't move REQ_F_CQE_SKIP out of io_free_req()
	- don't set .accept_group_kbuf for net send zc, in which members
	  need to be queued after buffer notification is got, and can be
	  enabled in future
	- add .grp_leader field via union, and share storage with .grp_link
	- move .grp_refs into one hole of io_kiocb, so that one extra
	cacheline isn't needed for io_kiocb
	- cleanup & document improvement

V3:
	- add IORING_FEAT_SQE_GROUP
	- simplify group completion, and minimize change on io_req_complete_defer()
	- simplify & cleanup io_queue_group_members()
	- fix many failure handling issues
	- cover failure handling code in added liburing tests
	- remove RFC

V2:
	- add generic sqe group, suggested by Kevin Wolf
	- add REQ_F_SQE_GROUP_DEP which is based on IOSQE_SQE_GROUP, for sharing
	  kernel resource in group wide, suggested by Kevin Wolf
	- remove sqe ext flag, and use the last bit for IOSQE_SQE_GROUP(Pavel),
	in future we still can extend sqe flags with one uring context flag
	- initialize group requests via submit state pattern, suggested by Pavel
	- all kinds of cleanup & bug fixes

Ming Lei (7):
  io_uring: rename io_mapped_ubuf as io_mapped_buf
  io_uring: rename ubuf of io_mapped_buf as start
  io_uring: shrink io_mapped_buf
  io_uring: reuse io_mapped_buf for kernel buffer
  io_uring: support leased group buffer with REQ_F_GROUP_BUF
  io_uring/uring_cmd: support leasing device kernel buffer to io_uring
  ublk: support leasing io buffer to io_uring

 drivers/block/ublk_drv.c       | 160 +++++++++++++++++++++++++++++++--
 include/linux/io_uring/cmd.h   |   7 ++
 include/linux/io_uring_types.h |  50 +++++++++++
 include/uapi/linux/ublk_cmd.h  |  11 ++-
 io_uring/fdinfo.c              |   4 +-
 io_uring/io_uring.c            |  32 +++++--
 io_uring/io_uring.h            |   5 ++
 io_uring/kbuf.c                |  83 +++++++++++++++++
 io_uring/kbuf.h                |  46 ++++++++++
 io_uring/net.c                 |  27 +++++-
 io_uring/rsrc.c                |  19 ++--
 io_uring/rsrc.h                |  16 +---
 io_uring/rw.c                  |  37 ++++++--
 io_uring/uring_cmd.c           |  10 +++
 14 files changed, 463 insertions(+), 44 deletions(-)

-- 
2.47.0

[PATCH V9 1/7] io_uring: rename io_mapped_ubuf as io_mapped_buf

[Ming Lei]

Rename io_mapped_ubuf so that the same structure can be used for
describing kernel buffer.

Signed-off-by: Ming Lei <ming.lei@redhat.com>
---
 io_uring/fdinfo.c |  2 +-
 io_uring/rsrc.c   | 10 +++++-----
 io_uring/rsrc.h   |  6 +++---
 3 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/io_uring/fdinfo.c b/io_uring/fdinfo.c
index b214e5a407b5..9ca95f877312 100644
--- a/io_uring/fdinfo.c
+++ b/io_uring/fdinfo.c
@@ -218,7 +218,7 @@ __cold void io_uring_show_fdinfo(struct seq_file *m, struct file *file)
 	}
 	seq_printf(m, "UserBufs:\t%u\n", ctx->buf_table.nr);
 	for (i = 0; has_lock && i < ctx->buf_table.nr; i++) {
-		struct io_mapped_ubuf *buf = NULL;
+		struct io_mapped_buf *buf = NULL;
 
 		if (ctx->buf_table.nodes[i])
 			buf = ctx->buf_table.nodes[i]->buf;
diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
index 2fb1791d7255..70dba4a4de1d 100644
--- a/io_uring/rsrc.c
+++ b/io_uring/rsrc.c
@@ -106,7 +106,7 @@ static void io_buffer_unmap(struct io_ring_ctx *ctx, struct io_rsrc_node *node)
 	unsigned int i;
 
 	if (node->buf) {
-		struct io_mapped_ubuf *imu = node->buf;
+		struct io_mapped_buf *imu = node->buf;
 
 		if (!refcount_dec_and_test(&imu->refs))
 			return;
@@ -580,7 +580,7 @@ static bool headpage_already_acct(struct io_ring_ctx *ctx, struct page **pages,
 	/* check previously registered pages */
 	for (i = 0; i < ctx->buf_table.nr; i++) {
 		struct io_rsrc_node *node = ctx->buf_table.nodes[i];
-		struct io_mapped_ubuf *imu;
+		struct io_mapped_buf *imu;
 
 		if (!node)
 			continue;
@@ -597,7 +597,7 @@ static bool headpage_already_acct(struct io_ring_ctx *ctx, struct page **pages,
 }
 
 static int io_buffer_account_pin(struct io_ring_ctx *ctx, struct page **pages,
-				 int nr_pages, struct io_mapped_ubuf *imu,
+				 int nr_pages, struct io_mapped_buf *imu,
 				 struct page **last_hpage)
 {
 	int i, ret;
@@ -724,7 +724,7 @@ static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx,
 						   struct iovec *iov,
 						   struct page **last_hpage)
 {
-	struct io_mapped_ubuf *imu = NULL;
+	struct io_mapped_buf *imu = NULL;
 	struct page **pages = NULL;
 	struct io_rsrc_node *node;
 	unsigned long off;
@@ -866,7 +866,7 @@ int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
 }
 
 int io_import_fixed(int ddir, struct iov_iter *iter,
-			   struct io_mapped_ubuf *imu,
+			   struct io_mapped_buf *imu,
 			   u64 buf_addr, size_t len)
 {
 	u64 buf_end;
diff --git a/io_uring/rsrc.h b/io_uring/rsrc.h
index bc3a863b14bb..9d55f9769c77 100644
--- a/io_uring/rsrc.h
+++ b/io_uring/rsrc.h
@@ -22,11 +22,11 @@ struct io_rsrc_node {
 	u64 tag;
 	union {
 		unsigned long file_ptr;
-		struct io_mapped_ubuf *buf;
+		struct io_mapped_buf *buf;
 	};
 };
 
-struct io_mapped_ubuf {
+struct io_mapped_buf {
 	u64		ubuf;
 	unsigned int	len;
 	unsigned int	nr_bvecs;
@@ -50,7 +50,7 @@ void io_rsrc_data_free(struct io_rsrc_data *data);
 int io_rsrc_data_alloc(struct io_rsrc_data *data, unsigned nr);
 
 int io_import_fixed(int ddir, struct iov_iter *iter,
-			   struct io_mapped_ubuf *imu,
+			   struct io_mapped_buf *imu,
 			   u64 buf_addr, size_t len);
 
 int io_register_clone_buffers(struct io_ring_ctx *ctx, void __user *arg);
-- 
2.47.0

[PATCH V9 2/7] io_uring: rename ubuf of io_mapped_buf as start

[Ming Lei]

->ubuf of `io_mapped_buf` stores the start address of userspace fixed
buffer. `io_mapped_buf` will be extended for covering kernel buffer,
so rename ->ubuf as ->start.

Signed-off-by: Ming Lei <ming.lei@redhat.com>
---
 io_uring/fdinfo.c | 2 +-
 io_uring/rsrc.c   | 6 +++---
 io_uring/rsrc.h   | 2 +-
 3 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/io_uring/fdinfo.c b/io_uring/fdinfo.c
index 9ca95f877312..9b39cb596136 100644
--- a/io_uring/fdinfo.c
+++ b/io_uring/fdinfo.c
@@ -223,7 +223,7 @@ __cold void io_uring_show_fdinfo(struct seq_file *m, struct file *file)
 		if (ctx->buf_table.nodes[i])
 			buf = ctx->buf_table.nodes[i]->buf;
 		if (buf)
-			seq_printf(m, "%5u: 0x%llx/%u\n", i, buf->ubuf, buf->len);
+			seq_printf(m, "%5u: 0x%llx/%u\n", i, buf->start, buf->len);
 		else
 			seq_printf(m, "%5u: <none>\n", i);
 	}
diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
index 70dba4a4de1d..9b8827c72230 100644
--- a/io_uring/rsrc.c
+++ b/io_uring/rsrc.c
@@ -765,7 +765,7 @@ static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx,
 
 	size = iov->iov_len;
 	/* store original address for later verification */
-	imu->ubuf = (unsigned long) iov->iov_base;
+	imu->start = (unsigned long) iov->iov_base;
 	imu->len = iov->iov_len;
 	imu->nr_bvecs = nr_pages;
 	imu->folio_shift = PAGE_SHIFT;
@@ -877,14 +877,14 @@ int io_import_fixed(int ddir, struct iov_iter *iter,
 	if (unlikely(check_add_overflow(buf_addr, (u64)len, &buf_end)))
 		return -EFAULT;
 	/* not inside the mapped region */
-	if (unlikely(buf_addr < imu->ubuf || buf_end > (imu->ubuf + imu->len)))
+	if (unlikely(buf_addr < imu->start || buf_end > (imu->start + imu->len)))
 		return -EFAULT;
 
 	/*
 	 * Might not be a start of buffer, set size appropriately
 	 * and advance us to the beginning.
 	 */
-	offset = buf_addr - imu->ubuf;
+	offset = buf_addr - imu->start;
 	iov_iter_bvec(iter, ddir, imu->bvec, imu->nr_bvecs, offset + len);
 
 	if (offset) {
diff --git a/io_uring/rsrc.h b/io_uring/rsrc.h
index 9d55f9769c77..887699400e29 100644
--- a/io_uring/rsrc.h
+++ b/io_uring/rsrc.h
@@ -27,7 +27,7 @@ struct io_rsrc_node {
 };
 
 struct io_mapped_buf {
-	u64		ubuf;
+	u64		start;
 	unsigned int	len;
 	unsigned int	nr_bvecs;
 	unsigned int    folio_shift;
-- 
2.47.0

[PATCH V9 3/7] io_uring: shrink io_mapped_buf

[Ming Lei]

`struct io_mapped_buf` will be extended to cover kernel buffer which
may be in fast IO path, and `struct io_mapped_buf` needs to be per-IO.

So shrink sizeof(struct io_mapped_buf) by the following ways:

- folio_shift is < 64, so 6bits are enough to hold it, the remained bits
  can be used for the coming kernel buffer

- define `acct_pages` as 'unsigned int', which is big enough for
  accounting pages in the buffer

Signed-off-by: Ming Lei <ming.lei@redhat.com>
---
 io_uring/rsrc.c | 2 ++
 io_uring/rsrc.h | 6 +++---
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
index 9b8827c72230..16f5abe03d10 100644
--- a/io_uring/rsrc.c
+++ b/io_uring/rsrc.c
@@ -685,6 +685,8 @@ static bool io_try_coalesce_buffer(struct page ***pages, int *nr_pages,
 		return false;
 
 	data->folio_shift = folio_shift(folio);
+	WARN_ON_ONCE(data->folio_shift >= 64);
+
 	/*
 	 * Check if pages are contiguous inside a folio, and all folios have
 	 * the same page count except for the head and tail.
diff --git a/io_uring/rsrc.h b/io_uring/rsrc.h
index 887699400e29..255ec94ea172 100644
--- a/io_uring/rsrc.h
+++ b/io_uring/rsrc.h
@@ -30,9 +30,9 @@ struct io_mapped_buf {
 	u64		start;
 	unsigned int	len;
 	unsigned int	nr_bvecs;
-	unsigned int    folio_shift;
 	refcount_t	refs;
-	unsigned long	acct_pages;
+	unsigned int	acct_pages;
+	unsigned int	folio_shift:6;
 	struct bio_vec	bvec[] __counted_by(nr_bvecs);
 };
 
@@ -41,7 +41,7 @@ struct io_imu_folio_data {
 	unsigned int	nr_pages_head;
 	/* For non-head/tail folios, has to be fully included */
 	unsigned int	nr_pages_mid;
-	unsigned int	folio_shift;
+	unsigned char	folio_shift;
 };
 
 struct io_rsrc_node *io_rsrc_node_alloc(struct io_ring_ctx *ctx, int type);
-- 
2.47.0

[PATCH V9 4/7] io_uring: reuse io_mapped_buf for kernel buffer

[Ming Lei]

Prepare for supporting kernel buffer in case of io group, in which group
leader leases kernel buffer to io_uring, and consumed by io_uring OPs.

So reuse io_mapped_buf for group kernel buffer, and unfortunately
io_import_fixed() can't be reused since userspace fixed buffer is
virt-contiguous, but it isn't true for kernel buffer.

Also kernel buffer lifetime is bound with group leader request, it isn't
necessary to use rsrc_node for tracking its lifetime, especially it needs
extra allocation of rsrc_node for each IO.

Signed-off-by: Ming Lei <ming.lei@redhat.com>
---
 include/linux/io_uring_types.h | 23 +++++++++++++++++++++++
 io_uring/kbuf.c                | 34 ++++++++++++++++++++++++++++++++++
 io_uring/kbuf.h                |  3 +++
 io_uring/rsrc.c                |  1 +
 io_uring/rsrc.h                | 10 ----------
 5 files changed, 61 insertions(+), 10 deletions(-)

diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h
index bc883078c1ed..9af83cf214c2 100644
--- a/include/linux/io_uring_types.h
+++ b/include/linux/io_uring_types.h
@@ -2,6 +2,7 @@
 #define IO_URING_TYPES_H
 
 #include <linux/blkdev.h>
+#include <linux/bvec.h>
 #include <linux/hashtable.h>
 #include <linux/task_work.h>
 #include <linux/bitmap.h>
@@ -39,6 +40,28 @@ enum io_uring_cmd_flags {
 	IO_URING_F_COMPAT		= (1 << 12),
 };
 
+struct io_mapped_buf {
+	u64		start;
+	unsigned int	len;
+	unsigned int	nr_bvecs;
+
+	/* kbuf hasn't refs and accounting, its lifetime is bound with req */
+	union {
+		struct {
+			refcount_t	refs;
+			unsigned int	acct_pages;
+		};
+		/* pbvec is only for kbuf */
+		const struct bio_vec	*pbvec;
+	};
+	unsigned int	folio_shift:6;
+	unsigned int	dir:1;		/* ITER_DEST or ITER_SOURCE */
+	unsigned int	kbuf:1;		/* kernel buffer or not */
+	/* offset in the 1st bvec, for kbuf only */
+	unsigned int	offset;
+	struct bio_vec	bvec[] __counted_by(nr_bvecs);
+};
+
 struct io_wq_work_node {
 	struct io_wq_work_node *next;
 };
diff --git a/io_uring/kbuf.c b/io_uring/kbuf.c
index d407576ddfb7..c4a776860cb4 100644
--- a/io_uring/kbuf.c
+++ b/io_uring/kbuf.c
@@ -838,3 +838,37 @@ int io_pbuf_mmap(struct file *file, struct vm_area_struct *vma)
 	io_put_bl(ctx, bl);
 	return ret;
 }
+
+/*
+ * kernel buffer is built over generic bvec, and can't be always
+ * virt-contiguous, which is different with userspace fixed buffer,
+ * so we can't reuse io_import_fixed() here
+ *
+ * Also kernel buffer lifetime is bound with request, and we needn't
+ * to use rsrc_node to track its lifetime
+ */
+int io_import_kbuf(int ddir, struct iov_iter *iter,
+		    const struct io_mapped_buf *kbuf,
+		    u64 buf_off, size_t len)
+{
+	unsigned long offset = kbuf->offset;
+
+	WARN_ON_ONCE(!kbuf->kbuf);
+
+	if (ddir != kbuf->dir)
+		return -EINVAL;
+
+	if (unlikely(buf_off > kbuf->len))
+		return -EFAULT;
+
+	if (unlikely(len > kbuf->len - buf_off))
+		return -EFAULT;
+
+	offset += buf_off;
+	iov_iter_bvec(iter, ddir, kbuf->pbvec, kbuf->nr_bvecs, offset + len);
+
+	if (offset)
+		iov_iter_advance(iter, offset);
+
+	return 0;
+}
diff --git a/io_uring/kbuf.h b/io_uring/kbuf.h
index 36aadfe5ac00..04ccd52dd0ad 100644
--- a/io_uring/kbuf.h
+++ b/io_uring/kbuf.h
@@ -88,6 +88,9 @@ void io_put_bl(struct io_ring_ctx *ctx, struct io_buffer_list *bl);
 struct io_buffer_list *io_pbuf_get_bl(struct io_ring_ctx *ctx,
 				      unsigned long bgid);
 int io_pbuf_mmap(struct file *file, struct vm_area_struct *vma);
+int io_import_kbuf(int ddir, struct iov_iter *iter,
+		    const struct io_mapped_buf *kbuf,
+		    u64 buf_off, size_t len);
 
 static inline bool io_kbuf_recycle_ring(struct io_kiocb *req)
 {
diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
index 16f5abe03d10..5f35641c55ab 100644
--- a/io_uring/rsrc.c
+++ b/io_uring/rsrc.c
@@ -771,6 +771,7 @@ static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx,
 	imu->len = iov->iov_len;
 	imu->nr_bvecs = nr_pages;
 	imu->folio_shift = PAGE_SHIFT;
+	imu->kbuf = 0;
 	if (coalesced)
 		imu->folio_shift = data.folio_shift;
 	refcount_set(&imu->refs, 1);
diff --git a/io_uring/rsrc.h b/io_uring/rsrc.h
index 255ec94ea172..d54a5f84b9ed 100644
--- a/io_uring/rsrc.h
+++ b/io_uring/rsrc.h
@@ -26,16 +26,6 @@ struct io_rsrc_node {
 	};
 };
 
-struct io_mapped_buf {
-	u64		start;
-	unsigned int	len;
-	unsigned int	nr_bvecs;
-	refcount_t	refs;
-	unsigned int	acct_pages;
-	unsigned int	folio_shift:6;
-	struct bio_vec	bvec[] __counted_by(nr_bvecs);
-};
-
 struct io_imu_folio_data {
 	/* Head folio can be partially included in the fixed buf */
 	unsigned int	nr_pages_head;
-- 
2.47.0

[PATCH V9 5/7] io_uring: support leased group buffer with REQ_F_GROUP_BUF

[Ming Lei]

SQE group introduces one new mechanism to share resource among one
group of requests, and all member requests can consume the resource
leased by group leader efficiently in parallel.

This patch uses the added SQE group to lease kernel buffer from group
leader(driver) to members(io_uring) in sqe group:

- this kernel buffer is owned by kernel device(driver), and has very
  short lifetime, such as, it is often aligned with block IO lifetime

- group leader leases the kernel buffer from driver to member requests
  of io_uring subsystem

- member requests uses the leased buffer to do FS or network IO,
  IOSQE_IO_DRAIN bit isn't used for group member IO, so it is mapped to
  GROUP_KBUF; the actual use becomes very similar with buffer select.

- this kernel buffer is returned back after all member requests are
  completed

io_uring builtin provide/register buffer isn't one good match for this
use case:

- complicated dependency on add/remove buffer
  this buffer has to be added/removed to one global table by add/remove OPs,
  and all consumer OPs have to sync with the add/remove OPs; either
  consumer OPs have to by issued one by one with IO_LINK; or two extra
  syscall are added for one time of buffer lease & consumption, this way
  slows down ublk io handling, and may lose zero copy value

- application becomes more complicated

- application may panic and the kernel buffer is left in io_uring, which
  complicates io_uring shutdown handling since returning back buffer
  needs to cowork with buffer owner

- big change is needed in io_uring provide/register buffer

- the requirement is just to lease the kernel buffer to io_uring subsystem for
  very short time, not necessary to move it into io_uring and make it global

This way looks a bit similar with kernel's pipe/splice, but there are some
important differences:

- splice is for transferring data between two FDs via pipe, and fd_out can
only read data from pipe, but data can't be written to; this feature can
lease buffer from group leader(driver subsystem) to members(io_uring subsystem),
so member request can write data to this buffer if the buffer direction is
allowed to write to.

- splice implements data transfer by moving pages between subsystem and
pipe, that means page ownership is transferred, and this way is one of the
most complicated thing of splice; this patch supports scenarios in which
the buffer can't be transferred, and buffer is only borrowed to member
requests for consumption, and is returned back after member requests
consume the leased buffer, so buffer lifetime is aligned with group leader
lifetime, and buffer lifetime is simplified a lot. Especially the buffer
is guaranteed to be returned back.

- splice can't run in async way basically

It can help to implement generic zero copy between device and related
operations, such as ublk, fuse.

Signed-off-by: Ming Lei <ming.lei@redhat.com>
---
 include/linux/io_uring_types.h | 29 +++++++++++++++++-
 io_uring/io_uring.c            | 32 ++++++++++++++++----
 io_uring/io_uring.h            |  5 ++++
 io_uring/kbuf.c                | 55 ++++++++++++++++++++++++++++++++--
 io_uring/kbuf.h                | 49 ++++++++++++++++++++++++++++--
 io_uring/net.c                 | 27 ++++++++++++++++-
 io_uring/rw.c                  | 37 +++++++++++++++++++----
 7 files changed, 216 insertions(+), 18 deletions(-)

diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h
index 9af83cf214c2..f3d891fded55 100644
--- a/include/linux/io_uring_types.h
+++ b/include/linux/io_uring_types.h
@@ -40,8 +40,16 @@ enum io_uring_cmd_flags {
 	IO_URING_F_COMPAT		= (1 << 12),
 };
 
+struct io_mapped_buf;
+typedef void (io_uring_kbuf_ack_t) (const struct io_mapped_buf *);
+
 struct io_mapped_buf {
-	u64		start;
+	/* start is always 0 for kernel buffer */
+	union {
+		u64			start;
+		/* called for returning back the kernel buffer */
+		io_uring_kbuf_ack_t	*kbuf_ack;
+	};
 	unsigned int	len;
 	unsigned int	nr_bvecs;
 
@@ -504,6 +512,7 @@ enum {
 	REQ_F_BUFFERS_COMMIT_BIT,
 	REQ_F_GROUP_LEADER_BIT,
 	REQ_F_BUF_NODE_BIT,
+	REQ_F_GROUP_BUF_BIT,
 
 	/* not a real bit, just to check we're not overflowing the space */
 	__REQ_F_LAST_BIT,
@@ -588,6 +597,16 @@ enum {
 	REQ_F_GROUP_LEADER	= IO_REQ_FLAG(REQ_F_GROUP_LEADER_BIT),
 	/* buf node is valid */
 	REQ_F_BUF_NODE		= IO_REQ_FLAG(REQ_F_BUF_NODE_BIT),
+	/*
+	 * Use group leader's buffer
+	 *
+	 * For group member, this flag is mapped from IOSQE_IO_DRAIN which
+	 * isn't used for group member
+	 *
+	 * Group buffer has to be imported in ->issue() since it depends on
+	 * group leader.
+	 */
+	REQ_F_GROUP_BUF	= IO_REQ_FLAG(REQ_F_GROUP_BUF_BIT),
 };
 
 typedef void (*io_req_tw_func_t)(struct io_kiocb *req, struct io_tw_state *ts);
@@ -670,6 +689,14 @@ struct io_kiocb {
 		struct io_buffer_list	*buf_list;
 
 		struct io_rsrc_node	*buf_node;
+
+		/* valid IFF REQ_F_GROUP_BUF is set */
+		union {
+			/* store group buffer for group leader */
+			const struct io_mapped_buf *grp_buf;
+			/* for group member */
+			bool	grp_buf_imported;
+		};
 	};
 
 	union {
diff --git a/io_uring/io_uring.c b/io_uring/io_uring.c
index 076171977d5e..0a87312083bd 100644
--- a/io_uring/io_uring.c
+++ b/io_uring/io_uring.c
@@ -114,7 +114,7 @@
 
 #define IO_REQ_CLEAN_FLAGS (REQ_F_BUFFER_SELECTED | REQ_F_NEED_CLEANUP | \
 				REQ_F_POLLED | REQ_F_INFLIGHT | REQ_F_CREDS | \
-				REQ_F_ASYNC_DATA)
+				REQ_F_ASYNC_DATA | REQ_F_GROUP_BUF)
 
 #define IO_REQ_CLEAN_SLOW_FLAGS (REQ_F_REFCOUNT | REQ_F_LINK | REQ_F_HARDLINK |\
 				 REQ_F_GROUP | IO_REQ_CLEAN_FLAGS)
@@ -391,6 +391,8 @@ static bool req_need_defer(struct io_kiocb *req, u32 seq)
 
 static void io_clean_op(struct io_kiocb *req)
 {
+	if (req->flags & REQ_F_GROUP_BUF)
+		io_drop_group_buf(req);
 	if (req->flags & REQ_F_BUFFER_SELECTED) {
 		spin_lock(&req->ctx->completion_lock);
 		io_kbuf_drop(req);
@@ -925,14 +927,20 @@ static void io_queue_group_members(struct io_kiocb *req)
 	req->grp_link = NULL;
 	while (member) {
 		struct io_kiocb *next = member->grp_link;
+		bool grp_buf = member->flags & REQ_F_GROUP_BUF;
 
 		member->grp_leader = req;
 		if (unlikely(member->flags & REQ_F_FAIL))
 			io_req_task_queue_fail(member, member->cqe.res);
+		else if (unlikely(grp_buf && member->flags & REQ_F_BUF_NODE))
+			io_req_task_queue_fail(member, -EINVAL);
 		else if (unlikely(req->flags & REQ_F_FAIL))
 			io_req_task_queue_fail(member, -ECANCELED);
-		else
+		else {
+			if (grp_buf)
+				io_req_mark_group_buf(member, false);
 			io_req_task_queue(member);
+		}
 		member = next;
 	}
 }
@@ -997,6 +1005,11 @@ static enum group_mem io_prep_free_group_req(struct io_kiocb *req,
 			io_queue_group_members(req);
 		return GROUP_LEADER;
 	}
+
+	/* we are done with leased group buffer */
+	if (req->flags & REQ_F_GROUP_BUF)
+		req->flags &= ~REQ_F_GROUP_BUF;
+
 	if (!req_is_last_group_member(req))
 		return GROUP_OTHER_MEMBER;
 
@@ -2196,9 +2209,18 @@ static int io_init_req(struct io_ring_ctx *ctx, struct io_kiocb *req,
 		if (sqe_flags & IOSQE_CQE_SKIP_SUCCESS)
 			ctx->drain_disabled = true;
 		if (sqe_flags & IOSQE_IO_DRAIN) {
-			if (ctx->drain_disabled)
-				return io_init_fail_req(req, -EOPNOTSUPP);
-			io_init_req_drain(req);
+			/* IO_DRAIN is mapped to GROUP_BUF for group members */
+			if (ctx->submit_state.group.head) {
+				/* can't do buffer select */
+				if (sqe_flags & IOSQE_BUFFER_SELECT)
+					return io_init_fail_req(req, -EINVAL);
+				req->flags &= ~REQ_F_IO_DRAIN;
+				req->flags |= REQ_F_GROUP_BUF;
+			} else {
+				if (ctx->drain_disabled)
+					return io_init_fail_req(req, -EOPNOTSUPP);
+				io_init_req_drain(req);
+			}
 		}
 	}
 	if (unlikely(ctx->restricted || ctx->drain_active || ctx->drain_next)) {
diff --git a/io_uring/io_uring.h b/io_uring/io_uring.h
index 57b0d0209097..dd61529cd382 100644
--- a/io_uring/io_uring.h
+++ b/io_uring/io_uring.h
@@ -364,6 +364,11 @@ static inline bool req_is_group_leader(struct io_kiocb *req)
 	return req->flags & REQ_F_GROUP_LEADER;
 }
 
+static inline bool req_is_group_member(struct io_kiocb *req)
+{
+	return (req->flags & REQ_F_GROUP) && !req_is_group_leader(req);
+}
+
 /*
  * Don't complete immediately but use deferred completion infrastructure.
  * Protected by ->uring_lock and can only be used either with
diff --git a/io_uring/kbuf.c b/io_uring/kbuf.c
index c4a776860cb4..1a8ed35d4d6c 100644
--- a/io_uring/kbuf.c
+++ b/io_uring/kbuf.c
@@ -847,9 +847,9 @@ int io_pbuf_mmap(struct file *file, struct vm_area_struct *vma)
  * Also kernel buffer lifetime is bound with request, and we needn't
  * to use rsrc_node to track its lifetime
  */
-int io_import_kbuf(int ddir, struct iov_iter *iter,
-		    const struct io_mapped_buf *kbuf,
-		    u64 buf_off, size_t len)
+static int io_import_kbuf(int ddir, struct iov_iter *iter,
+			  const struct io_mapped_buf *kbuf,
+			  u64 buf_off, size_t len)
 {
 	unsigned long offset = kbuf->offset;
 
@@ -872,3 +872,52 @@ int io_import_kbuf(int ddir, struct iov_iter *iter,
 
 	return 0;
 }
+
+int io_import_group_buf(struct io_kiocb *req, int dir, struct iov_iter *iter,
+			unsigned long buf_off, unsigned int len)
+{
+	struct io_kiocb *lead = req->grp_leader;
+	int ret;
+
+	if (!req_is_group_member(req))
+		return -EINVAL;
+
+	if (!lead || !(lead->flags & REQ_F_GROUP_BUF))
+		return -EINVAL;
+
+	/* buffer node may be assigned just before importing */
+	if (req->flags & REQ_F_BUF_NODE)
+		return -EINVAL;
+
+	if (io_req_group_buf_imported(req))
+		return 0;
+
+	ret = io_import_kbuf(dir, iter, lead->grp_buf, buf_off, len);
+	if (!ret)
+		io_req_mark_group_buf(req, true);
+	return ret;
+}
+
+int io_lease_group_kbuf(struct io_kiocb *req,
+			const struct io_mapped_buf *grp_buf)
+{
+	if (!(req->flags & REQ_F_GROUP_LEADER))
+		return -EINVAL;
+
+	if (req->flags & (REQ_F_BUFFER_SELECT | REQ_F_BUF_NODE))
+		return -EINVAL;
+
+	if (!grp_buf->kbuf_ack || !grp_buf->pbvec || !grp_buf->kbuf)
+		return -EINVAL;
+
+	/*
+	 * Allow io_uring OPs to borrow this leased kbuf, which is returned
+	 * back by calling `kbuf_ack` when the group leader is freed.
+	 *
+	 * Not like pipe/splice, this kernel buffer is always owned by the
+	 * provider, and has to be returned back.
+	 */
+	req->grp_buf = grp_buf;
+	req->flags |= REQ_F_GROUP_BUF;
+	return 0;
+}
diff --git a/io_uring/kbuf.h b/io_uring/kbuf.h
index 04ccd52dd0ad..f98e2d8dc48c 100644
--- a/io_uring/kbuf.h
+++ b/io_uring/kbuf.h
@@ -88,9 +88,11 @@ void io_put_bl(struct io_ring_ctx *ctx, struct io_buffer_list *bl);
 struct io_buffer_list *io_pbuf_get_bl(struct io_ring_ctx *ctx,
 				      unsigned long bgid);
 int io_pbuf_mmap(struct file *file, struct vm_area_struct *vma);
-int io_import_kbuf(int ddir, struct iov_iter *iter,
-		    const struct io_mapped_buf *kbuf,
-		    u64 buf_off, size_t len);
+
+int io_import_group_buf(struct io_kiocb *req, int dir, struct iov_iter *iter,
+			unsigned long buf_off, unsigned int len);
+int io_lease_group_kbuf(struct io_kiocb *req,
+			const struct io_mapped_buf *grp_buf);
 
 static inline bool io_kbuf_recycle_ring(struct io_kiocb *req)
 {
@@ -223,4 +225,45 @@ static inline unsigned int io_put_kbufs(struct io_kiocb *req, int len,
 {
 	return __io_put_kbufs(req, len, nbufs, issue_flags);
 }
+
+static inline bool io_use_group_buf(struct io_kiocb *req)
+{
+	return req->flags & REQ_F_GROUP_BUF;
+}
+
+static inline bool io_use_group_kbuf(struct io_kiocb *req)
+{
+	if (io_use_group_buf(req))
+		return req->grp_leader && req->grp_leader->grp_buf->kbuf;
+	return false;
+}
+
+static inline void io_drop_group_buf(struct io_kiocb *req)
+{
+	const struct io_mapped_buf *gbuf = req->grp_buf;
+
+	if (gbuf && gbuf->kbuf)
+		gbuf->kbuf_ack(gbuf);
+}
+
+/* zero remained bytes of kernel buffer for avoiding to leak data */
+static inline void io_req_zero_remained(struct io_kiocb *req, struct iov_iter *iter)
+{
+	size_t left = iov_iter_count(iter);
+
+	if (iov_iter_rw(iter) == READ && left > 0)
+		iov_iter_zero(left, iter);
+}
+
+/* For group member only */
+static inline void io_req_mark_group_buf(struct io_kiocb *req, bool imported)
+{
+	req->grp_buf_imported = imported;
+}
+
+/* For group member only */
+static inline bool io_req_group_buf_imported(struct io_kiocb *req)
+{
+	return req->grp_buf_imported;
+}
 #endif
diff --git a/io_uring/net.c b/io_uring/net.c
index 2ccc2b409431..e87b67498733 100644
--- a/io_uring/net.c
+++ b/io_uring/net.c
@@ -88,6 +88,13 @@ struct io_sr_msg {
  */
 #define MULTISHOT_MAX_RETRY	32
 
+#define user_ptr_to_u64(x) (		\
+{					\
+	typecheck(void __user *, (x));	\
+	(u64)(unsigned long)(x);	\
+}					\
+)
+
 int io_shutdown_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
 {
 	struct io_shutdown *shutdown = io_kiocb_to_cmd(req, struct io_shutdown);
@@ -384,7 +391,7 @@ static int io_send_setup(struct io_kiocb *req, const struct io_uring_sqe *sqe)
 		kmsg->msg.msg_name = &kmsg->addr;
 		kmsg->msg.msg_namelen = addr_len;
 	}
-	if (!io_do_buffer_select(req)) {
+	if (!io_do_buffer_select(req) && !io_use_group_buf(req)) {
 		ret = import_ubuf(ITER_SOURCE, sr->buf, sr->len,
 				  &kmsg->msg.msg_iter);
 		if (unlikely(ret < 0))
@@ -599,6 +606,15 @@ int io_send(struct io_kiocb *req, unsigned int issue_flags)
 	if (issue_flags & IO_URING_F_NONBLOCK)
 		flags |= MSG_DONTWAIT;
 
+	if (io_use_group_buf(req)) {
+		ret = io_import_group_buf(req, ITER_SOURCE,
+					  &kmsg->msg.msg_iter,
+					  user_ptr_to_u64(sr->buf),
+					  sr->len);
+		if (unlikely(ret))
+			return ret;
+	}
+
 retry_bundle:
 	if (io_do_buffer_select(req)) {
 		struct buf_sel_arg arg = {
@@ -889,6 +905,8 @@ static inline bool io_recv_finish(struct io_kiocb *req, int *ret,
 		*ret = IOU_STOP_MULTISHOT;
 	else
 		*ret = IOU_OK;
+	if (io_use_group_kbuf(req))
+		io_req_zero_remained(req, &kmsg->msg.msg_iter);
 	io_req_msg_cleanup(req, issue_flags);
 	return true;
 }
@@ -1161,6 +1179,13 @@ int io_recv(struct io_kiocb *req, unsigned int issue_flags)
 			goto out_free;
 		}
 		sr->buf = NULL;
+	} else if (io_use_group_buf(req)) {
+		ret = io_import_group_buf(req, ITER_DEST,
+					  &kmsg->msg.msg_iter,
+					  user_ptr_to_u64(sr->buf),
+					  sr->len);
+		if (unlikely(ret))
+			goto out_free;
 	}
 
 	kmsg->msg.msg_flags = 0;
diff --git a/io_uring/rw.c b/io_uring/rw.c
index e368b9afde03..c5ab464dd51d 100644
--- a/io_uring/rw.c
+++ b/io_uring/rw.c
@@ -488,6 +488,11 @@ static bool __io_complete_rw_common(struct io_kiocb *req, long res)
 		}
 		req_set_fail(req);
 		req->cqe.res = res;
+		if (io_use_group_kbuf(req)) {
+			struct io_async_rw *io = req->async_data;
+
+			io_req_zero_remained(req, &io->iter);
+		}
 	}
 	return false;
 }
@@ -629,11 +634,15 @@ static inline loff_t *io_kiocb_ppos(struct kiocb *kiocb)
  */
 static ssize_t loop_rw_iter(int ddir, struct io_rw *rw, struct iov_iter *iter)
 {
+	struct io_kiocb *req = cmd_to_io_kiocb(rw);
 	struct kiocb *kiocb = &rw->kiocb;
 	struct file *file = kiocb->ki_filp;
 	ssize_t ret = 0;
 	loff_t *ppos;
 
+	if (io_use_group_kbuf(req))
+		return -EOPNOTSUPP;
+
 	/*
 	 * Don't support polled IO through this interface, and we can't
 	 * support non-blocking either. For the latter, this just causes
@@ -832,20 +841,32 @@ static int io_rw_init_file(struct io_kiocb *req, fmode_t mode, int rw_type)
 	return 0;
 }
 
+static int rw_import_group_buf(struct io_kiocb *req, int dir,
+			       struct io_rw *rw, struct io_async_rw *io)
+{
+	int ret = io_import_group_buf(req, dir, &io->iter, rw->addr, rw->len);
+
+	if (!ret)
+		iov_iter_save_state(&io->iter, &io->iter_state);
+	return ret;
+}
+
 static int __io_read(struct io_kiocb *req, unsigned int issue_flags)
 {
 	bool force_nonblock = issue_flags & IO_URING_F_NONBLOCK;
 	struct io_rw *rw = io_kiocb_to_cmd(req, struct io_rw);
 	struct io_async_rw *io = req->async_data;
 	struct kiocb *kiocb = &rw->kiocb;
-	ssize_t ret;
+	ssize_t ret = 0;
 	loff_t *ppos;
 
-	if (io_do_buffer_select(req)) {
+	if (io_do_buffer_select(req))
 		ret = io_import_iovec(ITER_DEST, req, io, issue_flags);
-		if (unlikely(ret < 0))
-			return ret;
-	}
+	else if (io_use_group_buf(req))
+		ret = rw_import_group_buf(req, ITER_DEST, rw, io);
+	if (unlikely(ret < 0))
+		return ret;
+
 	ret = io_rw_init_file(req, FMODE_READ, READ);
 	if (unlikely(ret))
 		return ret;
@@ -1028,6 +1049,12 @@ int io_write(struct io_kiocb *req, unsigned int issue_flags)
 	ssize_t ret, ret2;
 	loff_t *ppos;
 
+	if (io_use_group_buf(req)) {
+		ret = rw_import_group_buf(req, ITER_SOURCE, rw, io);
+		if (unlikely(ret < 0))
+			return ret;
+	}
+
 	ret = io_rw_init_file(req, FMODE_WRITE, WRITE);
 	if (unlikely(ret))
 		return ret;
-- 
2.47.0

[PATCH V9 6/7] io_uring/uring_cmd: support leasing device kernel buffer to io_uring

[Ming Lei]

Add API of io_uring_cmd_lease_kbuf() for driver to lease its kernel
buffer to io_uring.

The leased buffer can only be consumed by io_uring OPs in group wide,
and the uring_cmd has to be one group leader.

This way can support generic device zero copy over device buffer in
userspace:

- create one sqe group
- lease one device buffer to io_uring by the group leader of uring_cmd
- io_uring member OPs consume this kernel buffer by passing IOSQE_IO_DRAIN
  which isn't used for group member, and mapped to GROUP_BUF.
- the kernel buffer is returned back after all member OPs are completed

Signed-off-by: Ming Lei <ming.lei@redhat.com>
---
 include/linux/io_uring/cmd.h |  7 +++++++
 io_uring/uring_cmd.c         | 10 ++++++++++
 2 files changed, 17 insertions(+)

diff --git a/include/linux/io_uring/cmd.h b/include/linux/io_uring/cmd.h
index 578a3fdf5c71..8a6f1db1ca84 100644
--- a/include/linux/io_uring/cmd.h
+++ b/include/linux/io_uring/cmd.h
@@ -60,6 +60,8 @@ void io_uring_cmd_mark_cancelable(struct io_uring_cmd *cmd,
 /* Execute the request from a blocking context */
 void io_uring_cmd_issue_blocking(struct io_uring_cmd *ioucmd);
 
+int io_uring_cmd_lease_kbuf(struct io_uring_cmd *ioucmd,
+		const struct io_mapped_buf *grp_kbuf);
 #else
 static inline int io_uring_cmd_import_fixed(u64 ubuf, unsigned long len, int rw,
 			      struct iov_iter *iter, void *ioucmd)
@@ -82,6 +84,11 @@ static inline void io_uring_cmd_mark_cancelable(struct io_uring_cmd *cmd,
 static inline void io_uring_cmd_issue_blocking(struct io_uring_cmd *ioucmd)
 {
 }
+static inline int io_uring_cmd_lease_kbuf(struct io_uring_cmd *ioucmd,
+		const struct io_mapped_buf *grp_kbuf)
+{
+	return -EOPNOTSUPP;
+}
 #endif
 
 /*
diff --git a/io_uring/uring_cmd.c b/io_uring/uring_cmd.c
index 40b8b777ba12..5d59183212f6 100644
--- a/io_uring/uring_cmd.c
+++ b/io_uring/uring_cmd.c
@@ -15,6 +15,7 @@
 #include "alloc_cache.h"
 #include "rsrc.h"
 #include "uring_cmd.h"
+#include "kbuf.h"
 
 static struct uring_cache *io_uring_async_get(struct io_kiocb *req)
 {
@@ -175,6 +176,15 @@ void io_uring_cmd_done(struct io_uring_cmd *ioucmd, ssize_t ret, ssize_t res2,
 }
 EXPORT_SYMBOL_GPL(io_uring_cmd_done);
 
+int io_uring_cmd_lease_kbuf(struct io_uring_cmd *ioucmd,
+			    const struct io_mapped_buf *grp_kbuf)
+{
+	struct io_kiocb *req = cmd_to_io_kiocb(ioucmd);
+
+	return io_lease_group_kbuf(req, grp_kbuf);
+}
+EXPORT_SYMBOL_GPL(io_uring_cmd_lease_kbuf);
+
 static int io_uring_cmd_prep_setup(struct io_kiocb *req,
 				   const struct io_uring_sqe *sqe)
 {
-- 
2.47.0

[PATCH V9 7/7] ublk: support leasing io buffer to io_uring

[Ming Lei]

Suopport to lease block IO buffer for userpace to run io_uring operations(FS,
network IO), then ublk zero copy can be supported.

userspace code:

	git clone https://github.com/ublk-org/ublksrv.git -b uring_group

And both loop and nbd zero copy(io_uring send and send zc) are covered.

Performance improvement is quite obvious in big block size test, such as
'loop --buffered_io' perf is doubled in 64KB block test("loop/007 vs
loop/009").

Signed-off-by: Ming Lei <ming.lei@redhat.com>
---
 drivers/block/ublk_drv.c      | 160 ++++++++++++++++++++++++++++++++--
 include/uapi/linux/ublk_cmd.h |  11 ++-
 2 files changed, 161 insertions(+), 10 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 6ba2c1dd1d87..5803c6418d1e 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -51,6 +51,8 @@
 /* private ioctl command mirror */
 #define UBLK_CMD_DEL_DEV_ASYNC	_IOC_NR(UBLK_U_CMD_DEL_DEV_ASYNC)
 
+#define UBLK_IO_PROVIDE_IO_BUF _IOC_NR(UBLK_U_IO_PROVIDE_IO_BUF)
+
 /* All UBLK_F_* have to be included into UBLK_F_ALL */
 #define UBLK_F_ALL (UBLK_F_SUPPORT_ZERO_COPY \
 		| UBLK_F_URING_CMD_COMP_IN_TASK \
@@ -71,6 +73,9 @@ struct ublk_rq_data {
 	struct llist_node node;
 
 	struct kref ref;
+
+	bool allocated_bvec;
+	struct io_mapped_buf buf[0];
 };
 
 struct ublk_uring_cmd_pdu {
@@ -189,11 +194,15 @@ struct ublk_params_header {
 	__u32	types;
 };
 
+static inline struct request *__ublk_check_and_get_req(struct ublk_device *ub,
+		struct ublk_queue *ubq, int tag, size_t offset);
 static bool ublk_abort_requests(struct ublk_device *ub, struct ublk_queue *ubq);
 
 static inline unsigned int ublk_req_build_flags(struct request *req);
 static inline struct ublksrv_io_desc *ublk_get_iod(struct ublk_queue *ubq,
 						   int tag);
+static void ublk_io_buf_giveback_cb(const struct io_mapped_buf *buf);
+
 static inline bool ublk_dev_is_user_copy(const struct ublk_device *ub)
 {
 	return ub->dev_info.flags & UBLK_F_USER_COPY;
@@ -588,6 +597,11 @@ static inline bool ublk_need_req_ref(const struct ublk_queue *ubq)
 	return ublk_support_user_copy(ubq);
 }
 
+static inline bool ublk_support_zc(const struct ublk_queue *ubq)
+{
+	return ubq->flags & UBLK_F_SUPPORT_ZERO_COPY;
+}
+
 static inline void ublk_init_req_ref(const struct ublk_queue *ubq,
 		struct request *req)
 {
@@ -851,6 +865,72 @@ static size_t ublk_copy_user_pages(const struct request *req,
 	return done;
 }
 
+/*
+ * The built command buffer is immutable, so it is fine to feed it to
+ * concurrent io_uring provide buf commands
+ */
+static int ublk_init_zero_copy_buffer(struct request *req)
+{
+	struct ublk_rq_data *data = blk_mq_rq_to_pdu(req);
+	struct io_mapped_buf *imu = data->buf;
+	struct req_iterator rq_iter;
+	unsigned int nr_bvecs = 0;
+	struct bio_vec *bvec;
+	unsigned int offset;
+	struct bio_vec bv;
+
+	if (!ublk_rq_has_data(req))
+		goto exit;
+
+	rq_for_each_bvec(bv, req, rq_iter)
+		nr_bvecs++;
+
+	if (!nr_bvecs)
+		goto exit;
+
+	if (req->bio != req->biotail) {
+		int idx = 0;
+
+		bvec = kvmalloc_array(nr_bvecs, sizeof(struct bio_vec),
+				GFP_NOIO);
+		if (!bvec)
+			return -ENOMEM;
+
+		offset = 0;
+		rq_for_each_bvec(bv, req, rq_iter)
+			bvec[idx++] = bv;
+		data->allocated_bvec = true;
+	} else {
+		struct bio *bio = req->bio;
+
+		offset = bio->bi_iter.bi_bvec_done;
+		bvec = __bvec_iter_bvec(bio->bi_io_vec, bio->bi_iter);
+	}
+	imu->kbuf = 1;
+	imu->pbvec = bvec;
+	imu->nr_bvecs = nr_bvecs;
+	imu->offset = offset;
+	imu->len = blk_rq_bytes(req);
+	imu->dir = req_op(req) == REQ_OP_READ ? ITER_DEST : ITER_SOURCE;
+	imu->kbuf_ack = ublk_io_buf_giveback_cb;
+
+	return 0;
+exit:
+	imu->pbvec = NULL;
+	return 0;
+}
+
+static void ublk_deinit_zero_copy_buffer(struct request *req)
+{
+	struct ublk_rq_data *data = blk_mq_rq_to_pdu(req);
+	struct io_mapped_buf *imu = data->buf;
+
+	if (data->allocated_bvec) {
+		kvfree(imu->pbvec);
+		data->allocated_bvec = false;
+	}
+}
+
 static inline bool ublk_need_map_req(const struct request *req)
 {
 	return ublk_rq_has_data(req) && req_op(req) == REQ_OP_WRITE;
@@ -862,13 +942,25 @@ static inline bool ublk_need_unmap_req(const struct request *req)
 	       (req_op(req) == REQ_OP_READ || req_op(req) == REQ_OP_DRV_IN);
 }
 
-static int ublk_map_io(const struct ublk_queue *ubq, const struct request *req,
+static int ublk_map_io(const struct ublk_queue *ubq, struct request *req,
 		struct ublk_io *io)
 {
 	const unsigned int rq_bytes = blk_rq_bytes(req);
 
-	if (ublk_support_user_copy(ubq))
+	if (ublk_support_user_copy(ubq)) {
+		if (ublk_support_zc(ubq)) {
+			int ret = ublk_init_zero_copy_buffer(req);
+
+			/*
+			 * The only failure is -ENOMEM for allocating providing
+			 * buffer command, return zero so that we can requeue
+			 * this req.
+			 */
+			if (unlikely(ret))
+				return 0;
+		}
 		return rq_bytes;
+	}
 
 	/*
 	 * no zero copy, we delay copy WRITE request data into ublksrv
@@ -886,13 +978,16 @@ static int ublk_map_io(const struct ublk_queue *ubq, const struct request *req,
 }
 
 static int ublk_unmap_io(const struct ublk_queue *ubq,
-		const struct request *req,
+		struct request *req,
 		struct ublk_io *io)
 {
 	const unsigned int rq_bytes = blk_rq_bytes(req);
 
-	if (ublk_support_user_copy(ubq))
+	if (ublk_support_user_copy(ubq)) {
+		if (ublk_support_zc(ubq))
+			ublk_deinit_zero_copy_buffer(req);
 		return rq_bytes;
+	}
 
 	if (ublk_need_unmap_req(req)) {
 		struct iov_iter iter;
@@ -1038,6 +1133,7 @@ static inline void __ublk_complete_rq(struct request *req)
 
 	return;
 exit:
+	ublk_deinit_zero_copy_buffer(req);
 	blk_mq_end_request(req, res);
 }
 
@@ -1680,6 +1776,45 @@ static inline void ublk_prep_cancel(struct io_uring_cmd *cmd,
 	io_uring_cmd_mark_cancelable(cmd, issue_flags);
 }
 
+static void ublk_io_buf_giveback_cb(const struct io_mapped_buf *buf)
+{
+	struct ublk_rq_data *data = container_of(buf, struct ublk_rq_data, buf[0]);
+	struct request *req = blk_mq_rq_from_pdu(data);
+	struct ublk_queue *ubq = req->mq_hctx->driver_data;
+
+	ublk_put_req_ref(ubq, req);
+}
+
+static int ublk_provide_io_buf(struct io_uring_cmd *cmd,
+		struct ublk_queue *ubq, int tag)
+{
+	struct ublk_device *ub = cmd->file->private_data;
+	struct ublk_rq_data *data;
+	struct request *req;
+
+	if (!ub)
+		return -EPERM;
+
+	req = __ublk_check_and_get_req(ub, ubq, tag, 0);
+	if (!req)
+		return -EINVAL;
+
+	pr_devel("%s: qid %d tag %u request bytes %u\n",
+			__func__, tag, ubq->q_id, blk_rq_bytes(req));
+
+	data = blk_mq_rq_to_pdu(req);
+
+	/*
+	 * io_uring guarantees that the callback will be called after
+	 * the provided buffer is consumed, and it is automatic removal
+	 * before this uring command is freed.
+	 *
+	 * This request won't be completed unless the callback is called,
+	 * so ublk module won't be unloaded too.
+	 */
+	return io_uring_cmd_lease_kbuf(cmd, data->buf);
+}
+
 static int __ublk_ch_uring_cmd(struct io_uring_cmd *cmd,
 			       unsigned int issue_flags,
 			       const struct ublksrv_io_cmd *ub_cmd)
@@ -1731,6 +1866,10 @@ static int __ublk_ch_uring_cmd(struct io_uring_cmd *cmd,
 
 	ret = -EINVAL;
 	switch (_IOC_NR(cmd_op)) {
+	case UBLK_IO_PROVIDE_IO_BUF:
+		if (unlikely(!ublk_support_zc(ubq)))
+			goto out;
+		return ublk_provide_io_buf(cmd, ubq, tag);
 	case UBLK_IO_FETCH_REQ:
 		/* UBLK_IO_FETCH_REQ is only allowed before queue is setup */
 		if (ublk_queue_ready(ubq)) {
@@ -2149,11 +2288,14 @@ static void ublk_align_max_io_size(struct ublk_device *ub)
 
 static int ublk_add_tag_set(struct ublk_device *ub)
 {
+	int zc = !!(ub->dev_info.flags & UBLK_F_SUPPORT_ZERO_COPY);
+	struct ublk_rq_data *data;
+
 	ub->tag_set.ops = &ublk_mq_ops;
 	ub->tag_set.nr_hw_queues = ub->dev_info.nr_hw_queues;
 	ub->tag_set.queue_depth = ub->dev_info.queue_depth;
 	ub->tag_set.numa_node = NUMA_NO_NODE;
-	ub->tag_set.cmd_size = sizeof(struct ublk_rq_data);
+	ub->tag_set.cmd_size = struct_size(data, buf, zc);
 	ub->tag_set.flags = BLK_MQ_F_SHOULD_MERGE;
 	ub->tag_set.driver_data = ub;
 	return blk_mq_alloc_tag_set(&ub->tag_set);
@@ -2458,8 +2600,12 @@ static int ublk_ctrl_add_dev(struct io_uring_cmd *cmd)
 		goto out_free_dev_number;
 	}
 
-	/* We are not ready to support zero copy */
-	ub->dev_info.flags &= ~UBLK_F_SUPPORT_ZERO_COPY;
+	/* zero copy depends on user copy */
+	if ((ub->dev_info.flags & UBLK_F_SUPPORT_ZERO_COPY) &&
+			!ublk_dev_is_user_copy(ub)) {
+		ret = -EINVAL;
+		goto out_free_dev_number;
+	}
 
 	ub->dev_info.nr_hw_queues = min_t(unsigned int,
 			ub->dev_info.nr_hw_queues, nr_cpu_ids);
diff --git a/include/uapi/linux/ublk_cmd.h b/include/uapi/linux/ublk_cmd.h
index 12873639ea96..04d73b349709 100644
--- a/include/uapi/linux/ublk_cmd.h
+++ b/include/uapi/linux/ublk_cmd.h
@@ -94,6 +94,8 @@
 	_IOWR('u', UBLK_IO_COMMIT_AND_FETCH_REQ, struct ublksrv_io_cmd)
 #define	UBLK_U_IO_NEED_GET_DATA		\
 	_IOWR('u', UBLK_IO_NEED_GET_DATA, struct ublksrv_io_cmd)
+#define	UBLK_U_IO_PROVIDE_IO_BUF	\
+	_IOWR('u', 0x23, struct ublksrv_io_cmd)
 
 /* only ABORT means that no re-fetch */
 #define UBLK_IO_RES_OK			0
@@ -127,9 +129,12 @@
 #define UBLKSRV_IO_BUF_TOTAL_SIZE	(1ULL << UBLKSRV_IO_BUF_TOTAL_BITS)
 
 /*
- * zero copy requires 4k block size, and can remap ublk driver's io
- * request into ublksrv's vm space
- */
+ * io_uring provide kbuf command based zero copy
+ *
+ * Not available for UBLK_F_UNPRIVILEGED_DEV, because we rely on ublk
+ * server to fill up request buffer for READ IO, and ublk server can't
+ * be trusted in case of UBLK_F_UNPRIVILEGED_DEV.
+*/
 #define UBLK_F_SUPPORT_ZERO_COPY	(1ULL << 0)
 
 /*
-- 
2.47.0

Re: [PATCH V9 2/7] io_uring: rename ubuf of io_mapped_buf as start

[Jens Axboe]

On 11/6/24 5:26 AM, Ming Lei wrote:
> ->ubuf of `io_mapped_buf` stores the start address of userspace fixed
> buffer. `io_mapped_buf` will be extended for covering kernel buffer,
> so rename ->ubuf as ->start.

Minor nit, but I'd probably rename it to 'addr' instead.

-- 
Jens Axboe

Re: [PATCH V9 3/7] io_uring: shrink io_mapped_buf

[Jens Axboe]

On 11/6/24 5:26 AM, Ming Lei wrote:
> `struct io_mapped_buf` will be extended to cover kernel buffer which
> may be in fast IO path, and `struct io_mapped_buf` needs to be per-IO.
> 
> So shrink sizeof(struct io_mapped_buf) by the following ways:
> 
> - folio_shift is < 64, so 6bits are enough to hold it, the remained bits
>   can be used for the coming kernel buffer
> 
> - define `acct_pages` as 'unsigned int', which is big enough for
>   accounting pages in the buffer
> 
> Signed-off-by: Ming Lei <ming.lei@redhat.com>
> ---
>  io_uring/rsrc.c | 2 ++
>  io_uring/rsrc.h | 6 +++---
>  2 files changed, 5 insertions(+), 3 deletions(-)
> 
> diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
> index 9b8827c72230..16f5abe03d10 100644
> --- a/io_uring/rsrc.c
> +++ b/io_uring/rsrc.c
> @@ -685,6 +685,8 @@ static bool io_try_coalesce_buffer(struct page ***pages, int *nr_pages,
>  		return false;
>  
>  	data->folio_shift = folio_shift(folio);
> +	WARN_ON_ONCE(data->folio_shift >= 64);

Since folio_shift is 6 bits, how can that be try?

I think you'd want:

	WARN_ON_ONCE(folio_shift(folio) >= 64);

instead.

And agree that acct_pages doesn't need to be an unsigned long.

-- 
Jens Axboe

Re: [PATCH V9 4/7] io_uring: reuse io_mapped_buf for kernel buffer

[Jens Axboe]

On 11/6/24 5:26 AM, Ming Lei wrote:
> Prepare for supporting kernel buffer in case of io group, in which group
> leader leases kernel buffer to io_uring, and consumed by io_uring OPs.
> 
> So reuse io_mapped_buf for group kernel buffer, and unfortunately
> io_import_fixed() can't be reused since userspace fixed buffer is
> virt-contiguous, but it isn't true for kernel buffer.
> 
> Also kernel buffer lifetime is bound with group leader request, it isn't
> necessary to use rsrc_node for tracking its lifetime, especially it needs
> extra allocation of rsrc_node for each IO.

While it isn't strictly necessary, I do think it'd clean up the io_kiocb
parts and hopefully unify the assign and put path more. So I'd strongly
suggest you do use an io_rsrc_node, even if it does just map the
io_mapped_buf for this.

> +struct io_mapped_buf {
> +	u64		start;
> +	unsigned int	len;
> +	unsigned int	nr_bvecs;
> +
> +	/* kbuf hasn't refs and accounting, its lifetime is bound with req */
> +	union {
> +		struct {
> +			refcount_t	refs;
> +			unsigned int	acct_pages;
> +		};
> +		/* pbvec is only for kbuf */
> +		const struct bio_vec	*pbvec;
> +	};
> +	unsigned int	folio_shift:6;
> +	unsigned int	dir:1;		/* ITER_DEST or ITER_SOURCE */
> +	unsigned int	kbuf:1;		/* kernel buffer or not */
> +	/* offset in the 1st bvec, for kbuf only */
> +	unsigned int	offset;
> +	struct bio_vec	bvec[] __counted_by(nr_bvecs);
> +};

And then I'd get rid of this union, and have it follow the normal rules
for an io_mapped_buf in that the refs are valid. Yes it'll take 8b more,
but honestly I think unifying these bits and keeping it consistent is a
LOT more important than saving a bit of space.

This is imho the last piece missing to make this conform more nicely
with how resource nodes are generally handled and used.

-- 
Jens Axboe

Re: [PATCH V9 5/7] io_uring: support leased group buffer with REQ_F_GROUP_BUF

[Jens Axboe]

On 11/6/24 5:26 AM, Ming Lei wrote:
> @@ -670,6 +689,14 @@ struct io_kiocb {
>  		struct io_buffer_list	*buf_list;
>  
>  		struct io_rsrc_node	*buf_node;
> +
> +		/* valid IFF REQ_F_GROUP_BUF is set */
> +		union {
> +			/* store group buffer for group leader */
> +			const struct io_mapped_buf *grp_buf;
> +			/* for group member */
> +			bool	grp_buf_imported;
> +		};
>  	};

Just add a REQ_F flag for this.

> +/* For group member only */
> +static inline void io_req_mark_group_buf(struct io_kiocb *req, bool imported)
> +{
> +	req->grp_buf_imported = imported;
> +}
> +
> +/* For group member only */
> +static inline bool io_req_group_buf_imported(struct io_kiocb *req)
> +{
> +	return req->grp_buf_imported;
> +}

And kill these useless helpers, should just set or clear the above
mentioned REQ_F flag instead.

-- 
Jens Axboe

Re: [PATCH V9 3/7] io_uring: shrink io_mapped_buf

[Ming Lei]

On Wed, Nov 06, 2024 at 08:09:38AM -0700, Jens Axboe wrote:
> On 11/6/24 5:26 AM, Ming Lei wrote:
> > `struct io_mapped_buf` will be extended to cover kernel buffer which
> > may be in fast IO path, and `struct io_mapped_buf` needs to be per-IO.
> > 
> > So shrink sizeof(struct io_mapped_buf) by the following ways:
> > 
> > - folio_shift is < 64, so 6bits are enough to hold it, the remained bits
> >   can be used for the coming kernel buffer
> > 
> > - define `acct_pages` as 'unsigned int', which is big enough for
> >   accounting pages in the buffer
> > 
> > Signed-off-by: Ming Lei <ming.lei@redhat.com>
> > ---
> >  io_uring/rsrc.c | 2 ++
> >  io_uring/rsrc.h | 6 +++---
> >  2 files changed, 5 insertions(+), 3 deletions(-)
> > 
> > diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
> > index 9b8827c72230..16f5abe03d10 100644
> > --- a/io_uring/rsrc.c
> > +++ b/io_uring/rsrc.c
> > @@ -685,6 +685,8 @@ static bool io_try_coalesce_buffer(struct page ***pages, int *nr_pages,
> >  		return false;
> >  
> >  	data->folio_shift = folio_shift(folio);
> > +	WARN_ON_ONCE(data->folio_shift >= 64);
> 
> Since folio_shift is 6 bits, how can that be try?
> 
> I think you'd want:
> 
> 	WARN_ON_ONCE(folio_shift(folio) >= 64);
> 
> instead.

imu->folio_shift is 6 bits, and it is only copied from data->folio_shift(char),
that is why the warning is added for data->folio_shift.

Thanks,
Ming

Re: [PATCH V9 4/7] io_uring: reuse io_mapped_buf for kernel buffer

[Ming Lei]

On Wed, Nov 06, 2024 at 08:15:13AM -0700, Jens Axboe wrote:
> On 11/6/24 5:26 AM, Ming Lei wrote:
> > Prepare for supporting kernel buffer in case of io group, in which group
> > leader leases kernel buffer to io_uring, and consumed by io_uring OPs.
> > 
> > So reuse io_mapped_buf for group kernel buffer, and unfortunately
> > io_import_fixed() can't be reused since userspace fixed buffer is
> > virt-contiguous, but it isn't true for kernel buffer.
> > 
> > Also kernel buffer lifetime is bound with group leader request, it isn't
> > necessary to use rsrc_node for tracking its lifetime, especially it needs
> > extra allocation of rsrc_node for each IO.
> 
> While it isn't strictly necessary, I do think it'd clean up the io_kiocb
> parts and hopefully unify the assign and put path more. So I'd strongly
> suggest you do use an io_rsrc_node, even if it does just map the
> io_mapped_buf for this.

Can you share your idea about how to unify buffer? I am also interested
in this area, so I may take it into account in this patch.

Will you plan to use io_rsrc_node for all buffer type(include buffer
select)?

> 
> > +struct io_mapped_buf {
> > +	u64		start;
> > +	unsigned int	len;
> > +	unsigned int	nr_bvecs;
> > +
> > +	/* kbuf hasn't refs and accounting, its lifetime is bound with req */
> > +	union {
> > +		struct {
> > +			refcount_t	refs;
> > +			unsigned int	acct_pages;
> > +		};
> > +		/* pbvec is only for kbuf */
> > +		const struct bio_vec	*pbvec;
> > +	};
> > +	unsigned int	folio_shift:6;
> > +	unsigned int	dir:1;		/* ITER_DEST or ITER_SOURCE */
> > +	unsigned int	kbuf:1;		/* kernel buffer or not */
> > +	/* offset in the 1st bvec, for kbuf only */
> > +	unsigned int	offset;
> > +	struct bio_vec	bvec[] __counted_by(nr_bvecs);
> > +};
> 
> And then I'd get rid of this union, and have it follow the normal rules
> for an io_mapped_buf in that the refs are valid. Yes it'll take 8b more,
> but honestly I think unifying these bits and keeping it consistent is a
> LOT more important than saving a bit of space.
> 
> This is imho the last piece missing to make this conform more nicely
> with how resource nodes are generally handled and used.

OK.


thanks,
Ming

Re: [PATCH V9 3/7] io_uring: shrink io_mapped_buf

[Jens Axboe]

On 11/6/24 6:04 PM, Ming Lei wrote:
> On Wed, Nov 06, 2024 at 08:09:38AM -0700, Jens Axboe wrote:
>> On 11/6/24 5:26 AM, Ming Lei wrote:
>>> `struct io_mapped_buf` will be extended to cover kernel buffer which
>>> may be in fast IO path, and `struct io_mapped_buf` needs to be per-IO.
>>>
>>> So shrink sizeof(struct io_mapped_buf) by the following ways:
>>>
>>> - folio_shift is < 64, so 6bits are enough to hold it, the remained bits
>>>   can be used for the coming kernel buffer
>>>
>>> - define `acct_pages` as 'unsigned int', which is big enough for
>>>   accounting pages in the buffer
>>>
>>> Signed-off-by: Ming Lei <ming.lei@redhat.com>
>>> ---
>>>  io_uring/rsrc.c | 2 ++
>>>  io_uring/rsrc.h | 6 +++---
>>>  2 files changed, 5 insertions(+), 3 deletions(-)
>>>
>>> diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
>>> index 9b8827c72230..16f5abe03d10 100644
>>> --- a/io_uring/rsrc.c
>>> +++ b/io_uring/rsrc.c
>>> @@ -685,6 +685,8 @@ static bool io_try_coalesce_buffer(struct page ***pages, int *nr_pages,
>>>  		return false;
>>>  
>>>  	data->folio_shift = folio_shift(folio);
>>> +	WARN_ON_ONCE(data->folio_shift >= 64);
>>
>> Since folio_shift is 6 bits, how can that be try?
>>
>> I think you'd want:
>>
>> 	WARN_ON_ONCE(folio_shift(folio) >= 64);
>>
>> instead.
> 
> imu->folio_shift is 6 bits, and it is only copied from data->folio_shift(char),
> that is why the warning is added for data->folio_shift.

Ah yes that is fine then, for some reason I read that as 'data' being
an io_mapped_buffer.

-- 
Jens Axboe

Re: [PATCH V9 4/7] io_uring: reuse io_mapped_buf for kernel buffer

[Jens Axboe]

On 11/6/24 6:22 PM, Ming Lei wrote:
> On Wed, Nov 06, 2024 at 08:15:13AM -0700, Jens Axboe wrote:
>> On 11/6/24 5:26 AM, Ming Lei wrote:
>>> Prepare for supporting kernel buffer in case of io group, in which group
>>> leader leases kernel buffer to io_uring, and consumed by io_uring OPs.
>>>
>>> So reuse io_mapped_buf for group kernel buffer, and unfortunately
>>> io_import_fixed() can't be reused since userspace fixed buffer is
>>> virt-contiguous, but it isn't true for kernel buffer.
>>>
>>> Also kernel buffer lifetime is bound with group leader request, it isn't
>>> necessary to use rsrc_node for tracking its lifetime, especially it needs
>>> extra allocation of rsrc_node for each IO.
>>
>> While it isn't strictly necessary, I do think it'd clean up the io_kiocb
>> parts and hopefully unify the assign and put path more. So I'd strongly
>> suggest you do use an io_rsrc_node, even if it does just map the
>> io_mapped_buf for this.
> 
> Can you share your idea about how to unify buffer? I am also interested
> in this area, so I may take it into account in this patch.

I just mean use an io_rsrc_node rather than an io_mapped_buf. The node
holds the buf, and then it should not need extra checking. Particularly
with the callback, which I think needs to go in the io_rsrc_node.

Hence I don't think you need to change much. Yes your mapping side will
need to allocate an io_rsrc_node for this, but that's really not too
bad. The benefits would be that the node assignment to an io_kiocb and
putting of the node would follow normal registered buffers.

> Will you plan to use io_rsrc_node for all buffer type(include buffer
> select)?

Probably not - the provided ring buffers are very low overhead and
shared between userspace and the kernel, so it would not make much sense
to shove an io_rsrc_node in between. But for anything else, yeah I'd
love to see it just use the resource node.

-- 
Jens Axboe