Re: [PATCH] io_uring: add io_uring_enter(2) fixed file support

Xiaoguang Wang
2022-03-04T15:17:08Z

hi,

> On 3/4/22 6:39 AM, Xiaoguang Wang wrote:
>> hi,
>>
>>> On 3/3/22 2:19 PM, Jens Axboe wrote:
>>>> On 3/3/22 1:41 PM, Jens Axboe wrote:
>>>>> On 3/3/22 10:18 AM, Jens Axboe wrote:
>>>>>> On 3/3/22 9:31 AM, Jens Axboe wrote:
>>>>>>> On 3/3/22 7:40 AM, Jens Axboe wrote:
>>>>>>>> On 3/3/22 7:36 AM, Jens Axboe wrote:
>>>>>>>>> The only potential oddity here is that the fd passed back is not a
>>>>>>>>> legitimate fd. io_uring does support poll(2) on its file descriptor, so
>>>>>>>>> that could cause some confusion even if I don't think anyone actually
>>>>>>>>> does poll(2) on io_uring.
>>>>>>>> Side note - the only implication here is that we then likely can't make
>>>>>>>> the optimized behavior the default, it has to be an IORING_SETUP_REG
>>>>>>>> flag which tells us that the application is aware of this limitation.
>>>>>>>> Though I guess close(2) might mess with that too... Hmm.
>>>>>>> Not sure I can find a good approach for that. Tried out your patch and
>>>>>>> made some fixes:
>>>>>>>
>>>>>>> - Missing free on final tctx free
>>>>>>> - Rename registered_files to registered_rings
>>>>>>> - Fix off-by-ones in checking max registration count
>>>>>>> - Use kcalloc
>>>>>>> - Rename ENTER_FIXED_FILE -> ENTER_REGISTERED_RING
>>>>>>> - Don't pass in tctx to io_uring_unreg_ringfd()
>>>>>>> - Get rid of forward declaration for adding tctx node
>>>>>>> - Get rid of extra file pointer in io_uring_enter()
>>>>>>> - Fix deadlock in io_ringfd_register()
>>>>>>> - Use io_uring_rsrc_update rather than add a new struct type
>>>>>> - Allow multiple register/unregister instead of enforcing just 1 at the
>>>>>> time
>>>>>> - Check for it really being a ring fd when registering
>>>>>>
>>>>>> For different batch counts, nice improvements are seen. Roughly:
>>>>>>
>>>>>> Batch==1 15% faster
>>>>>> Batch==2 13% faster
>>>>>> Batch==4 11% faster
>>>>>>
>>>>>> This is just in microbenchmarks where the fdget/fdput play a bigger
>>>>>> factor, but it will certainly help with real world situations where
>>>>>> batching is more limited than in benchmarks.
>>>>> In trying this out in applications, I think the better registration API
>>>>> is to allow io_uring to pick the offset. The application doesn't care,
>>>>> it's just a magic integer there. And if we allow io_uring to pick it,
>>>>> then that makes things a lot easier to deal with.
>>>>>
>>>>> For registration, pass in an array of io_uring_rsrc_update structs, just
>>>>> filling in the ring_fd in the data field. Return value is number of ring
>>>>> fds registered, and up->offset now contains the chosen offset for each
>>>>> of them.
>>>>>
>>>>> Unregister is the same struct, but just with offset filled in.
>>>>>
>>>>> For applications using io_uring, which is all of them as far as I'm
>>>>> aware, we can also easily hide this. This means we can get the optimized
>>>>> behavior by default.
>>>> Only complication here is if the ring is shared across threads or
>>>> processes. The thread one can be common, one thread doing submit and one
>>>> doing completions. That one is a bit harder to handle. Without
>>>> inheriting registered ring fds, not sure how it can be handled by
>>>> default. Should probably just introduce a new queue init helper, but
>>>> then that requires application changes to adopt...
>>>>
>>>> Ideas welcome!
>>> Don't see a way to do it by default, so I think it'll have to be opt-in.
>>> We could make it the default if you never shared a ring, which most
>>> people don't do, but we can't easily do so if it's ever shared between
>>> tasks or processes.
>> Before sending this patch, I also have thought about whether we can
>> make this enter_ring_fd be shared between threads and be default
>> feature, and found that it's hard :)
>>
>> 1) first we need to ensure this ring fd registration always can be
>> unregistered, so this cache is tied to io_uring_task, then when
>> thread exits, we can ensure fput called correspondingly.
>>
>> 2) we may also implement a file cache shared between threads in a
>> process, but this may introduce extra overhead, at least need locks
>> to protect modifications to this cache. If this cache is per thread,
>> it doesn't need any synchronizations.
>>
>> So I suggest to make it be just simple and opt-in, and the
>> registration interface seems not complicated, a thread trying to
>> submit sqes can adopt it easily.
> Yes, I pretty much have come to that same conclusion myself...
>
>>> With liburing, if you share, you share the io_uring struct as well. So
>>> it's hard to maintain a new ->enter_ring_fd in there. It'd be doable if
>>> we could reserve real fd == . Which is of course possible
>>> using xarray or similar, but that'll add extra overhead.
>> For liburing, we may need to add fixed file version for helpers which
>> submit sqes or reap cqes, for example, io_uring_submit_fixed(), which
>> passes a enter_ring_fd.
> I'll take a look at liburing and see what we need to do there. I think
> the sanest thing to do here is say that using a registered ring fd means
> you cannot share the ring, ever. And then just have a
> ring->enter_ring_fd which is normally just set to ring_fd when the ring
> is setup, and if you register the ring fd, then we set it to whatever
> the registered value is. Everything calling io_uring_enter() then just
> needs to be modified to use ->enter_ring_fd instead of ->ring_fd.
ok, look forward to use this api.
>
>>> Anyway, current version below. Only real change here is allowing either
>>> specific offset or generated offset, depending on what the
>>> io_uring_rsrc_update->offset is set to. If set to -1U, then io_uring
>>> will find a free offset. If set to anything else, io_uring will use that
>>> index (as long as it's >=0 && < MAX).
>> Seems you forgot to attach the newest version, and also don't see a
>> patch attachment. Finally, thanks for your quick response and many
>> code improvements, really appreciate it.
> Oops, below now. How do you want to handle this patch? It's now a bit of
> a mix of both of our stuff...
Since you have almost rewritten most of my original patch and now it
looks much better, so I would
suggest just adds my Reported-by :)

