/* *	An async IO implementation for Linux *	Written by Benjamin LaHaise <bcrl@kvack.org> * *	Implements an efficient asynchronous io interface. * *	Copyright 2000, 2001, 2002 Red Hat, Inc.  All Rights Reserved. * *	See ../COPYING for licensing terms. */#include <linux/kernel.h>#include <linux/init.h>#include <linux/errno.h>#include <linux/time.h>#include <linux/aio_abi.h>#include <linux/export.h>#include <linux/syscalls.h>#include <linux/backing-dev.h>#include <linux/uio.h> #define DEBUG 0 #include <linux/sched.h>#include <linux/fs.h>#include <linux/file.h>#include <linux/mm.h>#include <linux/mman.h>#include <linux/mmu_context.h>#include <linux/slab.h>#include <linux/timer.h>#include <linux/aio.h>#include <linux/highmem.h>#include <linux/workqueue.h>#include <linux/security.h>#include <linux/eventfd.h>#include <linux/blkdev.h>#include <linux/compat.h> #include <asm/kmap_types.h>#include <asm/uaccess.h> #if DEBUG > 1#define dprintk		printk#else#define dprintk(x...)	do { ; } while (0)#endif /*------ sysctl variables----*/static DEFINE_SPINLOCK(aio_nr_lock);unsigned long aio_nr;		/* current system wide number of aio requests */unsigned long aio_max_nr = 0x10000; /* system wide maximum number of aio requests *//*----end sysctl variables---*/ static struct kmem_cache	*kiocb_cachep;static struct kmem_cache	*kioctx_cachep; static struct workqueue_struct *aio_wq; /* Used for rare fput completion. */static void aio_fput_routine(struct work_struct *);static DECLARE_WORK(fput_work, aio_fput_routine); static DEFINE_SPINLOCK(fput_lock);static LIST_HEAD(fput_head); static void aio_kick_handler(struct work_struct *);static void aio_queue_work(struct kioctx *); /* aio_setup *	Creates the slab caches used by the aio routines, panic on *	failure as this is done early during the boot sequence. */static int __init aio_setup(void){	kiocb_cachep = KMEM_CACHE(kiocb, SLAB_HWCACHE_ALIGN|SLAB_PANIC);	kioctx_cachep = KMEM_CACHE(kioctx,SLAB_HWCACHE_ALIGN|SLAB_PANIC); 	aio_wq = alloc_workqueue("aio", 0, 1);	/* used to limit concurrency */	BUG_ON(!aio_wq); 	pr_debug("aio_setup: sizeof(struct page) = %d\n", (int)sizeof(struct page)); 	return 0;}__initcall(aio_setup); static void aio_free_ring(struct kioctx *ctx){	struct aio_ring_info *info = &ctx->ring_info;	long i; 	for (i=0; i<info->nr_pages; i++)		put_page(info->ring_pages[i]); 	if (info->mmap_size) {		BUG_ON(ctx->mm != current->mm);		vm_munmap(info->mmap_base, info->mmap_size);	} 	if (info->ring_pages && info->ring_pages != info->internal_pages)		kfree(info->ring_pages);	info->ring_pages = NULL;	info->nr = 0;} static int aio_setup_ring(struct kioctx *ctx){	struct aio_ring *ring;	struct aio_ring_info *info = &ctx->ring_info;	unsigned nr_events = ctx->max_reqs;	unsigned long size;	int nr_pages; 	/* Compensate for the ring buffer's head/tail overlap entry */	nr_events += 2;	/* 1 is required, 2 for good luck */ 	size = sizeof(struct aio_ring);	size += sizeof(struct io_event) * nr_events;	nr_pages = (size + PAGE_SIZE-1) >> PAGE_SHIFT; 	if (nr_pages < 0)		return -EINVAL; 	nr_events = (PAGE_SIZE * nr_pages - sizeof(struct aio_ring)) / sizeof(struct io_event); 	info->nr = 0;	info->ring_pages = info->internal_pages;	if (nr_pages > AIO_RING_PAGES) {		info->ring_pages = kcalloc(nr_pages, sizeof(struct page *), GFP_KERNEL);		if (!info->ring_pages)			return -ENOMEM;	} 	info->mmap_size = nr_pages * PAGE_SIZE;	dprintk("attempting mmap of %lu bytes\n", info->mmap_size);	down_write(&ctx->mm->mmap_sem);	info->mmap_base = do_mmap_pgoff(NULL, 0, info->mmap_size, 					PROT_READ|PROT_WRITE,					MAP_ANONYMOUS|MAP_PRIVATE, 0);	if (IS_ERR((void *)info->mmap_base)) {		up_write(&ctx->mm->mmap_sem);		info->mmap_size = 0;		aio_free_ring(ctx);		return -EAGAIN;	} 	dprintk("mmap address: 0x%08lx\n", info->mmap_base);	info->nr_pages = get_user_pages(current, ctx->mm,					info->mmap_base, nr_pages, 					1, 0, info->ring_pages, NULL);	up_write(&ctx->mm->mmap_sem); 	if (unlikely(info->nr_pages != nr_pages)) {		aio_free_ring(ctx);		return -EAGAIN;	} 	ctx->user_id = info->mmap_base; 	info->nr = nr_events;		/* trusted copy */ 	ring = kmap_atomic(info->ring_pages[0]);	ring->nr = nr_events;	/* user copy */	ring->id = ctx->user_id;	ring->head = ring->tail = 0;	ring->magic = AIO_RING_MAGIC;	ring->compat_features = AIO_RING_COMPAT_FEATURES;	ring->incompat_features = AIO_RING_INCOMPAT_FEATURES;	ring->header_length = sizeof(struct aio_ring);	kunmap_atomic(ring); 	return 0;}  /* aio_ring_event: returns a pointer to the event at the given index from * kmap_atomic().  Release the pointer with put_aio_ring_event(); */#define AIO_EVENTS_PER_PAGE	(PAGE_SIZE / sizeof(struct io_event))#define AIO_EVENTS_FIRST_PAGE	((PAGE_SIZE - sizeof(struct aio_ring)) / sizeof(struct io_event))#define AIO_EVENTS_OFFSET	(AIO_EVENTS_PER_PAGE - AIO_EVENTS_FIRST_PAGE) #define aio_ring_event(info, nr) ({					\	unsigned pos = (nr) + AIO_EVENTS_OFFSET;			\	struct io_event *__event;					\	__event = kmap_atomic(						\			(info)->ring_pages[pos / AIO_EVENTS_PER_PAGE]); \	__event += pos % AIO_EVENTS_PER_PAGE;				\	__event;							\}) #define put_aio_ring_event(event) do {		\	struct io_event *__event = (event);	\	(void)__event;				\	kunmap_atomic((void *)((unsigned long)__event & PAGE_MASK)); \} while(0) static void ctx_rcu_free(struct rcu_head *head){	struct kioctx *ctx = container_of(head, struct kioctx, rcu_head);	kmem_cache_free(kioctx_cachep, ctx);} /* __put_ioctx *	Called when the last user of an aio context has gone away, *	and the struct needs to be freed. */static void __put_ioctx(struct kioctx *ctx){	unsigned nr_events = ctx->max_reqs;	BUG_ON(ctx->reqs_active); 	cancel_delayed_work_sync(&ctx->wq);	aio_free_ring(ctx);	mmdrop(ctx->mm);	ctx->mm = NULL;	if (nr_events) {		spin_lock(&aio_nr_lock);		BUG_ON(aio_nr - nr_events > aio_nr);		aio_nr -= nr_events;		spin_unlock(&aio_nr_lock);	}	pr_debug("__put_ioctx: freeing %p\n", ctx);	call_rcu(&ctx->rcu_head, ctx_rcu_free);} static inline int try_get_ioctx(struct kioctx *kioctx){	return atomic_inc_not_zero(&kioctx->users);} static inline void put_ioctx(struct kioctx *kioctx){	BUG_ON(atomic_read(&kioctx->users) <= 0);	if (unlikely(atomic_dec_and_test(&kioctx->users)))		__put_ioctx(kioctx);} /* ioctx_alloc *	Allocates and initializes an ioctx.  Returns an ERR_PTR if it failed. */static struct kioctx *ioctx_alloc(unsigned nr_events){	struct mm_struct *mm;	struct kioctx *ctx;	int err = -ENOMEM; 	/* Prevent overflows */	if ((nr_events > (0x10000000U / sizeof(struct io_event))) ||	    (nr_events > (0x10000000U / sizeof(struct kiocb)))) {		pr_debug("ENOMEM: nr_events too high\n");		return ERR_PTR(-EINVAL);	} 	if (!nr_events || (unsigned long)nr_events > aio_max_nr)		return ERR_PTR(-EAGAIN); 	ctx = kmem_cache_zalloc(kioctx_cachep, GFP_KERNEL);	if (!ctx)		return ERR_PTR(-ENOMEM); 	ctx->max_reqs = nr_events;	mm = ctx->mm = current->mm;	atomic_inc(&mm->mm_count); 	atomic_set(&ctx->users, 2);	spin_lock_init(&ctx->ctx_lock);	spin_lock_init(&ctx->ring_info.ring_lock);	init_waitqueue_head(&ctx->wait); 	INIT_LIST_HEAD(&ctx->active_reqs);	INIT_LIST_HEAD(&ctx->run_list);	INIT_DELAYED_WORK(&ctx->wq, aio_kick_handler); 	if (aio_setup_ring(ctx) < 0)		goto out_freectx; 	/* limit the number of system wide aios */	spin_lock(&aio_nr_lock);	if (aio_nr + nr_events > aio_max_nr ||	    aio_nr + nr_events < aio_nr) {		spin_unlock(&aio_nr_lock);		goto out_cleanup;	}	aio_nr += ctx->max_reqs;	spin_unlock(&aio_nr_lock); 	/* now link into global list. */	spin_lock(&mm->ioctx_lock);	hlist_add_head_rcu(&ctx->list, &mm->ioctx_list);	spin_unlock(&mm->ioctx_lock); 	dprintk("aio: allocated ioctx %p[%ld]: mm=%p mask=0x%x\n",		ctx, ctx->user_id, current->mm, ctx->ring_info.nr);	return ctx; out_cleanup:	err = -EAGAIN;	aio_free_ring(ctx);out_freectx:	mmdrop(mm);	kmem_cache_free(kioctx_cachep, ctx);	dprintk("aio: error allocating ioctx %d\n", err);	return ERR_PTR(err);} /* kill_ctx *	Cancels all outstanding aio requests on an aio context.  Used  *	when the processes owning a context have all exited to encourage  *	the rapid destruction of the kioctx. */static void kill_ctx(struct kioctx *ctx){	int (*cancel)(struct kiocb *, struct io_event *);	struct task_struct *tsk = current;	DECLARE_WAITQUEUE(wait, tsk);	struct io_event res; 	spin_lock_irq(&ctx->ctx_lock);	ctx->dead = 1;	while (!list_empty(&ctx->active_reqs)) {		struct list_head *pos = ctx->active_reqs.next;		struct kiocb *iocb = list_kiocb(pos);		list_del_init(&iocb->ki_list);		cancel = iocb->ki_cancel;		kiocbSetCancelled(iocb);		if (cancel) {			iocb->ki_users++;			spin_unlock_irq(&ctx->ctx_lock);			cancel(iocb, &res);			spin_lock_irq(&ctx->ctx_lock);		}	} 	if (!ctx->reqs_active)		goto out; 	add_wait_queue(&ctx->wait, &wait);	set_task_state(tsk, TASK_UNINTERRUPTIBLE);	while (ctx->reqs_active) {		spin_unlock_irq(&ctx->ctx_lock);		io_schedule();		set_task_state(tsk, TASK_UNINTERRUPTIBLE);		spin_lock_irq(&ctx->ctx_lock);	}	__set_task_state(tsk, TASK_RUNNING);	remove_wait_queue(&ctx->wait, &wait); out:	spin_unlock_irq(&ctx->ctx_lock);} /* wait_on_sync_kiocb: *	Waits on the given sync kiocb to complete. */ssize_t wait_on_sync_kiocb(struct kiocb *iocb){	while (iocb->ki_users) {		set_current_state(TASK_UNINTERRUPTIBLE);		if (!iocb->ki_users)			break;		io_schedule();	}	__set_current_state(TASK_RUNNING);	return iocb->ki_user_data;}EXPORT_SYMBOL(wait_on_sync_kiocb); /* exit_aio: called when the last user of mm goes away.  At this point,  * there is no way for any new requests to be submited or any of the  * io_* syscalls to be called on the context.  However, there may be  * outstanding requests which hold references to the context; as they  * go away, they will call put_ioctx and release any pinned memory * associated with the request (held via struct page * references). */void exit_aio(struct mm_struct *mm){	struct kioctx *ctx; 	while (!hlist_empty(&mm->ioctx_list)) {		ctx = hlist_entry(mm->ioctx_list.first, struct kioctx, list);		hlist_del_rcu(&ctx->list); 		kill_ctx(ctx); 		if (1 != atomic_read(&ctx->users))			printk(KERN_DEBUG				"exit_aio:ioctx still alive: %d %d %d\n",				atomic_read(&ctx->users), ctx->dead,				ctx->reqs_active);		/*		 * We don't need to bother with munmap() here -		 * exit_mmap(mm) is coming and it'll unmap everything.		 * Since aio_free_ring() uses non-zero ->mmap_size		 * as indicator that it needs to unmap the area,		 * just set it to 0; aio_free_ring() is the only		 * place that uses ->mmap_size, so it's safe.		 * That way we get all munmap done to current->mm -		 * all other callers have ctx->mm == current->mm.		 */		ctx->ring_info.mmap_size = 0;		put_ioctx(ctx);	}} /* aio_get_req *	Allocate a slot for an aio request.  Increments the users count * of the kioctx so that the kioctx stays around until all requests are * complete.  Returns NULL if no requests are free. * * Returns with kiocb->users set to 2.  The io submit code path holds * an extra reference while submitting the i/o. * This prevents races between the aio code path referencing the * req (after submitting it) and aio_complete() freeing the req. */static struct kiocb *__aio_get_req(struct kioctx *ctx){	struct kiocb *req = NULL; 	req = kmem_cache_alloc(kiocb_cachep, GFP_KERNEL);	if (unlikely(!req))		return NULL; 	req->ki_flags = 0;	req->ki_users = 2;	req->ki_key = 0;	req->ki_ctx = ctx;	req->ki_cancel = NULL;	req->ki_retry = NULL;	req->ki_dtor = NULL;	req->private = NULL;	req->ki_iovec = NULL;	INIT_LIST_HEAD(&req->ki_run_list);	req->ki_eventfd = NULL; 	return req;} /* * struct kiocb's are allocated in batches to reduce the number of * times the ctx lock is acquired and released. */#define KIOCB_BATCH_SIZE	32Lstruct kiocb_batch {	struct list_head head;	long count; /* number of requests left to allocate */}; static void kiocb_batch_init(struct kiocb_batch *batch, long total){	INIT_LIST_HEAD(&batch->head);	batch->count = total;} static void kiocb_batch_free(struct kioctx *ctx, struct kiocb_batch *batch){	struct kiocb *req, *n; 	if (list_empty(&batch->head))		return; 	spin_lock_irq(&ctx->ctx_lock);	list_for_each_entry_safe(req, n, &batch->head, ki_batch) {		list_del(&req->ki_batch);		list_del(&req->ki_list);		kmem_cache_free(kiocb_cachep, req);		ctx->reqs_active--;	}	if (unlikely(!ctx->reqs_active && ctx->dead))		wake_up_all(&ctx->wait);	spin_unlock_irq(&ctx->ctx_lock);} /* * Allocate a batch of kiocbs.  This avoids taking and dropping the * context lock a lot during setup. */static int kiocb_batch_refill(struct kioctx *ctx, struct kiocb_batch *batch){	unsigned short allocated, to_alloc;	long avail;	bool called_fput = false;	struct kiocb *req, *n;	struct aio_ring *ring; 	to_alloc = min(batch->count, KIOCB_BATCH_SIZE);	for (allocated = 0; allocated < to_alloc; allocated++) {		req = __aio_get_req(ctx);		if (!req)			/* allocation failed, go with what we've got */			break;		list_add(&req->ki_batch, &batch->head);	} 	if (allocated == 0)		goto out; retry:	spin_lock_irq(&ctx->ctx_lock);	ring = kmap_atomic(ctx->ring_info.ring_pages[0]);

Lines 1–500 of 1861. Load more