2 * An async IO implementation for Linux
3 * Written by Benjamin LaHaise <bcrl@kvack.org>
5 * Implements an efficient asynchronous io interface.
7 * Copyright 2000, 2001, 2002 Red Hat, Inc. All Rights Reserved.
9 * See ../COPYING for licensing terms.
11 #include <linux/kernel.h>
12 #include <linux/init.h>
13 #include <linux/errno.h>
14 #include <linux/time.h>
15 #include <linux/aio_abi.h>
16 #include <linux/module.h>
17 #include <linux/syscalls.h>
18 #include <linux/backing-dev.h>
19 #include <linux/uio.h>
23 #include <linux/sched.h>
25 #include <linux/file.h>
27 #include <linux/mman.h>
28 #include <linux/mmu_context.h>
29 #include <linux/slab.h>
30 #include <linux/timer.h>
31 #include <linux/aio.h>
32 #include <linux/highmem.h>
33 #include <linux/workqueue.h>
34 #include <linux/security.h>
35 #include <linux/eventfd.h>
36 #include <linux/blkdev.h>
37 #include <linux/compat.h>
39 #include <asm/kmap_types.h>
40 #include <asm/uaccess.h>
43 #define dprintk printk
45 #define dprintk(x...) do { ; } while (0)
48 /*------ sysctl variables----*/
49 static DEFINE_SPINLOCK(aio_nr_lock
);
50 unsigned long aio_nr
; /* current system wide number of aio requests */
51 unsigned long aio_max_nr
= 0x10000; /* system wide maximum number of aio requests */
52 /*----end sysctl variables---*/
54 static struct kmem_cache
*kiocb_cachep
;
55 static struct kmem_cache
*kioctx_cachep
;
57 static struct workqueue_struct
*aio_wq
;
59 /* Used for rare fput completion. */
60 static void aio_fput_routine(struct work_struct
*);
61 static DECLARE_WORK(fput_work
, aio_fput_routine
);
63 static DEFINE_SPINLOCK(fput_lock
);
64 static LIST_HEAD(fput_head
);
66 static void aio_kick_handler(struct work_struct
*);
67 static void aio_queue_work(struct kioctx
*);
70 * Creates the slab caches used by the aio routines, panic on
71 * failure as this is done early during the boot sequence.
73 static int __init
aio_setup(void)
75 kiocb_cachep
= KMEM_CACHE(kiocb
, SLAB_HWCACHE_ALIGN
|SLAB_PANIC
);
76 kioctx_cachep
= KMEM_CACHE(kioctx
,SLAB_HWCACHE_ALIGN
|SLAB_PANIC
);
78 aio_wq
= alloc_workqueue("aio", 0, 1); /* used to limit concurrency */
81 pr_debug("aio_setup: sizeof(struct page) = %d\n", (int)sizeof(struct page
));
85 __initcall(aio_setup
);
87 static void aio_free_ring(struct kioctx
*ctx
)
89 struct aio_ring_info
*info
= &ctx
->ring_info
;
92 for (i
=0; i
<info
->nr_pages
; i
++)
93 put_page(info
->ring_pages
[i
]);
95 if (info
->mmap_size
) {
96 down_write(&ctx
->mm
->mmap_sem
);
97 do_munmap(ctx
->mm
, info
->mmap_base
, info
->mmap_size
);
98 up_write(&ctx
->mm
->mmap_sem
);
101 if (info
->ring_pages
&& info
->ring_pages
!= info
->internal_pages
)
102 kfree(info
->ring_pages
);
103 info
->ring_pages
= NULL
;
107 static int aio_setup_ring(struct kioctx
*ctx
)
109 struct aio_ring
*ring
;
110 struct aio_ring_info
*info
= &ctx
->ring_info
;
111 unsigned nr_events
= ctx
->max_reqs
;
115 /* Compensate for the ring buffer's head/tail overlap entry */
116 nr_events
+= 2; /* 1 is required, 2 for good luck */
118 size
= sizeof(struct aio_ring
);
119 size
+= sizeof(struct io_event
) * nr_events
;
120 nr_pages
= (size
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
125 nr_events
= (PAGE_SIZE
* nr_pages
- sizeof(struct aio_ring
)) / sizeof(struct io_event
);
128 info
->ring_pages
= info
->internal_pages
;
129 if (nr_pages
> AIO_RING_PAGES
) {
130 info
->ring_pages
= kcalloc(nr_pages
, sizeof(struct page
*), GFP_KERNEL
);
131 if (!info
->ring_pages
)
135 info
->mmap_size
= nr_pages
* PAGE_SIZE
;
136 dprintk("attempting mmap of %lu bytes\n", info
->mmap_size
);
137 down_write(&ctx
->mm
->mmap_sem
);
138 info
->mmap_base
= do_mmap(NULL
, 0, info
->mmap_size
,
139 PROT_READ
|PROT_WRITE
, MAP_ANONYMOUS
|MAP_PRIVATE
,
141 if (IS_ERR((void *)info
->mmap_base
)) {
142 up_write(&ctx
->mm
->mmap_sem
);
148 dprintk("mmap address: 0x%08lx\n", info
->mmap_base
);
149 info
->nr_pages
= get_user_pages(current
, ctx
->mm
,
150 info
->mmap_base
, nr_pages
,
151 1, 0, info
->ring_pages
, NULL
);
152 up_write(&ctx
->mm
->mmap_sem
);
154 if (unlikely(info
->nr_pages
!= nr_pages
)) {
159 ctx
->user_id
= info
->mmap_base
;
161 info
->nr
= nr_events
; /* trusted copy */
163 ring
= kmap_atomic(info
->ring_pages
[0]);
164 ring
->nr
= nr_events
; /* user copy */
165 ring
->id
= ctx
->user_id
;
166 ring
->head
= ring
->tail
= 0;
167 ring
->magic
= AIO_RING_MAGIC
;
168 ring
->compat_features
= AIO_RING_COMPAT_FEATURES
;
169 ring
->incompat_features
= AIO_RING_INCOMPAT_FEATURES
;
170 ring
->header_length
= sizeof(struct aio_ring
);
177 /* aio_ring_event: returns a pointer to the event at the given index from
178 * kmap_atomic(). Release the pointer with put_aio_ring_event();
180 #define AIO_EVENTS_PER_PAGE (PAGE_SIZE / sizeof(struct io_event))
181 #define AIO_EVENTS_FIRST_PAGE ((PAGE_SIZE - sizeof(struct aio_ring)) / sizeof(struct io_event))
182 #define AIO_EVENTS_OFFSET (AIO_EVENTS_PER_PAGE - AIO_EVENTS_FIRST_PAGE)
184 #define aio_ring_event(info, nr) ({ \
185 unsigned pos = (nr) + AIO_EVENTS_OFFSET; \
186 struct io_event *__event; \
187 __event = kmap_atomic( \
188 (info)->ring_pages[pos / AIO_EVENTS_PER_PAGE]); \
189 __event += pos % AIO_EVENTS_PER_PAGE; \
193 #define put_aio_ring_event(event) do { \
194 struct io_event *__event = (event); \
196 kunmap_atomic((void *)((unsigned long)__event & PAGE_MASK)); \
199 static void ctx_rcu_free(struct rcu_head
*head
)
201 struct kioctx
*ctx
= container_of(head
, struct kioctx
, rcu_head
);
202 kmem_cache_free(kioctx_cachep
, ctx
);
206 * Called when the last user of an aio context has gone away,
207 * and the struct needs to be freed.
209 static void __put_ioctx(struct kioctx
*ctx
)
211 unsigned nr_events
= ctx
->max_reqs
;
212 BUG_ON(ctx
->reqs_active
);
214 cancel_delayed_work_sync(&ctx
->wq
);
219 spin_lock(&aio_nr_lock
);
220 BUG_ON(aio_nr
- nr_events
> aio_nr
);
222 spin_unlock(&aio_nr_lock
);
224 pr_debug("__put_ioctx: freeing %p\n", ctx
);
225 call_rcu(&ctx
->rcu_head
, ctx_rcu_free
);
228 static inline int try_get_ioctx(struct kioctx
*kioctx
)
230 return atomic_inc_not_zero(&kioctx
->users
);
233 static inline void put_ioctx(struct kioctx
*kioctx
)
235 BUG_ON(atomic_read(&kioctx
->users
) <= 0);
236 if (unlikely(atomic_dec_and_test(&kioctx
->users
)))
241 * Allocates and initializes an ioctx. Returns an ERR_PTR if it failed.
243 static struct kioctx
*ioctx_alloc(unsigned nr_events
)
245 struct mm_struct
*mm
;
249 /* Prevent overflows */
250 if ((nr_events
> (0x10000000U
/ sizeof(struct io_event
))) ||
251 (nr_events
> (0x10000000U
/ sizeof(struct kiocb
)))) {
252 pr_debug("ENOMEM: nr_events too high\n");
253 return ERR_PTR(-EINVAL
);
256 if (!nr_events
|| (unsigned long)nr_events
> aio_max_nr
)
257 return ERR_PTR(-EAGAIN
);
259 ctx
= kmem_cache_zalloc(kioctx_cachep
, GFP_KERNEL
);
261 return ERR_PTR(-ENOMEM
);
263 ctx
->max_reqs
= nr_events
;
264 mm
= ctx
->mm
= current
->mm
;
265 atomic_inc(&mm
->mm_count
);
267 atomic_set(&ctx
->users
, 2);
268 spin_lock_init(&ctx
->ctx_lock
);
269 spin_lock_init(&ctx
->ring_info
.ring_lock
);
270 init_waitqueue_head(&ctx
->wait
);
272 INIT_LIST_HEAD(&ctx
->active_reqs
);
273 INIT_LIST_HEAD(&ctx
->run_list
);
274 INIT_DELAYED_WORK(&ctx
->wq
, aio_kick_handler
);
276 if (aio_setup_ring(ctx
) < 0)
279 /* limit the number of system wide aios */
280 spin_lock(&aio_nr_lock
);
281 if (aio_nr
+ nr_events
> aio_max_nr
||
282 aio_nr
+ nr_events
< aio_nr
) {
283 spin_unlock(&aio_nr_lock
);
286 aio_nr
+= ctx
->max_reqs
;
287 spin_unlock(&aio_nr_lock
);
289 /* now link into global list. */
290 spin_lock(&mm
->ioctx_lock
);
291 hlist_add_head_rcu(&ctx
->list
, &mm
->ioctx_list
);
292 spin_unlock(&mm
->ioctx_lock
);
294 dprintk("aio: allocated ioctx %p[%ld]: mm=%p mask=0x%x\n",
295 ctx
, ctx
->user_id
, current
->mm
, ctx
->ring_info
.nr
);
303 kmem_cache_free(kioctx_cachep
, ctx
);
304 dprintk("aio: error allocating ioctx %d\n", err
);
309 * Cancels all outstanding aio requests on an aio context. Used
310 * when the processes owning a context have all exited to encourage
311 * the rapid destruction of the kioctx.
313 static void aio_cancel_all(struct kioctx
*ctx
)
315 int (*cancel
)(struct kiocb
*, struct io_event
*);
317 spin_lock_irq(&ctx
->ctx_lock
);
319 while (!list_empty(&ctx
->active_reqs
)) {
320 struct list_head
*pos
= ctx
->active_reqs
.next
;
321 struct kiocb
*iocb
= list_kiocb(pos
);
322 list_del_init(&iocb
->ki_list
);
323 cancel
= iocb
->ki_cancel
;
324 kiocbSetCancelled(iocb
);
327 spin_unlock_irq(&ctx
->ctx_lock
);
329 spin_lock_irq(&ctx
->ctx_lock
);
332 spin_unlock_irq(&ctx
->ctx_lock
);
335 static void wait_for_all_aios(struct kioctx
*ctx
)
337 struct task_struct
*tsk
= current
;
338 DECLARE_WAITQUEUE(wait
, tsk
);
340 spin_lock_irq(&ctx
->ctx_lock
);
341 if (!ctx
->reqs_active
)
344 add_wait_queue(&ctx
->wait
, &wait
);
345 set_task_state(tsk
, TASK_UNINTERRUPTIBLE
);
346 while (ctx
->reqs_active
) {
347 spin_unlock_irq(&ctx
->ctx_lock
);
349 set_task_state(tsk
, TASK_UNINTERRUPTIBLE
);
350 spin_lock_irq(&ctx
->ctx_lock
);
352 __set_task_state(tsk
, TASK_RUNNING
);
353 remove_wait_queue(&ctx
->wait
, &wait
);
356 spin_unlock_irq(&ctx
->ctx_lock
);
359 /* wait_on_sync_kiocb:
360 * Waits on the given sync kiocb to complete.
362 ssize_t
wait_on_sync_kiocb(struct kiocb
*iocb
)
364 while (iocb
->ki_users
) {
365 set_current_state(TASK_UNINTERRUPTIBLE
);
370 __set_current_state(TASK_RUNNING
);
371 return iocb
->ki_user_data
;
373 EXPORT_SYMBOL(wait_on_sync_kiocb
);
375 /* exit_aio: called when the last user of mm goes away. At this point,
376 * there is no way for any new requests to be submited or any of the
377 * io_* syscalls to be called on the context. However, there may be
378 * outstanding requests which hold references to the context; as they
379 * go away, they will call put_ioctx and release any pinned memory
380 * associated with the request (held via struct page * references).
382 void exit_aio(struct mm_struct
*mm
)
386 while (!hlist_empty(&mm
->ioctx_list
)) {
387 ctx
= hlist_entry(mm
->ioctx_list
.first
, struct kioctx
, list
);
388 hlist_del_rcu(&ctx
->list
);
392 wait_for_all_aios(ctx
);
394 if (1 != atomic_read(&ctx
->users
))
396 "exit_aio:ioctx still alive: %d %d %d\n",
397 atomic_read(&ctx
->users
), ctx
->dead
,
404 * Allocate a slot for an aio request. Increments the users count
405 * of the kioctx so that the kioctx stays around until all requests are
406 * complete. Returns NULL if no requests are free.
408 * Returns with kiocb->users set to 2. The io submit code path holds
409 * an extra reference while submitting the i/o.
410 * This prevents races between the aio code path referencing the
411 * req (after submitting it) and aio_complete() freeing the req.
413 static struct kiocb
*__aio_get_req(struct kioctx
*ctx
)
415 struct kiocb
*req
= NULL
;
417 req
= kmem_cache_alloc(kiocb_cachep
, GFP_KERNEL
);
425 req
->ki_cancel
= NULL
;
426 req
->ki_retry
= NULL
;
429 req
->ki_iovec
= NULL
;
430 INIT_LIST_HEAD(&req
->ki_run_list
);
431 req
->ki_eventfd
= NULL
;
437 * struct kiocb's are allocated in batches to reduce the number of
438 * times the ctx lock is acquired and released.
440 #define KIOCB_BATCH_SIZE 32L
442 struct list_head head
;
443 long count
; /* number of requests left to allocate */
446 static void kiocb_batch_init(struct kiocb_batch
*batch
, long total
)
448 INIT_LIST_HEAD(&batch
->head
);
449 batch
->count
= total
;
452 static void kiocb_batch_free(struct kioctx
*ctx
, struct kiocb_batch
*batch
)
454 struct kiocb
*req
, *n
;
456 if (list_empty(&batch
->head
))
459 spin_lock_irq(&ctx
->ctx_lock
);
460 list_for_each_entry_safe(req
, n
, &batch
->head
, ki_batch
) {
461 list_del(&req
->ki_batch
);
462 list_del(&req
->ki_list
);
463 kmem_cache_free(kiocb_cachep
, req
);
466 if (unlikely(!ctx
->reqs_active
&& ctx
->dead
))
467 wake_up_all(&ctx
->wait
);
468 spin_unlock_irq(&ctx
->ctx_lock
);
472 * Allocate a batch of kiocbs. This avoids taking and dropping the
473 * context lock a lot during setup.
475 static int kiocb_batch_refill(struct kioctx
*ctx
, struct kiocb_batch
*batch
)
477 unsigned short allocated
, to_alloc
;
479 bool called_fput
= false;
480 struct kiocb
*req
, *n
;
481 struct aio_ring
*ring
;
483 to_alloc
= min(batch
->count
, KIOCB_BATCH_SIZE
);
484 for (allocated
= 0; allocated
< to_alloc
; allocated
++) {
485 req
= __aio_get_req(ctx
);
487 /* allocation failed, go with what we've got */
489 list_add(&req
->ki_batch
, &batch
->head
);
496 spin_lock_irq(&ctx
->ctx_lock
);
497 ring
= kmap_atomic(ctx
->ring_info
.ring_pages
[0]);
499 avail
= aio_ring_avail(&ctx
->ring_info
, ring
) - ctx
->reqs_active
;
501 if (avail
== 0 && !called_fput
) {
503 * Handle a potential starvation case. It is possible that
504 * we hold the last reference on a struct file, causing us
505 * to delay the final fput to non-irq context. In this case,
506 * ctx->reqs_active is artificially high. Calling the fput
507 * routine here may free up a slot in the event completion
508 * ring, allowing this allocation to succeed.
511 spin_unlock_irq(&ctx
->ctx_lock
);
512 aio_fput_routine(NULL
);
517 if (avail
< allocated
) {
518 /* Trim back the number of requests. */
519 list_for_each_entry_safe(req
, n
, &batch
->head
, ki_batch
) {
520 list_del(&req
->ki_batch
);
521 kmem_cache_free(kiocb_cachep
, req
);
522 if (--allocated
<= avail
)
527 batch
->count
-= allocated
;
528 list_for_each_entry(req
, &batch
->head
, ki_batch
) {
529 list_add(&req
->ki_list
, &ctx
->active_reqs
);
534 spin_unlock_irq(&ctx
->ctx_lock
);
540 static inline struct kiocb
*aio_get_req(struct kioctx
*ctx
,
541 struct kiocb_batch
*batch
)
545 if (list_empty(&batch
->head
))
546 if (kiocb_batch_refill(ctx
, batch
) == 0)
548 req
= list_first_entry(&batch
->head
, struct kiocb
, ki_batch
);
549 list_del(&req
->ki_batch
);
553 static inline void really_put_req(struct kioctx
*ctx
, struct kiocb
*req
)
555 assert_spin_locked(&ctx
->ctx_lock
);
557 if (req
->ki_eventfd
!= NULL
)
558 eventfd_ctx_put(req
->ki_eventfd
);
561 if (req
->ki_iovec
!= &req
->ki_inline_vec
)
562 kfree(req
->ki_iovec
);
563 kmem_cache_free(kiocb_cachep
, req
);
566 if (unlikely(!ctx
->reqs_active
&& ctx
->dead
))
567 wake_up_all(&ctx
->wait
);
570 static void aio_fput_routine(struct work_struct
*data
)
572 spin_lock_irq(&fput_lock
);
573 while (likely(!list_empty(&fput_head
))) {
574 struct kiocb
*req
= list_kiocb(fput_head
.next
);
575 struct kioctx
*ctx
= req
->ki_ctx
;
577 list_del(&req
->ki_list
);
578 spin_unlock_irq(&fput_lock
);
580 /* Complete the fput(s) */
581 if (req
->ki_filp
!= NULL
)
584 /* Link the iocb into the context's free list */
586 spin_lock_irq(&ctx
->ctx_lock
);
587 really_put_req(ctx
, req
);
589 * at that point ctx might've been killed, but actual
592 spin_unlock_irq(&ctx
->ctx_lock
);
595 spin_lock_irq(&fput_lock
);
597 spin_unlock_irq(&fput_lock
);
601 * Returns true if this put was the last user of the request.
603 static int __aio_put_req(struct kioctx
*ctx
, struct kiocb
*req
)
605 dprintk(KERN_DEBUG
"aio_put(%p): f_count=%ld\n",
606 req
, atomic_long_read(&req
->ki_filp
->f_count
));
608 assert_spin_locked(&ctx
->ctx_lock
);
611 BUG_ON(req
->ki_users
< 0);
612 if (likely(req
->ki_users
))
614 list_del(&req
->ki_list
); /* remove from active_reqs */
615 req
->ki_cancel
= NULL
;
616 req
->ki_retry
= NULL
;
619 * Try to optimize the aio and eventfd file* puts, by avoiding to
620 * schedule work in case it is not final fput() time. In normal cases,
621 * we would not be holding the last reference to the file*, so
622 * this function will be executed w/out any aio kthread wakeup.
624 if (unlikely(!fput_atomic(req
->ki_filp
))) {
625 spin_lock(&fput_lock
);
626 list_add(&req
->ki_list
, &fput_head
);
627 spin_unlock(&fput_lock
);
628 schedule_work(&fput_work
);
631 really_put_req(ctx
, req
);
637 * Returns true if this put was the last user of the kiocb,
638 * false if the request is still in use.
640 int aio_put_req(struct kiocb
*req
)
642 struct kioctx
*ctx
= req
->ki_ctx
;
644 spin_lock_irq(&ctx
->ctx_lock
);
645 ret
= __aio_put_req(ctx
, req
);
646 spin_unlock_irq(&ctx
->ctx_lock
);
649 EXPORT_SYMBOL(aio_put_req
);
651 static struct kioctx
*lookup_ioctx(unsigned long ctx_id
)
653 struct mm_struct
*mm
= current
->mm
;
654 struct kioctx
*ctx
, *ret
= NULL
;
655 struct hlist_node
*n
;
659 hlist_for_each_entry_rcu(ctx
, n
, &mm
->ioctx_list
, list
) {
661 * RCU protects us against accessing freed memory but
662 * we have to be careful not to get a reference when the
663 * reference count already dropped to 0 (ctx->dead test
664 * is unreliable because of races).
666 if (ctx
->user_id
== ctx_id
&& !ctx
->dead
&& try_get_ioctx(ctx
)){
677 * Queue up a kiocb to be retried. Assumes that the kiocb
678 * has already been marked as kicked, and places it on
679 * the retry run list for the corresponding ioctx, if it
680 * isn't already queued. Returns 1 if it actually queued
681 * the kiocb (to tell the caller to activate the work
682 * queue to process it), or 0, if it found that it was
685 static inline int __queue_kicked_iocb(struct kiocb
*iocb
)
687 struct kioctx
*ctx
= iocb
->ki_ctx
;
689 assert_spin_locked(&ctx
->ctx_lock
);
691 if (list_empty(&iocb
->ki_run_list
)) {
692 list_add_tail(&iocb
->ki_run_list
,
700 * This is the core aio execution routine. It is
701 * invoked both for initial i/o submission and
702 * subsequent retries via the aio_kick_handler.
703 * Expects to be invoked with iocb->ki_ctx->lock
704 * already held. The lock is released and reacquired
705 * as needed during processing.
707 * Calls the iocb retry method (already setup for the
708 * iocb on initial submission) for operation specific
709 * handling, but takes care of most of common retry
710 * execution details for a given iocb. The retry method
711 * needs to be non-blocking as far as possible, to avoid
712 * holding up other iocbs waiting to be serviced by the
713 * retry kernel thread.
715 * The trickier parts in this code have to do with
716 * ensuring that only one retry instance is in progress
717 * for a given iocb at any time. Providing that guarantee
718 * simplifies the coding of individual aio operations as
719 * it avoids various potential races.
721 static ssize_t
aio_run_iocb(struct kiocb
*iocb
)
723 struct kioctx
*ctx
= iocb
->ki_ctx
;
724 ssize_t (*retry
)(struct kiocb
*);
727 if (!(retry
= iocb
->ki_retry
)) {
728 printk("aio_run_iocb: iocb->ki_retry = NULL\n");
733 * We don't want the next retry iteration for this
734 * operation to start until this one has returned and
735 * updated the iocb state. However, wait_queue functions
736 * can trigger a kick_iocb from interrupt context in the
737 * meantime, indicating that data is available for the next
738 * iteration. We want to remember that and enable the
739 * next retry iteration _after_ we are through with
742 * So, in order to be able to register a "kick", but
743 * prevent it from being queued now, we clear the kick
744 * flag, but make the kick code *think* that the iocb is
745 * still on the run list until we are actually done.
746 * When we are done with this iteration, we check if
747 * the iocb was kicked in the meantime and if so, queue
751 kiocbClearKicked(iocb
);
754 * This is so that aio_complete knows it doesn't need to
755 * pull the iocb off the run list (We can't just call
756 * INIT_LIST_HEAD because we don't want a kick_iocb to
757 * queue this on the run list yet)
759 iocb
->ki_run_list
.next
= iocb
->ki_run_list
.prev
= NULL
;
760 spin_unlock_irq(&ctx
->ctx_lock
);
762 /* Quit retrying if the i/o has been cancelled */
763 if (kiocbIsCancelled(iocb
)) {
765 aio_complete(iocb
, ret
, 0);
766 /* must not access the iocb after this */
771 * Now we are all set to call the retry method in async
776 if (ret
!= -EIOCBRETRY
&& ret
!= -EIOCBQUEUED
) {
778 * There's no easy way to restart the syscall since other AIO's
779 * may be already running. Just fail this IO with EINTR.
781 if (unlikely(ret
== -ERESTARTSYS
|| ret
== -ERESTARTNOINTR
||
782 ret
== -ERESTARTNOHAND
|| ret
== -ERESTART_RESTARTBLOCK
))
784 aio_complete(iocb
, ret
, 0);
787 spin_lock_irq(&ctx
->ctx_lock
);
789 if (-EIOCBRETRY
== ret
) {
791 * OK, now that we are done with this iteration
792 * and know that there is more left to go,
793 * this is where we let go so that a subsequent
794 * "kick" can start the next iteration
797 /* will make __queue_kicked_iocb succeed from here on */
798 INIT_LIST_HEAD(&iocb
->ki_run_list
);
799 /* we must queue the next iteration ourselves, if it
800 * has already been kicked */
801 if (kiocbIsKicked(iocb
)) {
802 __queue_kicked_iocb(iocb
);
805 * __queue_kicked_iocb will always return 1 here, because
806 * iocb->ki_run_list is empty at this point so it should
807 * be safe to unconditionally queue the context into the
818 * Process all pending retries queued on the ioctx
820 * Assumes it is operating within the aio issuer's mm
823 static int __aio_run_iocbs(struct kioctx
*ctx
)
826 struct list_head run_list
;
828 assert_spin_locked(&ctx
->ctx_lock
);
830 list_replace_init(&ctx
->run_list
, &run_list
);
831 while (!list_empty(&run_list
)) {
832 iocb
= list_entry(run_list
.next
, struct kiocb
,
834 list_del(&iocb
->ki_run_list
);
836 * Hold an extra reference while retrying i/o.
838 iocb
->ki_users
++; /* grab extra reference */
840 __aio_put_req(ctx
, iocb
);
842 if (!list_empty(&ctx
->run_list
))
847 static void aio_queue_work(struct kioctx
* ctx
)
849 unsigned long timeout
;
851 * if someone is waiting, get the work started right
852 * away, otherwise, use a longer delay
855 if (waitqueue_active(&ctx
->wait
))
859 queue_delayed_work(aio_wq
, &ctx
->wq
, timeout
);
864 * Process all pending retries queued on the ioctx
865 * run list, and keep running them until the list
867 * Assumes it is operating within the aio issuer's mm context.
869 static inline void aio_run_all_iocbs(struct kioctx
*ctx
)
871 spin_lock_irq(&ctx
->ctx_lock
);
872 while (__aio_run_iocbs(ctx
))
874 spin_unlock_irq(&ctx
->ctx_lock
);
879 * Work queue handler triggered to process pending
880 * retries on an ioctx. Takes on the aio issuer's
881 * mm context before running the iocbs, so that
882 * copy_xxx_user operates on the issuer's address
884 * Run on aiod's context.
886 static void aio_kick_handler(struct work_struct
*work
)
888 struct kioctx
*ctx
= container_of(work
, struct kioctx
, wq
.work
);
889 mm_segment_t oldfs
= get_fs();
890 struct mm_struct
*mm
;
895 spin_lock_irq(&ctx
->ctx_lock
);
896 requeue
=__aio_run_iocbs(ctx
);
898 spin_unlock_irq(&ctx
->ctx_lock
);
902 * we're in a worker thread already; no point using non-zero delay
905 queue_delayed_work(aio_wq
, &ctx
->wq
, 0);
910 * Called by kick_iocb to queue the kiocb for retry
911 * and if required activate the aio work queue to process
914 static void try_queue_kicked_iocb(struct kiocb
*iocb
)
916 struct kioctx
*ctx
= iocb
->ki_ctx
;
920 spin_lock_irqsave(&ctx
->ctx_lock
, flags
);
921 /* set this inside the lock so that we can't race with aio_run_iocb()
922 * testing it and putting the iocb on the run list under the lock */
923 if (!kiocbTryKick(iocb
))
924 run
= __queue_kicked_iocb(iocb
);
925 spin_unlock_irqrestore(&ctx
->ctx_lock
, flags
);
932 * Called typically from a wait queue callback context
933 * to trigger a retry of the iocb.
934 * The retry is usually executed by aio workqueue
935 * threads (See aio_kick_handler).
937 void kick_iocb(struct kiocb
*iocb
)
939 /* sync iocbs are easy: they can only ever be executing from a
941 if (is_sync_kiocb(iocb
)) {
942 kiocbSetKicked(iocb
);
943 wake_up_process(iocb
->ki_obj
.tsk
);
947 try_queue_kicked_iocb(iocb
);
949 EXPORT_SYMBOL(kick_iocb
);
952 * Called when the io request on the given iocb is complete.
953 * Returns true if this is the last user of the request. The
954 * only other user of the request can be the cancellation code.
956 int aio_complete(struct kiocb
*iocb
, long res
, long res2
)
958 struct kioctx
*ctx
= iocb
->ki_ctx
;
959 struct aio_ring_info
*info
;
960 struct aio_ring
*ring
;
961 struct io_event
*event
;
967 * Special case handling for sync iocbs:
968 * - events go directly into the iocb for fast handling
969 * - the sync task with the iocb in its stack holds the single iocb
970 * ref, no other paths have a way to get another ref
971 * - the sync task helpfully left a reference to itself in the iocb
973 if (is_sync_kiocb(iocb
)) {
974 BUG_ON(iocb
->ki_users
!= 1);
975 iocb
->ki_user_data
= res
;
977 wake_up_process(iocb
->ki_obj
.tsk
);
981 info
= &ctx
->ring_info
;
983 /* add a completion event to the ring buffer.
984 * must be done holding ctx->ctx_lock to prevent
985 * other code from messing with the tail
986 * pointer since we might be called from irq
989 spin_lock_irqsave(&ctx
->ctx_lock
, flags
);
991 if (iocb
->ki_run_list
.prev
&& !list_empty(&iocb
->ki_run_list
))
992 list_del_init(&iocb
->ki_run_list
);
995 * cancelled requests don't get events, userland was given one
996 * when the event got cancelled.
998 if (kiocbIsCancelled(iocb
))
1001 ring
= kmap_atomic(info
->ring_pages
[0]);
1004 event
= aio_ring_event(info
, tail
);
1005 if (++tail
>= info
->nr
)
1008 event
->obj
= (u64
)(unsigned long)iocb
->ki_obj
.user
;
1009 event
->data
= iocb
->ki_user_data
;
1013 dprintk("aio_complete: %p[%lu]: %p: %p %Lx %lx %lx\n",
1014 ctx
, tail
, iocb
, iocb
->ki_obj
.user
, iocb
->ki_user_data
,
1017 /* after flagging the request as done, we
1018 * must never even look at it again
1020 smp_wmb(); /* make event visible before updating tail */
1025 put_aio_ring_event(event
);
1026 kunmap_atomic(ring
);
1028 pr_debug("added to ring %p at [%lu]\n", iocb
, tail
);
1031 * Check if the user asked us to deliver the result through an
1032 * eventfd. The eventfd_signal() function is safe to be called
1035 if (iocb
->ki_eventfd
!= NULL
)
1036 eventfd_signal(iocb
->ki_eventfd
, 1);
1039 /* everything turned out well, dispose of the aiocb. */
1040 ret
= __aio_put_req(ctx
, iocb
);
1043 * We have to order our ring_info tail store above and test
1044 * of the wait list below outside the wait lock. This is
1045 * like in wake_up_bit() where clearing a bit has to be
1046 * ordered with the unlocked test.
1050 if (waitqueue_active(&ctx
->wait
))
1051 wake_up(&ctx
->wait
);
1053 spin_unlock_irqrestore(&ctx
->ctx_lock
, flags
);
1056 EXPORT_SYMBOL(aio_complete
);
1059 * Pull an event off of the ioctx's event ring. Returns the number of
1060 * events fetched (0 or 1 ;-)
1061 * FIXME: make this use cmpxchg.
1062 * TODO: make the ringbuffer user mmap()able (requires FIXME).
1064 static int aio_read_evt(struct kioctx
*ioctx
, struct io_event
*ent
)
1066 struct aio_ring_info
*info
= &ioctx
->ring_info
;
1067 struct aio_ring
*ring
;
1071 ring
= kmap_atomic(info
->ring_pages
[0]);
1072 dprintk("in aio_read_evt h%lu t%lu m%lu\n",
1073 (unsigned long)ring
->head
, (unsigned long)ring
->tail
,
1074 (unsigned long)ring
->nr
);
1076 if (ring
->head
== ring
->tail
)
1079 spin_lock(&info
->ring_lock
);
1081 head
= ring
->head
% info
->nr
;
1082 if (head
!= ring
->tail
) {
1083 struct io_event
*evp
= aio_ring_event(info
, head
);
1085 head
= (head
+ 1) % info
->nr
;
1086 smp_mb(); /* finish reading the event before updatng the head */
1089 put_aio_ring_event(evp
);
1091 spin_unlock(&info
->ring_lock
);
1094 kunmap_atomic(ring
);
1095 dprintk("leaving aio_read_evt: %d h%lu t%lu\n", ret
,
1096 (unsigned long)ring
->head
, (unsigned long)ring
->tail
);
1100 struct aio_timeout
{
1101 struct timer_list timer
;
1103 struct task_struct
*p
;
1106 static void timeout_func(unsigned long data
)
1108 struct aio_timeout
*to
= (struct aio_timeout
*)data
;
1111 wake_up_process(to
->p
);
1114 static inline void init_timeout(struct aio_timeout
*to
)
1116 setup_timer_on_stack(&to
->timer
, timeout_func
, (unsigned long) to
);
1121 static inline void set_timeout(long start_jiffies
, struct aio_timeout
*to
,
1122 const struct timespec
*ts
)
1124 to
->timer
.expires
= start_jiffies
+ timespec_to_jiffies(ts
);
1125 if (time_after(to
->timer
.expires
, jiffies
))
1126 add_timer(&to
->timer
);
1131 static inline void clear_timeout(struct aio_timeout
*to
)
1133 del_singleshot_timer_sync(&to
->timer
);
1136 static int read_events(struct kioctx
*ctx
,
1137 long min_nr
, long nr
,
1138 struct io_event __user
*event
,
1139 struct timespec __user
*timeout
)
1141 long start_jiffies
= jiffies
;
1142 struct task_struct
*tsk
= current
;
1143 DECLARE_WAITQUEUE(wait
, tsk
);
1146 struct io_event ent
;
1147 struct aio_timeout to
;
1150 /* needed to zero any padding within an entry (there shouldn't be
1151 * any, but C is fun!
1153 memset(&ent
, 0, sizeof(ent
));
1156 while (likely(i
< nr
)) {
1157 ret
= aio_read_evt(ctx
, &ent
);
1158 if (unlikely(ret
<= 0))
1161 dprintk("read event: %Lx %Lx %Lx %Lx\n",
1162 ent
.data
, ent
.obj
, ent
.res
, ent
.res2
);
1164 /* Could we split the check in two? */
1166 if (unlikely(copy_to_user(event
, &ent
, sizeof(ent
)))) {
1167 dprintk("aio: lost an event due to EFAULT.\n");
1172 /* Good, event copied to userland, update counts. */
1184 /* racey check, but it gets redone */
1185 if (!retry
&& unlikely(!list_empty(&ctx
->run_list
))) {
1187 aio_run_all_iocbs(ctx
);
1195 if (unlikely(copy_from_user(&ts
, timeout
, sizeof(ts
))))
1198 set_timeout(start_jiffies
, &to
, &ts
);
1201 while (likely(i
< nr
)) {
1202 add_wait_queue_exclusive(&ctx
->wait
, &wait
);
1204 set_task_state(tsk
, TASK_INTERRUPTIBLE
);
1205 ret
= aio_read_evt(ctx
, &ent
);
1210 if (unlikely(ctx
->dead
)) {
1214 if (to
.timed_out
) /* Only check after read evt */
1216 /* Try to only show up in io wait if there are ops
1218 if (ctx
->reqs_active
)
1222 if (signal_pending(tsk
)) {
1226 /*ret = aio_read_evt(ctx, &ent);*/
1229 set_task_state(tsk
, TASK_RUNNING
);
1230 remove_wait_queue(&ctx
->wait
, &wait
);
1232 if (unlikely(ret
<= 0))
1236 if (unlikely(copy_to_user(event
, &ent
, sizeof(ent
)))) {
1237 dprintk("aio: lost an event due to EFAULT.\n");
1241 /* Good, event copied to userland, update counts. */
1249 destroy_timer_on_stack(&to
.timer
);
1253 /* Take an ioctx and remove it from the list of ioctx's. Protects
1254 * against races with itself via ->dead.
1256 static void io_destroy(struct kioctx
*ioctx
)
1258 struct mm_struct
*mm
= current
->mm
;
1261 /* delete the entry from the list is someone else hasn't already */
1262 spin_lock(&mm
->ioctx_lock
);
1263 was_dead
= ioctx
->dead
;
1265 hlist_del_rcu(&ioctx
->list
);
1266 spin_unlock(&mm
->ioctx_lock
);
1268 dprintk("aio_release(%p)\n", ioctx
);
1269 if (likely(!was_dead
))
1270 put_ioctx(ioctx
); /* twice for the list */
1272 aio_cancel_all(ioctx
);
1273 wait_for_all_aios(ioctx
);
1276 * Wake up any waiters. The setting of ctx->dead must be seen
1277 * by other CPUs at this point. Right now, we rely on the
1278 * locking done by the above calls to ensure this consistency.
1280 wake_up_all(&ioctx
->wait
);
1281 put_ioctx(ioctx
); /* once for the lookup */
1285 * Create an aio_context capable of receiving at least nr_events.
1286 * ctxp must not point to an aio_context that already exists, and
1287 * must be initialized to 0 prior to the call. On successful
1288 * creation of the aio_context, *ctxp is filled in with the resulting
1289 * handle. May fail with -EINVAL if *ctxp is not initialized,
1290 * if the specified nr_events exceeds internal limits. May fail
1291 * with -EAGAIN if the specified nr_events exceeds the user's limit
1292 * of available events. May fail with -ENOMEM if insufficient kernel
1293 * resources are available. May fail with -EFAULT if an invalid
1294 * pointer is passed for ctxp. Will fail with -ENOSYS if not
1297 SYSCALL_DEFINE2(io_setup
, unsigned, nr_events
, aio_context_t __user
*, ctxp
)
1299 struct kioctx
*ioctx
= NULL
;
1303 ret
= get_user(ctx
, ctxp
);
1308 if (unlikely(ctx
|| nr_events
== 0)) {
1309 pr_debug("EINVAL: io_setup: ctx %lu nr_events %u\n",
1314 ioctx
= ioctx_alloc(nr_events
);
1315 ret
= PTR_ERR(ioctx
);
1316 if (!IS_ERR(ioctx
)) {
1317 ret
= put_user(ioctx
->user_id
, ctxp
);
1330 * Destroy the aio_context specified. May cancel any outstanding
1331 * AIOs and block on completion. Will fail with -ENOSYS if not
1332 * implemented. May fail with -EINVAL if the context pointed to
1335 SYSCALL_DEFINE1(io_destroy
, aio_context_t
, ctx
)
1337 struct kioctx
*ioctx
= lookup_ioctx(ctx
);
1338 if (likely(NULL
!= ioctx
)) {
1342 pr_debug("EINVAL: io_destroy: invalid context id\n");
1346 static void aio_advance_iovec(struct kiocb
*iocb
, ssize_t ret
)
1348 struct iovec
*iov
= &iocb
->ki_iovec
[iocb
->ki_cur_seg
];
1352 while (iocb
->ki_cur_seg
< iocb
->ki_nr_segs
&& ret
> 0) {
1353 ssize_t
this = min((ssize_t
)iov
->iov_len
, ret
);
1354 iov
->iov_base
+= this;
1355 iov
->iov_len
-= this;
1356 iocb
->ki_left
-= this;
1358 if (iov
->iov_len
== 0) {
1364 /* the caller should not have done more io than what fit in
1365 * the remaining iovecs */
1366 BUG_ON(ret
> 0 && iocb
->ki_left
== 0);
1369 static ssize_t
aio_rw_vect_retry(struct kiocb
*iocb
)
1371 struct file
*file
= iocb
->ki_filp
;
1372 struct address_space
*mapping
= file
->f_mapping
;
1373 struct inode
*inode
= mapping
->host
;
1374 ssize_t (*rw_op
)(struct kiocb
*, const struct iovec
*,
1375 unsigned long, loff_t
);
1377 unsigned short opcode
;
1379 if ((iocb
->ki_opcode
== IOCB_CMD_PREADV
) ||
1380 (iocb
->ki_opcode
== IOCB_CMD_PREAD
)) {
1381 rw_op
= file
->f_op
->aio_read
;
1382 opcode
= IOCB_CMD_PREADV
;
1384 rw_op
= file
->f_op
->aio_write
;
1385 opcode
= IOCB_CMD_PWRITEV
;
1388 /* This matches the pread()/pwrite() logic */
1389 if (iocb
->ki_pos
< 0)
1393 ret
= rw_op(iocb
, &iocb
->ki_iovec
[iocb
->ki_cur_seg
],
1394 iocb
->ki_nr_segs
- iocb
->ki_cur_seg
,
1397 aio_advance_iovec(iocb
, ret
);
1399 /* retry all partial writes. retry partial reads as long as its a
1401 } while (ret
> 0 && iocb
->ki_left
> 0 &&
1402 (opcode
== IOCB_CMD_PWRITEV
||
1403 (!S_ISFIFO(inode
->i_mode
) && !S_ISSOCK(inode
->i_mode
))));
1405 /* This means we must have transferred all that we could */
1406 /* No need to retry anymore */
1407 if ((ret
== 0) || (iocb
->ki_left
== 0))
1408 ret
= iocb
->ki_nbytes
- iocb
->ki_left
;
1410 /* If we managed to write some out we return that, rather than
1411 * the eventual error. */
1412 if (opcode
== IOCB_CMD_PWRITEV
1413 && ret
< 0 && ret
!= -EIOCBQUEUED
&& ret
!= -EIOCBRETRY
1414 && iocb
->ki_nbytes
- iocb
->ki_left
)
1415 ret
= iocb
->ki_nbytes
- iocb
->ki_left
;
1420 static ssize_t
aio_fdsync(struct kiocb
*iocb
)
1422 struct file
*file
= iocb
->ki_filp
;
1423 ssize_t ret
= -EINVAL
;
1425 if (file
->f_op
->aio_fsync
)
1426 ret
= file
->f_op
->aio_fsync(iocb
, 1);
1430 static ssize_t
aio_fsync(struct kiocb
*iocb
)
1432 struct file
*file
= iocb
->ki_filp
;
1433 ssize_t ret
= -EINVAL
;
1435 if (file
->f_op
->aio_fsync
)
1436 ret
= file
->f_op
->aio_fsync(iocb
, 0);
1440 static ssize_t
aio_setup_vectored_rw(int type
, struct kiocb
*kiocb
, bool compat
)
1444 #ifdef CONFIG_COMPAT
1446 ret
= compat_rw_copy_check_uvector(type
,
1447 (struct compat_iovec __user
*)kiocb
->ki_buf
,
1448 kiocb
->ki_nbytes
, 1, &kiocb
->ki_inline_vec
,
1449 &kiocb
->ki_iovec
, 1);
1452 ret
= rw_copy_check_uvector(type
,
1453 (struct iovec __user
*)kiocb
->ki_buf
,
1454 kiocb
->ki_nbytes
, 1, &kiocb
->ki_inline_vec
,
1455 &kiocb
->ki_iovec
, 1);
1459 kiocb
->ki_nr_segs
= kiocb
->ki_nbytes
;
1460 kiocb
->ki_cur_seg
= 0;
1461 /* ki_nbytes/left now reflect bytes instead of segs */
1462 kiocb
->ki_nbytes
= ret
;
1463 kiocb
->ki_left
= ret
;
1470 static ssize_t
aio_setup_single_vector(struct kiocb
*kiocb
)
1472 kiocb
->ki_iovec
= &kiocb
->ki_inline_vec
;
1473 kiocb
->ki_iovec
->iov_base
= kiocb
->ki_buf
;
1474 kiocb
->ki_iovec
->iov_len
= kiocb
->ki_left
;
1475 kiocb
->ki_nr_segs
= 1;
1476 kiocb
->ki_cur_seg
= 0;
1482 * Performs the initial checks and aio retry method
1483 * setup for the kiocb at the time of io submission.
1485 static ssize_t
aio_setup_iocb(struct kiocb
*kiocb
, bool compat
)
1487 struct file
*file
= kiocb
->ki_filp
;
1490 switch (kiocb
->ki_opcode
) {
1491 case IOCB_CMD_PREAD
:
1493 if (unlikely(!(file
->f_mode
& FMODE_READ
)))
1496 if (unlikely(!access_ok(VERIFY_WRITE
, kiocb
->ki_buf
,
1499 ret
= security_file_permission(file
, MAY_READ
);
1502 ret
= aio_setup_single_vector(kiocb
);
1506 if (file
->f_op
->aio_read
)
1507 kiocb
->ki_retry
= aio_rw_vect_retry
;
1509 case IOCB_CMD_PWRITE
:
1511 if (unlikely(!(file
->f_mode
& FMODE_WRITE
)))
1514 if (unlikely(!access_ok(VERIFY_READ
, kiocb
->ki_buf
,
1517 ret
= security_file_permission(file
, MAY_WRITE
);
1520 ret
= aio_setup_single_vector(kiocb
);
1524 if (file
->f_op
->aio_write
)
1525 kiocb
->ki_retry
= aio_rw_vect_retry
;
1527 case IOCB_CMD_PREADV
:
1529 if (unlikely(!(file
->f_mode
& FMODE_READ
)))
1531 ret
= security_file_permission(file
, MAY_READ
);
1534 ret
= aio_setup_vectored_rw(READ
, kiocb
, compat
);
1538 if (file
->f_op
->aio_read
)
1539 kiocb
->ki_retry
= aio_rw_vect_retry
;
1541 case IOCB_CMD_PWRITEV
:
1543 if (unlikely(!(file
->f_mode
& FMODE_WRITE
)))
1545 ret
= security_file_permission(file
, MAY_WRITE
);
1548 ret
= aio_setup_vectored_rw(WRITE
, kiocb
, compat
);
1552 if (file
->f_op
->aio_write
)
1553 kiocb
->ki_retry
= aio_rw_vect_retry
;
1555 case IOCB_CMD_FDSYNC
:
1557 if (file
->f_op
->aio_fsync
)
1558 kiocb
->ki_retry
= aio_fdsync
;
1560 case IOCB_CMD_FSYNC
:
1562 if (file
->f_op
->aio_fsync
)
1563 kiocb
->ki_retry
= aio_fsync
;
1566 dprintk("EINVAL: io_submit: no operation provided\n");
1570 if (!kiocb
->ki_retry
)
1576 static int io_submit_one(struct kioctx
*ctx
, struct iocb __user
*user_iocb
,
1577 struct iocb
*iocb
, struct kiocb_batch
*batch
,
1584 /* enforce forwards compatibility on users */
1585 if (unlikely(iocb
->aio_reserved1
|| iocb
->aio_reserved2
)) {
1586 pr_debug("EINVAL: io_submit: reserve field set\n");
1590 /* prevent overflows */
1592 (iocb
->aio_buf
!= (unsigned long)iocb
->aio_buf
) ||
1593 (iocb
->aio_nbytes
!= (size_t)iocb
->aio_nbytes
) ||
1594 ((ssize_t
)iocb
->aio_nbytes
< 0)
1596 pr_debug("EINVAL: io_submit: overflow check\n");
1600 file
= fget(iocb
->aio_fildes
);
1601 if (unlikely(!file
))
1604 req
= aio_get_req(ctx
, batch
); /* returns with 2 references to req */
1605 if (unlikely(!req
)) {
1609 req
->ki_filp
= file
;
1610 if (iocb
->aio_flags
& IOCB_FLAG_RESFD
) {
1612 * If the IOCB_FLAG_RESFD flag of aio_flags is set, get an
1613 * instance of the file* now. The file descriptor must be
1614 * an eventfd() fd, and will be signaled for each completed
1615 * event using the eventfd_signal() function.
1617 req
->ki_eventfd
= eventfd_ctx_fdget((int) iocb
->aio_resfd
);
1618 if (IS_ERR(req
->ki_eventfd
)) {
1619 ret
= PTR_ERR(req
->ki_eventfd
);
1620 req
->ki_eventfd
= NULL
;
1625 ret
= put_user(req
->ki_key
, &user_iocb
->aio_key
);
1626 if (unlikely(ret
)) {
1627 dprintk("EFAULT: aio_key\n");
1631 req
->ki_obj
.user
= user_iocb
;
1632 req
->ki_user_data
= iocb
->aio_data
;
1633 req
->ki_pos
= iocb
->aio_offset
;
1635 req
->ki_buf
= (char __user
*)(unsigned long)iocb
->aio_buf
;
1636 req
->ki_left
= req
->ki_nbytes
= iocb
->aio_nbytes
;
1637 req
->ki_opcode
= iocb
->aio_lio_opcode
;
1639 ret
= aio_setup_iocb(req
, compat
);
1644 spin_lock_irq(&ctx
->ctx_lock
);
1646 * We could have raced with io_destroy() and are currently holding a
1647 * reference to ctx which should be destroyed. We cannot submit IO
1648 * since ctx gets freed as soon as io_submit() puts its reference. The
1649 * check here is reliable: io_destroy() sets ctx->dead before waiting
1650 * for outstanding IO and the barrier between these two is realized by
1651 * unlock of mm->ioctx_lock and lock of ctx->ctx_lock. Analogously we
1652 * increment ctx->reqs_active before checking for ctx->dead and the
1653 * barrier is realized by unlock and lock of ctx->ctx_lock. Thus if we
1654 * don't see ctx->dead set here, io_destroy() waits for our IO to
1658 spin_unlock_irq(&ctx
->ctx_lock
);
1663 if (!list_empty(&ctx
->run_list
)) {
1664 /* drain the run list */
1665 while (__aio_run_iocbs(ctx
))
1668 spin_unlock_irq(&ctx
->ctx_lock
);
1670 aio_put_req(req
); /* drop extra ref to req */
1674 aio_put_req(req
); /* drop extra ref to req */
1675 aio_put_req(req
); /* drop i/o ref to req */
1679 long do_io_submit(aio_context_t ctx_id
, long nr
,
1680 struct iocb __user
*__user
*iocbpp
, bool compat
)
1685 struct blk_plug plug
;
1686 struct kiocb_batch batch
;
1688 if (unlikely(nr
< 0))
1691 if (unlikely(nr
> LONG_MAX
/sizeof(*iocbpp
)))
1692 nr
= LONG_MAX
/sizeof(*iocbpp
);
1694 if (unlikely(!access_ok(VERIFY_READ
, iocbpp
, (nr
*sizeof(*iocbpp
)))))
1697 ctx
= lookup_ioctx(ctx_id
);
1698 if (unlikely(!ctx
)) {
1699 pr_debug("EINVAL: io_submit: invalid context id\n");
1703 kiocb_batch_init(&batch
, nr
);
1705 blk_start_plug(&plug
);
1708 * AKPM: should this return a partial result if some of the IOs were
1709 * successfully submitted?
1711 for (i
=0; i
<nr
; i
++) {
1712 struct iocb __user
*user_iocb
;
1715 if (unlikely(__get_user(user_iocb
, iocbpp
+ i
))) {
1720 if (unlikely(copy_from_user(&tmp
, user_iocb
, sizeof(tmp
)))) {
1725 ret
= io_submit_one(ctx
, user_iocb
, &tmp
, &batch
, compat
);
1729 blk_finish_plug(&plug
);
1731 kiocb_batch_free(ctx
, &batch
);
1737 * Queue the nr iocbs pointed to by iocbpp for processing. Returns
1738 * the number of iocbs queued. May return -EINVAL if the aio_context
1739 * specified by ctx_id is invalid, if nr is < 0, if the iocb at
1740 * *iocbpp[0] is not properly initialized, if the operation specified
1741 * is invalid for the file descriptor in the iocb. May fail with
1742 * -EFAULT if any of the data structures point to invalid data. May
1743 * fail with -EBADF if the file descriptor specified in the first
1744 * iocb is invalid. May fail with -EAGAIN if insufficient resources
1745 * are available to queue any iocbs. Will return 0 if nr is 0. Will
1746 * fail with -ENOSYS if not implemented.
1748 SYSCALL_DEFINE3(io_submit
, aio_context_t
, ctx_id
, long, nr
,
1749 struct iocb __user
* __user
*, iocbpp
)
1751 return do_io_submit(ctx_id
, nr
, iocbpp
, 0);
1755 * Finds a given iocb for cancellation.
1757 static struct kiocb
*lookup_kiocb(struct kioctx
*ctx
, struct iocb __user
*iocb
,
1760 struct list_head
*pos
;
1762 assert_spin_locked(&ctx
->ctx_lock
);
1764 /* TODO: use a hash or array, this sucks. */
1765 list_for_each(pos
, &ctx
->active_reqs
) {
1766 struct kiocb
*kiocb
= list_kiocb(pos
);
1767 if (kiocb
->ki_obj
.user
== iocb
&& kiocb
->ki_key
== key
)
1774 * Attempts to cancel an iocb previously passed to io_submit. If
1775 * the operation is successfully cancelled, the resulting event is
1776 * copied into the memory pointed to by result without being placed
1777 * into the completion queue and 0 is returned. May fail with
1778 * -EFAULT if any of the data structures pointed to are invalid.
1779 * May fail with -EINVAL if aio_context specified by ctx_id is
1780 * invalid. May fail with -EAGAIN if the iocb specified was not
1781 * cancelled. Will fail with -ENOSYS if not implemented.
1783 SYSCALL_DEFINE3(io_cancel
, aio_context_t
, ctx_id
, struct iocb __user
*, iocb
,
1784 struct io_event __user
*, result
)
1786 int (*cancel
)(struct kiocb
*iocb
, struct io_event
*res
);
1788 struct kiocb
*kiocb
;
1792 ret
= get_user(key
, &iocb
->aio_key
);
1796 ctx
= lookup_ioctx(ctx_id
);
1800 spin_lock_irq(&ctx
->ctx_lock
);
1802 kiocb
= lookup_kiocb(ctx
, iocb
, key
);
1803 if (kiocb
&& kiocb
->ki_cancel
) {
1804 cancel
= kiocb
->ki_cancel
;
1806 kiocbSetCancelled(kiocb
);
1809 spin_unlock_irq(&ctx
->ctx_lock
);
1811 if (NULL
!= cancel
) {
1812 struct io_event tmp
;
1813 pr_debug("calling cancel\n");
1814 memset(&tmp
, 0, sizeof(tmp
));
1815 tmp
.obj
= (u64
)(unsigned long)kiocb
->ki_obj
.user
;
1816 tmp
.data
= kiocb
->ki_user_data
;
1817 ret
= cancel(kiocb
, &tmp
);
1819 /* Cancellation succeeded -- copy the result
1820 * into the user's buffer.
1822 if (copy_to_user(result
, &tmp
, sizeof(tmp
)))
1834 * Attempts to read at least min_nr events and up to nr events from
1835 * the completion queue for the aio_context specified by ctx_id. If
1836 * it succeeds, the number of read events is returned. May fail with
1837 * -EINVAL if ctx_id is invalid, if min_nr is out of range, if nr is
1838 * out of range, if timeout is out of range. May fail with -EFAULT
1839 * if any of the memory specified is invalid. May return 0 or
1840 * < min_nr if the timeout specified by timeout has elapsed
1841 * before sufficient events are available, where timeout == NULL
1842 * specifies an infinite timeout. Note that the timeout pointed to by
1843 * timeout is relative and will be updated if not NULL and the
1844 * operation blocks. Will fail with -ENOSYS if not implemented.
1846 SYSCALL_DEFINE5(io_getevents
, aio_context_t
, ctx_id
,
1849 struct io_event __user
*, events
,
1850 struct timespec __user
*, timeout
)
1852 struct kioctx
*ioctx
= lookup_ioctx(ctx_id
);
1855 if (likely(ioctx
)) {
1856 if (likely(min_nr
<= nr
&& min_nr
>= 0))
1857 ret
= read_events(ioctx
, min_nr
, nr
, events
, timeout
);
1861 asmlinkage_protect(5, ret
, ctx_id
, min_nr
, nr
, events
, timeout
);