Regards,
Xiaoguang Wang
>
>
> diff --git a/fs/io_uring.c b/fs/io_uring.c
> index 8f26c4602384..9180f122ecd7 100644
> --- a/fs/io_uring.c
> +++ b/fs/io_uring.c
> @@ -466,6 +466,11 @@ struct io_ring_ctx {
> };
> };
>
> +/*
> + * Arbitrary limit, can be raised if need be
> + */
> +#define IO_RINGFD_REG_MAX 16
> +
> struct io_uring_task {
> /* submission side */
> int cached_refs;
> @@ -481,6 +486,7 @@ struct io_uring_task {
> struct io_wq_work_list task_list;
> struct io_wq_work_list prior_task_list;
> struct callback_head task_work;
> + struct file **registered_rings;
> bool task_running;
> };
>
> @@ -8779,8 +8785,16 @@ static __cold int io_uring_alloc_task_context(struct task_struct *task,
> if (unlikely(!tctx))
> return -ENOMEM;
>
> + tctx->registered_rings = kcalloc(IO_RINGFD_REG_MAX,
> + sizeof(struct file *), GFP_KERNEL);
> + if (unlikely(!tctx->registered_rings)) {
> + kfree(tctx);
> + return -ENOMEM;
> + }
> +
> ret = percpu_counter_init(&tctx->inflight, 0, GFP_KERNEL);
> if (unlikely(ret)) {
> + kfree(tctx->registered_rings);
> kfree(tctx);
> return ret;
> }
> @@ -8789,6 +8803,7 @@ static __cold int io_uring_alloc_task_context(struct task_struct *task,
> if (IS_ERR(tctx->io_wq)) {
> ret = PTR_ERR(tctx->io_wq);
> percpu_counter_destroy(&tctx->inflight);
> + kfree(tctx->registered_rings);
> kfree(tctx);
> return ret;
> }
> @@ -8813,6 +8828,7 @@ void __io_uring_free(struct task_struct *tsk)
> WARN_ON_ONCE(tctx->io_wq);
> WARN_ON_ONCE(tctx->cached_refs);
>
> + kfree(tctx->registered_rings);
> percpu_counter_destroy(&tctx->inflight);
> kfree(tctx);
> tsk->io_uring = NULL;
> @@ -10035,6 +10051,139 @@ void __io_uring_cancel(bool cancel_all)
> io_uring_cancel_generic(cancel_all, NULL);
> }
>
> +void io_uring_unreg_ringfd(void)
> +{
> + struct io_uring_task *tctx = current->io_uring;
> + int i;
> +
> + for (i = 0; i < IO_RINGFD_REG_MAX; i++) {
> + if (tctx->registered_rings[i]) {
> + fput(tctx->registered_rings[i]);
> + tctx->registered_rings[i] = NULL;
> + }
> + }
> +}
> +
> +static int io_ring_add_registered_fd(struct io_uring_task *tctx, int fd,
> + int start, int end)
> +{
> + struct file *file;
> + int offset;
> +
> + for (offset = start; offset < end; offset++) {
> + offset = array_index_nospec(offset, IO_RINGFD_REG_MAX);
> + if (tctx->registered_rings[offset])
> + continue;
> +
> + file = fget(fd);
> + if (!file) {
> + return -EBADF;
> + } else if (file->f_op != &io_uring_fops) {
> + fput(file);
> + return -EOPNOTSUPP;
> + }
> + tctx->registered_rings[offset] = file;
> + return offset;
> + }
> +
> + return -EBUSY;
> +}
> +
> +/*
> + * Register a ring fd to avoid fdget/fdput for each io_uring_enter()
> + * invocation. User passes in an array of struct io_uring_rsrc_update
> + * with ->data set to the ring_fd, and ->offset given for the desired
> + * index. If no index is desired, application may set ->offset == -1U
> + * and we'll find an available index. Returns number of entries
> + * successfully processed, or < 0 on error if none were processed.
> + */
> +static int io_ringfd_register(struct io_ring_ctx *ctx, void __user *__arg,
> + unsigned nr_args)
> +{
> + struct io_uring_rsrc_update __user *arg = __arg;
> + struct io_uring_rsrc_update reg;
> + struct io_uring_task *tctx;
> + int ret, i;
> +
> + if (!nr_args || nr_args > IO_RINGFD_REG_MAX)
> + return -EINVAL;
> +
> + mutex_unlock(&ctx->uring_lock);
> + ret = io_uring_add_tctx_node(ctx);
> + mutex_lock(&ctx->uring_lock);
> + if (ret)
> + return ret;
> +
> + tctx = current->io_uring;
> + for (i = 0; i < nr_args; i++) {
> + int start, end;
> +
> + if (copy_from_user(&reg, &arg[i], sizeof(reg))) {
> + ret = -EFAULT;
> + break;
> + }
> +
> + if (reg.offset == -1U) {
> + start = 0;
> + end = IO_RINGFD_REG_MAX;
> + } else {
> + if (reg.offset >= IO_RINGFD_REG_MAX) {
> + ret = -EINVAL;
> + break;
> + }
> + start = reg.offset;
> + end = start + 1;
> + }
> +
> + ret = io_ring_add_registered_fd(tctx, reg.data, start, end);
> + if (ret < 0)
> + break;
> +
> + reg.offset = ret;
> + if (copy_to_user(&arg[i], &reg, sizeof(reg))) {
> + fput(tctx->registered_rings[reg.offset]);
> + tctx->registered_rings[reg.offset] = NULL;
> + ret = -EFAULT;
> + break;
> + }
> + }
> +
> + return i ? i : ret;
> +}
> +
> +static int io_ringfd_unregister(struct io_ring_ctx *ctx, void __user *__arg,
> + unsigned nr_args)
> +{
> + struct io_uring_rsrc_update __user *arg = __arg;
> + struct io_uring_task *tctx = current->io_uring;
> + struct io_uring_rsrc_update reg;
> + int ret = 0, i;
> +
> + if (!nr_args || nr_args > IO_RINGFD_REG_MAX)
> + return -EINVAL;
> + if (!tctx)
> + return 0;
> +
> + for (i = 0; i < nr_args; i++) {
> + if (copy_from_user(&reg, &arg[i], sizeof(reg))) {
> + ret = -EFAULT;
> + break;
> + }
> + if (reg.offset >= IO_RINGFD_REG_MAX) {
> + ret = -EINVAL;
> + break;
> + }
> +
> + reg.offset = array_index_nospec(reg.offset, IO_RINGFD_REG_MAX);
> + if (tctx->registered_rings[reg.offset]) {
> + fput(tctx->registered_rings[reg.offset]);
> + tctx->registered_rings[reg.offset] = NULL;
> + }
> + }
> +
> + return i ? i : ret;
> +}
> +
> static void *io_uring_validate_mmap_request(struct file *file,
> loff_t pgoff, size_t sz)
> {
> @@ -10165,12 +10314,28 @@ SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
> io_run_task_work();
>
> if (unlikely(flags & ~(IORING_ENTER_GETEVENTS | IORING_ENTER_SQ_WAKEUP |
> - IORING_ENTER_SQ_WAIT | IORING_ENTER_EXT_ARG)))
> + IORING_ENTER_SQ_WAIT | IORING_ENTER_EXT_ARG |
> + IORING_ENTER_REGISTERED_RING)))
> return -EINVAL;
>
> - f = fdget(fd);
> - if (unlikely(!f.file))
> - return -EBADF;
> + /*
> + * Ring fd has been registered via IORING_REGISTER_RING_FDS, we
> + * need only dereference our task private array to find it.
> + */
> + if (flags & IORING_ENTER_REGISTERED_RING) {
> + struct io_uring_task *tctx = current->io_uring;
> +
> + if (fd < 0 || fd >= IO_RINGFD_REG_MAX || !tctx)
> + return -EINVAL;
> + fd = array_index_nospec(fd, IO_RINGFD_REG_MAX);
> + f.file = tctx->registered_rings[fd];
> + if (unlikely(!f.file))
> + return -EBADF;
> + } else {
> + f = fdget(fd);
> + if (unlikely(!f.file))
> + return -EBADF;
> + }
>
> ret = -EOPNOTSUPP;
> if (unlikely(f.file->f_op != &io_uring_fops))
> @@ -10244,7 +10409,8 @@ SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
> out:
> percpu_ref_put(&ctx->refs);
> out_fput:
> - fdput(f);
> + if (!(flags & IORING_ENTER_REGISTERED_RING))
> + fdput(f);
> return submitted ? submitted : ret;
> }
>
> @@ -11134,6 +11300,12 @@ static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
> break;
> ret = io_register_iowq_max_workers(ctx, arg);
> break;
> + case IORING_REGISTER_RING_FDS:
> + ret = io_ringfd_register(ctx, arg, nr_args);
> + break;
> + case IORING_UNREGISTER_RING_FDS:
> + ret = io_ringfd_unregister(ctx, arg, nr_args);
> + break;
> default:
> ret = -EINVAL;
> break;
> diff --git a/include/linux/io_uring.h b/include/linux/io_uring.h
> index 649a4d7c241b..1814e698d861 100644
> --- a/include/linux/io_uring.h
> +++ b/include/linux/io_uring.h
> @@ -9,11 +9,14 @@
> struct sock *io_uring_get_socket(struct file *file);
> void __io_uring_cancel(bool cancel_all);
> void __io_uring_free(struct task_struct *tsk);
> +void io_uring_unreg_ringfd(void);
>
> static inline void io_uring_files_cancel(void)
> {
> - if (current->io_uring)
> + if (current->io_uring) {
> + io_uring_unreg_ringfd();
> __io_uring_cancel(false);
> + }
> }
> static inline void io_uring_task_cancel(void)
> {
> diff --git a/include/uapi/linux/io_uring.h b/include/uapi/linux/io_uring.h
> index 787f491f0d2a..42b2fe84dbcd 100644
> --- a/include/uapi/linux/io_uring.h
> +++ b/include/uapi/linux/io_uring.h
> @@ -257,10 +257,11 @@ struct io_cqring_offsets {
> /*
> * io_uring_enter(2) flags
> */
> -#define IORING_ENTER_GETEVENTS (1U << 0)
> -#define IORING_ENTER_SQ_WAKEUP (1U << 1)
> -#define IORING_ENTER_SQ_WAIT (1U << 2)
> -#define IORING_ENTER_EXT_ARG (1U << 3)
> +#define IORING_ENTER_GETEVENTS (1U << 0)
> +#define IORING_ENTER_SQ_WAKEUP (1U << 1)
> +#define IORING_ENTER_SQ_WAIT (1U << 2)
> +#define IORING_ENTER_EXT_ARG (1U << 3)
> +#define IORING_ENTER_REGISTERED_RING (1U << 4)
>
> /*
> * Passed in for io_uring_setup(2). Copied back with updated info on success
> @@ -325,6 +326,10 @@ enum {
> /* set/get max number of io-wq workers */
> IORING_REGISTER_IOWQ_MAX_WORKERS = 19,
>
> + /* register/unregister io_uring fd with the ring */
> + IORING_REGISTER_RING_FDS = 20,
> + IORING_UNREGISTER_RING_FDS = 21,
> +
> /* this goes last */
> IORING_REGISTER_LAST
> };
>