@@ -330,17 +330,70 @@ ssize_t generic_file_splice_read(struct file *in, loff_t *ppos,
struct iov_iter to;
struct kiocb kiocb;
int ret;
+ struct bio_vec *array = NULL;
+ bool consumer;
+
+ /*
+ * So far, in->fops->splice_read() has to make sure the following
+ * simple page use model works.
+ *
+ * pipe->consumed_by_read is set by the in end of the pipe
+ */
+ if ((flags & SPLICE_F_READ_TO_READ) && pipe->consumed_by_read) {
+ unsigned int head, tail, mask;
+ int nbufs = pipe->max_usage;
+ size_t left = len;
+ int n;
+
+ if (WARN_ON_ONCE(!(flags & SPLICE_F_DIRECT)))
+ return -EINVAL;
+
+ head = pipe->head;
+ tail = pipe->tail;
+ mask = pipe->ring_size - 1;
+
+ array = kcalloc(nbufs, sizeof(struct bio_vec), GFP_KERNEL);
+ if (!array)
+ return -ENOMEM;
+
+ for (n = 0; !pipe_empty(head, tail) && left && n < nbufs;
+ tail++) {
+ struct pipe_buffer *buf = &pipe->bufs[tail & mask];
+ size_t this_len = buf->len;
+
+ /* zero-length bvecs are not supported, skip them */
+ if (!this_len)
+ continue;
+ this_len = min(this_len, left);
+
+ array[n].bv_page = buf->page;
+ array[n].bv_len = this_len;
+ array[n].bv_offset = buf->offset;
+ left -= this_len;
+ n++;
+ }
+
+ consumer = true;
+ iov_iter_bvec(&to, READ, array, n, len - left);
+ } else {
+ /* !consumer means one pipe buf producer */
+ consumer = false;
+ iov_iter_pipe(&to, READ, pipe, len);
+ }
- iov_iter_pipe(&to, READ, pipe, len);
init_sync_kiocb(&kiocb, in);
kiocb.ki_pos = *ppos;
ret = call_read_iter(in, &kiocb, &to);
if (ret > 0) {
*ppos = kiocb.ki_pos;
file_accessed(in);
+
+ if (consumer)
+ splice_dismiss_pipe(pipe, ret);
} else if (ret < 0) {
/* free what was emitted */
- pipe_discard_from(pipe, to.start_head);
+ if (!consumer)
+ pipe_discard_from(pipe, to.start_head);
/*
* callers of ->splice_read() expect -EAGAIN on
* "can't put anything in there", rather than -EFAULT.
@@ -349,6 +402,11 @@ ssize_t generic_file_splice_read(struct file *in, loff_t *ppos,
ret = -EAGAIN;
}
+ if (consumer) {
+ kfree(array);
+ pipe->consumed_by_read = false;
+ }
+
return ret;
}
EXPORT_SYMBOL(generic_file_splice_read);
@@ -782,7 +840,7 @@ static long do_splice_from(struct pipe_inode_info *pipe, struct file *out,
*/
static long do_splice_to(struct file *in, loff_t *ppos,
struct pipe_inode_info *pipe, size_t len,
- unsigned int flags)
+ unsigned int flags, bool consumer)
{
unsigned int p_space;
int ret;
@@ -790,8 +848,12 @@ static long do_splice_to(struct file *in, loff_t *ppos,
if (unlikely(!(in->f_mode & FMODE_READ)))
return -EBADF;
- /* Don't try to read more the pipe has space for. */
- p_space = pipe->max_usage - pipe_occupancy(pipe->head, pipe->tail);
+ if (consumer) /* read is consumer */
+ p_space = pipe_occupancy(pipe->head, pipe->tail);
+ else
+ /* Don't try to read more the pipe has space for. */
+ p_space = pipe->max_usage - pipe_occupancy(pipe->head,
+ pipe->tail);
len = min_t(size_t, len, p_space << PAGE_SHIFT);
ret = rw_verify_area(READ, in, ppos, len);
@@ -875,7 +937,7 @@ ssize_t splice_direct_to_actor(struct file *in, struct splice_desc *sd,
size_t read_len;
loff_t pos = sd->pos, prev_pos = pos;
- ret = do_splice_to(in, &pos, pipe, len, flags);
+ ret = do_splice_to(in, &pos, pipe, len, flags, false);
if (unlikely(ret <= 0))
goto out_release;
@@ -992,6 +1054,76 @@ long do_splice_direct(struct file *in, loff_t *ppos, struct file *out,
}
EXPORT_SYMBOL(do_splice_direct);
+static int direct_splice_read_consumer_actor(struct pipe_inode_info *pipe,
+ struct splice_desc *sd)
+{
+ struct file *file = sd->u.file;
+
+ /* Pipe in side has to notify us by ->consumed_by_read */
+ if (!pipe->consumed_by_read)
+ return -EINVAL;
+
+ return do_splice_to(file, sd->opos, pipe, sd->total_len,
+ sd->flags, true);
+}
+
+/**
+ * do_splice_direct_read_consumer - splices data directly with producer/
+ * consumer model
+ * @in: file to splice from
+ * @ppos: input file offset
+ * @out: file to splice to
+ * @opos: output file offset
+ * @len: number of bytes to splice
+ * @flags: splice modifier flags, SPLICE_F_READ_TO_READ is required
+ *
+ * Description:
+ * For use by ublk or fuse to implement zero copy for READ request, and
+ * splice directly over internal pipe from device to file, and device's
+ * ->splice_read() produces pipe buffers, and file's ->splice_read()
+ * consumes the buffers.
+ *
+ */
+long do_splice_direct_read_consumer(struct file *in, loff_t *ppos,
+ struct file *out, loff_t *opos,
+ size_t len, unsigned int flags)
+{
+ struct splice_desc sd = {
+ .len = len,
+ .total_len = len,
+ .flags = flags,
+ .pos = *ppos,
+ .u.file = out,
+ .opos = opos,
+ };
+ long ret;
+
+ if (!(flags & (SPLICE_F_DIRECT | SPLICE_F_READ_TO_READ)))
+ return -EINVAL;
+
+ if (unlikely(!(out->f_mode & FMODE_READ)))
+ return -EBADF;
+
+ /*
+ * So far we just support F_READ_TO_READ if it is one plain
+ * file which ->splice_read points to generic_file_splice_read
+ */
+ if (out->f_op->splice_read != generic_file_splice_read)
+ return -EINVAL;
+
+ ret = rw_verify_area(READ, out, opos, len);
+ if (unlikely(ret < 0))
+ return ret;
+
+ ret = splice_direct_to_actor(in, &sd,
+ direct_splice_read_consumer_actor);
+ if (ret > 0)
+ *ppos = sd.pos;
+
+ return ret;
+}
+EXPORT_SYMBOL(do_splice_direct_read_consumer);
+
static int wait_for_space(struct pipe_inode_info *pipe, unsigned flags)
{
for (;;) {
@@ -1023,7 +1155,7 @@ long splice_file_to_pipe(struct file *in,
pipe_lock(opipe);
ret = wait_for_space(opipe, flags);
if (!ret)
- ret = do_splice_to(in, offset, opipe, len, flags);
+ ret = do_splice_to(in, offset, opipe, len, flags, false);
pipe_unlock(opipe);
if (ret > 0)
wakeup_pipe_readers(opipe);
@@ -3169,6 +3169,8 @@ extern ssize_t generic_splice_sendpage(struct pipe_inode_info *pipe,
struct file *out, loff_t *, size_t len, unsigned int flags);
extern long do_splice_direct(struct file *in, loff_t *ppos, struct file *out,
loff_t *opos, size_t len, unsigned int flags);
+extern long do_splice_direct_read_consumer(struct file *in, loff_t *ppos,
+ struct file *out, loff_t *opos, size_t len, unsigned int flags);
extern void
@@ -72,6 +72,15 @@ struct pipe_inode_info {
unsigned int r_counter;
unsigned int w_counter;
bool poll_usage;
+
+ /*
+ * If SPLICE_F_READ_TO_READ is applied, in->fops->splice_read()
+ * should set this flag, so that out->fops->splice_read() can
+ * observe this flag, then consume buffers in the pipe.
+ *
+ * Used by do_splice_direct_read_consumer() only.
+ */
+ bool consumed_by_read;
struct page *tmp_page;
struct fasync_struct *fasync_readers;
struct fasync_struct *fasync_writers;
@@ -26,6 +26,17 @@
/* used for io_uring interface only */
#define SPLICE_F_DIRECT (0x10) /* direct splice and user needn't provide pipe */
+/*
+ * The usual splice is file-to-pipe and pipe-to-file, and this flag means the
+ * splice is file-to-pipe and file-to-pipe. Looks this way is stupid, but
+ * please understand from producer & consumer viewpoint, the 1st file-to-pipe
+ * is producer, and the 2nd file-to-pipe is consumer, so here the 2nd
+ * ->slice_read just consumes buffers stored in the pipe.
+ *
+ * And this flag is only valid in case of SPLICE_F_DIRECT.
+ */
+#define SPLICE_F_READ_TO_READ (0x20)
+
/*
* Passed to the actors
*/
@@ -28,7 +28,7 @@ static int __io_splice_prep(struct io_kiocb *req,
{
struct io_splice *sp = io_kiocb_to_cmd(req, struct io_splice);
unsigned int valid_flags = SPLICE_F_FD_IN_FIXED | SPLICE_F_ALL |
- SPLICE_F_DIRECT;
+ SPLICE_F_DIRECT | SPLICE_F_READ_TO_READ;
sp->len = READ_ONCE(sqe->len);
sp->flags = READ_ONCE(sqe->splice_flags);
@@ -111,12 +111,15 @@ int io_splice(struct io_kiocb *req, unsigned int issue_flags)
poff_out = (sp->off_out == -1) ? NULL : &sp->off_out;
if (sp->len) {
- if (flags & SPLICE_F_DIRECT)
- ret = do_splice_direct(in, poff_in, out, poff_out,
- sp->len, flags);
- else
+ if (!(flags & (SPLICE_F_DIRECT | SPLICE_F_READ_TO_READ)))
ret = do_splice(in, poff_in, out, poff_out, sp->len,
flags);
+ else if (flags & SPLICE_F_READ_TO_READ)
+ ret = do_splice_direct_read_consumer(in, poff_in, out,
+ poff_out, sp->len, flags);
+ else
+ ret = do_splice_direct(in, poff_in, out, poff_out,
+ sp->len, flags);
}
if (!(sp->flags & SPLICE_F_FD_IN_FIXED))