Support for polled and buffered aio (and more)
Subsystems (11)
Jens Axboe
maintainerPATCH v8 1/1 Support for polled and buffered aio (and more)
We're getting closer... Some good bug fixes in this release, fully stable for me.
I decided to make the SQ/CQ ring interface the primary innovation here, and build the new features on top of that. This means that the io_submit/io_getevents interface doesn't get any of the advanced features, outside of polled IO. If you want any of the new fancy, you really sould use the ring interface, it's more efficient.
As before, there's a sample application as an introduction to how to use the ring interface:
http://git.kernel.dk/cgit/fio/plain/t/aio-ring.c
Outside of that, I also built a fio engine for it yesterday. If you update your fio to current -git, you'll have an aio-ring engine. As an example, if you want to run buffered async IO, you would do:
[aio-buffered]
ioengine=aio-ring
sqwq=1
filename=/dev/nvme1n1
direct=0
ramp_time=2
runtime=10s
iodepth=64
rw=randread
norandommap
bs=4k
which enables the SQ/CQ workqueue offload. As before, cached data is served without punting to a workqueue.
aio-ring also supports the sqthread interface, and it supports fixed buffers, and (of course) polled IO. For a speed run, you'd do:
[aio-polled]
ioengine=aio-ring
hipri=1
fixedbufs=1
filename=/dev/nvme1n1
direct=1
ramp_time=2
runtime=10s
iodepth=64
iodepth_batch_complete_max=64
iodepth_batch=32
rw=randread
norandommap
bs=4k
cpus_allowed=0
which enables polling (hipri=1) and fixed buffers (fixedbufs=1).
Interface wise, I made an important change to how the SQ ring works. Previously the ring was just an index into an array of iocbs. This was somewhat inflexible, as I found out when I converted applications to use it, as it requires an ordered relationship between application IO units and iocbs. Now the ring is just indexes into the iocb array, so you can submit multiple iocbs in one go without having to rely on iocbs being ordered in the array.
As before, find this in my aio-poll branch:
http://git.kernel.dk/cgit/linux-block/log/?h=aio-poll
or clone it here:
git://git.kernel.dk/linux-block aio-poll
and the code is based on the pending 4.21 block changes, for-4.21/block.
Since v7:
- Fix fput error handling in aio_prep_rw()
- Dropped USERIOCB for non-ring interface
- Fix CQ ring sizing
- Change sq_ring to be ring of indexes, not iocbs
- Rename aio_iocb_ring -> aio_sq_ring, aio_io_event_ring -> aio_cq_ring
- Fix not passing submission state always, if setup
- Fix double mmget/use_mm for SQTHREAD for rare conditions
- Fix __kthread_bind_mask() warning for SQTHREAD exit
- Comments
- Various bug fixes
Documentation/filesystems/vfs.txt | 3 + Documentation/sysctl/fs.txt | 8 +- arch/x86/entry/syscalls/syscall_64.tbl | 2 + block/bio.c | 33 +- fs/aio.c | 1891 ++++++++++++++++++++++-- fs/block_dev.c | 34 +- fs/file.c | 15 +- fs/file_table.c | 10 +- fs/gfs2/file.c | 2 + fs/iomap.c | 57 +- fs/xfs/xfs_file.c | 1 + include/linux/bio.h | 1 + include/linux/blk_types.h | 2 + include/linux/file.h | 2 + include/linux/fs.h | 5 +- include/linux/iomap.h | 1 + include/linux/syscalls.h | 4 + include/uapi/asm-generic/unistd.h | 4 +- include/uapi/linux/aio_abi.h | 37 + kernel/sys_ni.c | 2 + 20 files changed, 1916 insertions(+), 198 deletions(-)
Signature · 1 line
Jens Axboe
Jens Axboe
maintainerPATCH 1/26 fs: add an iopoll method to struct file_operations
From: 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.
Reviewed-by Johannes Thumshirn <[email protected]>
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
Jens Axboe
maintainerPATCH 2/26 block: add REQ_HIPRI_ASYNC
In reply to Jens AxboeFor the upcoming async polled IO, we can't sleep allocating requests. If we do, then we introduce a deadlock where the submitter already has async polled IO in-flight, but can't wait for them to complete since polled requests must be active found and reaped.
Signed-off-by Jens Axboe <[email protected]>
--- include/linux/blk_types.h | 1 + 1 file changed, 1 insertion(+)
include/linux/blk_types.h +1 -0
@@ -347,6 +347,7 @@ enum req_flag_bits {#define REQ_NOWAIT (1ULL << __REQ_NOWAIT)#define REQ_NOUNMAP (1ULL << __REQ_NOUNMAP)#define REQ_HIPRI (1ULL << __REQ_HIPRI) +#define REQ_HIPRI_ASYNC (REQ_HIPRI | REQ_NOWAIT)#define REQ_DRV (1ULL << __REQ_DRV)#define REQ_SWAP (1ULL << __REQ_SWAP)
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 3/26 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.
Reviewed-by Johannes Thumshirn <[email protected]>
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;
@@ -398,6 +406,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 ; }
@@ -2070,6 +2079,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
Jens Axboe
maintainerPATCH 4/26 block: use REQ_HIPRI_ASYNC for non-sync polled IO
In reply to Jens AxboeTell the block layer if it's a sync or async polled request, so it can do the right thing.
Signed-off-by Jens Axboe <[email protected]>
--- fs/block_dev.c | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-)
fs/block_dev.c +6 -2
@@ -402,8 +402,12 @@ __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 ) - bio->bi_opf |=REQ_HIPRI ; +if (iocb->ki_flags &IOCB_HIPRI ) { +if (!is_sync) + bio->bi_opf |=REQ_HIPRI_ASYNC ; +else + bio->bi_opf |=REQ_HIPRI ; + } qc =submit_bio (bio);WRITE_ONCE (iocb->ki_cookie, qc);
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 5/26 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]>
Modified to use REQ_HIPRI_ASYNC for async polled IO.
Signed-off-by Jens Axboe <[email protected]>
--- fs/gfs2/file.c | 2 ++ fs/iomap.c | 47 +++++++++++++++++++++++++++++-------------- fs/xfs/xfs_file.c | 1 + include/linux/iomap.h | 1 + 4 files changed, 36 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 +32 -15
86 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 6/26 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 7/26 aio: separate out ring reservation from req allocation
In reply to Jens AxboeFrom: Christoph Hellwig <[email protected]>
This is in preparation for certain types of IO not needing a ring reserveration.
Signed-off-by Christoph Hellwig <[email protected]>
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 8/26 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.
Reviewed-by Christoph Hellwig <[email protected]>
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-)
fs/aio.c +7 -2
@@ -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); + req =kmem_cache_alloc (kiocb_cachep,GFP_KERNEL );if (unlikely (!req))return NULL ;percpu_ref_get (&ctx->reqs); + req->ki_ctx = ctx;INIT_LIST_HEAD (&req->ki_list);refcount_set (&req->ki_refcnt,0 ); - req->ki_ctx = ctx; + req->ki_eventfd =NULL ;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 9/26 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.
Reviewed-by Christoph Hellwig <[email protected]>
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-)
fs/aio.c +14 -4
@@ -69,6 +69,12 @@ struct aio_ring {struct io_event io_events[0 ]; };/* 128 bytes + ring size */ +/* +* 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 +#define AIO_RING_PAGES 8struct kioctx_table {
@@ -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 10/26 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.
Reviewed-by Christoph Hellwig <[email protected]>
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-)
fs/aio.c +1 -2
@@ -1884,10 +1884,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 11/26 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
@@ -1420,7 +1420,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;
@@ -1461,7 +1461,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;
@@ -1500,8 +1500,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;
@@ -1533,8 +1533,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;
@@ -1589,7 +1589,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))
@@ -1717,7 +1718,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;
@@ -1789,27 +1790,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 ;
@@ -1823,14 +1820,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 ;
@@ -1845,32 +1842,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 ; }
@@ -1892,6 +1889,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); +} +/* 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
Signature · 1 line
2.17.1
Re: [PATCH 11/26] aio: split out iocb copy from io_submit_one()
Quoted 5 lines · Jens Axboe
In 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]>
Looks fine,
Reviewed-by Christoph Hellwig <[email protected]>
Jens Axboe
maintainerPATCH 12/26 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
@@ -1063,6 +1063,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. */
@@ -1090,10 +1099,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
Re: [PATCH 12/26] aio: abstract out io_event filler helper
Quoted 1 line · Jens Axboe
Signed-off-by Jens Axboe <[email protected]>
Looks fine,
Reviewed-by Christoph Hellwig <[email protected]>
Jens Axboe
maintainerPATCH 13/26 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 the flags, we add an iocb array and two user pointers for future use.
Signed-off-by Jens Axboe <[email protected]>
--- Documentation/sysctl/fs.txt | 8 +-- arch/x86/entry/syscalls/syscall_64.tbl | 1 + fs/aio.c | 76 +++++++++++++++++--------- include/linux/syscalls.h | 2 + include/uapi/asm-generic/unistd.h | 4 +- kernel/sys_ni.c | 1 + 6 files changed, 62 insertions(+), 30 deletions(-)
Documentation/sysctl/fs.txt +4 -4
@@ -47,10 +47,10 @@ Currently, these files are in /proc/sys/fs: aio-nr & aio-max-nr: aio-nr is the running total of the number of events specified on the -io_setup system call for all currently active aio contexts. If aio-nr -reaches aio-max-nr then io_setup will fail with EAGAIN. Note that -raising aio-max-nr does not result in the pre-allocation or re-sizing -of any kernel data structures. +io_setup/io_setup2 system call for all currently active aio contexts. +If aio-nr reaches aio-max-nr then io_setup will fail with EAGAIN. +Note that raising aio-max-nr does not result in the pre-allocation or +re-sizing of any kernel data structures. ==============================================================
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 +51 -25
136 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 ,void __user *,void __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 14/26 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 | 419 +++++++++++++++++++++++++++++++---- include/uapi/linux/aio_abi.h | 4 + 2 files changed, 384 insertions(+), 39 deletions(-)
fs/aio.c +380 -39
@@ -146,6 +146,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 */
@@ -198,14 +210,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 KIOCB_F_POLL_COMPLETED 0/* polled IO has completed */ +#define KIOCB_F_POLL_EAGAIN 1/* polled submission got EAGAIN */ +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----*/
@@ -222,6 +247,8 @@ static struct vfsmount *aio_mnt;static const struct file_operations aio_ring_fops;static const struct address_space_operations aio_ctx_aops; +static void aio_iopoll_reap_events (struct kioctx *); +static struct file *aio_private_file (struct kioctx *ctx,loff_t nr_pages) {struct file *file;
@@ -459,11 +486,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 )
@@ -547,6 +578,14 @@ static int aio_setup_ring(struct kioctx *ctx, unsigned int nr_events)return 0 ; } +/* +* Don't support cancel on anything that isn't old aio +*/ +static bool aio_ctx_supports_cancel (struct kioctx *ctx) +{ +return (ctx->flags &IOCTX_FLAG_IOPOLL ) ==0 ; +} +#define AIO_EVENTS_PER_PAGE (PAGE_SIZE / sizeof(struct io_event))#define AIO_EVENTS_FIRST_PAGE ((PAGE_SIZE - sizeof(struct aio_ring)) / sizeof(struct io_event))#define AIO_EVENTS_OFFSET (AIO_EVENTS_PER_PAGE - AIO_EVENTS_FIRST_PAGE)
@@ -557,6 +596,8 @@ void kiocb_set_cancel_fn(struct kiocb *iocb, kiocb_cancel_fn *cancel)struct kioctx *ctx = req->ki_ctx;unsigned long flags; +if (WARN_ON_ONCE (!aio_ctx_supports_cancel (ctx))) +return ;if (WARN_ON_ONCE (!list_empty (&req->ki_list)))return ;
@@ -745,6 +786,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;
@@ -816,11 +862,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 +1079,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 ;return req;
@@ -1070,6 +1121,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) {
@@ -1259,6 +1319,185 @@ 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 ; + +/* Shouldn't happen... */ +if (*nr_events >= max) +return 0 ; + +list_for_each_entry_safe (iocb, n, &ctx->poll_completing, ki_list) { +if (*nr_events == max) +break ; +if (!test_bit (KIOCB_F_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; +} + +/* +* Poll for a mininum of 'min' events, and a maximum of 'max'. Note that if +* min == 0 we consider that a non-spinning poll check - we'll still enter +* the driver poll loop, but only as a non-spinning completion check. +*/ +static int aio_iopoll_getevents (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 ((min && *nr_events >= min) || *nr_events >= max) +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 ((min && *nr_events >= min) || *nr_events >= max) +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 (KIOCB_F_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_getevents (ctx,NULL , &nr_events,1 ,UINT_MAX ); + } +} + +static int __aio_iopoll_check (struct kioctx *ctx,struct io_event __user *event, +unsigned int *nr_events,long min_nr,long max_nr) +{ +int ret =0 ; + +while (!*nr_events || !need_resched ()) { +int tmin =0 ; + +if (*nr_events < min_nr) + tmin = min_nr - *nr_events; + + ret =aio_iopoll_getevents (ctx, event, nr_events, tmin, max_nr); +if (ret <=0 ) +break ; + ret =0 ; + } + +return ret; +} + +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; + +/* 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; + } + + ret =__aio_iopoll_check (ctx, event, &nr_events, min_nr, nr); +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)
@@ -1303,7 +1542,9 @@ SYSCALL_DEFINE5(io_setup2, u32, nr_events, u32, flags, void __user *, user1,unsigned long ctx;long ret; -if (flags || user1 || user2) +if (user1 || user2) +return -EINVAL ; +if (flags & ~IOCTX_FLAG_IOPOLL )return -EINVAL ; ret =get_user (ctx, ctxp);
@@ -1429,13 +1670,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);
@@ -1447,19 +1683,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 (KIOCB_F_POLL_EAGAIN , &iocb->ki_flags); +else { +aio_fill_event (&iocb->ki_ev, iocb, res, res2); +set_bit (KIOCB_F_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 )
@@ -1485,9 +1750,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;
@@ -1532,15 +1823,40 @@ static inline void aio_rw_done(struct kiocb *req, ssize_t ret) } } -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 =test_bit (KIOCB_F_POLL_COMPLETED , &kiocb->ki_flags); +struct kioctx *ctx = kiocb->ki_ctx; + +spin_lock (&ctx->poll_lock); +if (front) +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;
@@ -1565,15 +1881,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;
@@ -1844,7 +2161,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 ;
@@ -1867,35 +2185,44 @@ 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; +if (aio_ctx_supports_cancel (ctx)) { + 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; 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 :
@@ -1911,13 +2238,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 (KIOCB_F_POLL_EAGAIN , &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; }
@@ -2073,6 +2408,9 @@ SYSCALL_DEFINE3(io_cancel, aio_context_t, ctx_id, struct iocb __user *, iocb,if (unlikely (!ctx))return -EINVAL ; +if (!aio_ctx_supports_cancel (ctx)) + goto err; +spin_lock_irq (&ctx->ctx_lock); kiocb =lookup_kiocb (ctx, iocb);if (kiocb) {
@@ -2089,9 +2427,8 @@ SYSCALL_DEFINE3(io_cancel, aio_context_t, ctx_id, struct iocb __user *, iocb,*/ ret = -EINPROGRESS ; } - +err:percpu_ref_put (&ctx->users); -return ret; }
@@ -2106,8 +2443,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 +4 -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 {
@@ -106,6 +108,8 @@ struct iocb {__u32 aio_resfd; };/* 64 bytes */ +#define IOCTX_FLAG_IOPOLL (1 << 0)/* io_context is polled */ +#undef IFBIG#undef IFLITTLE
Signature · 1 line
2.17.1
Re: [PATCH 14/26] aio: support for IO polling
Quoted 9 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.
Quoted 8 lines
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]>
Looks good to me, thanks.
Benny
Jens Axboe
maintainerRe: [PATCH 14/26] aio: support for IO polling
Quoted 16 lines · Benny Halevy
Quoted 15 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.
Quoted 4 lines
Quoted 2 lines
Signed-off-by Jens Axboe <[email protected]>Looks good to me, thanks.
Is that a reviewed-by?
Signature · 1 line
Jens Axboe
Benny Halevy
Reviewed-byRe: [PATCH 14/26] aio: support for IO polling
Quoted 17 lines · Jens Axboe
Quoted 16 lines · Benny Halevy
Quoted 15 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.
Quoted 7 lines
Quoted 4 lines
Quoted 2 lines
Signed-off-by Jens Axboe <[email protected]>Looks good to me, thanks.
Is that a reviewed-by?
Yes.
Reviewed-by Benny Halevy <[email protected]>
Jens Axboe
maintainerRe: [PATCH 14/26] aio: support for IO polling
Quoted 21 lines · Benny Halevy
Quoted 20 lines · Jens Axboe
Quoted 19 lines · Benny Halevy
Quoted 17 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]>
Quoted 8 lines
Quoted 4 lines
Quoted 1 line
Looks good to me, thanks.
Is that a reviewed-by?
Yes.
Reviewed-by Benny Halevy <[email protected]>
Thanks Benny, added.
Signature · 1 line
Jens Axboe
Jens Axboe
maintainerPATCH 15/26 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
@@ -233,6 +233,21 @@ struct aio_kiocb { }; }; +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; +}; +/*------ sysctl variables----*/ static DEFINE_SPINLOCK (aio_nr_lock);unsigned long aio_nr;/* current system wide number of aio requests */
@@ -247,6 +262,15 @@ static struct vfsmount *aio_mnt;static const struct file_operations aio_ring_fops;static const struct address_space_operations aio_ctx_aops; +/* +* 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_iopoll_reap_events (struct kioctx *);static struct file *aio_private_file (struct kioctx *ctx,loff_t nr_pages)
@@ -1823,13 +1847,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
@@ -1839,12 +1878,21 @@ static void aio_iopoll_iocb_issued(struct aio_kiocb *kiocb)const int front =test_bit (KIOCB_F_POLL_COMPLETED , &kiocb->ki_flags);struct kioctx *ctx = kiocb->ki_ctx; -spin_lock (&ctx->poll_lock); -if (front) -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) +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) +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,
@@ -2140,7 +2188,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;
@@ -2243,7 +2292,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:
@@ -2257,14 +2306,51 @@ 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;if (unlikely (copy_from_user (&iocb, user_iocb,sizeof (iocb))))return -EFAULT ; -return __io_submit_one (ctx, &iocb, user_iocb, compat); +return __io_submit_one (ctx, &iocb, 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 }/* sys_io_submit:
@@ -2282,10 +2368,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 ;
@@ -2299,8 +2385,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;
@@ -2309,12 +2397,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;
@@ -2324,10 +2412,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 ;
@@ -2341,8 +2429,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;
@@ -2351,12 +2441,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 16/26 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
Re: [PATCH 16/26] fs: add fget_many() and fput_many()
[adding Al, I think we really want him to review anything fget
related]
Quoted 42 lines · Jens Axboe
Some 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 likefget(), except it takes an argument for how many references to get on the file. Dittofput_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 +2
-2@@ -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; }
fs/file.c static struct file *__fget(unsigned int fd, fmode_t mask) · line 709
Quoted 10 lines
fs/file.c +6
-1+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 );
Can we just call __fget directly here? That is a little easier to follow, and might actually generate better code if the compiler is inliner challenged (which they often seem to be).
fs/file.c static struct file *__fget(unsigned int fd, fmode_t mask) · line 715
Quoted 6 lines
fs/file.c +6
-0+void fput (struct file *file) +{ +fput_many (file,1 ); +} + +
Double empty line here.
fs/file.c static struct file *__fget(unsigned int fd, fmode_t mask) · line 716
Quoted 1 line
fs/file.c +1
-0+#define get_file_rcu_many (x, cnt) atomic_long_add_unless(&(x)->f_count, (cnt), 0)
This could use a line break to be easier readable and not spill over 80 chars. Otherwise this looks fine to me.
Jens Axboe
maintainerRe: [PATCH 16/26] fs: add fget_many() and fput_many()
Quoted 47 lines · Christoph Hellwig
[adding Al, I think we really want him to review anything fget
related]Quoted 43 lines · Jens Axboe
Some 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 likefget(), except it takes an argument for how many references to get on the file. Dittofput_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 +2
-2@@ -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; }
Quoted 13 lines
fs/file.c static struct file *__fget(unsigned int fd, fmode_t mask) · line 709
Quoted 9 lines
fs/file.c +6
-1+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 );Can we just call
__fgetdirectly here? That is a little easier to follow, and might actually generate better code if the compiler is inliner challenged (which they often seem to be).
Certainly, it'd be the same thing. I kind of like having it follow the natural path to get there, unless there are strong objections to bypassing that step.
Quoted 8 lines
fs/file.c static struct file *__fget(unsigned int fd, fmode_t mask) · line 715
Quoted 6 lines
fs/file.c +6
-0+void fput (struct file *file) +{ +fput_many (file,1 ); +} + +Double empty line here.
Fixed.
Quoted 4 lines
fs/file.c static struct file *__fget(unsigned int fd, fmode_t mask) · line 716
Quoted 1 line
fs/file.c +1
-0+#define get_file_rcu_many (x, cnt) atomic_long_add_unless(&(x)->f_count, (cnt), 0)This could use a line break to be easier readable and not spill over 80 chars. Otherwise this looks fine to me.
Fixed.
Signature · 1 line
Jens Axboe
Jens Axboe
maintainerPATCH 17/26 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 | 110 +++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 94 insertions(+), 16 deletions(-)
fs/aio.c +94 -16
@@ -246,6 +246,15 @@ 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; };/*------ sysctl variables----*/
@@ -1353,7 +1362,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 ;/* Shouldn't happen... */ if (*nr_events >= max)
@@ -1370,7 +1380,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))) {
@@ -1380,6 +1403,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);
@@ -1740,13 +1766,60 @@ static void aio_complete_rw_poll(struct kiocb *kiocb, long res, long res2) } } -static int aio_prep_rw (struct aio_kiocb *kiocb,const struct iocb *iocb) +static void aio_file_put (struct aio_submit_state *state,struct file *file) +{ +if (!state) { +fput (file); + }else 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,NULL ); + 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;
@@ -1804,7 +1877,7 @@ static int aio_prep_rw(struct aio_kiocb *kiocb, const struct iocb *iocb)return 0 ; out_fput: -fput (req->ki_filp); +aio_file_put (state, req->ki_filp);return ret; }
@@ -1896,7 +1969,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;
@@ -1904,7 +1978,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;
@@ -1930,7 +2004,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;
@@ -1938,7 +2013,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;
@@ -2248,16 +2323,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 )
@@ -2335,17 +2410,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,NULL ); }/* * 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);
@@ -2386,7 +2464,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++) {
@@ -2430,7 +2508,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
Jens Axboe
maintainerPATCH 18/26 aio: split iocb init from allocation
In reply to Jens AxboeIn preparation from having pre-allocated requests, that we then just need to initialize before use.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-)
fs/aio.c +11 -6
@@ -1097,6 +1097,16 @@ static bool get_reqs_available(struct kioctx *ctx)return __get_reqs_available (ctx); } +static void aio_iocb_init (struct kioctx *ctx,struct aio_kiocb *req) +{ +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 ; +} +/* aio_get_req * Allocate a slot for an aio request. * Returns NULL if no requests are free.
@@ -1109,12 +1119,7 @@ static inline struct aio_kiocb *aio_get_req(struct kioctx *ctx)if (unlikely (!req))return NULL ; -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 ; +aio_iocb_init (ctx, req);return req; }
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 19/26 aio: batch aio_kiocb allocation
In reply to Jens AxboeSimilarly to how we use the state->ios_left to know how many references to get to a file, we can use it to allocate the aio_kiocb's we need in bulk.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 46 ++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 38 insertions(+), 8 deletions(-)
fs/aio.c +38 -8
@@ -233,6 +233,8 @@ struct aio_kiocb { }; }; +#define AIO_IOPOLL_BATCH 8 +struct aio_submit_state {struct kioctx *ctx;
@@ -247,6 +249,13 @@ struct aio_submit_state {struct list_head req_list;unsigned int req_count; +/* +* aio_kiocb alloc cache +*/ +void *iocbs[AIO_IOPOLL_BATCH ]; +unsigned int free_iocbs; +unsigned int cur_iocb; +/* * File reference cache */
@@ -1111,15 +1120,34 @@ static void aio_iocb_init(struct kioctx *ctx, struct aio_kiocb *req)* Allocate a slot for an aio request. * Returns NULL if no requests are free. */ -static inline struct aio_kiocb *aio_get_req (struct kioctx *ctx) +static struct aio_kiocb *aio_get_req (struct kioctx *ctx, +struct aio_submit_state *state) {struct aio_kiocb *req; - req =kmem_cache_alloc (kiocb_cachep,GFP_KERNEL ); -if (unlikely (!req)) -return NULL ; +if (!state) + req =kmem_cache_alloc (kiocb_cachep,GFP_KERNEL ); +else if (!state->free_iocbs) { +size_t size; +int ret; + + size =min_t (size_t, state->ios_left,ARRAY_SIZE (state->iocbs)); + ret =kmem_cache_alloc_bulk (kiocb_cachep,GFP_KERNEL , size, + state->iocbs); +if (ret <=0 ) +return ERR_PTR (-ENOMEM ); + state->free_iocbs = ret -1 ; + state->cur_iocb =1 ; + req = state->iocbs[0 ]; + }else { + req = state->iocbs[state->cur_iocb]; + state->free_iocbs--; + state->cur_iocb++; + } + +if (req) +aio_iocb_init (ctx, req); -aio_iocb_init (ctx, req);return req; }
@@ -1357,8 +1385,6 @@ 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. */
@@ -2295,7 +2321,7 @@ static int __io_submit_one(struct kioctx *ctx, const struct iocb *iocb,return -EAGAIN ; ret = -EAGAIN ; - req =aio_get_req (ctx); + req =aio_get_req (ctx, state);if (unlikely (!req)) goto out_put_reqs_available;
@@ -2416,6 +2442,9 @@ static void aio_submit_state_end(struct aio_submit_state *state)if (!list_empty (&state->req_list))aio_flush_state_reqs (state->ctx, state);aio_file_put (state,NULL ); +if (state->free_iocbs) +kmem_cache_free_bulk (kiocb_cachep, state->free_iocbs, + &state->iocbs[state->cur_iocb]); }/*
@@ -2427,6 +2456,7 @@ static void aio_submit_state_start(struct aio_submit_state *state, state->ctx = ctx;INIT_LIST_HEAD (&state->req_list); state->req_count =0 ; + state->free_iocbs =0 ; state->file =NULL ; state->ios_left = max_ios;#ifdef CONFIG_BLOCK
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 20/26 aio: split old ring complete out from aio_complete()
In reply to Jens AxboeSigned-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-)
fs/aio.c +12 -5
@@ -1205,12 +1205,9 @@ static void aio_fill_event(struct io_event *ev, struct aio_kiocb *iocb, ev->res2 = res2; } -/* aio_complete -* Called when the io request on the given iocb is complete. -*/ -static void aio_complete (struct aio_kiocb *iocb,long res,long res2) +static void aio_ring_complete (struct kioctx *ctx,struct aio_kiocb *iocb, +long res,long res2) { -struct kioctx *ctx = iocb->ki_ctx;struct aio_ring *ring;struct io_event *ev_page, *event;unsigned tail, pos, head;
@@ -1260,6 +1257,16 @@ static void aio_complete(struct aio_kiocb *iocb, long res, long res2)spin_unlock_irqrestore (&ctx->completion_lock, flags);pr_debug ("added to ring %p at [%u]\n" , iocb, tail); +} + +/* aio_complete +* Called when the io request on the given iocb is complete. +*/ +static void aio_complete (struct aio_kiocb *iocb,long res,long res2) +{ +struct kioctx *ctx = iocb->ki_ctx; + +aio_ring_complete (ctx, iocb, res, res2);/* * Check if the user asked us to deliver the result through an
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 21/26 aio: add support for submission/completion rings
In reply to Jens AxboeExperimental support for submitting and completing IO through rings shared between the application and kernel.
The submission rings are struct iocb, like we would submit through io_submit(), and the completion rings are struct io_event, like we would pass in (and copy back) from io_getevents().
A new system call is added for this, io_ring_enter(). This system call submits IO that is queued in the SQ ring, and/or completes IO and stores the results in the CQ ring.
This could be augmented with a kernel thread that does the submission and polling, then the application would never have to enter the kernel to do IO.
Sample application: http://git.kernel.dk/cgit/fio/plain/t/aio-ring.c
Signed-off-by Jens Axboe <[email protected]>
--- arch/x86/entry/syscalls/syscall_64.tbl | 1 + fs/aio.c | 484 +++++++++++++++++++++++-- include/linux/syscalls.h | 4 +- include/uapi/linux/aio_abi.h | 29 ++ kernel/sys_ni.c | 1 + 5 files changed, 493 insertions(+), 26 deletions(-)
arch/x86/entry/syscalls/syscall_64.tbl +1 -0
@@ -344,6 +344,7 @@ 333 common io_pgetevents __x64_sys_io_pgetevents 334 common rseq __x64_sys_rseq 335 common io_setup2 __x64_sys_io_setup2 +336 common io_ring_enter __x64_sys_io_ring_enter # # x32-specific system call numbers start at 512 to avoid cache impact
fs/aio.c +459 -25
623 lines not on this page · raw message
include/linux/syscalls.h +3 -1
@@ -287,8 +287,10 @@ 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 ,void __user *,void __user *, +asmlinkage long sys_io_setup2 (unsigned ,unsigned ,struct aio_cq_ring __user *, +struct aio_sq_ring __user *,aio_context_t __user *); +asmlinkage long sys_io_ring_enter (aio_context_t ,unsigned ,unsigned ,unsigned );asmlinkage long sys_io_destroy (aio_context_t ctx);asmlinkage long sys_io_submit (aio_context_t ,long ,struct iocb __user *__user *);
include/uapi/linux/aio_abi.h +29 -0
@@ -109,6 +109,35 @@ struct iocb { };/* 64 bytes */ #define IOCTX_FLAG_IOPOLL (1 << 0)/* io_context is polled */ +#define IOCTX_FLAG_SCQRING (1 << 1)/* Use SQ/CQ rings */ + +struct aio_sq_ring { +union { +struct { +u32 head;/* kernel consumer head */ +u32 tail;/* app producer tail */ +u32 nr_events;/* max events in ring */ +u64 iocbs;/* setup pointer to app iocbs */ + }; +u32 pad[16 ]; + }; +u32 array[0 ];/* actual ring, index to iocbs */ +}; + +struct aio_cq_ring { +union { +struct { +u32 head;/* app consumer head */ +u32 tail;/* kernel producer tail */ +u32 nr_events;/* max events in ring */ + }; +struct io_event pad; + }; +struct io_event events[0 ];/* ring, array of io_events */ +}; + +#define IORING_FLAG_SUBMIT (1 << 0) +#define IORING_FLAG_GETEVENTS (1 << 1)#undef IFBIG#undef IFLITTLE
kernel/sys_ni.c +1 -0
@@ -38,6 +38,7 @@ asmlinkage long sys_ni_syscall(void)COND_SYSCALL (io_setup);COND_SYSCALL (io_setup2); +COND_SYSCALL (io_ring_enter);COND_SYSCALL_COMPAT (io_setup);COND_SYSCALL (io_destroy);COND_SYSCALL (io_submit);
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 22/26 block: add BIO_HOLD_PAGES flag
In reply to Jens AxboeFor user mapped IO, we do get_user_pages() upfront, and then do a put_page() on each page at end_io time to release the page reference. In preparation for having permanently mapped pages, add a BIO_HOLD_PAGES flag that tells us not to release the pages, the caller will do that.
Signed-off-by Jens Axboe <[email protected]>
--- block/bio.c | 6 ++++-- include/linux/blk_types.h | 1 + 2 files changed, 5 insertions(+), 2 deletions(-)
block/bio.c +4 -2
@@ -1636,7 +1636,8 @@ static void bio_dirty_fn(struct work_struct *work) next = bio->bi_private;bio_set_pages_dirty (bio); -bio_release_pages (bio); +if (!bio_flagged (bio,BIO_HOLD_PAGES )) +bio_release_pages (bio);bio_put (bio); } }
@@ -1652,7 +1653,8 @@ void bio_check_pages_dirty(struct bio *bio) goto defer; } -bio_release_pages (bio); +if (!bio_flagged (bio,BIO_HOLD_PAGES )) +bio_release_pages (bio);bio_put (bio);return ; defer:
include/linux/blk_types.h +1 -0
@@ -228,6 +228,7 @@ struct bio {#define BIO_TRACE_COMPLETION 10/* bio_endio() should trace the final completion * of this bio. */ #define BIO_QUEUE_ENTERED 11/* can use blk_queue_enter_live() */ +#define BIO_HOLD_PAGES 12/* don't put O_DIRECT pages */ /* See BVEC_POOL_OFFSET below before adding new flags */
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 23/26 block: implement bio helper to add iter bvec pages to bio
In reply to Jens AxboeFor an ITER_BVEC, we can just iterate the iov and add the pages to the bio directly.
Signed-off-by Jens Axboe <[email protected]>
--- block/bio.c | 27 +++++++++++++++++++++++++++ include/linux/bio.h | 1 + 2 files changed, 28 insertions(+)
block/bio.c +27 -0
@@ -904,6 +904,33 @@ int bio_iov_iter_get_pages(struct bio *bio, struct iov_iter *iter) }EXPORT_SYMBOL_GPL (bio_iov_iter_get_pages); +/** +* bio_iov_bvec_add_pages - add pages from an ITER_BVEC to a bio +* @bio: bio to add pages to +* @iter: iov iterator describing the region to be added +* +* Iterate pages in the @iter and add them to the bio. We flag the +* @bio with BIO_HOLD_PAGES, telling IO completion not to free them. +*/ +int bio_iov_bvec_add_pages (struct bio *bio,struct iov_iter *iter) +{ +unsigned short orig_vcnt = bio->bi_vcnt; +const struct bio_vec *bv; + +do { +size_t size; + + bv = iter->bvec + iter->iov_offset; + size =bio_add_page (bio, bv->bv_page, bv->bv_len, bv->bv_offset); +if (size != bv->bv_len) +break ; +iov_iter_advance (iter, size); + }while (iov_iter_count (iter) && !bio_full (bio)); + +bio_set_flag (bio,BIO_HOLD_PAGES ); +return bio->bi_vcnt > orig_vcnt ?0 : -EINVAL ; +} +static void submit_bio_wait_endio (struct bio *bio) {complete (bio->bi_private);
include/linux/bio.h +1 -0
@@ -434,6 +434,7 @@ bool __bio_try_merge_page(struct bio *bio, struct page *page,void __bio_add_page (struct bio *bio,struct page *page,unsigned int len,unsigned int off);int bio_iov_iter_get_pages (struct bio *bio,struct iov_iter *iter); +int bio_iov_bvec_add_pages (struct bio *bio,struct iov_iter *iter);struct rq_map_data ;extern struct bio *bio_map_user_iov (struct request_queue *,struct iov_iter *,gfp_t );
Signature · 1 line
2.17.1
Re: [PATCH 23/26] block: implement bio helper to add iter bvec pages to bio
block/bio.c int bio_iov_iter_get_pages(struct bio *bio, struct iov_iter *iter) · line 924
Quoted 5 lines
block/bio.c +5
-0+ do { + size_t size; + + bv = iter->bvec + iter->iov_offset; + size = bio_add_page(bio, bv->bv_page, bv->bv_len, bv->bv_offset);
Overly long line here.
block/bio.c int bio_iov_iter_get_pages(struct bio *bio, struct iov_iter *iter) · line 928
Quoted 4 lines
block/bio.c +4
-0+if (size != bv->bv_len) +break ; +iov_iter_advance (iter, size); + }while (iov_iter_count (iter) && !bio_full (bio));
bio_add_page already does the bio_full check for you.
Jens Axboe
maintainerRe: [PATCH 23/26] block: implement bio helper to add iter bvec pages to bio
Quoted 7 lines · Christoph Hellwig
block/bio.c int bio_iov_iter_get_pages(struct bio *bio, struct iov_iter *iter) · line 924
Quoted 5 lines
block/bio.c +5
-0+ do { + size_t size; + + bv = iter->bvec + iter->iov_offset; + size = bio_add_page(bio, bv->bv_page, bv->bv_len, bv->bv_offset);Overly long line here.
Will fix.
Quoted 6 lines
block/bio.c int bio_iov_iter_get_pages(struct bio *bio, struct iov_iter *iter) · line 928
Quoted 4 lines
block/bio.c +4
-0+if (size != bv->bv_len) +break ; +iov_iter_advance (iter, size); + }while (iov_iter_count (iter) && !bio_full (bio));
bio_add_pagealready does thebio_fullcheck for you.
get_pages does the same thing, but I guess it makes more sense there since we'd have to undo if we could not add.
Signature · 1 line
Jens Axboe
Jens Axboe
maintainerPATCH 24/26 fs: add support for mapping an ITER_BVEC for O_DIRECT
In reply to Jens AxboeThis adds support for sync/async O_DIRECT to make a bvec type iter for bdev access, as well as iomap.
Signed-off-by Jens Axboe <[email protected]>
--- fs/block_dev.c | 16 ++++++++++++---- fs/iomap.c | 10 +++++++--- 2 files changed, 19 insertions(+), 7 deletions(-)
fs/block_dev.c +12 -4
@@ -219,7 +219,10 @@ __blkdev_direct_IO_simple(struct kiocb *iocb, struct iov_iter *iter, bio.bi_end_io = blkdev_bio_end_io_simple; bio.bi_ioprio = iocb->ki_ioprio; - ret =bio_iov_iter_get_pages (&bio, iter); +if (iov_iter_is_bvec (iter)) + ret =bio_iov_bvec_add_pages (&bio, iter); +else + ret =bio_iov_iter_get_pages (&bio, iter);if (unlikely (ret)) goto out; ret = bio.bi_iter.bi_size;
@@ -326,8 +329,9 @@ static void blkdev_bio_end_io(struct bio *bio)struct bio_vec *bvec;int i; -bio_for_each_segment_all (bvec, bio, i) -put_page (bvec->bv_page); +if (!bio_flagged (bio,BIO_HOLD_PAGES )) +bio_for_each_segment_all (bvec, bio, i) +put_page (bvec->bv_page);bio_put (bio); } }
@@ -381,7 +385,11 @@ __blkdev_direct_IO(struct kiocb *iocb, struct iov_iter *iter, int nr_pages) bio->bi_end_io = blkdev_bio_end_io; bio->bi_ioprio = iocb->ki_ioprio; - ret =bio_iov_iter_get_pages (bio, iter); +if (iov_iter_is_bvec (iter)) + ret =bio_iov_bvec_add_pages (bio, iter); +else + ret =bio_iov_iter_get_pages (bio, iter); +if (unlikely (ret)) { bio->bi_status =BLK_STS_IOERR ;bio_endio (bio);
fs/iomap.c +7 -3
@@ -1573,8 +1573,9 @@ static void iomap_dio_bio_end_io(struct bio *bio)struct bio_vec *bvec;int i; -bio_for_each_segment_all (bvec, bio, i) -put_page (bvec->bv_page); +if (!bio_flagged (bio,BIO_HOLD_PAGES )) +bio_for_each_segment_all (bvec, bio, i) +put_page (bvec->bv_page);bio_put (bio); } }
@@ -1673,7 +1674,10 @@ iomap_dio_bio_actor(struct inode *inode, loff_t pos, loff_t length, bio->bi_private = dio; bio->bi_end_io = iomap_dio_bio_end_io; - ret =bio_iov_iter_get_pages (bio, &iter); +if (iov_iter_is_bvec (&iter)) + ret =bio_iov_bvec_add_pages (bio, &iter); +else + ret =bio_iov_iter_get_pages (bio, &iter);if (unlikely (ret)) {/* * We have to stop part way through an IO. We must fall
Signature · 1 line
2.17.1
Re: [PATCH 24/26] fs: add support for mapping an ITER_BVEC for O_DIRECT
fs/block_dev.c __blkdev_direct_IO_simple(struct kiocb *iocb, struct iov_iter *iter, · line 225
Quoted 4 lines
fs/block_dev.c +4
-0+if (iov_iter_is_bvec (iter)) + ret =bio_iov_bvec_add_pages (&bio, iter); +else + ret =bio_iov_iter_get_pages (&bio, iter);
Shouldn't this actually be done inside bio_iov_iter_get_pages?
Jens Axboe
maintainerRe: [PATCH 24/26] fs: add support for mapping an ITER_BVEC for O_DIRECT
Quoted 6 lines · Christoph Hellwig
fs/block_dev.c __blkdev_direct_IO_simple(struct kiocb *iocb, struct iov_iter *iter, · line 225
Quoted 4 lines
fs/block_dev.c +4
-0+if (iov_iter_is_bvec (iter)) + ret =bio_iov_bvec_add_pages (&bio, iter); +else + ret =bio_iov_iter_get_pages (&bio, iter);Shouldn't this actually be done inside
bio_iov_iter_get_pages?
It probably should... Should be trivial, I'll move it there instead.
Signature · 1 line
Jens Axboe
Jens Axboe
maintainerPATCH 25/26 aio: add support for pre-mapped user IO buffers
In reply to Jens AxboeIf we have fixed user buffers, we can map them into the kernel when we setup the io_context. That avoids the need to do get_user_pages() for each and every IO.
To utilize this feature, the application must set both IOCTX_FLAG_USERIOCB, to provide iocb's in userspace, and then IOCTX_FLAG_FIXEDBUFS. The latter tells aio that the iocbs that are mapped already contain valid destination and sizes. These buffers can then be mapped into the kernel for the life time of the io_context, as opposed to just the duration of the each single IO.
Only works with non-vectored read/write commands for now, not with PREADV/PWRITEV.
A limit of 4M is imposed as the largest buffer we currently support. There's nothing preventing us from going larger, but we need some cap, and 4M seemed like it would definitely be big enough. RLIMIT_MEMLOCK is used to cap the total amount of memory pinned.
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 199 +++++++++++++++++++++++++++++++---- include/uapi/linux/aio_abi.h | 1 + 2 files changed, 182 insertions(+), 18 deletions(-)
fs/aio.c +181 -18
@@ -42,6 +42,8 @@#include <linux/ramfs.h> #include <linux/percpu-refcount.h> #include <linux/mount.h> +#include <linux/sizes.h> +#include <linux/nospec.h> #include <asm/kmap_types.h> #include <linux/uaccess.h>
@@ -104,6 +106,11 @@ struct aio_iocb_ring {struct aio_mapped_range iocb_range;/* maps user iocbs */ }; +struct aio_mapped_ubuf { +struct bio_vec *bvec; +unsigned int nr_bvecs; +}; +struct kioctx {struct percpu_ref users;atomic_t dead;
@@ -139,6 +146,9 @@ struct kioctx {struct page **ring_pages;long nr_pages; +/* if used, fixed mapped user buffers */ +struct aio_mapped_ubuf *user_bufs; +/* if used, completion and submission rings */ struct aio_iocb_ring sq_ring;struct aio_mapped_range cq_ring;
@@ -313,8 +323,10 @@ static const bool aio_use_state_req_list = true;static const bool aio_use_state_req_list = false;#endif +static void aio_iocb_buffer_unmap (struct kioctx *);static void aio_scqring_unmap (struct kioctx *);static void aio_iopoll_reap_events (struct kioctx *); +static const struct iocb *aio_iocb_from_index (struct kioctx *ctx,unsigned idx);static struct file *aio_private_file (struct kioctx *ctx,loff_t nr_pages) {
@@ -693,6 +705,7 @@ static void free_ioctx(struct work_struct *work)pr_debug ("freeing %p\n" , ctx);aio_scqring_unmap (ctx); +aio_iocb_buffer_unmap (ctx);aio_free_ring (ctx);free_percpu (ctx->cpu);percpu_ref_exit (&ctx->reqs);
@@ -1830,6 +1843,122 @@ static int aio_scqring_map(struct kioctx *ctx,return ret; } +static void aio_iocb_buffer_unmap (struct kioctx *ctx) +{ +int i, j; + +if (!ctx->user_bufs) +return ; + +for (i =0 ; i < ctx->max_reqs; i++) { +struct aio_mapped_ubuf *amu = &ctx->user_bufs[i]; + +for (j =0 ; j < amu->nr_bvecs; j++) +put_page (amu->bvec[j].bv_page); + +kfree (amu->bvec); + amu->nr_bvecs =0 ; + } + +kfree (ctx->user_bufs); + ctx->user_bufs =NULL ; +} + +static int aio_iocb_buffer_map (struct kioctx *ctx) +{ +unsigned long total_pages, page_limit; +struct page **pages =NULL ; +int i, j, got_pages =0 ; +const struct iocb *iocb; +int ret = -EINVAL ; + + ctx->user_bufs =kzalloc (ctx->max_reqs *sizeof (struct aio_mapped_ubuf ), +GFP_KERNEL ); +if (!ctx->user_bufs) +return -ENOMEM ; + +/* Don't allow more pages than we can safely lock */ + total_pages =0 ; + page_limit =rlimit (RLIMIT_MEMLOCK ) >>PAGE_SHIFT ; + +for (i =0 ; i < ctx->max_reqs; i++) { +struct aio_mapped_ubuf *amu = &ctx->user_bufs[i]; +unsigned long off, start, end, ubuf; +int pret, nr_pages; +size_t size; + + iocb =aio_iocb_from_index (ctx, i); + +/* +* Don't impose further limits on the size and buffer +* constraints here, we'll -EINVAL later when IO is +* submitted if they are wrong. +*/ + ret = -EFAULT ; +if (!iocb->aio_buf) + goto err; + +/* arbitrary limit, but we need something */ +if (iocb->aio_nbytes >SZ_4M ) + goto err; + + ubuf = iocb->aio_buf; + end = (ubuf + iocb->aio_nbytes +PAGE_SIZE -1 ) >>PAGE_SHIFT ; + start = ubuf >>PAGE_SHIFT ; + nr_pages = end - start; + + ret = -ENOMEM ; +if (total_pages + nr_pages > page_limit) + goto err; + +if (!pages || nr_pages > got_pages) { +kfree (pages); + pages =kmalloc (nr_pages *sizeof (struct page *), +GFP_KERNEL ); +if (!pages) + goto err; + got_pages = nr_pages; + } + + amu->bvec =kmalloc (nr_pages *sizeof (struct bio_vec ), +GFP_KERNEL ); +if (!amu->bvec) + goto err; + +down_write (¤t->mm->mmap_sem); + pret =get_user_pages ((unsigned long ) iocb->aio_buf, nr_pages, +1 , pages,NULL ); +up_write (¤t->mm->mmap_sem); + +if (pret < nr_pages) { +if (pret <0 ) + ret = pret; + goto err; + } + + off = ubuf & ~PAGE_MASK ; + size = iocb->aio_nbytes; +for (j =0 ; j < nr_pages; j++) { +size_t vec_len; + + vec_len =min_t (size_t, size,PAGE_SIZE - off); + amu->bvec[j].bv_page = pages[j]; + amu->bvec[j].bv_len = vec_len; + amu->bvec[j].bv_offset = off; + off =0 ; + size -= vec_len; + } + amu->nr_bvecs = nr_pages; + total_pages += nr_pages; + } +kfree (pages); +return 0 ; +err: +kfree (pages); +aio_iocb_buffer_unmap (ctx); +return ret; +} +/* sys_io_setup2: * Like sys_io_setup(), except that it takes a set of flags * (IOCTX_FLAG_*), and some pointers to user structures:
@@ -1847,7 +1976,8 @@ SYSCALL_DEFINE5(io_setup2, u32, nr_events, u32, flags,unsigned long ctx;long ret; -if (flags & ~(IOCTX_FLAG_IOPOLL |IOCTX_FLAG_SCQRING )) +if (flags & ~(IOCTX_FLAG_IOPOLL |IOCTX_FLAG_SCQRING | +IOCTX_FLAG_FIXEDBUFS ))return -EINVAL ; ret =get_user (ctx, ctxp);
@@ -1863,6 +1993,15 @@ SYSCALL_DEFINE5(io_setup2, u32, nr_events, u32, flags, ret =aio_scqring_map (ioctx, sq_ring, cq_ring);if (ret) goto err; +if (flags &IOCTX_FLAG_FIXEDBUFS ) { + ret =aio_iocb_buffer_map (ioctx); +if (ret) + goto err; + } + }else if (flags &IOCTX_FLAG_FIXEDBUFS ) { +/* can only support fixed bufs with SQ/CQ ring */ + ret = -EINVAL ; + goto err; } ret =put_user (ioctx->user_id, ctxp);
@@ -2142,23 +2281,42 @@ static int aio_prep_rw(struct aio_kiocb *kiocb, const struct iocb *iocb,return ret; } -static int aio_setup_rw (int rw,const struct iocb *iocb,struct iovec **iovec, -bool vectored,bool compat,struct iov_iter *iter) +static int aio_setup_rw (int rw,struct aio_kiocb *kiocb, +const struct iocb *iocb,struct iovec **iovec,bool vectored, +bool compat,bool kaddr,struct iov_iter *iter) { -void __user *buf = (void __user *)(uintptr_t )iocb->aio_buf; +void __user *ubuf = (void __user *)(uintptr_t )iocb->aio_buf;size_t len = iocb->aio_nbytes;if (!vectored) { -ssize_t ret =import_single_range (rw, buf, len, *iovec, iter); +ssize_t ret; + +if (!kaddr) { + ret =import_single_range (rw, ubuf, len, *iovec, iter); + }else { +struct kioctx *ctx = kiocb->ki_ctx; +struct aio_mapped_ubuf *amu; +int index; + +/* __io_submit_one() already validated the index */ + index =array_index_nospec ((long )kiocb->ki_user_iocb, + ctx->max_reqs); + amu = &ctx->user_bufs[index]; +iov_iter_bvec (iter, rw, amu->bvec, amu->nr_bvecs, len); + ret =0 ; + + } *iovec =NULL ;return ret; } +if (kaddr) +return -EINVAL ;#ifdef CONFIG_COMPAT if (compat) -return compat_import_iovec (rw, buf, len,UIO_FASTIOV , iovec, +return compat_import_iovec (rw, ubuf, len,UIO_FASTIOV , iovec, iter);#endif -return import_iovec (rw, buf, len,UIO_FASTIOV , iovec, iter); +return import_iovec (rw, ubuf, len,UIO_FASTIOV , iovec, iter); }static inline void aio_rw_done (struct kiocb *req,ssize_t ret)
@@ -2231,7 +2389,7 @@ static void aio_iopoll_iocb_issued(struct aio_submit_state *state,static ssize_t aio_read (struct aio_kiocb *kiocb,const struct iocb *iocb,struct aio_submit_state *state,bool vectored, -bool compat) +bool compat,bool kaddr) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs;struct kiocb *req = &kiocb->rw;
@@ -2251,9 +2409,11 @@ static ssize_t aio_read(struct aio_kiocb *kiocb, const struct iocb *iocb,if (unlikely (!file->f_op->read_iter)) goto out_fput; - ret =aio_setup_rw (READ , iocb, &iovec, vectored, compat, &iter); + ret =aio_setup_rw (READ , kiocb, iocb, &iovec, vectored, compat, kaddr, + &iter);if (ret) goto out_fput; + ret =rw_verify_area (READ , file, &req->ki_pos,iov_iter_count (&iter));if (!ret)aio_rw_done (req,call_read_iter (file, req, &iter));
@@ -2266,7 +2426,7 @@ 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,struct aio_submit_state *state,bool vectored, -bool compat) +bool compat,bool kaddr) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs;struct kiocb *req = &kiocb->rw;
@@ -2286,7 +2446,8 @@ static ssize_t aio_write(struct aio_kiocb *kiocb, const struct iocb *iocb,if (unlikely (!file->f_op->write_iter)) goto out_fput; - ret =aio_setup_rw (WRITE , iocb, &iovec, vectored, compat, &iter); + ret =aio_setup_rw (WRITE , kiocb, iocb, &iovec, vectored, compat, kaddr, + &iter);if (ret) goto out_fput; ret =rw_verify_area (WRITE , file, &req->ki_pos,iov_iter_count (&iter));
@@ -2525,7 +2686,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, -struct aio_submit_state *state,bool compat) +struct aio_submit_state *state,bool compat, +bool kaddr) {struct aio_kiocb *req;ssize_t ret;
@@ -2583,16 +2745,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, state, false, compat); + ret =aio_read (req, iocb, state, false, compat, kaddr);break ;case IOCB_CMD_PWRITE : - ret =aio_write (req, iocb, state, false, compat); + ret =aio_write (req, iocb, state, false, compat, kaddr);break ;case IOCB_CMD_PREADV : - ret =aio_read (req, iocb, state, true, compat); + ret =aio_read (req, iocb, state, true, compat, kaddr);break ;case IOCB_CMD_PWRITEV : - ret =aio_write (req, iocb, state, true, compat); + ret =aio_write (req, iocb, state, true, compat, kaddr);break ;case IOCB_CMD_FSYNC :if (ctx->flags &IOCTX_FLAG_IOPOLL )
@@ -2648,7 +2810,7 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb,if (unlikely (copy_from_user (&iocb, user_iocb,sizeof (iocb))))return -EFAULT ; -return __io_submit_one (ctx, &iocb, user_iocb, state, compat); +return __io_submit_one (ctx, &iocb, user_iocb, state, compat, false); }#ifdef CONFIG_BLOCK
@@ -2750,6 +2912,7 @@ static const struct iocb *aio_peek_sqring(struct kioctx *ctx, unsigned *nhead)static int aio_ring_submit (struct kioctx *ctx,unsigned int to_submit) { +bool kaddr = (ctx->flags &IOCTX_FLAG_FIXEDBUFS ) !=0 ;struct aio_submit_state state, *statep =NULL ;int i, ret =0 , submit =0 ;
@@ -2766,7 +2929,7 @@ static int aio_ring_submit(struct kioctx *ctx, unsigned int to_submit)if (!iocb)break ; - ret =__io_submit_one (ctx, iocb,NULL , statep, false); + ret =__io_submit_one (ctx, iocb,NULL , statep, false, kaddr);if (ret)break ;
include/uapi/linux/aio_abi.h +1 -0
@@ -110,6 +110,7 @@ struct iocb {#define IOCTX_FLAG_IOPOLL (1 << 0)/* io_context is polled */ #define IOCTX_FLAG_SCQRING (1 << 1)/* Use SQ/CQ rings */ +#define IOCTX_FLAG_FIXEDBUFS (1 << 2)/* IO buffers are fixed */ struct aio_sq_ring { union {
Signature · 1 line
2.17.1
Jens Axboe
maintainerPATCH 26/26 aio: support kernel side submission for aio with SCQRING
In reply to Jens AxboeAdd support for backing the io_context with either a thread, or a workqueue and letting those handle the submission for us. This can be used to reduce overhead for submission, or to always make submission async. The latter is particularly useful for buffered aio, which is now fully async with this feature.
For polled IO, we could have the kernel side thread hammer on the SQ ring and submit when it finds IO. This would mean that an application would NEVER have to enter the kernel to do IO! Didn't add this yet, but it would be trivial to add.
If an application sets IOCTX_FLAG_SCQTHREAD, the io_context gets a single thread backing. If used with buffered IO, this will limit the device queue depth to 1, but it will be async, IOs will simply be serialized.
Or an application can set IOCTX_FLAG_SQWQ, in which case the io_context gets a work queue backing. The concurrency level is the mininum of twice the available CPUs, or the queue depth specific for the context. For this mode, we attempt to do buffered reads inline, in case they are cached. So we should only punt to a workqueue, if we would have to block to get our data.
Tested with polling, no polling, fixedbufs, no fixedbufs, buffered, O_DIRECT.
See this sample application for how to use it:
http://git.kernel.dk/cgit/fio/plain/t/aio-ring.c
Signed-off-by Jens Axboe <[email protected]>
--- fs/aio.c | 421 ++++++++++++++++++++++++++++++++--- include/uapi/linux/aio_abi.h | 3 + 2 files changed, 397 insertions(+), 27 deletions(-)
fs/aio.c +394 -27
@@ -25,6 +25,7 @@#include <linux/sched/signal.h> #include <linux/fs.h> #include <linux/file.h> +#include <linux/fdtable.h> #include <linux/mm.h> #include <linux/mman.h> #include <linux/mmu_context.h>
@@ -44,6 +45,7 @@#include <linux/mount.h> #include <linux/sizes.h> #include <linux/nospec.h> +#include <linux/sched/mm.h> #include <asm/kmap_types.h> #include <linux/uaccess.h>
@@ -111,6 +113,14 @@ struct aio_mapped_ubuf {unsigned int nr_bvecs; }; +struct aio_sq_offload { +struct task_struct *thread;/* if using a thread */ +struct workqueue_struct *wq;/* wq offload */ +struct mm_struct *mm; +struct files_struct *files; +wait_queue_head_t wait; +}; +struct kioctx {struct percpu_ref users;atomic_t dead;
@@ -153,6 +163,10 @@ struct kioctx {struct aio_iocb_ring sq_ring;struct aio_mapped_range cq_ring;int cq_ring_overflow; +int submit_eagain; + +/* sq ring submitter thread, if used */ +struct aio_sq_offload sq_offload;struct rcu_work free_rwork;/* see free_ioctx() */
@@ -243,6 +257,7 @@ struct aio_kiocb {unsigned long ki_flags;#define KIOCB_F_POLL_COMPLETED 0/* polled IO has completed */ #define KIOCB_F_POLL_EAGAIN 1/* polled submission got EAGAIN */ +#define KIOCB_F_FORCE_NONBLOCK 2/* inline submission attempt */ refcount_t ki_refcnt;
@@ -1350,19 +1365,31 @@ static void aio_complete(struct aio_kiocb *iocb, long res, long res2)unsigned int tail;/* -* If we can't get a cq entry, userspace overflowed the -* submission (by quite a lot). Flag it as an overflow -* condition, and next io_ring_enter(2) call will return -* -EOVERFLOW. +* Catch EAGAIN early if we've forced a nonblock attempt, as +* we don't want to pass that back down to userspace through +* the CQ ring. Just mark the ctx as such, so the caller will +* see it and punt to workqueue. This is just for buffered +* aio reads. */ -spin_lock_irqsave (&ctx->completion_lock, flags); - ev =aio_peek_cqring (ctx, &tail); -if (ev) { -aio_fill_event (ev, iocb, res, res2); -aio_commit_cqring (ctx, tail); - }else - ctx->cq_ring_overflow =1 ; -spin_unlock_irqrestore (&ctx->completion_lock, flags); +if (res == -EAGAIN && +test_bit (KIOCB_F_FORCE_NONBLOCK , &iocb->ki_flags)) { + ctx->submit_eagain =1 ; + }else { +/* +* If we can't get a cq entry, userspace overflowed the +* submission (by quite a lot). Flag it as an overflow +* condition, and next io_ring_enter(2) call will return +* -EOVERFLOW. +*/ +spin_lock_irqsave (&ctx->completion_lock, flags); + ev =aio_peek_cqring (ctx, &tail); +if (ev) { +aio_fill_event (ev, iocb, res, res2); +aio_commit_cqring (ctx, tail); + }else + ctx->cq_ring_overflow =1 ; +spin_unlock_irqrestore (&ctx->completion_lock, flags); + } }else {aio_ring_complete (ctx, iocb, res, res2);
@@ -1728,6 +1755,64 @@ static long read_events(struct kioctx *ctx, long min_nr, long nr,return ret; } +static int aio_sq_thread (void *); + +static int aio_sq_thread_start (struct kioctx *ctx) +{ +struct aio_sq_ring *ring = ctx->sq_ring.ring; +struct aio_sq_offload *aso = &ctx->sq_offload; +int ret; + +memset (aso,0 ,sizeof (*aso)); +init_waitqueue_head (&aso->wait); + +if (!(ctx->flags &IOCTX_FLAG_FIXEDBUFS )) + aso->mm = current->mm; + + ret = -EBADF ; + aso->files =get_files_struct (current); +if (!aso->files) + goto err; + +if (ctx->flags &IOCTX_FLAG_SQTHREAD ) { +char name[32 ]; + +snprintf (name,sizeof (name),"aio-sq-%lu/%d" , ctx->user_id, + ring->sq_thread_cpu); + aso->thread =kthread_create_on_cpu (aio_sq_thread, ctx, + ring->sq_thread_cpu, name); +if (IS_ERR (aso->thread)) { + ret =PTR_ERR (aso->thread); + aso->thread =NULL ; + goto err; + } +wake_up_process (aso->thread); + }else if (ctx->flags &IOCTX_FLAG_SQWQ ) { +int concurrency; + +/* Do QD, or 2 * CPUS, whatever is smallest */ + concurrency =min (ring->nr_events -1 ,2 *num_online_cpus ()); + aso->wq =alloc_workqueue ("aio-sq-%lu" , +WQ_UNBOUND |WQ_FREEZABLE |WQ_SYSFS , + concurrency, + ctx->user_id); +if (!aso->wq) { + ret = -ENOMEM ; + goto err; + } + } + +return 0 ; +err: +if (aso->files) { +put_files_struct (aso->files); + aso->files =NULL ; + } +if (aso->mm) + aso->mm =NULL ; +return ret; +} +static void aio_unmap_range (struct aio_mapped_range *range) {int i;
@@ -1773,6 +1858,20 @@ static int aio_map_range(struct aio_mapped_range *range, void __user *uaddr,static void aio_scqring_unmap (struct kioctx *ctx) { +struct aio_sq_offload *aso = &ctx->sq_offload; + +if (aso->thread) { +kthread_park (aso->thread); +kthread_stop (aso->thread); + aso->thread =NULL ; + }else if (aso->wq) { +destroy_workqueue (aso->wq); + aso->wq =NULL ; + } +if (aso->files) { +put_files_struct (aso->files); + aso->files =NULL ; + }aio_unmap_range (&ctx->sq_ring.ring_range);aio_unmap_range (&ctx->sq_ring.iocb_range);aio_unmap_range (&ctx->cq_ring);
@@ -1834,6 +1933,9 @@ static int aio_scqring_map(struct kioctx *ctx, kcq_ring->nr_events = cq_ring_size; kcq_ring->head = kcq_ring->tail =0 ; +if (ctx->flags & (IOCTX_FLAG_SQTHREAD |IOCTX_FLAG_SQWQ )) + ret =aio_sq_thread_start (ctx); + err:if (ret) {aio_unmap_range (&ctx->sq_ring.ring_range);
@@ -1977,7 +2079,8 @@ SYSCALL_DEFINE5(io_setup2, u32, nr_events, u32, flags,long ret;if (flags & ~(IOCTX_FLAG_IOPOLL |IOCTX_FLAG_SCQRING | -IOCTX_FLAG_FIXEDBUFS )) +IOCTX_FLAG_FIXEDBUFS |IOCTX_FLAG_SQTHREAD | +IOCTX_FLAG_SQWQ ))return -EINVAL ; ret =get_user (ctx, ctxp);
@@ -1998,8 +2101,9 @@ SYSCALL_DEFINE5(io_setup2, u32, nr_events, u32, flags,if (ret) goto err; } - }else if (flags &IOCTX_FLAG_FIXEDBUFS ) { -/* can only support fixed bufs with SQ/CQ ring */ + }else if (flags & (IOCTX_FLAG_FIXEDBUFS |IOCTX_FLAG_SQTHREAD | +IOCTX_FLAG_SQWQ )) { +/* These features only supported with SCQRING */ ret = -EINVAL ; goto err; }
@@ -2213,7 +2317,7 @@ static struct file *aio_file_get(struct aio_submit_state *state, int fd) }static int aio_prep_rw (struct aio_kiocb *kiocb,const struct iocb *iocb, -struct aio_submit_state *state) +struct aio_submit_state *state,bool force_nonblock) {struct kioctx *ctx = kiocb->ki_ctx;struct kiocb *req = &kiocb->rw;
@@ -2246,6 +2350,10 @@ static int aio_prep_rw(struct aio_kiocb *kiocb, const struct iocb *iocb, ret =kiocb_set_rw_flags (req, iocb->aio_rw_flags);if (unlikely (ret)) goto out_fput; +if (force_nonblock) { + req->ki_flags |=IOCB_NOWAIT ; +set_bit (KIOCB_F_FORCE_NONBLOCK , &kiocb->ki_flags); + }if (iocb->aio_flags &IOCB_FLAG_HIPRI ) {/* shares space in the union, and is rather pointless.. */
@@ -2389,7 +2497,7 @@ static void aio_iopoll_iocb_issued(struct aio_submit_state *state,static ssize_t aio_read (struct aio_kiocb *kiocb,const struct iocb *iocb,struct aio_submit_state *state,bool vectored, -bool compat,bool kaddr) +bool compat,bool kaddr,bool force_nonblock) {struct iovec inline_vecs[UIO_FASTIOV ], *iovec = inline_vecs;struct kiocb *req = &kiocb->rw;
@@ -2397,7 +2505,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, state); + ret =aio_prep_rw (kiocb, iocb, state, force_nonblock);if (ret)return ret; file = req->ki_filp;
@@ -2434,7 +2542,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, state); + ret =aio_prep_rw (kiocb, iocb, state, false);if (ret)return ret; file = req->ki_filp;
@@ -2687,7 +2795,7 @@ 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,struct aio_submit_state *state,bool compat, -bool kaddr) +bool kaddr,bool force_nonblock) {struct aio_kiocb *req;ssize_t ret;
@@ -2745,13 +2853,15 @@ 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, state, false, compat, kaddr); + ret =aio_read (req, iocb, state, false, compat, kaddr, + force_nonblock);break ;case IOCB_CMD_PWRITE : ret =aio_write (req, iocb, state, false, compat, kaddr);break ;case IOCB_CMD_PREADV : - ret =aio_read (req, iocb, state, true, compat, kaddr); + ret =aio_read (req, iocb, state, true, compat, kaddr, + force_nonblock);break ;case IOCB_CMD_PWRITEV : ret =aio_write (req, iocb, state, true, compat, kaddr);
@@ -2810,7 +2920,8 @@ static int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb,if (unlikely (copy_from_user (&iocb, user_iocb,sizeof (iocb))))return -EFAULT ; -return __io_submit_one (ctx, &iocb, user_iocb, state, compat, false); +return __io_submit_one (ctx, &iocb, user_iocb, state, compat, false, + false); }#ifdef CONFIG_BLOCK
@@ -2929,7 +3040,8 @@ static int aio_ring_submit(struct kioctx *ctx, unsigned int to_submit)if (!iocb)break ; - ret =__io_submit_one (ctx, iocb,NULL , statep, false, kaddr); + ret =__io_submit_one (ctx, iocb,NULL , statep, false, kaddr, + false);if (ret)break ;
@@ -2981,15 +3093,270 @@ static int aio_cqring_wait(struct kioctx *ctx, int min_events)return ret; } +static void aio_fill_cq_error (struct kioctx *ctx,const struct iocb *iocb, +long ret) +{ +struct io_event *ev; +unsigned tail; + +/* +* Only really need the lock for non-polled IO, but this is an error +* so not worth checking. Just lock it so we know kernel access to +* the CQ ring is serialized. +*/ +spin_lock_irq (&ctx->completion_lock); + ev =aio_peek_cqring (ctx, &tail); + ev->obj = iocb->aio_data; + ev->data =0 ; + ev->res = ret; + ev->res2 =0 ; +aio_commit_cqring (ctx, tail); +spin_unlock_irq (&ctx->completion_lock); + +/* +* for thread offload, app could already be sleeping in io_ring_enter() +* before we get to flag the error. wake them up, if needed. +*/ +if (ctx->flags & (IOCTX_FLAG_SQTHREAD |IOCTX_FLAG_SQWQ )) +if (waitqueue_active (&ctx->wait)) +wake_up (&ctx->wait); +} + +struct aio_io_work { +struct work_struct work; +struct kioctx *ctx; +struct iocb iocb; +}; + +/* +* sq thread only supports O_DIRECT or FIXEDBUFS IO +*/ +static int aio_sq_thread (void *data) +{ +const struct iocb *iocbs[AIO_IOPOLL_BATCH ]; +struct aio_submit_state state; +struct kioctx *ctx = data; +struct aio_sq_offload *aso = &ctx->sq_offload; +struct mm_struct *cur_mm =NULL ; +struct files_struct *old_files; +mm_segment_t old_fs; +DEFINE_WAIT (wait); + + old_files = current->files; + current->files = aso->files; + + old_fs =get_fs (); +set_fs (USER_DS ); + +while (!kthread_should_stop ()) { +struct aio_submit_state *statep =NULL ; +const struct iocb *iocb; +bool mm_fault = false; +unsigned int nhead; +int ret, i, j; + + iocb =aio_peek_sqring (ctx, &nhead); +if (!iocb) { +prepare_to_wait (&aso->wait, &wait,TASK_INTERRUPTIBLE ); + iocb =aio_peek_sqring (ctx, &nhead); +if (!iocb) { +/* +* Drop cur_mm before scheduler. We can't hold +* it for long periods, and it would also +* introduce a deadlock with kill_ioctx(). +*/ +if (cur_mm) { +unuse_mm (cur_mm); +mmput (cur_mm); + cur_mm =NULL ; + } +if (kthread_should_park ()) +kthread_parkme (); +if (kthread_should_stop ()) { +finish_wait (&aso->wait, &wait); +break ; + } +if (signal_pending (current)) +flush_signals (current); +schedule (); + } +finish_wait (&aso->wait, &wait); +if (!iocb) +continue ; + } + +/* If ->mm is set, we're not doing FIXEDBUFS */ +if (aso->mm && !cur_mm) { + mm_fault = !mmget_not_zero (aso->mm); +if (!mm_fault) { +use_mm (aso->mm); + cur_mm = aso->mm; + } + } + + i =0 ; +do { +if (i ==ARRAY_SIZE (iocbs)) +break ; + iocbs[i++] = iocb; +aio_commit_sqring (ctx, nhead); + }while ((iocb =aio_peek_sqring (ctx, &nhead)) !=NULL ); + +if (i >AIO_PLUG_THRESHOLD ) { +aio_submit_state_start (&state, ctx, i); + statep = &state; + } + +for (j =0 ; j < i; j++) { +if (unlikely (mm_fault)) + ret = -EFAULT ; +else + ret =__io_submit_one (ctx, iocbs[j],NULL , + statep, false, !cur_mm, + false); +if (!ret) +continue ; + +aio_fill_cq_error (ctx, iocbs[j], ret); + } + +if (statep) +aio_submit_state_end (&state); + } + current->files = old_files; +set_fs (old_fs); +if (cur_mm) { +unuse_mm (cur_mm); +mmput (cur_mm); + } +return 0 ; +} + +static void aio_sq_wq_submit_work (struct work_struct *work) +{ + struct aio_io_work *aiw = container_of(work, struct aio_io_work, work); +struct kioctx *ctx = aiw->ctx; +struct aio_sq_offload *aso = &ctx->sq_offload; +mm_segment_t old_fs =get_fs (); +struct files_struct *old_files; +int ret; + + old_files = current->files; + current->files = aso->files; + +if (aso->mm) { +if (!mmget_not_zero (aso->mm)) { + ret = -EFAULT ; + goto err; + } +use_mm (aso->mm); + } + +set_fs (USER_DS ); + + ret =__io_submit_one (ctx, &aiw->iocb,NULL ,NULL , false, !aso->mm, + false); + +set_fs (old_fs); +if (aso->mm) { +unuse_mm (aso->mm); +mmput (aso->mm); + } + +err: +if (ret) +aio_fill_cq_error (ctx, &aiw->iocb, ret); + current->files = old_files; +kfree (aiw); +} + +/* +* If this is a read, try a cached inline read first. If the IO is in the +* page cache, we can satisfy it without blocking and without having to +* punt to a threaded execution. This is much faster, particularly for +* lower queue depth IO, and it's always a lot more efficient. +*/ +static int aio_sq_try_inline (struct kioctx *ctx,struct aio_io_work *aiw) +{ +struct aio_sq_offload *aso = &ctx->sq_offload; +int ret; + +if (aiw->iocb.aio_lio_opcode !=IOCB_CMD_PREAD && + aiw->iocb.aio_lio_opcode !=IOCB_CMD_PREADV ) +return -EAGAIN ; + + ret =__io_submit_one (ctx, &aiw->iocb,NULL ,NULL , false, !aso->mm, + true); + +if (ret == -EAGAIN || ctx->submit_eagain) { + ctx->submit_eagain =0 ; +return -EAGAIN ; + } + +/* +* We're done - even if this was an error, return 0. The error will +* be in the CQ ring for the application. +*/ +kfree (aiw); +return 0 ; +} + +static int aio_sq_wq_submit (struct kioctx *ctx,unsigned int to_submit) +{ +struct aio_io_work *work; +const struct iocb *iocb; +unsigned nhead; +int ret, queued; + + ret = queued =0 ; +while ((iocb =aio_peek_sqring (ctx, &nhead)) !=NULL ) { + work =kmalloc (sizeof (*work),GFP_KERNEL ); +if (!work) { + ret = -ENOMEM ; +break ; + } +memcpy (&work->iocb, iocb,sizeof (*iocb)); +aio_commit_sqring (ctx, nhead); + ret =aio_sq_try_inline (ctx, work); +if (ret == -EAGAIN ) { +INIT_WORK (&work->work, aio_sq_wq_submit_work); + work->ctx = ctx; +queue_work (ctx->sq_offload.wq, &work->work); + ret =0 ; + } + queued++; +if (queued == to_submit) +break ; + } + +return queued ? queued : ret; +} +static int __io_ring_enter (struct kioctx *ctx,unsigned int to_submit,unsigned int min_complete,unsigned int flags) {int ret =0 ;if (flags &IORING_FLAG_SUBMIT ) { - ret =aio_ring_submit (ctx, to_submit); -if (ret <0 ) -return ret; +if (!to_submit) +return 0 ; + +/* +* Three options here: +* 1) We have an sq thread, just wake it up to do submissions +* 2) We have an sq wq, queue a work item for each iocb +* 3) Submit directly +*/ +if (ctx->flags &IOCTX_FLAG_SQTHREAD ) { +wake_up (&ctx->sq_offload.wait); + ret = to_submit; + }else if (ctx->flags &IOCTX_FLAG_SQWQ ) { + ret =aio_sq_wq_submit (ctx, to_submit); + }else { + ret =aio_ring_submit (ctx, to_submit); +if (ret <0 ) +return ret; + } }if (flags &IORING_FLAG_GETEVENTS ) {unsigned int nr_events =0 ;
include/uapi/linux/aio_abi.h +3 -0
@@ -111,6 +111,8 @@ struct iocb {#define IOCTX_FLAG_IOPOLL (1 << 0)/* io_context is polled */ #define IOCTX_FLAG_SCQRING (1 << 1)/* Use SQ/CQ rings */ #define IOCTX_FLAG_FIXEDBUFS (1 << 2)/* IO buffers are fixed */ +#define IOCTX_FLAG_SQTHREAD (1 << 3)/* Use SQ thread */ +#define IOCTX_FLAG_SQWQ (1 << 4)/* Use SQ workqueue */ struct aio_sq_ring { union {
@@ -118,6 +120,7 @@ struct aio_sq_ring {u32 head;/* kernel consumer head */ u32 tail;/* app producer tail */ u32 nr_events;/* max events in ring */ +u16 sq_thread_cpu;u64 iocbs;/* setup pointer to app iocbs */ };u32 pad[16 ];
Signature · 1 line
2.17.1
Any chance you could send the series that just optimize the existing aio code and don't need new interface as an initial prep series?
Batching things up should make it a little easier to review, and with a little luck those "simple" patches might still make 4.21. Make sure to include Al on the series, please.
Jens Axboe
maintainerQuoted 6 lines · Christoph Hellwig
Any chance you could send the series that just optimize the existing aio code and don't need new interface as an initial prep series?
Batching things up should make it a little easier to review, and with a little luck those "simple" patches might still make 4.21. Make sure to include Al on the series, please.
Sure, I can do that. So basically all of the optimizations, but none of the features (like polling)? Would require serious refactoring though, but not impossible.
Or do you mean everything up to polled IO for old style aio, just excluding the new ring interfaces?
Signature · 1 line
Jens Axboe
Quoted 8 lines · Jens Axboe
Quoted 6 lines · Christoph Hellwig
Any chance you could send the series that just optimize the existing aio code and don't need new interface as an initial prep series?
Batching things up should make it a little easier to review, and with a little luck those "simple" patches might still make 4.21. Make sure to include Al on the series, please.
Quoted 6 lines
Sure, I can do that. So basically all of the optimizations, but none of the features (like polling)? Would require serious refactoring though, but not impossible.
Or do you mean everything up to polled IO for old style aio, just excluding the new ring interfaces?
No, just the various little optimizations and cleanups towards the beginning of the series. Shouldn't really require any refactoring.
Quoted 3 lines
Signature · 2 lines
-- Jens Axboe
---end quoted text---
Jens Axboe
maintainerQuoted 13 lines · Christoph Hellwig
Quoted 12 lines · Jens Axboe
Quoted 6 lines · Christoph Hellwig
Any chance you could send the series that just optimize the existing aio code and don't need new interface as an initial prep series?
Batching things up should make it a little easier to review, and with a little luck those "simple" patches might still make 4.21. Make sure to include Al on the series, please.
Sure, I can do that. So basically all of the optimizations, but none of the features (like polling)? Would require serious refactoring though, but not impossible.
Quoted 5 lines
Quoted 2 lines
Or do you mean everything up to polled IO for old style aio, just excluding the new ring interfaces?
No, just the various little optimizations and cleanups towards the beginning of the series. Shouldn't really require any refactoring.
OK, that'd be trivial enough. I'll do that and send it out separately.
Signature · 1 line
Jens Axboe
Quoted 14 lines · Jens Axboe
Quoted 13 lines · Christoph Hellwig
Quoted 12 lines · Jens Axboe
Quoted 6 lines · Christoph Hellwig
Any chance you could send the series that just optimize the existing aio code and don't need new interface as an initial prep series?
Batching things up should make it a little easier to review, and with a little luck those "simple" patches might still make 4.21. Make sure to include Al on the series, please.
Sure, I can do that. So basically all of the optimizations, but none of the features (like polling)? Would require serious refactoring though, but not impossible.
Quoted 7 lines
Quoted 5 lines
Quoted 2 lines
Or do you mean everything up to polled IO for old style aio, just excluding the new ring interfaces?
No, just the various little optimizations and cleanups towards the beginning of the series. Shouldn't really require any refactoring.
OK, that'd be trivial enough. I'll do that and send it out separately.
That would be roughly patches 6-12 in the current series as far as I can tell.