io_uring: add IORING_REGISTER_COPY_BUFFERS method

Buffers can get registered with io_uring, which allows to skip the
repeated pin_pages, unpin/unref pages for each O_DIRECT operation. This
reduces the overhead of O_DIRECT IO.

However, registrering buffers can take some time. Normally this isn't an
issue as it's done at initialization time (and hence less critical), but
for cases where rings can be created and destroyed as part of an IO
thread pool, registering the same buffers for multiple rings become a
more time sensitive proposition. As an example, let's say an application
has an IO memory pool of 500G. Initial registration takes:

Got 500 huge pages (each 1024MB)
Registered 500 pages in 409 msec

or about 0.4 seconds. If we go higher to 900 1GB huge pages being
registered:

Registered 900 pages in 738 msec

which is, as expected, a fully linear scaling.

Rather than have each ring pin/map/register the same buffer pool,
provide an io_uring_register(2) opcode to simply duplicate the buffers
that are registered with another ring. Adding the same 900GB of
registered buffers to the target ring can then be accomplished in:

Copied 900 pages in 17 usec

While timing differs a bit, this provides around a 25,000-40,000x
speedup for this use case.

Signed-off-by: Jens Axboe <axboe@kernel.dk>

diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
index a275f91d2ac0851ee60628ba46cf8753967e6e49..9dc5bb428c8a332df541e053b7affb9bb63cdfac 100644 (file)
--- a/include/uapi/linux/io_uring.h
+++ b/include/uapi/linux/io_uring.h
@@ -609,6 +609,9 @@ enum io_uring_register_op {
 
        IORING_REGISTER_CLOCK                   = 29,
 
+       /* copy registered buffers from source ring to current ring */
+       IORING_REGISTER_COPY_BUFFERS            = 30,
+
        /* this goes last */
        IORING_REGISTER_LAST,
 
@@ -694,6 +697,16 @@ struct io_uring_clock_register {
        __u32   __resv[3];
 };
 
+enum {
+       IORING_REGISTER_SRC_REGISTERED = 1,
+};
+
+struct io_uring_copy_buffers {
+       __u32   src_fd;
+       __u32   flags;
+       __u32   pad[6];
+};
+
 struct io_uring_buf {
        __u64   addr;
        __u32   len;

diff --git a/io_uring/register.c b/io_uring/register.c
index d90159478045c1a243ac6158b986cc7727aca2c9..dab0f8024ddf35e998e6229df933c59cd946a1ce 100644 (file)
--- a/io_uring/register.c
+++ b/io_uring/register.c
@@ -542,6 +542,12 @@ static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
                        break;
                ret = io_register_clock(ctx, arg);
                break;
+       case IORING_REGISTER_COPY_BUFFERS:
+               ret = -EINVAL;
+               if (!arg || nr_args != 1)
+                       break;
+               ret = io_register_copy_buffers(ctx, arg);
+               break;
        default:
                ret = -EINVAL;
                break;

diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
index 28f98de3c3042a25f88e5c824c86b4ebded938a7..40696a395f0afd1668bc6d17001e6535cc971212 100644 (file)
--- a/io_uring/rsrc.c
+++ b/io_uring/rsrc.c
@@ -17,6 +17,7 @@
 #include "openclose.h"
 #include "rsrc.h"
 #include "memmap.h"
+#include "register.h"
 
 struct io_rsrc_update {
        struct file                     *file;
@@ -1137,3 +1138,93 @@ int io_import_fixed(int ddir, struct iov_iter *iter,
 
        return 0;
 }
+
+static int io_copy_buffers(struct io_ring_ctx *ctx, struct io_ring_ctx *src_ctx)
+{
+       struct io_mapped_ubuf **user_bufs;
+       struct io_rsrc_data *data;
+       int i, ret, nbufs;
+
+       /*
+        * Drop our own lock here. We'll setup the data we need and reference
+        * the source buffers, then re-grab, check, and assign at the end.
+        */
+       mutex_unlock(&ctx->uring_lock);
+
+       mutex_lock(&src_ctx->uring_lock);
+       ret = -ENXIO;
+       nbufs = src_ctx->nr_user_bufs;
+       if (!nbufs)
+               goto out_unlock;
+       ret = io_rsrc_data_alloc(ctx, IORING_RSRC_BUFFER, NULL, nbufs, &data);
+       if (ret)
+               goto out_unlock;
+
+       ret = -ENOMEM;
+       user_bufs = kcalloc(nbufs, sizeof(*ctx->user_bufs), GFP_KERNEL);
+       if (!user_bufs)
+               goto out_free_data;
+
+       for (i = 0; i < nbufs; i++) {
+               struct io_mapped_ubuf *src = src_ctx->user_bufs[i];
+
+               refcount_inc(&src->refs);
+               user_bufs[i] = src;
+       }
+
+       /* Have a ref on the bufs now, drop src lock and re-grab our own lock */
+       mutex_unlock(&src_ctx->uring_lock);
+       mutex_lock(&ctx->uring_lock);
+       if (!ctx->user_bufs) {
+               ctx->user_bufs = user_bufs;
+               ctx->buf_data = data;
+               ctx->nr_user_bufs = nbufs;
+               return 0;
+       }
+
+       /* someone raced setting up buffers, dump ours */
+       for (i = 0; i < nbufs; i++)
+               io_buffer_unmap(ctx, &user_bufs[i]);
+       io_rsrc_data_free(data);
+       kfree(user_bufs);
+       return -EBUSY;
+out_free_data:
+       io_rsrc_data_free(data);
+out_unlock:
+       mutex_unlock(&src_ctx->uring_lock);
+       mutex_lock(&ctx->uring_lock);
+       return ret;
+}
+
+/*
+ * Copy the registered buffers from the source ring whose file descriptor
+ * is given in the src_fd to the current ring. This is identical to registering
+ * the buffers with ctx, except faster as mappings already exist.
+ *
+ * Since the memory is already accounted once, don't account it again.
+ */
+int io_register_copy_buffers(struct io_ring_ctx *ctx, void __user *arg)
+{
+       struct io_uring_copy_buffers buf;
+       bool registered_src;
+       struct file *file;
+       int ret;
+
+       if (ctx->user_bufs || ctx->nr_user_bufs)
+               return -EBUSY;
+       if (copy_from_user(&buf, arg, sizeof(buf)))
+               return -EFAULT;
+       if (buf.flags & ~IORING_REGISTER_SRC_REGISTERED)
+               return -EINVAL;
+       if (memchr_inv(buf.pad, 0, sizeof(buf.pad)))
+               return -EINVAL;
+
+       registered_src = (buf.flags & IORING_REGISTER_SRC_REGISTERED) != 0;
+       file = io_uring_register_get_file(buf.src_fd, registered_src);
+       if (IS_ERR(file))
+               return PTR_ERR(file);
+       ret = io_copy_buffers(ctx, file->private_data);
+       if (!registered_src)
+               fput(file);
+       return ret;
+}

diff --git a/io_uring/rsrc.h b/io_uring/rsrc.h
index 98a253172c2739d4a71955d627b1f90ac0ffdbb2..93546ab337a687118a26f087238e22deefd39d4e 100644 (file)
--- a/io_uring/rsrc.h
+++ b/io_uring/rsrc.h
@@ -68,6 +68,7 @@ int io_import_fixed(int ddir, struct iov_iter *iter,
                           struct io_mapped_ubuf *imu,
                           u64 buf_addr, size_t len);
 
+int io_register_copy_buffers(struct io_ring_ctx *ctx, void __user *arg);
 void __io_sqe_buffers_unregister(struct io_ring_ctx *ctx);
 int io_sqe_buffers_unregister(struct io_ring_ctx *ctx);
 int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,