Support for polled aio
Subsystems (11)
Jens Axboe
maintainerv3 cover Support for polled aio
For the grand introduction to this feature, see my original posting here:
https://lore.kernel.org/linux-block/[email protected]/
The patchset continues to evolve, and has grown some optimizations that benefit non-polled aio as well. One such example is the user mapped iocbs, which avoid copying a full iocb for each IO. If you look at memory performance, we copy 72b for each IO at submission time (8 byte pointer, 64 byte iocb), then copy back 16 bytes at completion for the io_event. That's 88 bytes of memory copies, for something that's potentially a 512b IO.
In terms of observed performance, I've been able to do 1070K 4k IOPS with this on a single device, using a single thread for both IO submissions and completions. For actual real life use cases, on a flash storage box permance is up 25% in terms of IOPS/throughput.
As before, patches are on top of my mq-perf branch
Changes since v2:
- Fix an off-by-one causing QD=1 aio polls to sometimes stall.
- Add fget_many/fput_many optimization.
- Ensure that io_cancel() is never called on a polled iocb.
- Add support for user mapped iocbs.
- IOCTX_FLAG_IOPOLL is now (1 << 1), since it was reordered after the user iocb support.
- Kill full aio_kiocb zero on alloc.
- Move fops->iopoll() storage back to the iocb (->ki_cookie). The usage in the dio private storage was inherently racy, making the store of the submission cookie open to use-after-free.
- Fix narrow use-before-submit race in fops->iopoll() access.
- Batch move submitted iocbs to poll_submitted list.
- Pass back EAGAIN at submission time, don't punt to getevents() time.
- Split some patches up.
- Address review concerns.
- Various little fixes and optimizations.
Documentation/filesystems/vfs.txt | 3 + arch/x86/entry/syscalls/syscall_64.tbl | 1 + fs/aio.c | 841 +++++++++++++++++++++---- fs/block_dev.c | 20 +- fs/file.c | 15 +- fs/file_table.c | 10 +- fs/gfs2/file.c | 2 + fs/iomap.c | 51 +- fs/xfs/xfs_file.c | 1 + include/linux/file.h | 2 + include/linux/fs.h | 5 +- include/linux/iomap.h | 1 + include/linux/syscalls.h | 2 + include/uapi/asm-generic/unistd.h | 4 +- include/uapi/linux/aio_abi.h | 5 + kernel/sys_ni.c | 1 + 16 files changed, 819 insertions(+), 145 deletions(-)
Signature · 1 line
Jens Axboe
Jens Axboe
maintainerPATCH 1/20 aio: fix failure to put the file pointer
If the ioprio capability check fails, we return without putting the file pointer.
Fixes d9a08a9e616b ("fs: Add aio iopriority support")
Reviewed-by Christoph Hellwig <[email protected]>
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 1 + 1 file changed, 1 insertion(+)
fs/aio.c +1 -0
@@ -1436,6 +1436,7 @@ static int aio_prep_rw(struct kiocb *req, struct iocb *iocb) ret =ioprio_check_cap (iocb->aio_reqprio);if (ret) {pr_debug ("aio ioprio check cap error: %d\n" , ret); +fput (req->ki_filp);return ret; }
Signature · 1 line
2.17.1
Re: [PATCH 01/20] aio: fix failure to put the file pointer
Reivewed-by Johannes Thumshirn <[email protected]>
Signature · 6 lines
Johannes Thumshirn SUSE Labs [email protected] +49 911 74053 689 SUSE LINUX GmbH, Maxfeldstr. 5, 90409 Nürnberg GF: Felix Imendörffer, Jane Smithard, Graham Norton HRB 21284 (AG Nürnberg) Key fingerprint = EC38 9CAB C2C4 F25D 8600 D0D0 0393 969D 2D76 0850
Jens Axboe
maintainerPATCH 2/20 aio: clear IOCB_HIPRI
In reply to Jens AxboeFrom: Christoph Hellwig <[email protected]>
No one is going to poll for aio (yet), so we must clear the HIPRI flag, as we would otherwise send it down the poll queues, where no one will be polling for completions.
Signed-off-by Christoph Hellwig <[email protected]>
IOCB_HIPRI, not RWF_HIPRI.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-)
fs/aio.c +8 -3
@@ -1436,8 +1436,7 @@ static int aio_prep_rw(struct kiocb *req, struct iocb *iocb) ret =ioprio_check_cap (iocb->aio_reqprio);if (ret) {pr_debug ("aio ioprio check cap error: %d\n" , ret); -fput (req->ki_filp); -return ret; + goto out_fput; } req->ki_ioprio = iocb->aio_reqprio;
@@ -1446,7 +1445,13 @@ static int aio_prep_rw(struct kiocb *req, struct iocb *iocb) ret =kiocb_set_rw_flags (req, iocb->aio_rw_flags);if (unlikely (ret)) -fput (req->ki_filp); + goto out_fput; + + req->ki_flags &= ~IOCB_HIPRI ;/* no one is going to poll for this I/O */ +return 0 ; + +out_fput: +fput (req->ki_filp);return ret; }
Signature · 1 line
2.17.1
Re: [PATCH 02/20] aio: clear IOCB_HIPRI
Reviewed-by Johannes Thumshirn <[email protected]>
Signature · 6 lines
Johannes Thumshirn SUSE Labs [email protected] +49 911 74053 689 SUSE LINUX GmbH, Maxfeldstr. 5, 90409 Nürnberg GF: Felix Imendörffer, Jane Smithard, Graham Norton HRB 21284 (AG Nürnberg) Key fingerprint = EC38 9CAB C2C4 F25D 8600 D0D0 0393 969D 2D76 0850
Jens Axboe
maintainerPATCH 3/20 fs: add an iopoll method to struct file_operations
In reply to Jens AxboeFrom: Christoph Hellwig <[email protected]>
This new methods is used to explicitly poll for I/O completion for an iocb. It must be called for any iocb submitted asynchronously (that is with a non-null ki_complete) which has the IOCB_HIPRI flag set.
The method is assisted by a new ki_cookie field in struct iocb to store the polling cookie.
TODO: we can probably union ki_cookie with the existing hint and I/O priority fields to avoid struct kiocb growth.
Signed-off-by Christoph Hellwig <[email protected]>
Signed-off-by Jens Axboe <[email protected]>
--- Documentation/filesystems/vfs.txt | 3 +++ include/linux/fs.h | 2 ++ 2 files changed, 5 insertions(+)
Documentation/filesystems/vfs.txt +3 -0
@@ -857,6 +857,7 @@ struct file_operations { ssize_t (*write) (struct file *, const char __user *, size_t, loff_t *); ssize_t (*read_iter) (struct kiocb *, struct iov_iter *); ssize_t (*write_iter) (struct kiocb *, struct iov_iter *); + int (*iopoll)(struct kiocb *kiocb, bool spin); int (*iterate) (struct file *, struct dir_context *); int (*iterate_shared) (struct file *, struct dir_context *); __poll_t (*poll) (struct file *, struct poll_table_struct *);
@@ -902,6 +903,8 @@ otherwise noted. write_iter: possibly asynchronous write with iov_iter as source + iopoll: called when aio wants to poll for completions on HIPRI iocbs + iterate: called when the VFS needs to read the directory contents iterate_shared: called when the VFS needs to read the directory contents
include/linux/fs.h +2 -0
@@ -310,6 +310,7 @@ struct kiocb {int ki_flags;u16 ki_hint;u16 ki_ioprio;/* See linux/ioprio.h */ +unsigned int ki_cookie;/* for ->iopoll */ } __randomize_layout;static inline bool is_sync_kiocb (struct kiocb *kiocb)
@@ -1781,6 +1782,7 @@ struct file_operations {ssize_t (*write) (struct file *,const char __user *,size_t ,loff_t *);ssize_t (*read_iter) (struct kiocb *,struct iov_iter *);ssize_t (*write_iter) (struct kiocb *,struct iov_iter *); +int (*iopoll)(struct kiocb *kiocb,bool spin);int (*iterate) (struct file *,struct dir_context *);int (*iterate_shared) (struct file *,struct dir_context *);__poll_t (*poll) (struct file *,struct poll_table_struct *);
Signature · 1 line
2.17.1
Re: [PATCH 03/20] fs: add an iopoll method to struct file_operations
Reviewed-by Johannes Thumshirn <[email protected]>
Signature · 6 lines
Johannes Thumshirn SUSE Labs [email protected] +49 911 74053 689 SUSE LINUX GmbH, Maxfeldstr. 5, 90409 Nürnberg GF: Felix Imendörffer, Jane Smithard, Graham Norton HRB 21284 (AG Nürnberg) Key fingerprint = EC38 9CAB C2C4 F25D 8600 D0D0 0393 969D 2D76 0850
Jens Axboe
maintainerPATCH 4/20 block: wire up block device iopoll method
In reply to Jens AxboeFrom: Christoph Hellwig <[email protected]>
Just call blk_poll on the iocb cookie, we can derive the block device from the inode trivially.
Signed-off-by Christoph Hellwig <[email protected]>
Signed-off-by Jens Axboe <[email protected]>
--- fs/block_dev.c | 10 ++++++++++ 1 file changed, 10 insertions(+)
fs/block_dev.c +10 -0
@@ -281,6 +281,14 @@ struct blkdev_dio {static struct bio_set blkdev_dio_pool; +static int blkdev_iopoll (struct kiocb *kiocb,bool wait) +{ +struct block_device *bdev =I_BDEV (kiocb->ki_filp->f_mapping->host); +struct request_queue *q =bdev_get_queue (bdev); + +return blk_poll (q,READ_ONCE (kiocb->ki_cookie), wait); +} +static void blkdev_bio_end_io (struct bio *bio) {struct blkdev_dio *dio = bio->bi_private;
@@ -396,6 +404,7 @@ __blkdev_direct_IO(struct kiocb *iocb, struct iov_iter *iter, int nr_pages) bio->bi_opf |=REQ_HIPRI ; qc =submit_bio (bio); +WRITE_ONCE (iocb->ki_cookie, qc);break ; }
@@ -2061,6 +2070,7 @@ const struct file_operations def_blk_fops = { .llseek = block_llseek, .read_iter = blkdev_read_iter, .write_iter = blkdev_write_iter, + .iopoll = blkdev_iopoll, .mmap = generic_file_mmap, .fsync = blkdev_fsync, .unlocked_ioctl = block_ioctl,
Signature · 1 line
2.17.1
Re: [PATCH 04/20] block: wire up block device iopoll method
Reviewed-by Johannes Thumshirn <[email protected]>
Signature · 6 lines
Johannes Thumshirn SUSE Labs [email protected] +49 911 74053 689 SUSE LINUX GmbH, Maxfeldstr. 5, 90409 Nürnberg GF: Felix Imendörffer, Jane Smithard, Graham Norton HRB 21284 (AG Nürnberg) Key fingerprint = EC38 9CAB C2C4 F25D 8600 D0D0 0393 969D 2D76 0850
Jens Axboe
maintainerPATCH 5/20 block: ensure that async polled IO is marked REQ_NOWAIT
In reply to Jens AxboeWe can't wait for polled events to complete, as they may require active polling from whoever submitted it. If that is the same task that is submitting new IO, we could deadlock waiting for IO to complete that this task is supposed to be completing itself.
Signed-off-by Jens Axboe <[email protected]>
--- fs/block_dev.c | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-)
fs/block_dev.c +9 -1
@@ -400,8 +400,16 @@ __blkdev_direct_IO(struct kiocb *iocb, struct iov_iter *iter, int nr_pages) nr_pages =iov_iter_npages (iter,BIO_MAX_PAGES );if (!nr_pages) { -if (iocb->ki_flags &IOCB_HIPRI ) +if (iocb->ki_flags &IOCB_HIPRI ) { bio->bi_opf |=REQ_HIPRI ; +/* +* For async polled IO, we can't wait for +* requests to complete, as they may also be +* polled and require active reaping. +*/ +if (!is_sync) + bio->bi_opf |=REQ_NOWAIT ; + } qc =submit_bio (bio);WRITE_ONCE (iocb->ki_cookie, qc);
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 6/20 iomap: wire up the iopoll method
In reply to Jens AxboeFrom: Christoph Hellwig <[email protected]>
Store the request queue the last bio was submitted to in the iocb private data in addition to the cookie so that we find the right block device. Also refactor the common direct I/O bio submission code into a nice little helper.
Signed-off-by Christoph Hellwig <[email protected]>
Signed-off-by Jens Axboe <[email protected]>
--- fs/gfs2/file.c | 2 ++ fs/iomap.c | 43 ++++++++++++++++++++++++++++--------------- fs/xfs/xfs_file.c | 1 + include/linux/iomap.h | 1 + 4 files changed, 32 insertions(+), 15 deletions(-)
fs/gfs2/file.c +2 -0
@@ -1280,6 +1280,7 @@ const struct file_operations gfs2_file_fops = { .llseek = gfs2_llseek, .read_iter = gfs2_file_read_iter, .write_iter = gfs2_file_write_iter, + .iopoll = iomap_dio_iopoll, .unlocked_ioctl = gfs2_ioctl, .mmap = gfs2_mmap, .open = gfs2_open,
@@ -1310,6 +1311,7 @@ const struct file_operations gfs2_file_fops_nolock = { .llseek = gfs2_llseek, .read_iter = gfs2_file_read_iter, .write_iter = gfs2_file_write_iter, + .iopoll = iomap_dio_iopoll, .unlocked_ioctl = gfs2_ioctl, .mmap = gfs2_mmap, .open = gfs2_open,
fs/iomap.c +28 -15
82 lines not on this page · raw message
fs/xfs/xfs_file.c +1 -0
@@ -1203,6 +1203,7 @@ const struct file_operations xfs_file_operations = { .write_iter = xfs_file_write_iter, .splice_read = generic_file_splice_read, .splice_write = iter_file_splice_write, + .iopoll = iomap_dio_iopoll, .unlocked_ioctl = xfs_file_ioctl, #ifdef CONFIG_COMPAT .compat_ioctl = xfs_file_compat_ioctl,
include/linux/iomap.h +1 -0
@@ -162,6 +162,7 @@ typedef int (iomap_dio_end_io_t)(struct kiocb *iocb, ssize_t ret, unsigned flags);ssize_t iomap_dio_rw (struct kiocb *iocb,struct iov_iter *iter,const struct iomap_ops *ops,iomap_dio_end_io_t end_io); +int iomap_dio_iopoll (struct kiocb *kiocb,bool spin);#ifdef CONFIG_SWAP struct file ;
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 7/20 iomap: ensure that async polled IO is marked REQ_NOWAIT
In reply to Jens AxboeWe can't wait for polled events to complete, as they may require active polling from whoever submitted it. If that is the same task that is submitting new IO, we could deadlock waiting for IO to complete that this task is supposed to be completing itself.
Signed-off-by Jens Axboe <[email protected]>
--- fs/iomap.c | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-)
fs/iomap.c +9 -1
@@ -1451,8 +1451,16 @@ static void iomap_dio_submit_bio(struct iomap_dio *dio, struct iomap *iomap, {atomic_inc (&dio->ref); -if (dio->iocb->ki_flags &IOCB_HIPRI ) +if (dio->iocb->ki_flags &IOCB_HIPRI ) { bio->bi_opf |=REQ_HIPRI ; +/* +* For async polled IO, we can't wait for requests to +* complete, as they may also be polled and require active +* reaping. +*/ +if (!dio->wait_for_completion) + bio->bi_opf |=REQ_NOWAIT ; + } dio->submit.last_queue =bdev_get_queue (iomap->bdev); dio->submit.cookie =submit_bio (bio);
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 8/20 aio: use assigned completion handler
In reply to Jens AxboeWe know this is a read/write request, but in preparation for having different kinds of those, ensure that we call the assigned handler instead of assuming it's aio_complete_rq().
Reviewed-by Christoph Hellwig <[email protected]>
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-)
fs/aio.c +1 -1
@@ -1490,7 +1490,7 @@ static inline void aio_rw_done(struct kiocb *req, ssize_t ret) ret = -EINTR ;/*FALLTHRU*/ default : -aio_complete_rw (req, ret,0 ); + req->ki_complete (req, ret,0 ); } }
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 9/20 aio: separate out ring reservation from req allocation
In reply to Jens AxboeThis is in preparation for certain types of IO not needing a ring reserveration.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-)
fs/aio.c +17 -13
@@ -901,7 +901,7 @@ static void put_reqs_available(struct kioctx *ctx, unsigned nr)local_irq_restore (flags); } -static bool get_reqs_available (struct kioctx *ctx) +static bool __get_reqs_available (struct kioctx *ctx) {struct kioctx_cpu *kcpu;bool ret = false;
@@ -993,6 +993,14 @@ static void user_refill_reqs_available(struct kioctx *ctx)spin_unlock_irq (&ctx->completion_lock); } +static bool get_reqs_available (struct kioctx *ctx) +{ +if (__get_reqs_available (ctx)) +return true; +user_refill_reqs_available (ctx); +return __get_reqs_available (ctx); +} +/* aio_get_req * Allocate a slot for an aio request. * Returns NULL if no requests are free.
@@ -1001,24 +1009,15 @@ static inline struct aio_kiocb *aio_get_req(struct kioctx *ctx) {struct aio_kiocb *req; -if (!get_reqs_available (ctx)) { -user_refill_reqs_available (ctx); -if (!get_reqs_available (ctx)) -return NULL ; - } - req =kmem_cache_alloc (kiocb_cachep,GFP_KERNEL |__GFP_ZERO);if (unlikely (!req)) - goto out_put; +return NULL ;percpu_ref_get (&ctx->reqs);INIT_LIST_HEAD (&req->ki_list);refcount_set (&req->ki_refcnt,0 ); req->ki_ctx = ctx;return req; -out_put: -put_reqs_available (ctx,1 ); -return NULL ; }static struct kioctx *lookup_ioctx (unsigned long ctx_id)
@@ -1805,9 +1804,13 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb,return -EINVAL ; } +if (!get_reqs_available (ctx)) +return -EAGAIN ; + + ret = -EAGAIN ; req =aio_get_req (ctx);if (unlikely (!req)) -return -EAGAIN ; + goto out_put_reqs_available;if (iocb.aio_flags &IOCB_FLAG_RESFD ) {/*
@@ -1870,11 +1873,12 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb, goto out_put_req;return 0 ; out_put_req: -put_reqs_available (ctx,1 );percpu_ref_put (&ctx->reqs);if (req->ki_eventfd)eventfd_ctx_put (req->ki_eventfd);kmem_cache_free (kiocb_cachep, req); +out_put_reqs_available: +put_reqs_available (ctx,1 );return ret; }
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 10/20 aio: don't zero entire aio_kiocb aio_get_req()
In reply to Jens AxboeIt's 192 bytes, fairly substantial. Most items don't need to be cleared, especially not upfront. Clear the ones we do need to clear, and leave the other ones for setup when the iocb is prepared and submitted.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-)
fs/aio.c +12 -7
@@ -1009,14 +1009,15 @@ static inline struct aio_kiocb *aio_get_req(struct kioctx *ctx) {struct aio_kiocb *req; - req =kmem_cache_alloc (kiocb_cachep,GFP_KERNEL |__GFP_ZERO); -if (unlikely (!req)) -return NULL ; + req =kmem_cache_alloc (kiocb_cachep,GFP_KERNEL ); +if (req) { +percpu_ref_get (&ctx->reqs); + req->ki_ctx = ctx; +INIT_LIST_HEAD (&req->ki_list); +refcount_set (&req->ki_refcnt,0 ); + req->ki_eventfd =NULL ; + } -percpu_ref_get (&ctx->reqs); -INIT_LIST_HEAD (&req->ki_list); -refcount_set (&req->ki_refcnt,0 ); - req->ki_ctx = ctx;return req; }
@@ -1730,6 +1731,10 @@ static ssize_t aio_poll(struct aio_kiocb *aiocb, struct iocb *iocb)if (unlikely (!req->file))return -EBADF ; + req->head =NULL ; + req->woken = false; + req->cancelled = false; + apt.pt._qproc = aio_poll_queue_proc; apt.pt._key = req->events; apt.iocb = aiocb;
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 11/20 aio: only use blk plugs for > 2 depth submissions
In reply to Jens AxboePlugging is meant to optimize submission of a string of IOs, if we don't have more than 2 being submitted, don't bother setting up a plug.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-)
fs/aio.c +14 -4
@@ -1887,6 +1887,12 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb,return ret; } +/* +* Plugging is meant to work with larger batches of IOs. If we don't +* have more than the below, then don't bother setting up a plug. +*/ +#define AIO_PLUG_THRESHOLD 2 +/* sys_io_submit: * Queue the nr iocbs pointed to by iocbpp for processing. Returns * the number of iocbs queued. May return -EINVAL if the aio_context
@@ -1919,7 +1925,8 @@ SYSCALL_DEFINE3(io_submit, aio_context_t, ctx_id, long, nr,if (nr > ctx->nr_events) nr = ctx->nr_events; -blk_start_plug (&plug); +if (nr >AIO_PLUG_THRESHOLD ) +blk_start_plug (&plug);for (i =0 ; i < nr; i++) {struct iocb __user *user_iocb;
@@ -1932,7 +1939,8 @@ SYSCALL_DEFINE3(io_submit, aio_context_t, ctx_id, long, nr,if (ret)break ; } -blk_finish_plug (&plug); +if (nr >AIO_PLUG_THRESHOLD ) +blk_finish_plug (&plug);percpu_ref_put (&ctx->users);return i ? i : ret;
@@ -1959,7 +1967,8 @@ COMPAT_SYSCALL_DEFINE3(io_submit, compat_aio_context_t, ctx_id,if (nr > ctx->nr_events) nr = ctx->nr_events; -blk_start_plug (&plug); +if (nr >AIO_PLUG_THRESHOLD ) +blk_start_plug (&plug);for (i =0 ; i < nr; i++) {compat_uptr_t user_iocb;
@@ -1972,7 +1981,8 @@ COMPAT_SYSCALL_DEFINE3(io_submit, compat_aio_context_t, ctx_id,if (ret)break ; } -blk_finish_plug (&plug); +if (nr >AIO_PLUG_THRESHOLD ) +blk_finish_plug (&plug);percpu_ref_put (&ctx->users);return i ? i : ret;
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 12/20 aio: use iocb_put() instead of open coding it
In reply to Jens AxboeReplace the percpu_ref_put() + kmem_cache_free() with a call to iocb_put() instead.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-)
fs/aio.c +1 -2
@@ -1878,10 +1878,9 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb, goto out_put_req;return 0 ; out_put_req: -percpu_ref_put (&ctx->reqs);if (req->ki_eventfd)eventfd_ctx_put (req->ki_eventfd); -kmem_cache_free (kiocb_cachep, req); +iocb_put (req); out_put_reqs_available:put_reqs_available (ctx,1 );return ret;
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 13/20 aio: split out iocb copy from io_submit_one()
In reply to Jens AxboeIn preparation of handing in iocbs in a different fashion as well. Also make it clear that the iocb being passed in isn't modified, by marking it const throughout.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 68 +++++++++++++++++++++++++++++++------------------------- 1 file changed, 38 insertions(+), 30 deletions(-)
fs/aio.c +38 -30
@@ -1414,7 +1414,7 @@ static void aio_complete_rw(struct kiocb *kiocb, long res, long res2)aio_complete (iocb, res, res2); } -static int aio_prep_rw (struct kiocb *req,struct iocb *iocb) +static int aio_prep_rw (struct kiocb *req,const struct iocb *iocb) {int ret;
@@ -1455,7 +1455,7 @@ static int aio_prep_rw(struct kiocb *req, struct iocb *iocb)return ret; } -static int aio_setup_rw (int rw,struct iocb *iocb,struct iovec **iovec, +static int aio_setup_rw (int rw,const struct iocb *iocb,struct iovec **iovec,bool vectored,bool compat,struct iov_iter *iter) {void __user *buf = (void __user *)(uintptr_t )iocb->aio_buf;
@@ -1494,8 +1494,8 @@ static inline void aio_rw_done(struct kiocb *req, ssize_t ret) } } -static ssize_t aio_read (struct kiocb *req,struct iocb *iocb,bool vectored, -bool compat) +static ssize_t aio_read (struct kiocb *req,const struct iocb *iocb, +bool vectored,bool compat) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs;struct iov_iter iter;
@@ -1527,8 +1527,8 @@ static ssize_t aio_read(struct kiocb *req, struct iocb *iocb, bool vectored,return ret; } -static ssize_t aio_write (struct kiocb *req,struct iocb *iocb,bool vectored, -bool compat) +static ssize_t aio_write (struct kiocb *req,const struct iocb *iocb, +bool vectored,bool compat) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs;struct iov_iter iter;
@@ -1583,7 +1583,8 @@ static void aio_fsync_work(struct work_struct *work) aio_complete(container_of(req, struct aio_kiocb, fsync), ret, 0); } -static int aio_fsync (struct fsync_iocb *req,struct iocb *iocb,bool datasync) +static int aio_fsync (struct fsync_iocb *req,const struct iocb *iocb, +bool datasync) {if (unlikely (iocb->aio_buf || iocb->aio_offset || iocb->aio_nbytes || iocb->aio_rw_flags))
@@ -1711,7 +1712,7 @@ aio_poll_queue_proc(struct file *file, struct wait_queue_head *head,add_wait_queue (head, &pt->iocb->poll.wait); } -static ssize_t aio_poll (struct aio_kiocb *aiocb,struct iocb *iocb) +static ssize_t aio_poll (struct aio_kiocb *aiocb,const struct iocb *iocb) {struct kioctx *ctx = aiocb->ki_ctx;struct poll_iocb *req = &aiocb->poll;
@@ -1783,27 +1784,23 @@ static ssize_t aio_poll(struct aio_kiocb *aiocb, struct iocb *iocb)return 0 ; } -static int io_submit_one (struct kioctx *ctx,struct iocb __user *user_iocb, -bool compat) +static int __io_submit_one (struct kioctx *ctx,const struct iocb *iocb, +struct iocb __user *user_iocb,bool compat) {struct aio_kiocb *req; -struct iocb iocb;ssize_t ret; -if (unlikely (copy_from_user (&iocb, user_iocb,sizeof (iocb)))) -return -EFAULT ; -/* enforce forwards compatibility on users */ -if (unlikely (iocb.aio_reserved2)) { +if (unlikely (iocb->aio_reserved2)) {pr_debug ("EINVAL: reserve field set\n" );return -EINVAL ; }/* prevent overflows */ if (unlikely ( - (iocb.aio_buf != (unsigned long )iocb.aio_buf) || - (iocb.aio_nbytes != (size_t )iocb.aio_nbytes) || - ((ssize_t )iocb.aio_nbytes <0 ) + (iocb->aio_buf != (unsigned long )iocb->aio_buf) || + (iocb->aio_nbytes != (size_t )iocb->aio_nbytes) || + ((ssize_t )iocb->aio_nbytes <0 ) )) {pr_debug ("EINVAL: overflow check\n" );return -EINVAL ;
@@ -1817,14 +1814,14 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb,if (unlikely (!req)) goto out_put_reqs_available; -if (iocb.aio_flags &IOCB_FLAG_RESFD ) { +if (iocb->aio_flags &IOCB_FLAG_RESFD ) {/* * If the IOCB_FLAG_RESFD flag of aio_flags is set, get an * instance of the file* now. The file descriptor must be * an eventfd() fd, and will be signaled for each completed * event using the eventfd_signal() function. */ - req->ki_eventfd =eventfd_ctx_fdget ((int ) iocb.aio_resfd); + req->ki_eventfd =eventfd_ctx_fdget ((int ) iocb->aio_resfd);if (IS_ERR (req->ki_eventfd)) { ret =PTR_ERR (req->ki_eventfd); req->ki_eventfd =NULL ;
@@ -1839,32 +1836,32 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb, } req->ki_user_iocb = user_iocb; - req->ki_user_data = iocb.aio_data; + req->ki_user_data = iocb->aio_data; -switch (iocb.aio_lio_opcode) { +switch (iocb->aio_lio_opcode) {case IOCB_CMD_PREAD : - ret =aio_read (&req->rw, &iocb, false, compat); + ret =aio_read (&req->rw, iocb, false, compat);break ;case IOCB_CMD_PWRITE : - ret =aio_write (&req->rw, &iocb, false, compat); + ret =aio_write (&req->rw, iocb, false, compat);break ;case IOCB_CMD_PREADV : - ret =aio_read (&req->rw, &iocb, true, compat); + ret =aio_read (&req->rw, iocb, true, compat);break ;case IOCB_CMD_PWRITEV : - ret =aio_write (&req->rw, &iocb, true, compat); + ret =aio_write (&req->rw, iocb, true, compat);break ;case IOCB_CMD_FSYNC : - ret =aio_fsync (&req->fsync, &iocb, false); + ret =aio_fsync (&req->fsync, iocb, false);break ;case IOCB_CMD_FDSYNC : - ret =aio_fsync (&req->fsync, &iocb, true); + ret =aio_fsync (&req->fsync, iocb, true);break ;case IOCB_CMD_POLL : - ret =aio_poll (req, &iocb); + ret =aio_poll (req, iocb);break ;default : -pr_debug ("invalid aio operation %d\n" , iocb.aio_lio_opcode); +pr_debug ("invalid aio operation %d\n" , iocb->aio_lio_opcode); ret = -EINVAL ;break ; }
@@ -1886,6 +1883,17 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb,return ret; } +static int io_submit_one (struct kioctx *ctx,struct iocb __user *user_iocb, +bool compat) +{ +struct iocb iocb; + +if (unlikely (copy_from_user (&iocb, user_iocb,sizeof (iocb)))) +return -EFAULT ; + +return __io_submit_one (ctx, &iocb, user_iocb, compat); +} +/* * Plugging is meant to work with larger batches of IOs. If we don't * have more than the below, then don't bother setting up a plug.
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 14/20 aio: abstract out io_event filler helper
In reply to Jens AxboeSigned-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-)
fs/aio.c +10 -4
@@ -1057,6 +1057,15 @@ static inline void iocb_put(struct aio_kiocb *iocb) } } +static void aio_fill_event (struct io_event *ev,struct aio_kiocb *iocb, +long res,long res2) +{ + ev->obj = (u64 )(unsigned long )iocb->ki_user_iocb; + ev->data = iocb->ki_user_data; + ev->res = res; + ev->res2 = res2; +} +/* aio_complete * Called when the io request on the given iocb is complete. */
@@ -1084,10 +1093,7 @@ static void aio_complete(struct aio_kiocb *iocb, long res, long res2) ev_page =kmap_atomic (ctx->ring_pages[pos /AIO_EVENTS_PER_PAGE ]); event = ev_page + pos %AIO_EVENTS_PER_PAGE ; - event->obj = (u64 )(unsigned long )iocb->ki_user_iocb; - event->data = iocb->ki_user_data; - event->res = res; - event->res2 = res2; +aio_fill_event (event, iocb, res, res2);kunmap_atomic (ev_page);flush_dcache_page (ctx->ring_pages[pos /AIO_EVENTS_PER_PAGE ]);
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 15/20 aio: add io_setup2() system call
In reply to Jens AxboeThis is just like io_setup(), except add a flags argument to let the caller control/define some of the io_context behavior. Outside of that, we pass in an iocb array for future use.
Signed-off-by Jens Axboe <[email protected]>
--- arch/x86/entry/syscalls/syscall_64.tbl | 1 + fs/aio.c | 70 ++++++++++++++++---------- include/linux/syscalls.h | 2 + include/uapi/asm-generic/unistd.h | 4 +- kernel/sys_ni.c | 1 + 5 files changed, 50 insertions(+), 28 deletions(-)
arch/x86/entry/syscalls/syscall_64.tbl +1 -0
@@ -343,6 +343,7 @@ 332 common statx __x64_sys_statx 333 common io_pgetevents __x64_sys_io_pgetevents 334 common rseq __x64_sys_rseq +335 common io_setup2 __x64_sys_io_setup2 # # x32-specific system call numbers start at 512 to avoid cache impact
fs/aio.c +43 -27
140 lines not on this page · raw message
include/linux/syscalls.h +2 -0
@@ -287,6 +287,8 @@ static inline void addr_limit_user_check(void)*/ #ifndef CONFIG_ARCH_HAS_SYSCALL_WRAPPER asmlinkage long sys_io_setup (unsigned nr_reqs,aio_context_t __user *ctx); +asmlinkage long sys_io_setup2 (unsigned ,unsigned ,struct iocb __user *, +aio_context_t __user *);asmlinkage long sys_io_destroy (aio_context_t ctx);asmlinkage long sys_io_submit (aio_context_t ,long ,struct iocb __user *__user *);
include/uapi/asm-generic/unistd.h +3 -1
@@ -738,9 +738,11 @@ __SYSCALL(__NR_statx, sys_statx) __SC_COMP(__NR_io_pgetevents, sys_io_pgetevents, compat_sys_io_pgetevents)#define __NR_rseq 293 __SYSCALL(__NR_rseq, sys_rseq) +#define __NR_io_setup2 294 +__SYSCALL(__NR_io_setup2, sys_io_setup2)#undef __NR_syscalls -#define __NR_syscalls 294 +#define __NR_syscalls 295/* * 32 bit systems traditionally used different
kernel/sys_ni.c +1 -0
@@ -37,6 +37,7 @@ asmlinkage long sys_ni_syscall(void)*/ COND_SYSCALL (io_setup); +COND_SYSCALL (io_setup2);COND_SYSCALL_COMPAT (io_setup);COND_SYSCALL (io_destroy);COND_SYSCALL (io_submit);
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 16/20 aio: add support for having user mapped iocbs
In reply to Jens AxboeFor io_submit(), we have to first copy each pointer to an iocb, then copy the iocb. The latter is 64 bytes in size, and that's a lot of copying for a single IO.
Add support for setting IOCTX_FLAG_USERIOCB through the new io_setup2() system call, which allows the iocbs to reside in userspace. If this flag is used, then io_submit() doesn't take pointers to iocbs anymore, it takes an index value into the array of iocbs instead. Similary, for io_getevents(), the iocb ->obj will be the index, not the pointer to the iocb.
See the change made to fio to support this feature, it's pretty trivialy to adapt to. For applications, like fio, that previously embedded the iocb inside a application private structure, some sort of lookup table/structure is needed to find the private IO structure from the index at io_getevents() time.
http://git.kernel.dk/cgit/fio/commit/?id=3c3168e91329c83880c91e5abc28b9d6b940fd95
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 99 +++++++++++++++++++++++++++++++----- include/uapi/linux/aio_abi.h | 2 + 2 files changed, 89 insertions(+), 12 deletions(-)
fs/aio.c +87 -12
@@ -121,6 +121,9 @@ struct kioctx {struct page **ring_pages;long nr_pages; +struct page **iocb_pages; +long iocb_nr_pages; +struct rcu_work free_rwork;/* see free_ioctx() */ /*
@@ -216,6 +219,11 @@ static struct vfsmount *aio_mnt;static const struct file_operations aio_ring_fops;static const struct address_space_operations aio_ctx_aops; +static const unsigned int iocb_page_shift = +ilog2 (PAGE_SIZE /sizeof (struct iocb )); + +static void aio_useriocb_free (struct kioctx *); +static struct file *aio_private_file (struct kioctx *ctx,loff_t nr_pages) {struct file *file;
@@ -572,6 +580,7 @@ static void free_ioctx(struct work_struct *work) free_rwork);pr_debug ("freeing %p\n" , ctx); +aio_useriocb_free (ctx);aio_free_ring (ctx);free_percpu (ctx->cpu);percpu_ref_exit (&ctx->reqs);
@@ -1281,6 +1290,45 @@ static long read_events(struct kioctx *ctx, long min_nr, long nr,return ret; } +static void aio_useriocb_free (struct kioctx *ctx) +{ +int i; + +if (!ctx->iocb_nr_pages) +return ; + +for (i =0 ; i < ctx->iocb_nr_pages; i++) +put_page (ctx->iocb_pages[i]); + +kfree (ctx->iocb_pages); +} + +static int aio_useriocb_map (struct kioctx *ctx,struct iocb __user *iocbs, +unsigned int nr_events) +{ +int nr_pages, ret; + +if ((unsigned long ) iocbs & ~PAGE_MASK ) +return -EINVAL ; + + nr_pages =sizeof (struct iocb ) * nr_events; + nr_pages = (nr_pages +PAGE_SIZE -1 ) >>PAGE_SHIFT ; + + ctx->iocb_pages =kzalloc (nr_pages *sizeof (struct page *),GFP_KERNEL ); +if (!ctx->iocb_pages) +return -ENOMEM ; + + ret =get_user_pages_fast ((unsigned long ) iocbs, nr_pages,0 , + ctx->iocb_pages); +if (ret < nr_pages) { +kfree (ctx->iocb_pages); +return -ENOMEM ; + } + + ctx->iocb_nr_pages = nr_pages; +return 0 ; +} + SYSCALL_DEFINE4(io_setup2, u32, nr_events, u32, flags, struct iocb * __user, iocbs, aio_context_t __user *, ctxp) {
@@ -1288,7 +1336,7 @@ SYSCALL_DEFINE4(io_setup2, u32, nr_events, u32, flags, struct iocb * __user,unsigned long ctx;long ret; -if (flags) +if (flags & ~IOCTX_FLAG_USERIOCB )return -EINVAL ; ret =get_user (ctx, ctxp);
@@ -1300,9 +1348,17 @@ SYSCALL_DEFINE4(io_setup2, u32, nr_events, u32, flags, struct iocb * __user,if (IS_ERR (ioctx)) goto out; +if (flags &IOCTX_FLAG_USERIOCB ) { + ret =aio_useriocb_map (ioctx, iocbs, nr_events); +if (ret) + goto err; + } + ret =put_user (ioctx->user_id, ctxp); -if (ret) +if (ret) { +err:kill_ioctx (current->mm, ioctx,NULL ); + }percpu_ref_put (&ioctx->users); out:return ret;
@@ -1851,10 +1907,13 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb, } } - ret =put_user (KIOCB_KEY , &user_iocb->aio_key); -if (unlikely (ret)) { -pr_debug ("EFAULT: aio_key\n" ); - goto out_put_req; +/* Don't support cancel on user mapped iocbs */ +if (!(ctx->flags &IOCTX_FLAG_USERIOCB )) { + ret =put_user (KIOCB_KEY , &user_iocb->aio_key); +if (unlikely (ret)) { +pr_debug ("EFAULT: aio_key\n" ); + goto out_put_req; + } } req->ki_user_iocb = user_iocb;
@@ -1908,12 +1967,26 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb,static int io_submit_one (struct kioctx *ctx,struct iocb __user *user_iocb,bool compat) { -struct iocb iocb; +struct iocb iocb, *iocbp; -if (unlikely (copy_from_user (&iocb, user_iocb,sizeof (iocb)))) -return -EFAULT ; +if (ctx->flags &IOCTX_FLAG_USERIOCB ) { +unsigned long iocb_index = (unsigned long ) user_iocb; +unsigned int page_index; -return __io_submit_one (ctx, &iocb, user_iocb, compat); +if (iocb_index >= ctx->nr_events) +return -EINVAL ; + + page_index = iocb_index >> iocb_page_shift; + iocb_index &= ((1 << iocb_page_shift) -1 ); + iocbp =page_address (ctx->iocb_pages[page_index]); + iocbp += iocb_index; + }else { +if (unlikely (copy_from_user (&iocb, user_iocb,sizeof (iocb)))) +return -EFAULT ; + iocbp = &iocb; + } + +return __io_submit_one (ctx, iocbp, user_iocb, compat); }/*
@@ -2063,6 +2136,9 @@ SYSCALL_DEFINE3(io_cancel, aio_context_t, ctx_id, struct iocb __user *, iocb,if (unlikely (!ctx))return -EINVAL ; +if (ctx->flags &IOCTX_FLAG_USERIOCB ) + goto err; +spin_lock_irq (&ctx->ctx_lock); kiocb =lookup_kiocb (ctx, iocb);if (kiocb) {
@@ -2079,9 +2155,8 @@ SYSCALL_DEFINE3(io_cancel, aio_context_t, ctx_id, struct iocb __user *, iocb,*/ ret = -EINPROGRESS ; } - +err:percpu_ref_put (&ctx->users); -return ret; }
include/uapi/linux/aio_abi.h +2 -0
@@ -106,6 +106,8 @@ struct iocb {__u32 aio_resfd; };/* 64 bytes */ +#define IOCTX_FLAG_USERIOCB (1 << 0)/* iocbs are user mapped */ +#undef IFBIG#undef IFLITTLE
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 17/20 aio: support for IO polling
In reply to Jens AxboeAdd polled variants of PREAD/PREADV and PWRITE/PWRITEV. These act like their non-polled counterparts, except we expect to poll for completion of them. The polling happens at io_getevent() time, and works just like non-polled IO.
To setup an io_context for polled IO, the application must call io_setup2() with IOCTX_FLAG_IOPOLL as one of the flags. It is illegal to mix and match polled and non-polled IO on an io_context.
Polled IO doesn't support the user mapped completion ring. Events must be reaped through the io_getevents() system call. For non-irq driven poll devices, there's no way to support completion reaping from userspace by just looking at the ring. The application itself is the one that pulls completion entries.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 378 +++++++++++++++++++++++++++++++---- include/uapi/linux/aio_abi.h | 3 + 2 files changed, 345 insertions(+), 36 deletions(-)
fs/aio.c +342 -36
@@ -143,6 +143,18 @@ struct kioctx {atomic_t reqs_available; }____cacheline_aligned_in_smp ; +/* iopoll submission state */ +struct { +spinlock_t poll_lock; +struct list_head poll_submitted; + }____cacheline_aligned_in_smp ; + +/* iopoll completion state */ +struct { +struct list_head poll_completing; +struct mutex getevents_lock; + }____cacheline_aligned_in_smp ; +struct {spinlock_t ctx_lock;struct list_head active_reqs;/* used for cancellation */
@@ -195,14 +207,27 @@ struct aio_kiocb {__u64 ki_user_data;/* user's data for completion */ struct list_head ki_list;/* the aio core uses this -* for cancellation */ +* for cancellation, or for +* polled IO */ + +unsigned long ki_flags; +#define IOCB_POLL_COMPLETED 0 +#define IOCB_POLL_BUSY 1 +refcount_t ki_refcnt; -/* -* If the aio_resfd field of the userspace iocb is not zero, -* this is the underlying eventfd context to deliver events to. -*/ -struct eventfd_ctx *ki_eventfd; +union { +/* +* If the aio_resfd field of the userspace iocb is not zero, +* this is the underlying eventfd context to deliver events to. +*/ +struct eventfd_ctx *ki_eventfd; + +/* +* For polled IO, stash completion info here +*/ +struct io_event ki_ev; + }; };/*------ sysctl variables----*/
@@ -223,6 +248,7 @@ static const unsigned int iocb_page_shift =ilog2 (PAGE_SIZE /sizeof (struct iocb ));static void aio_useriocb_free (struct kioctx *); +static void aio_iopoll_reap_events (struct kioctx *);static struct file *aio_private_file (struct kioctx *ctx,loff_t nr_pages) {
@@ -461,11 +487,15 @@ static int aio_setup_ring(struct kioctx *ctx, unsigned int nr_events)int i;struct file *file; -/* Compensate for the ring buffer's head/tail overlap entry */ - nr_events +=2 ;/* 1 is required, 2 for good luck */ - +/* +* Compensate for the ring buffer's head/tail overlap entry. +* IO polling doesn't require any io event entries +*/ size =sizeof (struct aio_ring ); - size +=sizeof (struct io_event ) * nr_events; +if (!(ctx->flags &IOCTX_FLAG_IOPOLL )) { + nr_events +=2 ;/* 1 is required, 2 for good luck */ + size +=sizeof (struct io_event ) * nr_events; + } nr_pages =PFN_UP (size);if (nr_pages <0 )
@@ -747,6 +777,11 @@ static struct kioctx *io_setup_flags(unsigned long ctxid,INIT_LIST_HEAD (&ctx->active_reqs); +spin_lock_init (&ctx->poll_lock); +INIT_LIST_HEAD (&ctx->poll_submitted); +INIT_LIST_HEAD (&ctx->poll_completing); +mutex_init (&ctx->getevents_lock); +if (percpu_ref_init (&ctx->users, free_ioctx_users,0 ,GFP_KERNEL )) goto err;
@@ -818,11 +853,15 @@ static int kill_ioctx(struct mm_struct *mm, struct kioctx *ctx, {struct kioctx_table *table; +mutex_lock (&ctx->getevents_lock);spin_lock (&mm->ioctx_lock);if (atomic_xchg (&ctx->dead,1 )) {spin_unlock (&mm->ioctx_lock); +mutex_unlock (&ctx->getevents_lock);return -EINVAL ; } +aio_iopoll_reap_events (ctx); +mutex_unlock (&ctx->getevents_lock); table =rcu_dereference_raw (mm->ioctx_table);WARN_ON (ctx !=rcu_access_pointer (table->table[ctx->id]));
@@ -1029,6 +1068,7 @@ static inline struct aio_kiocb *aio_get_req(struct kioctx *ctx)percpu_ref_get (&ctx->reqs); req->ki_ctx = ctx;INIT_LIST_HEAD (&req->ki_list); + req->ki_flags =0 ;refcount_set (&req->ki_refcnt,0 ); req->ki_eventfd =NULL ; }
@@ -1072,6 +1112,15 @@ static inline void iocb_put(struct aio_kiocb *iocb) } } +static void iocb_put_many (struct kioctx *ctx,void **iocbs,int *nr) +{ +if (nr) { +percpu_ref_put_many (&ctx->reqs, *nr); +kmem_cache_free_bulk (kiocb_cachep, *nr, iocbs); + *nr =0 ; + } +} +static void aio_fill_event (struct io_event *ev,struct aio_kiocb *iocb,long res,long res2) {
@@ -1261,6 +1310,166 @@ static bool aio_read_events(struct kioctx *ctx, long min_nr, long nr,return ret <0 || *i >= min_nr; } +#define AIO_IOPOLL_BATCH 8 + +/* +* Process completed iocb iopoll entries, copying the result to userspace. +*/ +static long aio_iopoll_reap (struct kioctx *ctx,struct io_event __user *evs, +unsigned int *nr_events,long max) +{ +void *iocbs[AIO_IOPOLL_BATCH ]; +struct aio_kiocb *iocb, *n; +int to_free =0 , ret =0 ; + +list_for_each_entry_safe (iocb, n, &ctx->poll_completing, ki_list) { +if (*nr_events == max) +break ; +if (!test_bit (IOCB_POLL_COMPLETED , &iocb->ki_flags)) +continue ; +if (to_free ==AIO_IOPOLL_BATCH ) +iocb_put_many (ctx, iocbs, &to_free); + +list_del (&iocb->ki_list); + iocbs[to_free++] = iocb; + +fput (iocb->rw.ki_filp); + +if (evs &©_to_user (evs + *nr_events, &iocb->ki_ev, +sizeof (iocb->ki_ev))) { + ret = -EFAULT ; +break ; + } + (*nr_events)++; + } + +if (to_free) +iocb_put_many (ctx, iocbs, &to_free); + +return ret; +} + +static int __aio_iopoll_check (struct kioctx *ctx,struct io_event __user *event, +unsigned int *nr_events,long min,long max) +{ +struct aio_kiocb *iocb; +int to_poll, polled, ret; + +/* +* Check if we already have done events that satisfy what we need +*/ +if (!list_empty (&ctx->poll_completing)) { + ret =aio_iopoll_reap (ctx, event, nr_events, max); +if (ret <0 ) +return ret; +if (*nr_events >= min) +return 0 ; + } + +/* +* Take in a new working set from the submitted list, if possible. +*/ +if (!list_empty_careful (&ctx->poll_submitted)) { +spin_lock (&ctx->poll_lock); +list_splice_init (&ctx->poll_submitted, &ctx->poll_completing); +spin_unlock (&ctx->poll_lock); + } + +if (list_empty (&ctx->poll_completing)) +return 0 ; + +/* +* Check again now that we have a new batch. +*/ + ret =aio_iopoll_reap (ctx, event, nr_events, max); +if (ret <0 ) +return ret; +if (*nr_events >= min) +return 0 ; + +/* +* Find up to 'max' worth of events to poll for, including the +* events we already successfully polled +*/ + polled = to_poll =0 ; +list_for_each_entry (iocb, &ctx->poll_completing, ki_list) { +/* +* Poll for needed events with spin == true, anything after +* that we just check if we have more, up to max. +*/ +bool spin = polled + *nr_events >= min; +struct kiocb *kiocb = &iocb->rw; + +if (test_bit (IOCB_POLL_COMPLETED , &iocb->ki_flags)) +break ; +if (++to_poll + *nr_events > max) +break ; + + ret = kiocb->ki_filp->f_op->iopoll (kiocb, spin); +if (ret <0 ) +return ret; + + polled += ret; +if (polled + *nr_events >= max) +break ; + } + + ret =aio_iopoll_reap (ctx, event, nr_events, max); +if (ret <0 ) +return ret; +if (*nr_events >= min) +return 0 ; +return to_poll; +} + +/* +* We can't just wait for polled events to come to us, we have to actively +* find and complete them. +*/ +static void aio_iopoll_reap_events (struct kioctx *ctx) +{ +if (!(ctx->flags &IOCTX_FLAG_IOPOLL )) +return ; + +while (!list_empty_careful (&ctx->poll_submitted) || + !list_empty (&ctx->poll_completing)) { +unsigned int nr_events =0 ; + +__aio_iopoll_check (ctx,NULL , &nr_events,1 ,UINT_MAX ); + } +} + +static int aio_iopoll_check (struct kioctx *ctx,long min_nr,long nr, +struct io_event __user *event) +{ +unsigned int nr_events =0 ; +int ret =0 ; + +/* Only allow one thread polling at a time */ +if (!mutex_trylock (&ctx->getevents_lock)) +return -EBUSY ; +if (unlikely (atomic_read (&ctx->dead))) { + ret = -EINVAL ; + goto err; + } + +while (!nr_events || !need_resched ()) { +int tmin =0 ; + +if (nr_events < min_nr) + tmin = min_nr - nr_events; + + ret =__aio_iopoll_check (ctx, event, &nr_events, tmin, nr); +if (ret <=0 ) +break ; + ret =0 ; + } + +err: +mutex_unlock (&ctx->getevents_lock); +return nr_events ? nr_events : ret; +} +static long read_events (struct kioctx *ctx,long min_nr,long nr,struct io_event __user *event,ktime_t until)
@@ -1336,7 +1545,7 @@ SYSCALL_DEFINE4(io_setup2, u32, nr_events, u32, flags, struct iocb * __user,unsigned long ctx;long ret; -if (flags & ~IOCTX_FLAG_USERIOCB ) +if (flags & ~(IOCTX_FLAG_USERIOCB |IOCTX_FLAG_IOPOLL ))return -EINVAL ; ret =get_user (ctx, ctxp);
@@ -1469,13 +1678,8 @@ static void aio_remove_iocb(struct aio_kiocb *iocb)spin_unlock_irqrestore (&ctx->ctx_lock, flags); } -static void aio_complete_rw (struct kiocb *kiocb,long res,long res2) +static void kiocb_end_write (struct kiocb *kiocb) { - struct aio_kiocb *iocb = container_of(kiocb, struct aio_kiocb, rw); - -if (!list_empty_careful (&iocb->ki_list)) -aio_remove_iocb (iocb); -if (kiocb->ki_flags &IOCB_WRITE ) {struct inode *inode =file_inode (kiocb->ki_filp);
@@ -1487,19 +1691,48 @@ static void aio_complete_rw(struct kiocb *kiocb, long res, long res2)__sb_writers_acquired (inode->i_sb,SB_FREEZE_WRITE );file_end_write (kiocb->ki_filp); } +} + +static void aio_complete_rw (struct kiocb *kiocb,long res,long res2) +{ + struct aio_kiocb *iocb = container_of(kiocb, struct aio_kiocb, rw); + +if (!list_empty_careful (&iocb->ki_list)) +aio_remove_iocb (iocb); + +kiocb_end_write (kiocb);fput (kiocb->ki_filp);aio_complete (iocb, res, res2); } -static int aio_prep_rw (struct kiocb *req,const struct iocb *iocb) +static void aio_complete_rw_poll (struct kiocb *kiocb,long res,long res2) { + struct aio_kiocb *iocb = container_of(kiocb, struct aio_kiocb, rw); + +kiocb_end_write (kiocb); + +/* +* Handle EAGAIN from resource limits with polled IO inline, don't +* pass the event back to userspace. +*/ +if (unlikely (res == -EAGAIN )) +set_bit (IOCB_POLL_BUSY , &iocb->ki_flags); +else { +aio_fill_event (&iocb->ki_ev, iocb, res, res2); +set_bit (IOCB_POLL_COMPLETED , &iocb->ki_flags); + } +} + +static int aio_prep_rw (struct aio_kiocb *kiocb,const struct iocb *iocb) +{ +struct kioctx *ctx = kiocb->ki_ctx; +struct kiocb *req = &kiocb->rw;int ret; req->ki_filp =fget (iocb->aio_fildes);if (unlikely (!req->ki_filp))return -EBADF ; - req->ki_complete = aio_complete_rw; req->ki_pos = iocb->aio_offset; req->ki_flags =iocb_flags (req->ki_filp);if (iocb->aio_flags &IOCB_FLAG_RESFD )
@@ -1525,9 +1758,35 @@ static int aio_prep_rw(struct kiocb *req, const struct iocb *iocb)if (unlikely (ret)) goto out_fput; - req->ki_flags &= ~IOCB_HIPRI ;/* no one is going to poll for this I/O */ -return 0 ; +if (iocb->aio_flags &IOCB_FLAG_HIPRI ) { +/* shares space in the union, and is rather pointless.. */ + ret = -EINVAL ; +if (iocb->aio_flags &IOCB_FLAG_RESFD ) + goto out_fput; + +/* can't submit polled IO to a non-polled ctx */ +if (!(ctx->flags &IOCTX_FLAG_IOPOLL )) + goto out_fput; + + ret = -EOPNOTSUPP ; +if (!(req->ki_flags &IOCB_DIRECT ) || + !req->ki_filp->f_op->iopoll) + goto out_fput; + + req->ki_flags |=IOCB_HIPRI ; + req->ki_complete = aio_complete_rw_poll; + }else { +/* can't submit non-polled IO to a polled ctx */ + ret = -EINVAL ; +if (ctx->flags &IOCTX_FLAG_IOPOLL ) + goto out_fput; + +/* no one is going to poll for this I/O */ + req->ki_flags &= ~IOCB_HIPRI ; + req->ki_complete = aio_complete_rw; + } +return 0 ; out_fput:fput (req->ki_filp);return ret;
@@ -1570,17 +1829,43 @@ static inline void aio_rw_done(struct kiocb *req, ssize_t ret)default : req->ki_complete (req, ret,0 ); } + } -static ssize_t aio_read (struct kiocb *req,const struct iocb *iocb, +/* +* After the iocb has been issued, it's safe to be found on the poll list. +* Adding the kiocb to the list AFTER submission ensures that we don't +* find it from a io_getevents() thread before the issuer is done accessing +* the kiocb cookie. +*/ +static void aio_iopoll_iocb_issued (struct aio_kiocb *kiocb) +{ +/* +* For fast devices, IO may have already completed. If it has, add +* it to the front so we find it first. We can't add to the poll_done +* list as that's unlocked from the completion side. +*/ +const int front_add =test_bit (IOCB_POLL_COMPLETED , &kiocb->ki_flags); +struct kioctx *ctx = kiocb->ki_ctx; + +spin_lock (&ctx->poll_lock); +if (front_add) +list_add (&kiocb->ki_list, &ctx->poll_submitted); +else +list_add_tail (&kiocb->ki_list, &ctx->poll_submitted); +spin_unlock (&ctx->poll_lock); +} + +static ssize_t aio_read (struct aio_kiocb *kiocb,const struct iocb *iocb,bool vectored,bool compat) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs; +struct kiocb *req = &kiocb->rw;struct iov_iter iter;struct file *file;ssize_t ret; - ret =aio_prep_rw (req, iocb); + ret =aio_prep_rw (kiocb, iocb);if (ret)return ret; file = req->ki_filp;
@@ -1605,15 +1890,16 @@ static ssize_t aio_read(struct kiocb *req, const struct iocb *iocb,return ret; } -static ssize_t aio_write (struct kiocb *req,const struct iocb *iocb, +static ssize_t aio_write (struct aio_kiocb *kiocb,const struct iocb *iocb,bool vectored,bool compat) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs; +struct kiocb *req = &kiocb->rw;struct iov_iter iter;struct file *file;ssize_t ret; - ret =aio_prep_rw (req, iocb); + ret =aio_prep_rw (kiocb, iocb);if (ret)return ret; file = req->ki_filp;
@@ -1884,7 +2170,8 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb,return -EINVAL ; } -if (!get_reqs_available (ctx)) +/* Poll IO doesn't need ring reservations */ +if (!(ctx->flags &IOCTX_FLAG_IOPOLL ) && !get_reqs_available (ctx))return -EAGAIN ; ret = -EAGAIN ;
@@ -1907,8 +2194,8 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb, } } -/* Don't support cancel on user mapped iocbs */ -if (!(ctx->flags &IOCTX_FLAG_USERIOCB )) { +/* Don't support cancel on user mapped iocbs or polled context */ +if (!(ctx->flags & (IOCTX_FLAG_USERIOCB |IOCTX_FLAG_IOPOLL ))) { ret =put_user (KIOCB_KEY , &user_iocb->aio_key);if (unlikely (ret)) {pr_debug ("EFAULT: aio_key\n" );
@@ -1919,26 +2206,33 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb, req->ki_user_iocb = user_iocb; req->ki_user_data = iocb->aio_data; + ret = -EINVAL ;switch (iocb->aio_lio_opcode) {case IOCB_CMD_PREAD : - ret =aio_read (&req->rw, iocb, false, compat); + ret =aio_read (req, iocb, false, compat);break ;case IOCB_CMD_PWRITE : - ret =aio_write (&req->rw, iocb, false, compat); + ret =aio_write (req, iocb, false, compat);break ;case IOCB_CMD_PREADV : - ret =aio_read (&req->rw, iocb, true, compat); + ret =aio_read (req, iocb, true, compat);break ;case IOCB_CMD_PWRITEV : - ret =aio_write (&req->rw, iocb, true, compat); + ret =aio_write (req, iocb, true, compat);break ;case IOCB_CMD_FSYNC : +if (ctx->flags &IOCTX_FLAG_IOPOLL ) +break ; ret =aio_fsync (&req->fsync, iocb, false);break ;case IOCB_CMD_FDSYNC : +if (ctx->flags &IOCTX_FLAG_IOPOLL ) +break ; ret =aio_fsync (&req->fsync, iocb, true);break ;case IOCB_CMD_POLL : +if (ctx->flags &IOCTX_FLAG_IOPOLL ) +break ; ret =aio_poll (req, iocb);break ;default :
@@ -1954,13 +2248,21 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb,*/ if (ret) goto out_put_req; +if (ctx->flags &IOCTX_FLAG_IOPOLL ) { +if (test_bit (IOCB_POLL_BUSY , &req->ki_flags)) { + ret = -EAGAIN ; + goto out_put_req; + } +aio_iopoll_iocb_issued (req); + }return 0 ; out_put_req:if (req->ki_eventfd)eventfd_ctx_put (req->ki_eventfd);iocb_put (req); out_put_reqs_available: -put_reqs_available (ctx,1 ); +if (!(ctx->flags &IOCTX_FLAG_IOPOLL )) +put_reqs_available (ctx,1 );return ret; }
@@ -2136,7 +2438,7 @@ SYSCALL_DEFINE3(io_cancel, aio_context_t, ctx_id, struct iocb __user *, iocb,if (unlikely (!ctx))return -EINVAL ; -if (ctx->flags &IOCTX_FLAG_USERIOCB ) +if (ctx->flags & (IOCTX_FLAG_USERIOCB |IOCTX_FLAG_IOPOLL )) goto err;spin_lock_irq (&ctx->ctx_lock);
@@ -2171,8 +2473,12 @@ static long do_io_getevents(aio_context_t ctx_id,long ret = -EINVAL ;if (likely (ioctx)) { -if (likely (min_nr <= nr && min_nr >=0 )) - ret =read_events (ioctx, min_nr, nr, events, until); +if (likely (min_nr <= nr && min_nr >=0 )) { +if (ioctx->flags &IOCTX_FLAG_IOPOLL ) + ret =aio_iopoll_check (ioctx, min_nr, nr, events); +else + ret =read_events (ioctx, min_nr, nr, events, until); + }percpu_ref_put (&ioctx->users); }
include/uapi/linux/aio_abi.h +3 -0
@@ -52,9 +52,11 @@ enum {* is valid. * IOCB_FLAG_IOPRIO - Set if the "aio_reqprio" member of the "struct iocb" * is valid. +* IOCB_FLAG_HIPRI - Use IO completion polling */ #define IOCB_FLAG_RESFD (1 << 0)#define IOCB_FLAG_IOPRIO (1 << 1) +#define IOCB_FLAG_HIPRI (1 << 2)/* read() from /dev/aio returns these structures. */ struct io_event {
@@ -107,6 +109,7 @@ struct iocb { };/* 64 bytes */ #define IOCTX_FLAG_USERIOCB (1 << 0)/* iocbs are user mapped */ +#define IOCTX_FLAG_IOPOLL (1 << 1)/* io_context is polled */ #undef IFBIG#undef IFLITTLE
Signature · 1 line
2.17.1
Re: [PATCH 17/20] aio: support for IO polling
Quoted 121 lines · Jens Axboe
Add polled variants of PREAD/PREADV and PWRITE/PWRITEV. These act like their non-polled counterparts, except we expect to poll for completion of them. The polling happens at
io_getevent()time, and works just like non-polled IO.To setup an
io_contextfor polled IO, the application must callio_setup2()withIOCTX_FLAG_IOPOLLas one of the flags. It is illegal to mix and match polled and non-polled IO on anio_context.Polled IO doesn't support the user mapped completion ring. Events must be reaped through the
io_getevents()system call. For non-irq driven poll devices, there's no way to support completion reaping from userspace by just looking at the ring. The application itself is the one that pulls completion entries.Signed-off-by Jens Axboe <[email protected]>--- fs/aio.c | 378 +++++++++++++++++++++++++++++++---- include/uapi/linux/aio_abi.h | 3 + 2 files changed, 345 insertions(+), 36 deletions(-)fs/aio.c +45
-10@@ -143,6 +143,18 @@ struct kioctx {atomic_t reqs_available; }____cacheline_aligned_in_smp ; +/* iopoll submission state */ +struct { +spinlock_t poll_lock; +struct list_head poll_submitted; + }____cacheline_aligned_in_smp ; + +/* iopoll completion state */ +struct { +struct list_head poll_completing; +struct mutex getevents_lock; + }____cacheline_aligned_in_smp ; +struct {spinlock_t ctx_lock;struct list_head active_reqs;/* used for cancellation */ @@ -195,14 +207,27 @@ struct aio_kiocb {__u64 ki_user_data;/* user's data for completion */ struct list_head ki_list;/* the aio core uses this -* for cancellation */ +* for cancellation, or for +* polled IO */ + +unsigned long ki_flags; +#define IOCB_POLL_COMPLETED 0 +#define IOCB_POLL_BUSY 1 +refcount_t ki_refcnt; -/* -* If the aio_resfd field of the userspace iocb is not zero, -* this is the underlying eventfd context to deliver events to. -*/ -struct eventfd_ctx *ki_eventfd; +union { +/* +* If the aio_resfd field of the userspace iocb is not zero, +* this is the underlying eventfd context to deliver events to. +*/ +struct eventfd_ctx *ki_eventfd; + +/* +* For polled IO, stash completion info here +*/ +struct io_event ki_ev; + }; };/*------ sysctl variables----*/ @@ -223,6 +248,7 @@ static const unsigned int iocb_page_shift =ilog2 (PAGE_SIZE /sizeof (struct iocb ));static void aio_useriocb_free (struct kioctx *); +static void aio_iopoll_reap_events (struct kioctx *);static struct file *aio_private_file (struct kioctx *ctx,loff_t nr_pages) {@@ -461,11 +487,15 @@ static int aio_setup_ring(struct kioctx *ctx, unsigned int nr_events)int i;struct file *file; -/* Compensate for the ring buffer's head/tail overlap entry */ - nr_events +=2 ;/* 1 is required, 2 for good luck */ - +/* +* Compensate for the ring buffer's head/tail overlap entry. +* IO polling doesn't require any io event entries +*/ size =sizeof (struct aio_ring ); - size +=sizeof (struct io_event ) * nr_events; +if (!(ctx->flags &IOCTX_FLAG_IOPOLL )) { + nr_events +=2 ;/* 1 is required, 2 for good luck */ + size +=sizeof (struct io_event ) * nr_events; + } nr_pages =PFN_UP (size);if (nr_pages <0 )@@ -747,6 +777,11 @@ static struct kioctx *io_setup_flags(unsigned long ctxid,INIT_LIST_HEAD (&ctx->active_reqs); +spin_lock_init (&ctx->poll_lock); +INIT_LIST_HEAD (&ctx->poll_submitted); +INIT_LIST_HEAD (&ctx->poll_completing); +mutex_init (&ctx->getevents_lock); +if (percpu_ref_init (&ctx->users, free_ioctx_users,0 ,GFP_KERNEL )) goto err;@@ -818,11 +853,15 @@ static int kill_ioctx(struct mm_struct *mm, struct kioctx *ctx, {struct kioctx_table *table;
fs/aio.c static int kill_ioctx(struct mm_struct *mm, struct kioctx *ctx, · line 864
Quoted 10 lines
fs/aio.c +4
-0+mutex_lock (&ctx->getevents_lock);spin_lock (&mm->ioctx_lock);if (atomic_xchg (&ctx->dead,1 )) {spin_unlock (&mm->ioctx_lock); +mutex_unlock (&ctx->getevents_lock);return -EINVAL ; } +aio_iopoll_reap_events (ctx); +mutex_unlock (&ctx->getevents_lock);
Is it worth handling the mutex lock and calling aio_iopoll_reap_events only if (ctx->flags & IOCTX_FLAG_IOPOLL)? If so, testing it can be removed from aio_iopoll_reap_events() (and maybe it could even be open coded
here since this is its only call site apparently)
Quoted 12 lines
fs/aio.c +1
-0table =rcu_dereference_raw (mm->ioctx_table);WARN_ON (ctx !=rcu_access_pointer (table->table[ctx->id]));@@ -1029,6 +1068,7 @@ static inline struct aio_kiocb *aio_get_req(struct kioctx *ctx)percpu_ref_get (&ctx->reqs); req->ki_ctx = ctx;INIT_LIST_HEAD (&req->ki_list); + req->ki_flags =0 ;refcount_set (&req->ki_refcnt,0 ); req->ki_eventfd =NULL ; }
fs/aio.c static inline void iocb_put(struct aio_kiocb *iocb) · line 1117
Quoted 6 lines
fs/aio.c +3
-0@@ -1072,6 +1112,15 @@ static inline void iocb_put(struct aio_kiocb *iocb) } } +static void iocb_put_many (struct kioctx *ctx,void **iocbs,int *nr) +{ +if (nr) {
How can nr by NULL?
And what's the point of supporting this case?
Did you mean: if (*nr)?
(In this case, if safe to call the functions below with *nr==0, I'm not sure it's worth optimizing... especially since this is a static function and its callers make sure to call it only when *nr > 0)
Quoted 17 lines
fs/aio.c +10
-0+percpu_ref_put_many (&ctx->reqs, *nr); +kmem_cache_free_bulk (kiocb_cachep, *nr, iocbs); + *nr =0 ; + } +} +static void aio_fill_event (struct io_event *ev,struct aio_kiocb *iocb,long res,long res2) {@@ -1261,6 +1310,166 @@ static bool aio_read_events(struct kioctx *ctx, long min_nr, long nr,return ret <0 || *i >= min_nr; } +#define AIO_IOPOLL_BATCH 8 + +/* +* Process completed iocb iopoll entries, copying the result to userspace.
fs/aio.c static bool aio_read_events(struct kioctx *ctx, long min_nr, long nr, · line 1326
Quoted 10 lines
fs/aio.c +10
-0+*/ +static long aio_iopoll_reap (struct kioctx *ctx,struct io_event __user *evs, +unsigned int *nr_events,long max) +{ +void *iocbs[AIO_IOPOLL_BATCH ]; +struct aio_kiocb *iocb, *n; +int to_free =0 , ret =0 ; + +list_for_each_entry_safe (iocb, n, &ctx->poll_completing, ki_list) { +if (*nr_events == max)
*nr_events >= max would be safer.
Quoted 263 lines
fs/aio.c +207
-11+break ; +if (!test_bit (IOCB_POLL_COMPLETED , &iocb->ki_flags)) +continue ; +if (to_free ==AIO_IOPOLL_BATCH ) +iocb_put_many (ctx, iocbs, &to_free); + +list_del (&iocb->ki_list); + iocbs[to_free++] = iocb; + +fput (iocb->rw.ki_filp); + +if (evs &©_to_user (evs + *nr_events, &iocb->ki_ev, +sizeof (iocb->ki_ev))) { + ret = -EFAULT ; +break ; + } + (*nr_events)++; + } + +if (to_free) +iocb_put_many (ctx, iocbs, &to_free); + +return ret; +} + +static int __aio_iopoll_check (struct kioctx *ctx,struct io_event __user *event, +unsigned int *nr_events,long min,long max) +{ +struct aio_kiocb *iocb; +int to_poll, polled, ret; + +/* +* Check if we already have done events that satisfy what we need +*/ +if (!list_empty (&ctx->poll_completing)) { + ret =aio_iopoll_reap (ctx, event, nr_events, max); +if (ret <0 ) +return ret; +if (*nr_events >= min) +return 0 ; + } + +/* +* Take in a new working set from the submitted list, if possible. +*/ +if (!list_empty_careful (&ctx->poll_submitted)) { +spin_lock (&ctx->poll_lock); +list_splice_init (&ctx->poll_submitted, &ctx->poll_completing); +spin_unlock (&ctx->poll_lock); + } + +if (list_empty (&ctx->poll_completing)) +return 0 ; + +/* +* Check again now that we have a new batch. +*/ + ret =aio_iopoll_reap (ctx, event, nr_events, max); +if (ret <0 ) +return ret; +if (*nr_events >= min) +return 0 ; + +/* +* Find up to 'max' worth of events to poll for, including the +* events we already successfully polled +*/ + polled = to_poll =0 ; +list_for_each_entry (iocb, &ctx->poll_completing, ki_list) { +/* +* Poll for needed events with spin == true, anything after +* that we just check if we have more, up to max. +*/ +bool spin = polled + *nr_events >= min; +struct kiocb *kiocb = &iocb->rw; + +if (test_bit (IOCB_POLL_COMPLETED , &iocb->ki_flags)) +break ; +if (++to_poll + *nr_events > max) +break ; + + ret = kiocb->ki_filp->f_op->iopoll (kiocb, spin); +if (ret <0 ) +return ret; + + polled += ret; +if (polled + *nr_events >= max) +break ; + } + + ret =aio_iopoll_reap (ctx, event, nr_events, max); +if (ret <0 ) +return ret; +if (*nr_events >= min) +return 0 ; +return to_poll; +} + +/* +* We can't just wait for polled events to come to us, we have to actively +* find and complete them. +*/ +static void aio_iopoll_reap_events (struct kioctx *ctx) +{ +if (!(ctx->flags &IOCTX_FLAG_IOPOLL )) +return ; + +while (!list_empty_careful (&ctx->poll_submitted) || + !list_empty (&ctx->poll_completing)) { +unsigned int nr_events =0 ; + +__aio_iopoll_check (ctx,NULL , &nr_events,1 ,UINT_MAX ); + } +} + +static int aio_iopoll_check (struct kioctx *ctx,long min_nr,long nr, +struct io_event __user *event) +{ +unsigned int nr_events =0 ; +int ret =0 ; + +/* Only allow one thread polling at a time */ +if (!mutex_trylock (&ctx->getevents_lock)) +return -EBUSY ; +if (unlikely (atomic_read (&ctx->dead))) { + ret = -EINVAL ; + goto err; + } + +while (!nr_events || !need_resched ()) { +int tmin =0 ; + +if (nr_events < min_nr) + tmin = min_nr - nr_events; + + ret =__aio_iopoll_check (ctx, event, &nr_events, tmin, nr); +if (ret <=0 ) +break ; + ret =0 ; + } + +err: +mutex_unlock (&ctx->getevents_lock); +return nr_events ? nr_events : ret; +} +static long read_events (struct kioctx *ctx,long min_nr,long nr,struct io_event __user *event,ktime_t until)@@ -1336,7 +1545,7 @@ SYSCALL_DEFINE4(io_setup2, u32, nr_events, u32, flags, struct iocb * __user,unsigned long ctx;long ret; -if (flags & ~IOCTX_FLAG_USERIOCB ) +if (flags & ~(IOCTX_FLAG_USERIOCB |IOCTX_FLAG_IOPOLL ))return -EINVAL ; ret =get_user (ctx, ctxp);@@ -1469,13 +1678,8 @@ static void aio_remove_iocb(struct aio_kiocb *iocb)spin_unlock_irqrestore (&ctx->ctx_lock, flags); } -static void aio_complete_rw (struct kiocb *kiocb,long res,long res2) +static void kiocb_end_write (struct kiocb *kiocb) { - struct aio_kiocb *iocb = container_of(kiocb, struct aio_kiocb, rw); - -if (!list_empty_careful (&iocb->ki_list)) -aio_remove_iocb (iocb); -if (kiocb->ki_flags &IOCB_WRITE ) {struct inode *inode =file_inode (kiocb->ki_filp);@@ -1487,19 +1691,48 @@ static void aio_complete_rw(struct kiocb *kiocb, long res, long res2)__sb_writers_acquired (inode->i_sb,SB_FREEZE_WRITE );file_end_write (kiocb->ki_filp); } +} + +static void aio_complete_rw (struct kiocb *kiocb,long res,long res2) +{ + struct aio_kiocb *iocb = container_of(kiocb, struct aio_kiocb, rw); + +if (!list_empty_careful (&iocb->ki_list)) +aio_remove_iocb (iocb); + +kiocb_end_write (kiocb);fput (kiocb->ki_filp);aio_complete (iocb, res, res2); } -static int aio_prep_rw (struct kiocb *req,const struct iocb *iocb) +static void aio_complete_rw_poll (struct kiocb *kiocb,long res,long res2) { + struct aio_kiocb *iocb = container_of(kiocb, struct aio_kiocb, rw); + +kiocb_end_write (kiocb); + +/* +* Handle EAGAIN from resource limits with polled IO inline, don't +* pass the event back to userspace. +*/ +if (unlikely (res == -EAGAIN )) +set_bit (IOCB_POLL_BUSY , &iocb->ki_flags); +else { +aio_fill_event (&iocb->ki_ev, iocb, res, res2); +set_bit (IOCB_POLL_COMPLETED , &iocb->ki_flags); + } +} + +static int aio_prep_rw (struct aio_kiocb *kiocb,const struct iocb *iocb) +{ +struct kioctx *ctx = kiocb->ki_ctx; +struct kiocb *req = &kiocb->rw;int ret; req->ki_filp =fget (iocb->aio_fildes);if (unlikely (!req->ki_filp))return -EBADF ; - req->ki_complete = aio_complete_rw; req->ki_pos = iocb->aio_offset; req->ki_flags =iocb_flags (req->ki_filp);if (iocb->aio_flags &IOCB_FLAG_RESFD )@@ -1525,9 +1758,35 @@ static int aio_prep_rw(struct kiocb *req, const struct iocb *iocb)if (unlikely (ret)) goto out_fput; - req->ki_flags &= ~IOCB_HIPRI ;/* no one is going to poll for this I/O */ -return 0 ; +if (iocb->aio_flags &IOCB_FLAG_HIPRI ) { +/* shares space in the union, and is rather pointless.. */ + ret = -EINVAL ; +if (iocb->aio_flags &IOCB_FLAG_RESFD ) + goto out_fput; + +/* can't submit polled IO to a non-polled ctx */ +if (!(ctx->flags &IOCTX_FLAG_IOPOLL )) + goto out_fput; + + ret = -EOPNOTSUPP ; +if (!(req->ki_flags &IOCB_DIRECT ) || + !req->ki_filp->f_op->iopoll) + goto out_fput; + + req->ki_flags |=IOCB_HIPRI ; + req->ki_complete = aio_complete_rw_poll; + }else { +/* can't submit non-polled IO to a polled ctx */ + ret = -EINVAL ; +if (ctx->flags &IOCTX_FLAG_IOPOLL ) + goto out_fput; + +/* no one is going to poll for this I/O */ + req->ki_flags &= ~IOCB_HIPRI ; + req->ki_complete = aio_complete_rw; + } +return 0 ; out_fput:fput (req->ki_filp);return ret;
fs/aio.c static inline void aio_rw_done(struct kiocb *req, ssize_t ret) · line 1832
Quoted 4 lines
fs/aio.c +1
-0@@ -1570,17 +1829,43 @@ static inline void aio_rw_done(struct kiocb *req, ssize_t ret)default : req->ki_complete (req, ret,0 ); } +
nit: this hunk is probably unintentional
Quoted 190 lines
fs/aio.c +61
-15} -static ssize_t aio_read (struct kiocb *req,const struct iocb *iocb, +/* +* After the iocb has been issued, it's safe to be found on the poll list. +* Adding the kiocb to the list AFTER submission ensures that we don't +* find it from a io_getevents() thread before the issuer is done accessing +* the kiocb cookie. +*/ +static void aio_iopoll_iocb_issued (struct aio_kiocb *kiocb) +{ +/* +* For fast devices, IO may have already completed. If it has, add +* it to the front so we find it first. We can't add to the poll_done +* list as that's unlocked from the completion side. +*/ +const int front_add =test_bit (IOCB_POLL_COMPLETED , &kiocb->ki_flags); +struct kioctx *ctx = kiocb->ki_ctx; + +spin_lock (&ctx->poll_lock); +if (front_add) +list_add (&kiocb->ki_list, &ctx->poll_submitted); +else +list_add_tail (&kiocb->ki_list, &ctx->poll_submitted); +spin_unlock (&ctx->poll_lock); +} + +static ssize_t aio_read (struct aio_kiocb *kiocb,const struct iocb *iocb,bool vectored,bool compat) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs; +struct kiocb *req = &kiocb->rw;struct iov_iter iter;struct file *file;ssize_t ret; - ret =aio_prep_rw (req, iocb); + ret =aio_prep_rw (kiocb, iocb);if (ret)return ret; file = req->ki_filp;@@ -1605,15 +1890,16 @@ static ssize_t aio_read(struct kiocb *req, const struct iocb *iocb,return ret; } -static ssize_t aio_write (struct kiocb *req,const struct iocb *iocb, +static ssize_t aio_write (struct aio_kiocb *kiocb,const struct iocb *iocb,bool vectored,bool compat) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs; +struct kiocb *req = &kiocb->rw;struct iov_iter iter;struct file *file;ssize_t ret; - ret =aio_prep_rw (req, iocb); + ret =aio_prep_rw (kiocb, iocb);if (ret)return ret; file = req->ki_filp;@@ -1884,7 +2170,8 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb,return -EINVAL ; } -if (!get_reqs_available (ctx)) +/* Poll IO doesn't need ring reservations */ +if (!(ctx->flags &IOCTX_FLAG_IOPOLL ) && !get_reqs_available (ctx))return -EAGAIN ; ret = -EAGAIN ;@@ -1907,8 +2194,8 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb, } } -/* Don't support cancel on user mapped iocbs */ -if (!(ctx->flags &IOCTX_FLAG_USERIOCB )) { +/* Don't support cancel on user mapped iocbs or polled context */ +if (!(ctx->flags & (IOCTX_FLAG_USERIOCB |IOCTX_FLAG_IOPOLL ))) { ret =put_user (KIOCB_KEY , &user_iocb->aio_key);if (unlikely (ret)) {pr_debug ("EFAULT: aio_key\n" );@@ -1919,26 +2206,33 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb, req->ki_user_iocb = user_iocb; req->ki_user_data = iocb->aio_data; + ret = -EINVAL ;switch (iocb->aio_lio_opcode) {case IOCB_CMD_PREAD : - ret =aio_read (&req->rw, iocb, false, compat); + ret =aio_read (req, iocb, false, compat);break ;case IOCB_CMD_PWRITE : - ret =aio_write (&req->rw, iocb, false, compat); + ret =aio_write (req, iocb, false, compat);break ;case IOCB_CMD_PREADV : - ret =aio_read (&req->rw, iocb, true, compat); + ret =aio_read (req, iocb, true, compat);break ;case IOCB_CMD_PWRITEV : - ret =aio_write (&req->rw, iocb, true, compat); + ret =aio_write (req, iocb, true, compat);break ;case IOCB_CMD_FSYNC : +if (ctx->flags &IOCTX_FLAG_IOPOLL ) +break ; ret =aio_fsync (&req->fsync, iocb, false);break ;case IOCB_CMD_FDSYNC : +if (ctx->flags &IOCTX_FLAG_IOPOLL ) +break ; ret =aio_fsync (&req->fsync, iocb, true);break ;case IOCB_CMD_POLL : +if (ctx->flags &IOCTX_FLAG_IOPOLL ) +break ; ret =aio_poll (req, iocb);break ;default :@@ -1954,13 +2248,21 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb,*/ if (ret) goto out_put_req; +if (ctx->flags &IOCTX_FLAG_IOPOLL ) { +if (test_bit (IOCB_POLL_BUSY , &req->ki_flags)) { + ret = -EAGAIN ; + goto out_put_req; + } +aio_iopoll_iocb_issued (req); + }return 0 ; out_put_req:if (req->ki_eventfd)eventfd_ctx_put (req->ki_eventfd);iocb_put (req); out_put_reqs_available: -put_reqs_available (ctx,1 ); +if (!(ctx->flags &IOCTX_FLAG_IOPOLL )) +put_reqs_available (ctx,1 );return ret; }@@ -2136,7 +2438,7 @@ SYSCALL_DEFINE3(io_cancel, aio_context_t, ctx_id, struct iocb __user *, iocb,if (unlikely (!ctx))return -EINVAL ; -if (ctx->flags &IOCTX_FLAG_USERIOCB ) +if (ctx->flags & (IOCTX_FLAG_USERIOCB |IOCTX_FLAG_IOPOLL )) goto err;spin_lock_irq (&ctx->ctx_lock);@@ -2171,8 +2473,12 @@ static long do_io_getevents(aio_context_t ctx_id,long ret = -EINVAL ;if (likely (ioctx)) { -if (likely (min_nr <= nr && min_nr >=0 )) - ret =read_events (ioctx, min_nr, nr, events, until); +if (likely (min_nr <= nr && min_nr >=0 )) { +if (ioctx->flags &IOCTX_FLAG_IOPOLL ) + ret =aio_iopoll_check (ioctx, min_nr, nr, events); +else + ret =read_events (ioctx, min_nr, nr, events, until); + }percpu_ref_put (&ioctx->users); }include/uapi/linux/aio_abi.h +3
-0@@ -52,9 +52,11 @@ enum {* is valid. * IOCB_FLAG_IOPRIO - Set if the "aio_reqprio" member of the "struct iocb" * is valid. +* IOCB_FLAG_HIPRI - Use IO completion polling */ #define IOCB_FLAG_RESFD (1 << 0)#define IOCB_FLAG_IOPRIO (1 << 1) +#define IOCB_FLAG_HIPRI (1 << 2)/* read() from /dev/aio returns these structures. */ struct io_event {@@ -107,6 +109,7 @@ struct iocb { };/* 64 bytes */ #define IOCTX_FLAG_USERIOCB (1 << 0)/* iocbs are user mapped */ +#define IOCTX_FLAG_IOPOLL (1 << 1)/* io_context is polled */ #undef IFBIG#undef IFLITTLE
Jens Axboe
maintainerRe: [PATCH 17/20] aio: support for IO polling
Quoted 4 lines · Benny Halevy
Quoted 4 lines
fs/aio.c +0
-0@@ -818,11 +853,15 @@ static int kill_ioctx(struct mm_struct *mm, struct kioctx *ctx, {struct kioctx_table *table;
Quoted 15 lines
fs/aio.c static int kill_ioctx(struct mm_struct *mm, struct kioctx *ctx, · line 864
Quoted 9 lines
fs/aio.c +4
-0+mutex_lock (&ctx->getevents_lock);spin_lock (&mm->ioctx_lock);if (atomic_xchg (&ctx->dead,1 )) {spin_unlock (&mm->ioctx_lock); +mutex_unlock (&ctx->getevents_lock);return -EINVAL ; } +aio_iopoll_reap_events (ctx); +mutex_unlock (&ctx->getevents_lock);Is it worth handling the mutex lock and calling
aio_iopoll_reap_eventsonly if (ctx->flags&IOCTX_FLAG_IOPOLL)? If so, testing it can be removed fromaio_iopoll_reap_events()(and maybe it could even be open coded
here since this is its only call site apparently)
I don't think it really matters, this only happens when you tear down an io_context. FWIW, I think it's cleaner to retain the test in the function, not outside it.
Quoted 4 lines
Quoted 4 lines
(quoted diff) +0
-0@@ -1072,6 +1112,15 @@ static inline void iocb_put(struct aio_kiocb *iocb) } }
Quoted 10 lines
fs/aio.c static inline void iocb_put(struct aio_kiocb *iocb) · line 1117
Quoted 3 lines
fs/aio.c +3
-0+static void iocb_put_many (struct kioctx *ctx,void **iocbs,int *nr) +{ +if (nr) {How can nr by NULL?
And what's the point of supporting this case?
Did you mean: if (*nr)?
(In this case, if safe to call the functions below with *nr==0, I'm not sure it's worth optimizing... especially since this is a static function and its callers make sure to call it only when *nr > 0)
Indeed, that should be if (*nr), thanks! The slub implementation of the bulk free complains if you pass in nr == 0. Outside of that, a single check should be better than checking in multiple spots.
Quoted 4 lines
Quoted 4 lines
fs/aio.c +0
-0@@ -1261,6 +1310,166 @@ static bool aio_read_events(struct kioctx *ctx, long min_nr, long nr,return ret <0 || *i >= min_nr; }
Quoted 16 lines
Quoted 4 lines
fs/aio.c +4
-0+#define AIO_IOPOLL_BATCH 8 + +/* +* Process completed iocb iopoll entries, copying the result to userspace. fs/aio.c static bool aio_read_events(struct kioctx *ctx, long min_nr, long nr, · line 1326
Quoted 10 lines
fs/aio.c +10
-0+*/ +static long aio_iopoll_reap (struct kioctx *ctx,struct io_event __user *evs, +unsigned int *nr_events,long max) +{ +void *iocbs[AIO_IOPOLL_BATCH ]; +struct aio_kiocb *iocb, *n; +int to_free =0 , ret =0 ; + +list_for_each_entry_safe (iocb, n, &ctx->poll_completing, ki_list) { +if (*nr_events == max)
*nr_events>= max would be safer.
I don't see how we can get there with it being larger than already, that would be a big bug if we fill more events than userspace asked for.
Quoted 7 lines
fs/aio.c static inline void aio_rw_done(struct kiocb *req, ssize_t ret) · line 1832
Quoted 5 lines
fs/aio.c +1
-0@@ -1570,17 +1829,43 @@ static inline void aio_rw_done(struct kiocb *req, ssize_t ret)default : req->ki_complete (req, ret,0 ); } +nit: this hunk is probably unintentional
Looks like it, I'll kill it.
Signature · 1 line
Jens Axboe
Re: [PATCH 17/20] aio: support for IO polling
Quoted 40 lines · Jens Axboe
Quoted 4 lines · Benny Halevy
Quoted 4 lines
fs/aio.c +0
-0@@ -818,11 +853,15 @@ static int kill_ioctx(struct mm_struct *mm, struct kioctx *ctx, {struct kioctx_table *table;Quoted 15 lines
fs/aio.c static int kill_ioctx(struct mm_struct *mm, struct kioctx *ctx, · line 864
Quoted 9 lines
fs/aio.c +4
-0+mutex_lock (&ctx->getevents_lock);spin_lock (&mm->ioctx_lock);if (atomic_xchg (&ctx->dead,1 )) {spin_unlock (&mm->ioctx_lock); +mutex_unlock (&ctx->getevents_lock);return -EINVAL ; } +aio_iopoll_reap_events (ctx); +mutex_unlock (&ctx->getevents_lock);Is it worth handling the mutex lock and calling
aio_iopoll_reap_eventsonly if (ctx->flags&IOCTX_FLAG_IOPOLL)? If so, testing it can be removed fromaio_iopoll_reap_events()(and maybe it could even be open coded
here since this is its only call site apparently)I don't think it really matters, this only happens when you tear down an
io_context. FWIW, I think it's cleaner to retain the test in the function, not outside it.Quoted 4 lines
Quoted 4 lines
(quoted diff) +0
-0@@ -1072,6 +1112,15 @@ static inline void iocb_put(struct aio_kiocb *iocb) } }Quoted 10 lines
fs/aio.c static inline void iocb_put(struct aio_kiocb *iocb) · line 1117
Quoted 3 lines
fs/aio.c +3
-0+static void iocb_put_many (struct kioctx *ctx,void **iocbs,int *nr) +{ +if (nr) {How can nr by NULL?
And what's the point of supporting this case?
Did you mean: if (*nr)?
(In this case, if safe to call the functions below with *nr==0, I'm not sure it's worth optimizing... especially since this is a static function and its callers make sure to call it only when *nr > 0)
Quoted 4 lines
Indeed, that should be if (*nr), thanks! The slub implementation of the bulk free complains if you pass in nr == 0. Outside of that, a single check should be better than checking in multiple spots.
Cool. The compiler might also optimize it away when inlining this function if the caller tests *nr for being non-zero too.
Quoted 21 lines
Quoted 4 lines
Quoted 4 lines
fs/aio.c +0
-0@@ -1261,6 +1310,166 @@ static bool aio_read_events(struct kioctx *ctx, long min_nr, long nr,return ret <0 || *i >= min_nr; }Quoted 16 lines
Quoted 4 lines
fs/aio.c +4
-0+#define AIO_IOPOLL_BATCH 8 + +/* +* Process completed iocb iopoll entries, copying the result to userspace. fs/aio.c static bool aio_read_events(struct kioctx *ctx, long min_nr, long nr, · line 1326
Quoted 10 lines
fs/aio.c +10
-0+*/ +static long aio_iopoll_reap (struct kioctx *ctx,struct io_event __user *evs, +unsigned int *nr_events,long max) +{ +void *iocbs[AIO_IOPOLL_BATCH ]; +struct aio_kiocb *iocb, *n; +int to_free =0 , ret =0 ; + +list_for_each_entry_safe (iocb, n, &ctx->poll_completing, ki_list) { +if (*nr_events == max)
*nr_events>= max would be safer.
Quoted 3 lines
I don't see how we can get there with it being larger than already, that would be a big bug if we fill more events than userspace asked for.
Currently we indeed can't, but if the code changes in the future and we do, this will reduce the damage - hence being safer (and it costs nothing in terms of performance).
Quoted 11 lines
Quoted 7 lines
fs/aio.c static inline void aio_rw_done(struct kiocb *req, ssize_t ret) · line 1832
Quoted 5 lines
fs/aio.c +1
-0@@ -1570,17 +1829,43 @@ static inline void aio_rw_done(struct kiocb *req, ssize_t ret)default : req->ki_complete (req, ret,0 ); } +nit: this hunk is probably unintentional
Looks like it, I'll kill it.
Jens Axboe
maintainerRe: [PATCH 17/20] aio: support for IO polling
Quoted 8 lines · Benny Halevy
Quoted 4 lines
I don't see how we can get there with it being larger than already, that would be a big bug if we fill more events than userspace asked for.
Currently we indeed can't, but if the code changes in the future and we do, this will reduce the damage - hence being safer (and it costs nothing in terms of performance).
The thing is, if we're ever over max, we have potentially corrupted user space memory by copying back too many events. So if anything, it should be a BUG() condition, not just a check.
Signature · 1 line
Jens Axboe
Re: [PATCH 17/20] aio: support for IO polling
Quoted 6 lines · Jens Axboe
Quoted 5 lines · Benny Halevy
Quoted 4 lines
I don't see how we can get there with it being larger than already, that would be a big bug if we fill more events than userspace asked for.
Quoted 8 lines
Quoted 3 lines
Currently we indeed can't, but if the code changes in the future and we do, this will reduce the damage - hence being safer (and it costs nothing in terms of performance).
The thing is, if we're ever over max, we have potentially corrupted user space memory by copying back too many events. So if anything, it should be a
BUG()condition, not just a check.
Agreed.
Jens Axboe
maintainerPATCH 18/20 aio: add submission side request cache
In reply to Jens AxboeWe have to add each submitted polled request to the io_context poll_submitted list, which means we have to grab the poll_lock. We already use the block plug to batch submissions if we're doing a batch of IO submissions, extend that to cover the poll requests internally as well.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 136 +++++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 113 insertions(+), 23 deletions(-)
fs/aio.c +113 -23
@@ -247,6 +247,15 @@ static const struct address_space_operations aio_ctx_aops;static const unsigned int iocb_page_shift =ilog2 (PAGE_SIZE /sizeof (struct iocb )); +/* +* We rely on block level unplugs to flush pending requests, if we schedule +*/ +#ifdef CONFIG_BLOCK +static const bool aio_use_state_req_list = true; +#else +static const bool aio_use_state_req_list = false; +#endif +static void aio_useriocb_free (struct kioctx *);static void aio_iopoll_reap_events (struct kioctx *);
@@ -1724,6 +1733,21 @@ static void aio_complete_rw_poll(struct kiocb *kiocb, long res, long res2) } } +struct aio_submit_state { +struct kioctx *ctx; + +struct blk_plug plug; +#ifdef CONFIG_BLOCK +struct blk_plug_cb plug_cb; +#endif + +/* +* Polled iocbs that have been submitted, but not added to the ctx yet +*/ +struct list_head req_list; +unsigned int req_count; +}; +static int aio_prep_rw (struct aio_kiocb *kiocb,const struct iocb *iocb) {struct kioctx *ctx = kiocb->ki_ctx;
@@ -1832,13 +1856,28 @@ static inline void aio_rw_done(struct kiocb *req, ssize_t ret) } +/* +* Called either at the end of IO submission, or through a plug callback +* because we're going to schedule. Moves out local batch of requests to +* the ctx poll list, so they can be found for polling + reaping. +*/ +static void aio_flush_state_reqs (struct kioctx *ctx, +struct aio_submit_state *state) +{ +spin_lock (&ctx->poll_lock); +list_splice_tail_init (&state->req_list, &ctx->poll_submitted); +spin_unlock (&ctx->poll_lock); + state->req_count =0 ; +} +/* * After the iocb has been issued, it's safe to be found on the poll list. * Adding the kiocb to the list AFTER submission ensures that we don't * find it from a io_getevents() thread before the issuer is done accessing * the kiocb cookie. */ -static void aio_iopoll_iocb_issued (struct aio_kiocb *kiocb) +static void aio_iopoll_iocb_issued (struct aio_submit_state *state, +struct aio_kiocb *kiocb) {/* * For fast devices, IO may have already completed. If it has, add
@@ -1848,12 +1887,21 @@ static void aio_iopoll_iocb_issued(struct aio_kiocb *kiocb)const int front_add =test_bit (IOCB_POLL_COMPLETED , &kiocb->ki_flags);struct kioctx *ctx = kiocb->ki_ctx; -spin_lock (&ctx->poll_lock); -if (front_add) -list_add (&kiocb->ki_list, &ctx->poll_submitted); -else -list_add_tail (&kiocb->ki_list, &ctx->poll_submitted); -spin_unlock (&ctx->poll_lock); +if (!state || !aio_use_state_req_list) { +spin_lock (&ctx->poll_lock); +if (front_add) +list_add (&kiocb->ki_list, &ctx->poll_submitted); +else +list_add_tail (&kiocb->ki_list, &ctx->poll_submitted); +spin_unlock (&ctx->poll_lock); + }else { +if (front_add) +list_add (&kiocb->ki_list, &state->req_list); +else +list_add_tail (&kiocb->ki_list, &state->req_list); +if (++state->req_count >=AIO_IOPOLL_BATCH ) +aio_flush_state_reqs (ctx, state); + } } static ssize_t aio_read(struct aio_kiocb *kiocb, const struct iocb *iocb,
@@ -2149,7 +2197,8 @@ static ssize_t aio_poll(struct aio_kiocb *aiocb, const struct iocb *iocb) }static int __io_submit_one (struct kioctx *ctx,const struct iocb *iocb, -struct iocb __user *user_iocb,bool compat) +struct iocb __user *user_iocb, +struct aio_submit_state *state,bool compat) {struct aio_kiocb *req;ssize_t ret;
@@ -2253,7 +2302,7 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb, ret = -EAGAIN ; goto out_put_req; } -aio_iopoll_iocb_issued (req); +aio_iopoll_iocb_issued (state, req); }return 0 ; out_put_req:
@@ -2267,7 +2316,7 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb, }static int io_submit_one (struct kioctx *ctx,struct iocb __user *user_iocb, -bool compat) +struct aio_submit_state *state,bool compat) {struct iocb iocb, *iocbp;
@@ -2288,7 +2337,44 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb, iocbp = &iocb; } -return __io_submit_one (ctx, iocbp, user_iocb, compat); +return __io_submit_one (ctx, iocbp, user_iocb, state, compat); +} + +#ifdef CONFIG_BLOCK +static void aio_state_unplug (struct blk_plug_cb *cb,bool from_schedule) +{ +struct aio_submit_state *state; + + state = container_of(cb, struct aio_submit_state, plug_cb); +if (!list_empty (&state->req_list)) +aio_flush_state_reqs (state->ctx, state); +} +#endif + +/* +* Batched submission is done, ensure local IO is flushed out. +*/ +static void aio_submit_state_end (struct aio_submit_state *state) +{ +blk_finish_plug (&state->plug); +if (!list_empty (&state->req_list)) +aio_flush_state_reqs (state->ctx, state); +} + +/* +* Start submission side cache. +*/ +static void aio_submit_state_start (struct aio_submit_state *state, +struct kioctx *ctx) +{ + state->ctx = ctx; +INIT_LIST_HEAD (&state->req_list); + state->req_count =0 ; +#ifdef CONFIG_BLOCK + state->plug_cb.callback = aio_state_unplug; +blk_start_plug (&state->plug); +list_add (&state->plug_cb.list, &state->plug.cb_list); +#endif }/*
@@ -2312,10 +2398,10 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb,SYSCALL_DEFINE3 (io_submit, aio_context_t, ctx_id, long, nr, struct iocb __user * __user *, iocbpp) { +struct aio_submit_state state, *statep =NULL ;struct kioctx *ctx;long ret =0 ;int i =0 ; -struct blk_plug plug;if (unlikely (nr <0 ))return -EINVAL ;
@@ -2329,8 +2415,10 @@ SYSCALL_DEFINE3(io_submit, aio_context_t, ctx_id, long, nr,if (nr > ctx->nr_events) nr = ctx->nr_events; -if (nr >AIO_PLUG_THRESHOLD ) -blk_start_plug (&plug); +if (nr >AIO_PLUG_THRESHOLD ) { +aio_submit_state_start (&state, ctx); + statep = &state; + }for (i =0 ; i < nr; i++) {struct iocb __user *user_iocb;
@@ -2339,12 +2427,12 @@ SYSCALL_DEFINE3(io_submit, aio_context_t, ctx_id, long, nr,break ; } - ret =io_submit_one (ctx, user_iocb, false); + ret =io_submit_one (ctx, user_iocb, statep, false);if (ret)break ; } -if (nr >AIO_PLUG_THRESHOLD ) -blk_finish_plug (&plug); +if (statep) +aio_submit_state_end (statep);percpu_ref_put (&ctx->users);return i ? i : ret;
@@ -2354,10 +2442,10 @@ SYSCALL_DEFINE3(io_submit, aio_context_t, ctx_id, long, nr,COMPAT_SYSCALL_DEFINE3 (io_submit, compat_aio_context_t, ctx_id, int, nr, compat_uptr_t __user *, iocbpp) { +struct aio_submit_state state, *statep =NULL ;struct kioctx *ctx;long ret =0 ;int i =0 ; -struct blk_plug plug;if (unlikely (nr <0 ))return -EINVAL ;
@@ -2371,8 +2459,10 @@ COMPAT_SYSCALL_DEFINE3(io_submit, compat_aio_context_t, ctx_id,if (nr > ctx->nr_events) nr = ctx->nr_events; -if (nr >AIO_PLUG_THRESHOLD ) -blk_start_plug (&plug); +if (nr >AIO_PLUG_THRESHOLD ) { +aio_submit_state_start (&state, ctx); + statep = &state; + }for (i =0 ; i < nr; i++) {compat_uptr_t user_iocb;
@@ -2381,12 +2471,12 @@ COMPAT_SYSCALL_DEFINE3(io_submit, compat_aio_context_t, ctx_id,break ; } - ret =io_submit_one (ctx,compat_ptr (user_iocb), true); + ret =io_submit_one (ctx,compat_ptr (user_iocb), statep, true);if (ret)break ; } -if (nr >AIO_PLUG_THRESHOLD ) -blk_finish_plug (&plug); +if (statep) +aio_submit_state_end (statep);percpu_ref_put (&ctx->users);return i ? i : ret;
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 19/20 fs: add fget_many() and fput_many()
In reply to Jens AxboeSome uses cases repeatedly get and put references to the same file, but the only exposed interface is doing these one at the time. As each of these entail an atomic inc or dec on a shared structure, that cost can add up.
Add fget_many(), which works just like fget(), except it takes an argument for how many references to get on the file. Ditto fput_many(), which can drop an arbitrary number of references to a file.
Signed-off-by Jens Axboe <[email protected]>
--- fs/file.c | 15 ++++++++++----- fs/file_table.c | 10 ++++++++-- include/linux/file.h | 2 ++ include/linux/fs.h | 3 ++- 4 files changed, 22 insertions(+), 8 deletions(-)
fs/file.c +10 -5
@@ -676,7 +676,7 @@ void do_close_on_exec(struct files_struct *files)spin_unlock (&files->file_lock); } -static struct file *__fget (unsigned int fd,fmode_t mask) +static struct file *__fget (unsigned int fd,fmode_t mask,unsigned int refs) {struct files_struct *files = current->files;struct file *file;
@@ -691,7 +691,7 @@ static struct file *__fget(unsigned int fd, fmode_t mask)*/ if (file->f_mode & mask) file =NULL ; -else if (!get_file_rcu (file)) +else if (!get_file_rcu_many (file, refs)) goto loop; }rcu_read_unlock ();
@@ -699,15 +699,20 @@ static struct file *__fget(unsigned int fd, fmode_t mask)return file; } +struct file *fget_many (unsigned int fd,unsigned int refs) +{ +return __fget (fd,FMODE_PATH , refs); +} +struct file *fget (unsigned int fd) { -return __fget (fd,FMODE_PATH ); +return fget_many (fd,1 ); }EXPORT_SYMBOL (fget);struct file *fget_raw (unsigned int fd) { -return __fget (fd,0 ); +return __fget (fd,0 ,1 ); }EXPORT_SYMBOL (fget_raw);
@@ -738,7 +743,7 @@ static unsigned long __fget_light(unsigned int fd, fmode_t mask)return 0 ;return (unsigned long )file; }else { - file =__fget (fd, mask); + file =__fget (fd, mask,1 );if (!file)return 0 ;return FDPUT_FPUT | (unsigned long )file;
fs/file_table.c +8 -2
@@ -326,9 +326,9 @@ void flush_delayed_fput(void) static DECLARE_DELAYED_WORK(delayed_fput_work, delayed_fput); -void fput (struct file *file) +void fput_many (struct file *file,unsigned int refs) { -if (atomic_long_dec_and_test (&file->f_count)) { +if (atomic_long_sub_and_test (refs, &file->f_count)) {struct task_struct *task = current;if (likely (!in_interrupt () && !(task->flags &PF_KTHREAD ))) {
@@ -347,6 +347,12 @@ void fput(struct file *file) } } +void fput (struct file *file) +{ +fput_many (file,1 ); +} + +/* * synchronous analog of fput(); for kernel threads that might be needed * in some umount() (and thus can't use flush_delayed_fput() without
include/linux/file.h +2 -0
@@ -13,6 +13,7 @@struct file ;extern void fput (struct file *); +extern void fput_many (struct file *,unsigned int );struct file_operations ;struct vfsmount ;
@@ -44,6 +45,7 @@ static inline void fdput(struct fd fd) }extern struct file *fget (unsigned int fd); +extern struct file *fget_many (unsigned int fd,unsigned int refs);extern struct file *fget_raw (unsigned int fd);extern unsigned long __fdget (unsigned int fd);extern unsigned long __fdget_raw (unsigned int fd);
include/linux/fs.h +2 -1
@@ -952,7 +952,8 @@ static inline struct file *get_file(struct file *f)atomic_long_inc (&f->f_count);return f; } -#define get_file_rcu (x) atomic_long_inc_not_zero(&(x)->f_count) +#define get_file_rcu_many (x, cnt) atomic_long_add_unless(&(x)->f_count, (cnt), 0) +#define get_file_rcu (x) get_file_rcu_many((x), 1)#define fput_atomic (x) atomic_long_add_unless(&(x)->f_count, -1, 1)#define file_count (x) atomic_long_read(&(x)->f_count)
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 20/20 aio: use fget/fput_many() for file references
In reply to Jens AxboeOn the submission side, add file reference batching to the aio_submit_state. We get as many references as the number of iocbs we are submitting, and drop unused ones if we end up switching files. The assumption here is that we're usually only dealing with one fd, and if there are multiple, hopefuly they are at least somewhat ordered. Could trivially be extended to cover multiple fds, if needed.
On the completion side we do the same thing, except this is trivially done just locally in aio_iopoll_reap().
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 106 +++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 91 insertions(+), 15 deletions(-)
fs/aio.c +91 -15
@@ -1329,7 +1329,8 @@ static long aio_iopoll_reap(struct kioctx *ctx, struct io_event __user *evs, {void *iocbs[AIO_IOPOLL_BATCH ];struct aio_kiocb *iocb, *n; -int to_free =0 , ret =0 ; +int file_count, to_free =0 , ret =0 ; +struct file *file =NULL ;list_for_each_entry_safe (iocb, n, &ctx->poll_completing, ki_list) {if (*nr_events == max)
@@ -1342,7 +1343,20 @@ static long aio_iopoll_reap(struct kioctx *ctx, struct io_event __user *evs,list_del (&iocb->ki_list); iocbs[to_free++] = iocb; -fput (iocb->rw.ki_filp); +/* +* Batched puts of the same file, to avoid dirtying the +* file usage count multiple times, if avoidable. +*/ +if (!file) { + file = iocb->rw.ki_filp; + file_count =1 ; + }else if (file == iocb->rw.ki_filp) { + file_count++; + }else { +fput_many (file, file_count); + file = iocb->rw.ki_filp; + file_count =1 ; + }if (evs &©_to_user (evs + *nr_events, &iocb->ki_ev,sizeof (iocb->ki_ev))) {
@@ -1352,6 +1366,9 @@ static long aio_iopoll_reap(struct kioctx *ctx, struct io_event __user *evs, (*nr_events)++; } +if (file) +fput_many (file, file_count); +if (to_free)iocb_put_many (ctx, iocbs, &to_free);
@@ -1746,15 +1763,69 @@ struct aio_submit_state {*/ struct list_head req_list;unsigned int req_count; + +/* +* File reference cache +*/ +struct file *file; +unsigned int fd; +unsigned int has_refs; +unsigned int used_refs; +unsigned int ios_left; }; -static int aio_prep_rw (struct aio_kiocb *kiocb,const struct iocb *iocb) +static void aio_file_put (struct aio_submit_state *state) +{ +if (state->file) { +int diff = state->has_refs - state->used_refs; + +if (diff) +fput_many (state->file, diff); + state->file =NULL ; + } +} + +/* +* Get as many references to a file as we have IOs left in this submission, +* assuming most submissions are for one file, or at least that each file +* has more than one submission. +*/ +static struct file *aio_file_get (struct aio_submit_state *state,int fd) +{ +if (!state) +return fget (fd); + +if (!state->file) { +get_file: + state->file =fget_many (fd, state->ios_left); +if (!state->file) +return NULL ; + + state->fd = fd; + state->has_refs = state->ios_left; + state->used_refs =1 ; + state->ios_left--; +return state->file; + } + +if (state->fd == fd) { + state->used_refs++; + state->ios_left--; +return state->file; + } + +aio_file_put (state); + goto get_file; +} + +static int aio_prep_rw (struct aio_kiocb *kiocb,const struct iocb *iocb, +struct aio_submit_state *state) {struct kioctx *ctx = kiocb->ki_ctx;struct kiocb *req = &kiocb->rw;int ret; - req->ki_filp =fget (iocb->aio_fildes); + req->ki_filp =aio_file_get (state, iocb->aio_fildes);if (unlikely (!req->ki_filp))return -EBADF ; req->ki_pos = iocb->aio_offset;
@@ -1905,7 +1976,8 @@ static void aio_iopoll_iocb_issued(struct aio_submit_state *state, }static ssize_t aio_read (struct aio_kiocb *kiocb,const struct iocb *iocb, -bool vectored,bool compat) +struct aio_submit_state *state,bool vectored, +bool compat) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs;struct kiocb *req = &kiocb->rw;
@@ -1913,7 +1985,7 @@ static ssize_t aio_read(struct aio_kiocb *kiocb, const struct iocb *iocb,struct file *file;ssize_t ret; - ret =aio_prep_rw (kiocb, iocb); + ret =aio_prep_rw (kiocb, iocb, state);if (ret)return ret; file = req->ki_filp;
@@ -1939,7 +2011,8 @@ static ssize_t aio_read(struct aio_kiocb *kiocb, const struct iocb *iocb, }static ssize_t aio_write (struct aio_kiocb *kiocb,const struct iocb *iocb, -bool vectored,bool compat) +struct aio_submit_state *state,bool vectored, +bool compat) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs;struct kiocb *req = &kiocb->rw;
@@ -1947,7 +2020,7 @@ static ssize_t aio_write(struct aio_kiocb *kiocb, const struct iocb *iocb,struct file *file;ssize_t ret; - ret =aio_prep_rw (kiocb, iocb); + ret =aio_prep_rw (kiocb, iocb, state);if (ret)return ret; file = req->ki_filp;
@@ -2258,16 +2331,16 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb, ret = -EINVAL ;switch (iocb->aio_lio_opcode) {case IOCB_CMD_PREAD : - ret =aio_read (req, iocb, false, compat); + ret =aio_read (req, iocb, state, false, compat);break ;case IOCB_CMD_PWRITE : - ret =aio_write (req, iocb, false, compat); + ret =aio_write (req, iocb, state, false, compat);break ;case IOCB_CMD_PREADV : - ret =aio_read (req, iocb, true, compat); + ret =aio_read (req, iocb, state, true, compat);break ;case IOCB_CMD_PWRITEV : - ret =aio_write (req, iocb, true, compat); + ret =aio_write (req, iocb, state, true, compat);break ;case IOCB_CMD_FSYNC :if (ctx->flags &IOCTX_FLAG_IOPOLL )
@@ -2359,17 +2432,20 @@ static void aio_submit_state_end(struct aio_submit_state *state)blk_finish_plug (&state->plug);if (!list_empty (&state->req_list))aio_flush_state_reqs (state->ctx, state); +aio_file_put (state); }/* * Start submission side cache. */ static void aio_submit_state_start (struct aio_submit_state *state, -struct kioctx *ctx) +struct kioctx *ctx,int max_ios) { state->ctx = ctx;INIT_LIST_HEAD (&state->req_list); state->req_count =0 ; + state->file =NULL ; + state->ios_left = max_ios;#ifdef CONFIG_BLOCK state->plug_cb.callback = aio_state_unplug;blk_start_plug (&state->plug);
@@ -2416,7 +2492,7 @@ SYSCALL_DEFINE3(io_submit, aio_context_t, ctx_id, long, nr, nr = ctx->nr_events;if (nr >AIO_PLUG_THRESHOLD ) { -aio_submit_state_start (&state, ctx); +aio_submit_state_start (&state, ctx, nr); statep = &state; }for (i =0 ; i < nr; i++) {
@@ -2460,7 +2536,7 @@ COMPAT_SYSCALL_DEFINE3(io_submit, compat_aio_context_t, ctx_id, nr = ctx->nr_events;if (nr >AIO_PLUG_THRESHOLD ) { -aio_submit_state_start (&state, ctx); +aio_submit_state_start (&state, ctx, nr); statep = &state; }for (i =0 ; i < nr; i++) {
Signature · 1 line
2.17.1