Kernel

Kernel Git Source Tree

Root/fs/aio.c

Source at commit 9845c1745d3d531a5b9544f5322c62bfb4d4e9bc created 1 year 2 months ago. By Xiangfu, rtc: jz4740 fix hwclock give time out
1	/*
2	* An async IO implementation for Linux
3	* Written by Benjamin LaHaise <bcrl@kvack.org>
4	*
5	* Implements an efficient asynchronous io interface.
6	*
7	* Copyright 2000, 2001, 2002 Red Hat, Inc. All Rights Reserved.
8	*
9	* See ../COPYING for licensing terms.
10	*/
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>
20
21	#define DEBUG 0
22
23	#include <linux/sched.h>
24	#include <linux/fs.h>
25	#include <linux/file.h>
26	#include <linux/mm.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>
38
39	#include <asm/kmap_types.h>
40	#include <asm/uaccess.h>
41
42	#if DEBUG > 1
43	#define dprintk printk
44	#else
45	#define dprintk(x...) do { ; } while (0)
46	#endif
47
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---/
53
54	static struct kmem_cache *kiocb_cachep;
55	static struct kmem_cache *kioctx_cachep;
56
57	static struct workqueue_struct *aio_wq;
58
59	/* Used for rare fput completion. */
60	static void aio_fput_routine(struct work_struct *);
61	static DECLARE_WORK(fput_work, aio_fput_routine);
62
63	static DEFINE_SPINLOCK(fput_lock);
64	static LIST_HEAD(fput_head);
65
66	static void aio_kick_handler(struct work_struct *);
67	static void aio_queue_work(struct kioctx *);
68
69	/* aio_setup
70	* Creates the slab caches used by the aio routines, panic on
71	* failure as this is done early during the boot sequence.
72	*/
73	static int __init aio_setup(void)
74	{
75	kiocb_cachep = KMEM_CACHE(kiocb, SLAB_HWCACHE_ALIGN\|SLAB_PANIC);
76	kioctx_cachep = KMEM_CACHE(kioctx,SLAB_HWCACHE_ALIGN\|SLAB_PANIC);
77
78	aio_wq = alloc_workqueue("aio", 0, 1); /* used to limit concurrency */
79	BUG_ON(!aio_wq);
80
81	pr_debug("aio_setup: sizeof(struct page) = %d\n", (int)sizeof(struct page));
82
83	return 0;
84	}
85	__initcall(aio_setup);
86
87	static void aio_free_ring(struct kioctx *ctx)
88	{
89	struct aio_ring_info *info = &ctx->ring_info;
90	long i;
91
92	for (i=0; i<info->nr_pages; i++)
93	put_page(info->ring_pages[i]);
94
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);
99	}
100
101	if (info->ring_pages && info->ring_pages != info->internal_pages)
102	kfree(info->ring_pages);
103	info->ring_pages = NULL;
104	info->nr = 0;
105	}
106
107	static int aio_setup_ring(struct kioctx *ctx)
108	{
109	struct aio_ring *ring;
110	struct aio_ring_info *info = &ctx->ring_info;
111	unsigned nr_events = ctx->max_reqs;
112	unsigned long size;
113	int nr_pages;
114
115	/* Compensate for the ring buffer's head/tail overlap entry */
116	nr_events += 2; /* 1 is required, 2 for good luck */
117
118	size = sizeof(struct aio_ring);
119	size += sizeof(struct io_event) * nr_events;
120	nr_pages = (size + PAGE_SIZE-1) >> PAGE_SHIFT;
121
122	if (nr_pages < 0)
123	return -EINVAL;
124
125	nr_events = (PAGE_SIZE * nr_pages - sizeof(struct aio_ring)) / sizeof(struct io_event);
126
127	info->nr = 0;
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)
132	return -ENOMEM;
133	}
134
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,
140	0);
141	if (IS_ERR((void *)info->mmap_base)) {
142	up_write(&ctx->mm->mmap_sem);
143	info->mmap_size = 0;
144	aio_free_ring(ctx);
145	return -EAGAIN;
146	}
147
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);
153
154	if (unlikely(info->nr_pages != nr_pages)) {
155	aio_free_ring(ctx);
156	return -EAGAIN;
157	}
158
159	ctx->user_id = info->mmap_base;
160
161	info->nr = nr_events; /* trusted copy */
162
163	ring = kmap_atomic(info->ring_pages[0], KM_USER0);
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);
171	kunmap_atomic(ring, KM_USER0);
172
173	return 0;
174	}
175
176
177	/* aio_ring_event: returns a pointer to the event at the given index from
178	* kmap_atomic(, km). Release the pointer with put_aio_ring_event();
179	*/
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)
183
184	#define aio_ring_event(info, nr, km) ({ \
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], km); \
189	__event += pos % AIO_EVENTS_PER_PAGE; \
190	__event; \
191	})
192
193	#define put_aio_ring_event(event, km) do { \
194	struct io_event *__event = (event); \
195	(void)__event; \
196	kunmap_atomic((void *)((unsigned long)__event & PAGE_MASK), km); \
197	} while(0)
198
199	static void ctx_rcu_free(struct rcu_head *head)
200	{
201	struct kioctx *ctx = container_of(head, struct kioctx, rcu_head);
202	unsigned nr_events = ctx->max_reqs;
203
204	kmem_cache_free(kioctx_cachep, ctx);
205
206	if (nr_events) {
207	spin_lock(&aio_nr_lock);
208	BUG_ON(aio_nr - nr_events > aio_nr);
209	aio_nr -= nr_events;
210	spin_unlock(&aio_nr_lock);
211	}
212	}
213
214	/* __put_ioctx
215	* Called when the last user of an aio context has gone away,
216	* and the struct needs to be freed.
217	*/
218	static void __put_ioctx(struct kioctx *ctx)
219	{
220	BUG_ON(ctx->reqs_active);
221
222	cancel_delayed_work(&ctx->wq);
223	cancel_work_sync(&ctx->wq.work);
224	aio_free_ring(ctx);
225	mmdrop(ctx->mm);
226	ctx->mm = NULL;
227	pr_debug("__put_ioctx: freeing %p\n", ctx);
228	call_rcu(&ctx->rcu_head, ctx_rcu_free);
229	}
230
231	static inline void get_ioctx(struct kioctx *kioctx)
232	{
233	BUG_ON(atomic_read(&kioctx->users) <= 0);
234	atomic_inc(&kioctx->users);
235	}
236
237	static inline int try_get_ioctx(struct kioctx *kioctx)
238	{
239	return atomic_inc_not_zero(&kioctx->users);
240	}
241
242	static inline void put_ioctx(struct kioctx *kioctx)
243	{
244	BUG_ON(atomic_read(&kioctx->users) <= 0);
245	if (unlikely(atomic_dec_and_test(&kioctx->users)))
246	__put_ioctx(kioctx);
247	}
248
249	/* ioctx_alloc
250	* Allocates and initializes an ioctx. Returns an ERR_PTR if it failed.
251	*/
252	static struct kioctx *ioctx_alloc(unsigned nr_events)
253	{
254	struct mm_struct *mm;
255	struct kioctx *ctx;
256	int did_sync = 0;
257
258	/* Prevent overflows */
259	if ((nr_events > (0x10000000U / sizeof(struct io_event))) \|\|
260	(nr_events > (0x10000000U / sizeof(struct kiocb)))) {
261	pr_debug("ENOMEM: nr_events too high\n");
262	return ERR_PTR(-EINVAL);
263	}
264
265	if ((unsigned long)nr_events > aio_max_nr)
266	return ERR_PTR(-EAGAIN);
267
268	ctx = kmem_cache_zalloc(kioctx_cachep, GFP_KERNEL);
269	if (!ctx)
270	return ERR_PTR(-ENOMEM);
271
272	ctx->max_reqs = nr_events;
273	mm = ctx->mm = current->mm;
274	atomic_inc(&mm->mm_count);
275
276	atomic_set(&ctx->users, 1);
277	spin_lock_init(&ctx->ctx_lock);
278	spin_lock_init(&ctx->ring_info.ring_lock);
279	init_waitqueue_head(&ctx->wait);
280
281	INIT_LIST_HEAD(&ctx->active_reqs);
282	INIT_LIST_HEAD(&ctx->run_list);
283	INIT_DELAYED_WORK(&ctx->wq, aio_kick_handler);
284
285	if (aio_setup_ring(ctx) < 0)
286	goto out_freectx;
287
288	/* limit the number of system wide aios */
289	do {
290	spin_lock_bh(&aio_nr_lock);
291	if (aio_nr + nr_events > aio_max_nr \|\|
292	aio_nr + nr_events < aio_nr)
293	ctx->max_reqs = 0;
294	else
295	aio_nr += ctx->max_reqs;
296	spin_unlock_bh(&aio_nr_lock);
297	if (ctx->max_reqs \|\| did_sync)
298	break;
299
300	/* wait for rcu callbacks to have completed before giving up */
301	synchronize_rcu();
302	did_sync = 1;
303	ctx->max_reqs = nr_events;
304	} while (1);
305
306	if (ctx->max_reqs == 0)
307	goto out_cleanup;
308
309	/* now link into global list. */
310	spin_lock(&mm->ioctx_lock);
311	hlist_add_head_rcu(&ctx->list, &mm->ioctx_list);
312	spin_unlock(&mm->ioctx_lock);
313
314	dprintk("aio: allocated ioctx %p[%ld]: mm=%p mask=0x%x\n",
315	ctx, ctx->user_id, current->mm, ctx->ring_info.nr);
316	return ctx;
317
318	out_cleanup:
319	__put_ioctx(ctx);
320	return ERR_PTR(-EAGAIN);
321
322	out_freectx:
323	mmdrop(mm);
324	kmem_cache_free(kioctx_cachep, ctx);
325	ctx = ERR_PTR(-ENOMEM);
326
327	dprintk("aio: error allocating ioctx %p\n", ctx);
328	return ctx;
329	}
330
331	/* aio_cancel_all
332	* Cancels all outstanding aio requests on an aio context. Used
333	* when the processes owning a context have all exited to encourage
334	* the rapid destruction of the kioctx.
335	*/
336	static void aio_cancel_all(struct kioctx *ctx)
337	{
338	int (cancel)(struct kiocb , struct io_event *);
339	struct io_event res;
340	spin_lock_irq(&ctx->ctx_lock);
341	ctx->dead = 1;
342	while (!list_empty(&ctx->active_reqs)) {
343	struct list_head *pos = ctx->active_reqs.next;
344	struct kiocb *iocb = list_kiocb(pos);
345	list_del_init(&iocb->ki_list);
346	cancel = iocb->ki_cancel;
347	kiocbSetCancelled(iocb);
348	if (cancel) {
349	iocb->ki_users++;
350	spin_unlock_irq(&ctx->ctx_lock);
351	cancel(iocb, &res);
352	spin_lock_irq(&ctx->ctx_lock);
353	}
354	}
355	spin_unlock_irq(&ctx->ctx_lock);
356	}
357
358	static void wait_for_all_aios(struct kioctx *ctx)
359	{
360	struct task_struct *tsk = current;
361	DECLARE_WAITQUEUE(wait, tsk);
362
363	spin_lock_irq(&ctx->ctx_lock);
364	if (!ctx->reqs_active)
365	goto out;
366
367	add_wait_queue(&ctx->wait, &wait);
368	set_task_state(tsk, TASK_UNINTERRUPTIBLE);
369	while (ctx->reqs_active) {
370	spin_unlock_irq(&ctx->ctx_lock);
371	io_schedule();
372	set_task_state(tsk, TASK_UNINTERRUPTIBLE);
373	spin_lock_irq(&ctx->ctx_lock);
374	}
375	__set_task_state(tsk, TASK_RUNNING);
376	remove_wait_queue(&ctx->wait, &wait);
377
378	out:
379	spin_unlock_irq(&ctx->ctx_lock);
380	}
381
382	/* wait_on_sync_kiocb:
383	* Waits on the given sync kiocb to complete.
384	*/
385	ssize_t wait_on_sync_kiocb(struct kiocb *iocb)
386	{
387	while (iocb->ki_users) {
388	set_current_state(TASK_UNINTERRUPTIBLE);
389	if (!iocb->ki_users)
390	break;
391	io_schedule();
392	}
393	__set_current_state(TASK_RUNNING);
394	return iocb->ki_user_data;
395	}
396	EXPORT_SYMBOL(wait_on_sync_kiocb);
397
398	/* exit_aio: called when the last user of mm goes away. At this point,
399	* there is no way for any new requests to be submited or any of the
400	* io_* syscalls to be called on the context. However, there may be
401	* outstanding requests which hold references to the context; as they
402	* go away, they will call put_ioctx and release any pinned memory
403	* associated with the request (held via struct page * references).
404	*/
405	void exit_aio(struct mm_struct *mm)
406	{
407	struct kioctx *ctx;
408
409	while (!hlist_empty(&mm->ioctx_list)) {
410	ctx = hlist_entry(mm->ioctx_list.first, struct kioctx, list);
411	hlist_del_rcu(&ctx->list);
412
413	aio_cancel_all(ctx);
414
415	wait_for_all_aios(ctx);
416	/*
417	* Ensure we don't leave the ctx on the aio_wq
418	*/
419	cancel_work_sync(&ctx->wq.work);
420
421	if (1 != atomic_read(&ctx->users))
422	printk(KERN_DEBUG
423	"exit_aio:ioctx still alive: %d %d %d\n",
424	atomic_read(&ctx->users), ctx->dead,
425	ctx->reqs_active);
426	put_ioctx(ctx);
427	}
428	}
429
430	/* aio_get_req
431	* Allocate a slot for an aio request. Increments the users count
432	* of the kioctx so that the kioctx stays around until all requests are
433	* complete. Returns NULL if no requests are free.
434	*
435	* Returns with kiocb->users set to 2. The io submit code path holds
436	* an extra reference while submitting the i/o.
437	* This prevents races between the aio code path referencing the
438	* req (after submitting it) and aio_complete() freeing the req.
439	*/
440	static struct kiocb __aio_get_req(struct kioctx ctx)
441	{
442	struct kiocb *req = NULL;
443
444	req = kmem_cache_alloc(kiocb_cachep, GFP_KERNEL);
445	if (unlikely(!req))
446	return NULL;
447
448	req->ki_flags = 0;
449	req->ki_users = 2;
450	req->ki_key = 0;
451	req->ki_ctx = ctx;
452	req->ki_cancel = NULL;
453	req->ki_retry = NULL;
454	req->ki_dtor = NULL;
455	req->private = NULL;
456	req->ki_iovec = NULL;
457	INIT_LIST_HEAD(&req->ki_run_list);
458	req->ki_eventfd = NULL;
459
460	return req;
461	}
462
463	/*
464	* struct kiocb's are allocated in batches to reduce the number of
465	* times the ctx lock is acquired and released.
466	*/
467	#define KIOCB_BATCH_SIZE 32L
468	struct kiocb_batch {
469	struct list_head head;
470	long count; /* number of requests left to allocate */
471	};
472
473	static void kiocb_batch_init(struct kiocb_batch *batch, long total)
474	{
475	INIT_LIST_HEAD(&batch->head);
476	batch->count = total;
477	}
478
479	static void kiocb_batch_free(struct kiocb_batch *batch)
480	{
481	struct kiocb req, n;
482
483	list_for_each_entry_safe(req, n, &batch->head, ki_batch) {
484	list_del(&req->ki_batch);
485	kmem_cache_free(kiocb_cachep, req);
486	}
487	}
488
489	/*
490	* Allocate a batch of kiocbs. This avoids taking and dropping the
491	* context lock a lot during setup.
492	*/
493	static int kiocb_batch_refill(struct kioctx ctx, struct kiocb_batch batch)
494	{
495	unsigned short allocated, to_alloc;
496	long avail;
497	bool called_fput = false;
498	struct kiocb req, n;
499	struct aio_ring *ring;
500
501	to_alloc = min(batch->count, KIOCB_BATCH_SIZE);
502	for (allocated = 0; allocated < to_alloc; allocated++) {
503	req = __aio_get_req(ctx);
504	if (!req)
505	/* allocation failed, go with what we've got */
506	break;
507	list_add(&req->ki_batch, &batch->head);
508	}
509
510	if (allocated == 0)
511	goto out;
512
513	retry:
514	spin_lock_irq(&ctx->ctx_lock);
515	ring = kmap_atomic(ctx->ring_info.ring_pages[0]);
516
517	avail = aio_ring_avail(&ctx->ring_info, ring) - ctx->reqs_active;
518	BUG_ON(avail < 0);
519	if (avail == 0 && !called_fput) {
520	/*
521	* Handle a potential starvation case. It is possible that
522	* we hold the last reference on a struct file, causing us
523	* to delay the final fput to non-irq context. In this case,
524	* ctx->reqs_active is artificially high. Calling the fput
525	* routine here may free up a slot in the event completion
526	* ring, allowing this allocation to succeed.
527	*/
528	kunmap_atomic(ring);
529	spin_unlock_irq(&ctx->ctx_lock);
530	aio_fput_routine(NULL);
531	called_fput = true;
532	goto retry;
533	}
534
535	if (avail < allocated) {
536	/* Trim back the number of requests. */
537	list_for_each_entry_safe(req, n, &batch->head, ki_batch) {
538	list_del(&req->ki_batch);
539	kmem_cache_free(kiocb_cachep, req);
540	if (--allocated <= avail)
541	break;
542	}
543	}
544
545	batch->count -= allocated;
546	list_for_each_entry(req, &batch->head, ki_batch) {
547	list_add(&req->ki_list, &ctx->active_reqs);
548	ctx->reqs_active++;
549	}
550
551	kunmap_atomic(ring);
552	spin_unlock_irq(&ctx->ctx_lock);
553
554	out:
555	return allocated;
556	}
557
558	static inline struct kiocb aio_get_req(struct kioctx ctx,
559	struct kiocb_batch *batch)
560	{
561	struct kiocb *req;
562
563	if (list_empty(&batch->head))
564	if (kiocb_batch_refill(ctx, batch) == 0)
565	return NULL;
566	req = list_first_entry(&batch->head, struct kiocb, ki_batch);
567	list_del(&req->ki_batch);
568	return req;
569	}
570
571	static inline void really_put_req(struct kioctx ctx, struct kiocb req)
572	{
573	assert_spin_locked(&ctx->ctx_lock);
574
575	if (req->ki_eventfd != NULL)
576	eventfd_ctx_put(req->ki_eventfd);
577	if (req->ki_dtor)
578	req->ki_dtor(req);
579	if (req->ki_iovec != &req->ki_inline_vec)
580	kfree(req->ki_iovec);
581	kmem_cache_free(kiocb_cachep, req);
582	ctx->reqs_active--;
583
584	if (unlikely(!ctx->reqs_active && ctx->dead))
585	wake_up_all(&ctx->wait);
586	}
587
588	static void aio_fput_routine(struct work_struct *data)
589	{
590	spin_lock_irq(&fput_lock);
591	while (likely(!list_empty(&fput_head))) {
592	struct kiocb *req = list_kiocb(fput_head.next);
593	struct kioctx *ctx = req->ki_ctx;
594
595	list_del(&req->ki_list);
596	spin_unlock_irq(&fput_lock);
597
598	/* Complete the fput(s) */
599	if (req->ki_filp != NULL)
600	fput(req->ki_filp);
601
602	/* Link the iocb into the context's free list */
603	spin_lock_irq(&ctx->ctx_lock);
604	really_put_req(ctx, req);
605	spin_unlock_irq(&ctx->ctx_lock);
606
607	put_ioctx(ctx);
608	spin_lock_irq(&fput_lock);
609	}
610	spin_unlock_irq(&fput_lock);
611	}
612
613	/* __aio_put_req
614	* Returns true if this put was the last user of the request.
615	*/
616	static int __aio_put_req(struct kioctx ctx, struct kiocb req)
617	{
618	dprintk(KERN_DEBUG "aio_put(%p): f_count=%ld\n",
619	req, atomic_long_read(&req->ki_filp->f_count));
620
621	assert_spin_locked(&ctx->ctx_lock);
622
623	req->ki_users--;
624	BUG_ON(req->ki_users < 0);
625	if (likely(req->ki_users))
626	return 0;
627	list_del(&req->ki_list); /* remove from active_reqs */
628	req->ki_cancel = NULL;
629	req->ki_retry = NULL;
630
631	/*
632	* Try to optimize the aio and eventfd file* puts, by avoiding to
633	* schedule work in case it is not final fput() time. In normal cases,
634	* we would not be holding the last reference to the file*, so
635	* this function will be executed w/out any aio kthread wakeup.
636	*/
637	if (unlikely(!fput_atomic(req->ki_filp))) {
638	get_ioctx(ctx);
639	spin_lock(&fput_lock);
640	list_add(&req->ki_list, &fput_head);
641	spin_unlock(&fput_lock);
642	schedule_work(&fput_work);
643	} else {
644	req->ki_filp = NULL;
645	really_put_req(ctx, req);
646	}
647	return 1;
648	}
649
650	/* aio_put_req
651	* Returns true if this put was the last user of the kiocb,
652	* false if the request is still in use.
653	*/
654	int aio_put_req(struct kiocb *req)
655	{
656	struct kioctx *ctx = req->ki_ctx;
657	int ret;
658	spin_lock_irq(&ctx->ctx_lock);
659	ret = __aio_put_req(ctx, req);
660	spin_unlock_irq(&ctx->ctx_lock);
661	return ret;
662	}
663	EXPORT_SYMBOL(aio_put_req);
664
665	static struct kioctx *lookup_ioctx(unsigned long ctx_id)
666	{
667	struct mm_struct *mm = current->mm;
668	struct kioctx ctx, ret = NULL;
669	struct hlist_node *n;
670
671	rcu_read_lock();
672
673	hlist_for_each_entry_rcu(ctx, n, &mm->ioctx_list, list) {
674	/*
675	* RCU protects us against accessing freed memory but
676	* we have to be careful not to get a reference when the
677	* reference count already dropped to 0 (ctx->dead test
678	* is unreliable because of races).
679	*/
680	if (ctx->user_id == ctx_id && !ctx->dead && try_get_ioctx(ctx)){
681	ret = ctx;
682	break;
683	}
684	}
685
686	rcu_read_unlock();
687	return ret;
688	}
689
690	/*
691	* Queue up a kiocb to be retried. Assumes that the kiocb
692	* has already been marked as kicked, and places it on
693	* the retry run list for the corresponding ioctx, if it
694	* isn't already queued. Returns 1 if it actually queued
695	* the kiocb (to tell the caller to activate the work
696	* queue to process it), or 0, if it found that it was
697	* already queued.
698	*/
699	static inline int __queue_kicked_iocb(struct kiocb *iocb)
700	{
701	struct kioctx *ctx = iocb->ki_ctx;
702
703	assert_spin_locked(&ctx->ctx_lock);
704
705	if (list_empty(&iocb->ki_run_list)) {
706	list_add_tail(&iocb->ki_run_list,
707	&ctx->run_list);
708	return 1;
709	}
710	return 0;
711	}
712
713	/* aio_run_iocb
714	* This is the core aio execution routine. It is
715	* invoked both for initial i/o submission and
716	* subsequent retries via the aio_kick_handler.
717	* Expects to be invoked with iocb->ki_ctx->lock
718	* already held. The lock is released and reacquired
719	* as needed during processing.
720	*
721	* Calls the iocb retry method (already setup for the
722	* iocb on initial submission) for operation specific
723	* handling, but takes care of most of common retry
724	* execution details for a given iocb. The retry method
725	* needs to be non-blocking as far as possible, to avoid
726	* holding up other iocbs waiting to be serviced by the
727	* retry kernel thread.
728	*
729	* The trickier parts in this code have to do with
730	* ensuring that only one retry instance is in progress
731	* for a given iocb at any time. Providing that guarantee
732	* simplifies the coding of individual aio operations as
733	* it avoids various potential races.
734	*/
735	static ssize_t aio_run_iocb(struct kiocb *iocb)
736	{
737	struct kioctx *ctx = iocb->ki_ctx;
738	ssize_t (retry)(struct kiocb );
739	ssize_t ret;
740
741	if (!(retry = iocb->ki_retry)) {
742	printk("aio_run_iocb: iocb->ki_retry = NULL\n");
743	return 0;
744	}
745
746	/*
747	* We don't want the next retry iteration for this
748	* operation to start until this one has returned and
749	* updated the iocb state. However, wait_queue functions
750	* can trigger a kick_iocb from interrupt context in the
751	* meantime, indicating that data is available for the next
752	* iteration. We want to remember that and enable the
753	* next retry iteration _after_ we are through with
754	* this one.
755	*
756	* So, in order to be able to register a "kick", but
757	* prevent it from being queued now, we clear the kick
758	* flag, but make the kick code think that the iocb is
759	* still on the run list until we are actually done.
760	* When we are done with this iteration, we check if
761	* the iocb was kicked in the meantime and if so, queue
762	* it up afresh.
763	*/
764
765	kiocbClearKicked(iocb);
766
767	/*
768	* This is so that aio_complete knows it doesn't need to
769	* pull the iocb off the run list (We can't just call
770	* INIT_LIST_HEAD because we don't want a kick_iocb to
771	* queue this on the run list yet)
772	*/
773	iocb->ki_run_list.next = iocb->ki_run_list.prev = NULL;
774	spin_unlock_irq(&ctx->ctx_lock);
775
776	/* Quit retrying if the i/o has been cancelled */
777	if (kiocbIsCancelled(iocb)) {
778	ret = -EINTR;
779	aio_complete(iocb, ret, 0);
780	/* must not access the iocb after this */
781	goto out;
782	}
783
784	/*
785	* Now we are all set to call the retry method in async
786	* context.
787	*/
788	ret = retry(iocb);
789
790	if (ret != -EIOCBRETRY && ret != -EIOCBQUEUED) {
791	/*
792	* There's no easy way to restart the syscall since other AIO's
793	* may be already running. Just fail this IO with EINTR.
794	*/
795	if (unlikely(ret == -ERESTARTSYS \|\| ret == -ERESTARTNOINTR \|\|
796	ret == -ERESTARTNOHAND \|\| ret == -ERESTART_RESTARTBLOCK))
797	ret = -EINTR;
798	aio_complete(iocb, ret, 0);
799	}
800	out:
801	spin_lock_irq(&ctx->ctx_lock);
802
803	if (-EIOCBRETRY == ret) {
804	/*
805	* OK, now that we are done with this iteration
806	* and know that there is more left to go,
807	* this is where we let go so that a subsequent
808	* "kick" can start the next iteration
809	*/
810
811	/* will make __queue_kicked_iocb succeed from here on */
812	INIT_LIST_HEAD(&iocb->ki_run_list);
813	/* we must queue the next iteration ourselves, if it
814	* has already been kicked */
815	if (kiocbIsKicked(iocb)) {
816	__queue_kicked_iocb(iocb);
817
818	/*
819	* __queue_kicked_iocb will always return 1 here, because
820	* iocb->ki_run_list is empty at this point so it should
821	* be safe to unconditionally queue the context into the
822	* work queue.
823	*/
824	aio_queue_work(ctx);
825	}
826	}
827	return ret;
828	}
829
830	/*
831	* __aio_run_iocbs:
832	* Process all pending retries queued on the ioctx
833	* run list.
834	* Assumes it is operating within the aio issuer's mm
835	* context.
836	*/
837	static int __aio_run_iocbs(struct kioctx *ctx)
838	{
839	struct kiocb *iocb;
840	struct list_head run_list;
841
842	assert_spin_locked(&ctx->ctx_lock);
843
844	list_replace_init(&ctx->run_list, &run_list);
845	while (!list_empty(&run_list)) {
846	iocb = list_entry(run_list.next, struct kiocb,
847	ki_run_list);
848	list_del(&iocb->ki_run_list);
849	/*
850	* Hold an extra reference while retrying i/o.
851	*/
852	iocb->ki_users++; /* grab extra reference */
853	aio_run_iocb(iocb);
854	__aio_put_req(ctx, iocb);
855	}
856	if (!list_empty(&ctx->run_list))
857	return 1;
858	return 0;
859	}
860
861	static void aio_queue_work(struct kioctx * ctx)
862	{
863	unsigned long timeout;
864	/*
865	* if someone is waiting, get the work started right
866	* away, otherwise, use a longer delay
867	*/
868	smp_mb();
869	if (waitqueue_active(&ctx->wait))
870	timeout = 1;
871	else
872	timeout = HZ/10;
873	queue_delayed_work(aio_wq, &ctx->wq, timeout);
874	}
875
876	/*
877	* aio_run_all_iocbs:
878	* Process all pending retries queued on the ioctx
879	* run list, and keep running them until the list
880	* stays empty.
881	* Assumes it is operating within the aio issuer's mm context.
882	*/
883	static inline void aio_run_all_iocbs(struct kioctx *ctx)
884	{
885	spin_lock_irq(&ctx->ctx_lock);
886	while (__aio_run_iocbs(ctx))
887	;
888	spin_unlock_irq(&ctx->ctx_lock);
889	}
890
891	/*
892	* aio_kick_handler:
893	* Work queue handler triggered to process pending
894	* retries on an ioctx. Takes on the aio issuer's
895	* mm context before running the iocbs, so that
896	* copy_xxx_user operates on the issuer's address
897	* space.
898	* Run on aiod's context.
899	*/
900	static void aio_kick_handler(struct work_struct *work)
901	{
902	struct kioctx *ctx = container_of(work, struct kioctx, wq.work);
903	mm_segment_t oldfs = get_fs();
904	struct mm_struct *mm;
905	int requeue;
906
907	set_fs(USER_DS);
908	use_mm(ctx->mm);
909	spin_lock_irq(&ctx->ctx_lock);
910	requeue =__aio_run_iocbs(ctx);
911	mm = ctx->mm;
912	spin_unlock_irq(&ctx->ctx_lock);
913	unuse_mm(mm);
914	set_fs(oldfs);
915	/*
916	* we're in a worker thread already, don't use queue_delayed_work,
917	*/
918	if (requeue)
919	queue_delayed_work(aio_wq, &ctx->wq, 0);
920	}
921
922
923	/*
924	* Called by kick_iocb to queue the kiocb for retry
925	* and if required activate the aio work queue to process
926	* it
927	*/
928	static void try_queue_kicked_iocb(struct kiocb *iocb)
929	{
930	struct kioctx *ctx = iocb->ki_ctx;
931	unsigned long flags;
932	int run = 0;
933
934	spin_lock_irqsave(&ctx->ctx_lock, flags);
935	/* set this inside the lock so that we can't race with aio_run_iocb()
936	* testing it and putting the iocb on the run list under the lock */
937	if (!kiocbTryKick(iocb))
938	run = __queue_kicked_iocb(iocb);
939	spin_unlock_irqrestore(&ctx->ctx_lock, flags);
940	if (run)
941	aio_queue_work(ctx);
942	}
943
944	/*
945	* kick_iocb:
946	* Called typically from a wait queue callback context
947	* to trigger a retry of the iocb.
948	* The retry is usually executed by aio workqueue
949	* threads (See aio_kick_handler).
950	*/
951	void kick_iocb(struct kiocb *iocb)
952	{
953	/* sync iocbs are easy: they can only ever be executing from a
954	* single context. */
955	if (is_sync_kiocb(iocb)) {
956	kiocbSetKicked(iocb);
957	wake_up_process(iocb->ki_obj.tsk);
958	return;
959	}
960
961	try_queue_kicked_iocb(iocb);
962	}
963	EXPORT_SYMBOL(kick_iocb);
964
965	/* aio_complete
966	* Called when the io request on the given iocb is complete.
967	* Returns true if this is the last user of the request. The
968	* only other user of the request can be the cancellation code.
969	*/
970	int aio_complete(struct kiocb *iocb, long res, long res2)
971	{
972	struct kioctx *ctx = iocb->ki_ctx;
973	struct aio_ring_info *info;
974	struct aio_ring *ring;
975	struct io_event *event;
976	unsigned long flags;
977	unsigned long tail;
978	int ret;
979
980	/*
981	* Special case handling for sync iocbs:
982	* - events go directly into the iocb for fast handling
983	* - the sync task with the iocb in its stack holds the single iocb
984	* ref, no other paths have a way to get another ref
985	* - the sync task helpfully left a reference to itself in the iocb
986	*/
987	if (is_sync_kiocb(iocb)) {
988	BUG_ON(iocb->ki_users != 1);
989	iocb->ki_user_data = res;
990	iocb->ki_users = 0;
991	wake_up_process(iocb->ki_obj.tsk);
992	return 1;
993	}
994
995	info = &ctx->ring_info;
996
997	/* add a completion event to the ring buffer.
998	* must be done holding ctx->ctx_lock to prevent
999	* other code from messing with the tail
1000	* pointer since we might be called from irq
1001	* context.
1002	*/
1003	spin_lock_irqsave(&ctx->ctx_lock, flags);
1004
1005	if (iocb->ki_run_list.prev && !list_empty(&iocb->ki_run_list))
1006	list_del_init(&iocb->ki_run_list);
1007
1008	/*
1009	* cancelled requests don't get events, userland was given one
1010	* when the event got cancelled.
1011	*/
1012	if (kiocbIsCancelled(iocb))
1013	goto put_rq;
1014
1015	ring = kmap_atomic(info->ring_pages[0], KM_IRQ1);
1016
1017	tail = info->tail;
1018	event = aio_ring_event(info, tail, KM_IRQ0);
1019	if (++tail >= info->nr)
1020	tail = 0;
1021
1022	event->obj = (u64)(unsigned long)iocb->ki_obj.user;
1023	event->data = iocb->ki_user_data;
1024	event->res = res;
1025	event->res2 = res2;
1026
1027	dprintk("aio_complete: %p[%lu]: %p: %p %Lx %lx %lx\n",
1028	ctx, tail, iocb, iocb->ki_obj.user, iocb->ki_user_data,
1029	res, res2);
1030
1031	/* after flagging the request as done, we
1032	* must never even look at it again
1033	*/
1034	smp_wmb(); /* make event visible before updating tail */
1035
1036	info->tail = tail;
1037	ring->tail = tail;
1038
1039	put_aio_ring_event(event, KM_IRQ0);
1040	kunmap_atomic(ring, KM_IRQ1);
1041
1042	pr_debug("added to ring %p at [%lu]\n", iocb, tail);
1043
1044	/*
1045	* Check if the user asked us to deliver the result through an
1046	* eventfd. The eventfd_signal() function is safe to be called
1047	* from IRQ context.
1048	*/
1049	if (iocb->ki_eventfd != NULL)
1050	eventfd_signal(iocb->ki_eventfd, 1);
1051
1052	put_rq:
1053	/* everything turned out well, dispose of the aiocb. */
1054	ret = __aio_put_req(ctx, iocb);
1055
1056	/*
1057	* We have to order our ring_info tail store above and test
1058	* of the wait list below outside the wait lock. This is
1059	* like in wake_up_bit() where clearing a bit has to be
1060	* ordered with the unlocked test.
1061	*/
1062	smp_mb();
1063
1064	if (waitqueue_active(&ctx->wait))
1065	wake_up(&ctx->wait);
1066
1067	spin_unlock_irqrestore(&ctx->ctx_lock, flags);
1068	return ret;
1069	}
1070	EXPORT_SYMBOL(aio_complete);
1071
1072	/* aio_read_evt
1073	* Pull an event off of the ioctx's event ring. Returns the number of
1074	* events fetched (0 or 1 ;-)
1075	* FIXME: make this use cmpxchg.
1076	* TODO: make the ringbuffer user mmap()able (requires FIXME).
1077	*/
1078	static int aio_read_evt(struct kioctx ioctx, struct io_event ent)
1079	{
1080	struct aio_ring_info *info = &ioctx->ring_info;
1081	struct aio_ring *ring;
1082	unsigned long head;
1083	int ret = 0;
1084
1085	ring = kmap_atomic(info->ring_pages[0], KM_USER0);
1086	dprintk("in aio_read_evt h%lu t%lu m%lu\n",
1087	(unsigned long)ring->head, (unsigned long)ring->tail,
1088	(unsigned long)ring->nr);
1089
1090	if (ring->head == ring->tail)
1091	goto out;
1092
1093	spin_lock(&info->ring_lock);
1094
1095	head = ring->head % info->nr;
1096	if (head != ring->tail) {
1097	struct io_event *evp = aio_ring_event(info, head, KM_USER1);
1098	ent = evp;
1099	head = (head + 1) % info->nr;
1100	smp_mb(); /* finish reading the event before updatng the head */
1101	ring->head = head;
1102	ret = 1;
1103	put_aio_ring_event(evp, KM_USER1);
1104	}
1105	spin_unlock(&info->ring_lock);
1106
1107	out:
1108	kunmap_atomic(ring, KM_USER0);
1109	dprintk("leaving aio_read_evt: %d h%lu t%lu\n", ret,
1110	(unsigned long)ring->head, (unsigned long)ring->tail);
1111	return ret;
1112	}
1113
1114	struct aio_timeout {
1115	struct timer_list timer;
1116	int timed_out;
1117	struct task_struct *p;
1118	};
1119
1120	static void timeout_func(unsigned long data)
1121	{
1122	struct aio_timeout to = (struct aio_timeout )data;
1123
1124	to->timed_out = 1;
1125	wake_up_process(to->p);
1126	}
1127
1128	static inline void init_timeout(struct aio_timeout *to)
1129	{
1130	setup_timer_on_stack(&to->timer, timeout_func, (unsigned long) to);
1131	to->timed_out = 0;
1132	to->p = current;
1133	}
1134
1135	static inline void set_timeout(long start_jiffies, struct aio_timeout *to,
1136	const struct timespec *ts)
1137	{
1138	to->timer.expires = start_jiffies + timespec_to_jiffies(ts);
1139	if (time_after(to->timer.expires, jiffies))
1140	add_timer(&to->timer);
1141	else
1142	to->timed_out = 1;
1143	}
1144
1145	static inline void clear_timeout(struct aio_timeout *to)
1146	{
1147	del_singleshot_timer_sync(&to->timer);
1148	}
1149
1150	static int read_events(struct kioctx *ctx,
1151	long min_nr, long nr,
1152	struct io_event __user *event,
1153	struct timespec __user *timeout)
1154	{
1155	long start_jiffies = jiffies;
1156	struct task_struct *tsk = current;
1157	DECLARE_WAITQUEUE(wait, tsk);
1158	int ret;
1159	int i = 0;
1160	struct io_event ent;
1161	struct aio_timeout to;
1162	int retry = 0;
1163
1164	/* needed to zero any padding within an entry (there shouldn't be
1165	* any, but C is fun!
1166	*/
1167	memset(&ent, 0, sizeof(ent));
1168	retry:
1169	ret = 0;
1170	while (likely(i < nr)) {
1171	ret = aio_read_evt(ctx, &ent);
1172	if (unlikely(ret <= 0))
1173	break;
1174
1175	dprintk("read event: %Lx %Lx %Lx %Lx\n",
1176	ent.data, ent.obj, ent.res, ent.res2);
1177
1178	/* Could we split the check in two? */
1179	ret = -EFAULT;
1180	if (unlikely(copy_to_user(event, &ent, sizeof(ent)))) {
1181	dprintk("aio: lost an event due to EFAULT.\n");
1182	break;
1183	}
1184	ret = 0;
1185
1186	/* Good, event copied to userland, update counts. */
1187	event ++;
1188	i ++;
1189	}
1190
1191	if (min_nr <= i)
1192	return i;
1193	if (ret)
1194	return ret;
1195
1196	/* End fast path */
1197
1198	/* racey check, but it gets redone */
1199	if (!retry && unlikely(!list_empty(&ctx->run_list))) {
1200	retry = 1;
1201	aio_run_all_iocbs(ctx);
1202	goto retry;
1203	}
1204
1205	init_timeout(&to);
1206	if (timeout) {
1207	struct timespec ts;
1208	ret = -EFAULT;
1209	if (unlikely(copy_from_user(&ts, timeout, sizeof(ts))))
1210	goto out;
1211
1212	set_timeout(start_jiffies, &to, &ts);
1213	}
1214
1215	while (likely(i < nr)) {
1216	add_wait_queue_exclusive(&ctx->wait, &wait);
1217	do {
1218	set_task_state(tsk, TASK_INTERRUPTIBLE);
1219	ret = aio_read_evt(ctx, &ent);
1220	if (ret)
1221	break;
1222	if (min_nr <= i)
1223	break;
1224	if (unlikely(ctx->dead)) {
1225	ret = -EINVAL;
1226	break;
1227	}
1228	if (to.timed_out) /* Only check after read evt */
1229	break;
1230	/* Try to only show up in io wait if there are ops
1231	* in flight */
1232	if (ctx->reqs_active)
1233	io_schedule();
1234	else
1235	schedule();
1236	if (signal_pending(tsk)) {
1237	ret = -EINTR;
1238	break;
1239	}
1240	/ret = aio_read_evt(ctx, &ent);/
1241	} while (1) ;
1242
1243	set_task_state(tsk, TASK_RUNNING);
1244	remove_wait_queue(&ctx->wait, &wait);
1245
1246	if (unlikely(ret <= 0))
1247	break;
1248
1249	ret = -EFAULT;
1250	if (unlikely(copy_to_user(event, &ent, sizeof(ent)))) {
1251	dprintk("aio: lost an event due to EFAULT.\n");
1252	break;
1253	}
1254
1255	/* Good, event copied to userland, update counts. */
1256	event ++;
1257	i ++;
1258	}
1259
1260	if (timeout)
1261	clear_timeout(&to);
1262	out:
1263	destroy_timer_on_stack(&to.timer);
1264	return i ? i : ret;
1265	}
1266
1267	/* Take an ioctx and remove it from the list of ioctx's. Protects
1268	* against races with itself via ->dead.
1269	*/
1270	static void io_destroy(struct kioctx *ioctx)
1271	{
1272	struct mm_struct *mm = current->mm;
1273	int was_dead;
1274
1275	/* delete the entry from the list is someone else hasn't already */
1276	spin_lock(&mm->ioctx_lock);
1277	was_dead = ioctx->dead;
1278	ioctx->dead = 1;
1279	hlist_del_rcu(&ioctx->list);
1280	spin_unlock(&mm->ioctx_lock);
1281
1282	dprintk("aio_release(%p)\n", ioctx);
1283	if (likely(!was_dead))
1284	put_ioctx(ioctx); /* twice for the list */
1285
1286	aio_cancel_all(ioctx);
1287	wait_for_all_aios(ioctx);
1288
1289	/*
1290	* Wake up any waiters. The setting of ctx->dead must be seen
1291	* by other CPUs at this point. Right now, we rely on the
1292	* locking done by the above calls to ensure this consistency.
1293	*/
1294	wake_up_all(&ioctx->wait);
1295	put_ioctx(ioctx); /* once for the lookup */
1296	}
1297
1298	/* sys_io_setup:
1299	* Create an aio_context capable of receiving at least nr_events.
1300	* ctxp must not point to an aio_context that already exists, and
1301	* must be initialized to 0 prior to the call. On successful
1302	* creation of the aio_context, *ctxp is filled in with the resulting
1303	* handle. May fail with -EINVAL if *ctxp is not initialized,
1304	* if the specified nr_events exceeds internal limits. May fail
1305	* with -EAGAIN if the specified nr_events exceeds the user's limit
1306	* of available events. May fail with -ENOMEM if insufficient kernel
1307	* resources are available. May fail with -EFAULT if an invalid
1308	* pointer is passed for ctxp. Will fail with -ENOSYS if not
1309	* implemented.
1310	*/
1311	SYSCALL_DEFINE2(io_setup, unsigned, nr_events, aio_context_t __user *, ctxp)
1312	{
1313	struct kioctx *ioctx = NULL;
1314	unsigned long ctx;
1315	long ret;
1316
1317	ret = get_user(ctx, ctxp);
1318	if (unlikely(ret))
1319	goto out;
1320
1321	ret = -EINVAL;
1322	if (unlikely(ctx \|\| nr_events == 0)) {
1323	pr_debug("EINVAL: io_setup: ctx %lu nr_events %u\n",
1324	ctx, nr_events);
1325	goto out;
1326	}
1327
1328	ioctx = ioctx_alloc(nr_events);
1329	ret = PTR_ERR(ioctx);
1330	if (!IS_ERR(ioctx)) {
1331	ret = put_user(ioctx->user_id, ctxp);
1332	if (!ret)
1333	return 0;
1334
1335	get_ioctx(ioctx); /* io_destroy() expects us to hold a ref */
1336	io_destroy(ioctx);
1337	}
1338
1339	out:
1340	return ret;
1341	}
1342
1343	/* sys_io_destroy:
1344	* Destroy the aio_context specified. May cancel any outstanding
1345	* AIOs and block on completion. Will fail with -ENOSYS if not
1346	* implemented. May fail with -EINVAL if the context pointed to
1347	* is invalid.
1348	*/
1349	SYSCALL_DEFINE1(io_destroy, aio_context_t, ctx)
1350	{
1351	struct kioctx *ioctx = lookup_ioctx(ctx);
1352	if (likely(NULL != ioctx)) {
1353	io_destroy(ioctx);
1354	return 0;
1355	}
1356	pr_debug("EINVAL: io_destroy: invalid context id\n");
1357	return -EINVAL;
1358	}
1359
1360	static void aio_advance_iovec(struct kiocb *iocb, ssize_t ret)
1361	{
1362	struct iovec *iov = &iocb->ki_iovec[iocb->ki_cur_seg];
1363
1364	BUG_ON(ret <= 0);
1365
1366	while (iocb->ki_cur_seg < iocb->ki_nr_segs && ret > 0) {
1367	ssize_t this = min((ssize_t)iov->iov_len, ret);
1368	iov->iov_base += this;
1369	iov->iov_len -= this;
1370	iocb->ki_left -= this;
1371	ret -= this;
1372	if (iov->iov_len == 0) {
1373	iocb->ki_cur_seg++;
1374	iov++;
1375	}
1376	}
1377
1378	/* the caller should not have done more io than what fit in
1379	* the remaining iovecs */
1380	BUG_ON(ret > 0 && iocb->ki_left == 0);
1381	}
1382
1383	static ssize_t aio_rw_vect_retry(struct kiocb *iocb)
1384	{
1385	struct file *file = iocb->ki_filp;
1386	struct address_space *mapping = file->f_mapping;
1387	struct inode *inode = mapping->host;
1388	ssize_t (rw_op)(struct kiocb , const struct iovec *,
1389	unsigned long, loff_t);
1390	ssize_t ret = 0;
1391	unsigned short opcode;
1392
1393	if ((iocb->ki_opcode == IOCB_CMD_PREADV) \|\|
1394	(iocb->ki_opcode == IOCB_CMD_PREAD)) {
1395	rw_op = file->f_op->aio_read;
1396	opcode = IOCB_CMD_PREADV;
1397	} else {
1398	rw_op = file->f_op->aio_write;
1399	opcode = IOCB_CMD_PWRITEV;
1400	}
1401
1402	/* This matches the pread()/pwrite() logic */
1403	if (iocb->ki_pos < 0)
1404	return -EINVAL;
1405
1406	do {
1407	ret = rw_op(iocb, &iocb->ki_iovec[iocb->ki_cur_seg],
1408	iocb->ki_nr_segs - iocb->ki_cur_seg,
1409	iocb->ki_pos);
1410	if (ret > 0)
1411	aio_advance_iovec(iocb, ret);
1412
1413	/* retry all partial writes. retry partial reads as long as its a
1414	* regular file. */
1415	} while (ret > 0 && iocb->ki_left > 0 &&
1416	(opcode == IOCB_CMD_PWRITEV \|\|
1417	(!S_ISFIFO(inode->i_mode) && !S_ISSOCK(inode->i_mode))));
1418
1419	/* This means we must have transferred all that we could */
1420	/* No need to retry anymore */
1421	if ((ret == 0) \|\| (iocb->ki_left == 0))
1422	ret = iocb->ki_nbytes - iocb->ki_left;
1423
1424	/* If we managed to write some out we return that, rather than
1425	* the eventual error. */
1426	if (opcode == IOCB_CMD_PWRITEV
1427	&& ret < 0 && ret != -EIOCBQUEUED && ret != -EIOCBRETRY
1428	&& iocb->ki_nbytes - iocb->ki_left)
1429	ret = iocb->ki_nbytes - iocb->ki_left;
1430
1431	return ret;
1432	}
1433
1434	static ssize_t aio_fdsync(struct kiocb *iocb)
1435	{
1436	struct file *file = iocb->ki_filp;
1437	ssize_t ret = -EINVAL;
1438
1439	if (file->f_op->aio_fsync)
1440	ret = file->f_op->aio_fsync(iocb, 1);
1441	return ret;
1442	}
1443
1444	static ssize_t aio_fsync(struct kiocb *iocb)
1445	{
1446	struct file *file = iocb->ki_filp;
1447	ssize_t ret = -EINVAL;
1448
1449	if (file->f_op->aio_fsync)
1450	ret = file->f_op->aio_fsync(iocb, 0);
1451	return ret;
1452	}
1453
1454	static ssize_t aio_setup_vectored_rw(int type, struct kiocb *kiocb, bool compat)
1455	{
1456	ssize_t ret;
1457
1458	#ifdef CONFIG_COMPAT
1459	if (compat)
1460	ret = compat_rw_copy_check_uvector(type,
1461	(struct compat_iovec __user *)kiocb->ki_buf,
1462	kiocb->ki_nbytes, 1, &kiocb->ki_inline_vec,
1463	&kiocb->ki_iovec, 1);
1464	else
1465	#endif
1466	ret = rw_copy_check_uvector(type,
1467	(struct iovec __user *)kiocb->ki_buf,
1468	kiocb->ki_nbytes, 1, &kiocb->ki_inline_vec,
1469	&kiocb->ki_iovec, 1);
1470	if (ret < 0)
1471	goto out;
1472
1473	kiocb->ki_nr_segs = kiocb->ki_nbytes;
1474	kiocb->ki_cur_seg = 0;
1475	/* ki_nbytes/left now reflect bytes instead of segs */
1476	kiocb->ki_nbytes = ret;
1477	kiocb->ki_left = ret;
1478
1479	ret = 0;
1480	out:
1481	return ret;
1482	}
1483
1484	static ssize_t aio_setup_single_vector(struct kiocb *kiocb)
1485	{
1486	kiocb->ki_iovec = &kiocb->ki_inline_vec;
1487	kiocb->ki_iovec->iov_base = kiocb->ki_buf;
1488	kiocb->ki_iovec->iov_len = kiocb->ki_left;
1489	kiocb->ki_nr_segs = 1;
1490	kiocb->ki_cur_seg = 0;
1491	return 0;
1492	}
1493
1494	/*
1495	* aio_setup_iocb:
1496	* Performs the initial checks and aio retry method
1497	* setup for the kiocb at the time of io submission.
1498	*/
1499	static ssize_t aio_setup_iocb(struct kiocb *kiocb, bool compat)
1500	{
1501	struct file *file = kiocb->ki_filp;
1502	ssize_t ret = 0;
1503
1504	switch (kiocb->ki_opcode) {
1505	case IOCB_CMD_PREAD:
1506	ret = -EBADF;
1507	if (unlikely(!(file->f_mode & FMODE_READ)))
1508	break;
1509	ret = -EFAULT;
1510	if (unlikely(!access_ok(VERIFY_WRITE, kiocb->ki_buf,
1511	kiocb->ki_left)))
1512	break;
1513	ret = security_file_permission(file, MAY_READ);
1514	if (unlikely(ret))
1515	break;
1516	ret = aio_setup_single_vector(kiocb);
1517	if (ret)
1518	break;
1519	ret = -EINVAL;
1520	if (file->f_op->aio_read)
1521	kiocb->ki_retry = aio_rw_vect_retry;
1522	break;
1523	case IOCB_CMD_PWRITE:
1524	ret = -EBADF;
1525	if (unlikely(!(file->f_mode & FMODE_WRITE)))
1526	break;
1527	ret = -EFAULT;
1528	if (unlikely(!access_ok(VERIFY_READ, kiocb->ki_buf,
1529	kiocb->ki_left)))
1530	break;
1531	ret = security_file_permission(file, MAY_WRITE);
1532	if (unlikely(ret))
1533	break;
1534	ret = aio_setup_single_vector(kiocb);
1535	if (ret)
1536	break;
1537	ret = -EINVAL;
1538	if (file->f_op->aio_write)
1539	kiocb->ki_retry = aio_rw_vect_retry;
1540	break;
1541	case IOCB_CMD_PREADV:
1542	ret = -EBADF;
1543	if (unlikely(!(file->f_mode & FMODE_READ)))
1544	break;
1545	ret = security_file_permission(file, MAY_READ);
1546	if (unlikely(ret))
1547	break;
1548	ret = aio_setup_vectored_rw(READ, kiocb, compat);
1549	if (ret)
1550	break;
1551	ret = -EINVAL;
1552	if (file->f_op->aio_read)
1553	kiocb->ki_retry = aio_rw_vect_retry;
1554	break;
1555	case IOCB_CMD_PWRITEV:
1556	ret = -EBADF;
1557	if (unlikely(!(file->f_mode & FMODE_WRITE)))
1558	break;
1559	ret = security_file_permission(file, MAY_WRITE);
1560	if (unlikely(ret))
1561	break;
1562	ret = aio_setup_vectored_rw(WRITE, kiocb, compat);
1563	if (ret)
1564	break;
1565	ret = -EINVAL;
1566	if (file->f_op->aio_write)
1567	kiocb->ki_retry = aio_rw_vect_retry;
1568	break;
1569	case IOCB_CMD_FDSYNC:
1570	ret = -EINVAL;
1571	if (file->f_op->aio_fsync)
1572	kiocb->ki_retry = aio_fdsync;
1573	break;
1574	case IOCB_CMD_FSYNC:
1575	ret = -EINVAL;
1576	if (file->f_op->aio_fsync)
1577	kiocb->ki_retry = aio_fsync;
1578	break;
1579	default:
1580	dprintk("EINVAL: io_submit: no operation provided\n");
1581	ret = -EINVAL;
1582	}
1583
1584	if (!kiocb->ki_retry)
1585	return ret;
1586
1587	return 0;
1588	}
1589
1590	static int io_submit_one(struct kioctx ctx, struct iocb __user user_iocb,
1591	struct iocb iocb, struct kiocb_batch batch,
1592	bool compat)
1593	{
1594	struct kiocb *req;
1595	struct file *file;
1596	ssize_t ret;
1597
1598	/* enforce forwards compatibility on users */
1599	if (unlikely(iocb->aio_reserved1 \|\| iocb->aio_reserved2)) {
1600	pr_debug("EINVAL: io_submit: reserve field set\n");
1601	return -EINVAL;
1602	}
1603
1604	/* prevent overflows */
1605	if (unlikely(
1606	(iocb->aio_buf != (unsigned long)iocb->aio_buf) \|\|
1607	(iocb->aio_nbytes != (size_t)iocb->aio_nbytes) \|\|
1608	((ssize_t)iocb->aio_nbytes < 0)
1609	)) {
1610	pr_debug("EINVAL: io_submit: overflow check\n");
1611	return -EINVAL;
1612	}
1613
1614	file = fget(iocb->aio_fildes);
1615	if (unlikely(!file))
1616	return -EBADF;
1617
1618	req = aio_get_req(ctx, batch); /* returns with 2 references to req */
1619	if (unlikely(!req)) {
1620	fput(file);
1621	return -EAGAIN;
1622	}
1623	req->ki_filp = file;
1624	if (iocb->aio_flags & IOCB_FLAG_RESFD) {
1625	/*
1626	* If the IOCB_FLAG_RESFD flag of aio_flags is set, get an
1627	* instance of the file* now. The file descriptor must be
1628	* an eventfd() fd, and will be signaled for each completed
1629	* event using the eventfd_signal() function.
1630	*/
1631	req->ki_eventfd = eventfd_ctx_fdget((int) iocb->aio_resfd);
1632	if (IS_ERR(req->ki_eventfd)) {
1633	ret = PTR_ERR(req->ki_eventfd);
1634	req->ki_eventfd = NULL;
1635	goto out_put_req;
1636	}
1637	}
1638
1639	ret = put_user(req->ki_key, &user_iocb->aio_key);
1640	if (unlikely(ret)) {
1641	dprintk("EFAULT: aio_key\n");
1642	goto out_put_req;
1643	}
1644
1645	req->ki_obj.user = user_iocb;
1646	req->ki_user_data = iocb->aio_data;
1647	req->ki_pos = iocb->aio_offset;
1648
1649	req->ki_buf = (char __user *)(unsigned long)iocb->aio_buf;
1650	req->ki_left = req->ki_nbytes = iocb->aio_nbytes;
1651	req->ki_opcode = iocb->aio_lio_opcode;
1652
1653	ret = aio_setup_iocb(req, compat);
1654
1655	if (ret)
1656	goto out_put_req;
1657
1658	spin_lock_irq(&ctx->ctx_lock);
1659	/*
1660	* We could have raced with io_destroy() and are currently holding a
1661	* reference to ctx which should be destroyed. We cannot submit IO
1662	* since ctx gets freed as soon as io_submit() puts its reference. The
1663	* check here is reliable: io_destroy() sets ctx->dead before waiting
1664	* for outstanding IO and the barrier between these two is realized by
1665	* unlock of mm->ioctx_lock and lock of ctx->ctx_lock. Analogously we
1666	* increment ctx->reqs_active before checking for ctx->dead and the
1667	* barrier is realized by unlock and lock of ctx->ctx_lock. Thus if we
1668	* don't see ctx->dead set here, io_destroy() waits for our IO to
1669	* finish.
1670	*/
1671	if (ctx->dead) {
1672	spin_unlock_irq(&ctx->ctx_lock);
1673	ret = -EINVAL;
1674	goto out_put_req;
1675	}
1676	aio_run_iocb(req);
1677	if (!list_empty(&ctx->run_list)) {
1678	/* drain the run list */
1679	while (__aio_run_iocbs(ctx))
1680	;
1681	}
1682	spin_unlock_irq(&ctx->ctx_lock);
1683
1684	aio_put_req(req); /* drop extra ref to req */
1685	return 0;
1686
1687	out_put_req:
1688	aio_put_req(req); /* drop extra ref to req */
1689	aio_put_req(req); /* drop i/o ref to req */
1690	return ret;
1691	}
1692
1693	long do_io_submit(aio_context_t ctx_id, long nr,
1694	struct iocb __user __user iocbpp, bool compat)
1695	{
1696	struct kioctx *ctx;
1697	long ret = 0;
1698	int i = 0;
1699	struct blk_plug plug;
1700	struct kiocb_batch batch;
1701
1702	if (unlikely(nr < 0))
1703	return -EINVAL;
1704
1705	if (unlikely(nr > LONG_MAX/sizeof(*iocbpp)))
1706	nr = LONG_MAX/sizeof(*iocbpp);
1707
1708	if (unlikely(!access_ok(VERIFY_READ, iocbpp, (nrsizeof(iocbpp)))))
1709	return -EFAULT;
1710
1711	ctx = lookup_ioctx(ctx_id);
1712	if (unlikely(!ctx)) {
1713	pr_debug("EINVAL: io_submit: invalid context id\n");
1714	return -EINVAL;
1715	}
1716
1717	kiocb_batch_init(&batch, nr);
1718
1719	blk_start_plug(&plug);
1720
1721	/*
1722	* AKPM: should this return a partial result if some of the IOs were
1723	* successfully submitted?
1724	*/
1725	for (i=0; i<nr; i++) {
1726	struct iocb __user *user_iocb;
1727	struct iocb tmp;
1728
1729	if (unlikely(__get_user(user_iocb, iocbpp + i))) {
1730	ret = -EFAULT;
1731	break;
1732	}
1733
1734	if (unlikely(copy_from_user(&tmp, user_iocb, sizeof(tmp)))) {
1735	ret = -EFAULT;
1736	break;
1737	}
1738
1739	ret = io_submit_one(ctx, user_iocb, &tmp, &batch, compat);
1740	if (ret)
1741	break;
1742	}
1743	blk_finish_plug(&plug);
1744
1745	kiocb_batch_free(&batch);
1746	put_ioctx(ctx);
1747	return i ? i : ret;
1748	}
1749
1750	/* sys_io_submit:
1751	* Queue the nr iocbs pointed to by iocbpp for processing. Returns
1752	* the number of iocbs queued. May return -EINVAL if the aio_context
1753	* specified by ctx_id is invalid, if nr is < 0, if the iocb at
1754	* *iocbpp[0] is not properly initialized, if the operation specified
1755	* is invalid for the file descriptor in the iocb. May fail with
1756	* -EFAULT if any of the data structures point to invalid data. May
1757	* fail with -EBADF if the file descriptor specified in the first
1758	* iocb is invalid. May fail with -EAGAIN if insufficient resources
1759	* are available to queue any iocbs. Will return 0 if nr is 0. Will
1760	* fail with -ENOSYS if not implemented.
1761	*/
1762	SYSCALL_DEFINE3(io_submit, aio_context_t, ctx_id, long, nr,
1763	struct iocb __user * __user *, iocbpp)
1764	{
1765	return do_io_submit(ctx_id, nr, iocbpp, 0);
1766	}
1767
1768	/* lookup_kiocb
1769	* Finds a given iocb for cancellation.
1770	*/
1771	static struct kiocb lookup_kiocb(struct kioctx ctx, struct iocb __user *iocb,
1772	u32 key)
1773	{
1774	struct list_head *pos;
1775
1776	assert_spin_locked(&ctx->ctx_lock);
1777
1778	/* TODO: use a hash or array, this sucks. */
1779	list_for_each(pos, &ctx->active_reqs) {
1780	struct kiocb *kiocb = list_kiocb(pos);
1781	if (kiocb->ki_obj.user == iocb && kiocb->ki_key == key)
1782	return kiocb;
1783	}
1784	return NULL;
1785	}
1786
1787	/* sys_io_cancel:
1788	* Attempts to cancel an iocb previously passed to io_submit. If
1789	* the operation is successfully cancelled, the resulting event is
1790	* copied into the memory pointed to by result without being placed
1791	* into the completion queue and 0 is returned. May fail with
1792	* -EFAULT if any of the data structures pointed to are invalid.
1793	* May fail with -EINVAL if aio_context specified by ctx_id is
1794	* invalid. May fail with -EAGAIN if the iocb specified was not
1795	* cancelled. Will fail with -ENOSYS if not implemented.
1796	*/
1797	SYSCALL_DEFINE3(io_cancel, aio_context_t, ctx_id, struct iocb __user *, iocb,
1798	struct io_event __user *, result)
1799	{
1800	int (cancel)(struct kiocb iocb, struct io_event *res);
1801	struct kioctx *ctx;
1802	struct kiocb *kiocb;
1803	u32 key;
1804	int ret;
1805
1806	ret = get_user(key, &iocb->aio_key);
1807	if (unlikely(ret))
1808	return -EFAULT;
1809
1810	ctx = lookup_ioctx(ctx_id);
1811	if (unlikely(!ctx))
1812	return -EINVAL;
1813
1814	spin_lock_irq(&ctx->ctx_lock);
1815	ret = -EAGAIN;
1816	kiocb = lookup_kiocb(ctx, iocb, key);
1817	if (kiocb && kiocb->ki_cancel) {
1818	cancel = kiocb->ki_cancel;
1819	kiocb->ki_users ++;
1820	kiocbSetCancelled(kiocb);
1821	} else
1822	cancel = NULL;
1823	spin_unlock_irq(&ctx->ctx_lock);
1824
1825	if (NULL != cancel) {
1826	struct io_event tmp;
1827	pr_debug("calling cancel\n");
1828	memset(&tmp, 0, sizeof(tmp));
1829	tmp.obj = (u64)(unsigned long)kiocb->ki_obj.user;
1830	tmp.data = kiocb->ki_user_data;
1831	ret = cancel(kiocb, &tmp);
1832	if (!ret) {
1833	/* Cancellation succeeded -- copy the result
1834	* into the user's buffer.
1835	*/
1836	if (copy_to_user(result, &tmp, sizeof(tmp)))
1837	ret = -EFAULT;
1838	}
1839	} else
1840	ret = -EINVAL;
1841
1842	put_ioctx(ctx);
1843
1844	return ret;
1845	}
1846
1847	/* io_getevents:
1848	* Attempts to read at least min_nr events and up to nr events from
1849	* the completion queue for the aio_context specified by ctx_id. If
1850	* it succeeds, the number of read events is returned. May fail with
1851	* -EINVAL if ctx_id is invalid, if min_nr is out of range, if nr is
1852	* out of range, if timeout is out of range. May fail with -EFAULT
1853	* if any of the memory specified is invalid. May return 0 or
1854	* < min_nr if the timeout specified by timeout has elapsed
1855	* before sufficient events are available, where timeout == NULL
1856	* specifies an infinite timeout. Note that the timeout pointed to by
1857	* timeout is relative and will be updated if not NULL and the
1858	* operation blocks. Will fail with -ENOSYS if not implemented.
1859	*/
1860	SYSCALL_DEFINE5(io_getevents, aio_context_t, ctx_id,
1861	long, min_nr,
1862	long, nr,
1863	struct io_event __user *, events,
1864	struct timespec __user *, timeout)
1865	{
1866	struct kioctx *ioctx = lookup_ioctx(ctx_id);
1867	long ret = -EINVAL;
1868
1869	if (likely(ioctx)) {
1870	if (likely(min_nr <= nr && min_nr >= 0))
1871	ret = read_events(ioctx, min_nr, nr, events, timeout);
1872	put_ioctx(ioctx);
1873	}
1874
1875	asmlinkage_protect(5, ret, ctx_id, min_nr, nr, events, timeout);
1876	return ret;
1877	}
1878

Download this file

Branches:
ben-wpan
ben-wpan-stefan
javiroman/ks7010
jz-2.6.34
jz-2.6.34-rc5
jz-2.6.34-rc6
jz-2.6.34-rc7
jz-2.6.35
jz-2.6.36
jz-2.6.37
jz-2.6.38
jz-2.6.39
jz-3.0
jz-3.1
jz-3.2
jz-3.3
jz-3.4
jz-3.5
jz-3.6
jz-3.6-rc2-pwm
jz-3.9
jz-3.9-rc8
jz47xx
jz47xx-2.6.38
master

Tags:
od-2011-09-04
od-2011-09-18
v2.6.34-rc5
v2.6.34-rc6
v2.6.34-rc7
v3.9

Kernel

Kernel Git Source Tree

Root/fs/aio.c

interactive