/* * QEMU posix-aio emulation * * Copyright IBM, Corp. 2008 * * Authors: * Anthony Liguori <aliguori@us.ibm.com> * * This work is licensed under the terms of the GNU GPL, version 2. See * the COPYING file in the top-level directory. * */ #include <sys/ioctl.h> #include <pthread.h> #include <unistd.h> #include <errno.h> #include <time.h> #include <string.h> #include <stdlib.h> #include <stdio.h> #include "osdep.h" #include "qemu-common.h" #include "posix-aio-compat.h" static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t cond = PTHREAD_COND_INITIALIZER; static pthread_t thread_id; static pthread_attr_t attr; static int max_threads = 64; static int cur_threads = 0; static int idle_threads = 0; static TAILQ_HEAD(, qemu_paiocb) request_list; #ifdef HAVE_PREADV static int preadv_present = 1; #else static int preadv_present = 0; #endif static void die2(int err, const char *what) { fprintf(stderr, "%s failed: %s\n", what, strerror(err)); abort(); } static void die(const char *what) { die2(errno, what); } static void mutex_lock(pthread_mutex_t *mutex) { int ret = pthread_mutex_lock(mutex); if (ret) die2(ret, "pthread_mutex_lock"); } static void mutex_unlock(pthread_mutex_t *mutex) { int ret = pthread_mutex_unlock(mutex); if (ret) die2(ret, "pthread_mutex_unlock"); } static int cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, struct timespec *ts) { int ret = pthread_cond_timedwait(cond, mutex, ts); if (ret && ret != ETIMEDOUT) die2(ret, "pthread_cond_timedwait"); return ret; } static void cond_signal(pthread_cond_t *cond) { int ret = pthread_cond_signal(cond); if (ret) die2(ret, "pthread_cond_signal"); } static void thread_create(pthread_t *thread, pthread_attr_t *attr, void *(*start_routine)(void*), void *arg) { int ret = pthread_create(thread, attr, start_routine, arg); if (ret) die2(ret, "pthread_create"); } static size_t handle_aiocb_ioctl(struct qemu_paiocb *aiocb) { int ret; ret = ioctl(aiocb->aio_fildes, aiocb->aio_ioctl_cmd, aiocb->aio_ioctl_buf); if (ret == -1) return -errno; /* * This looks weird, but the aio code only consideres a request * successfull if it has written the number full number of bytes. * * Now we overload aio_nbytes as aio_ioctl_cmd for the ioctl command, * so in fact we return the ioctl command here to make posix_aio_read() * happy.. */ return aiocb->aio_nbytes; } #ifdef HAVE_PREADV static ssize_t qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset) { return preadv(fd, iov, nr_iov, offset); } static ssize_t qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset) { return pwritev(fd, iov, nr_iov, offset); } #else static ssize_t qemu_preadv(int fd, const struct iovec *iov, int nr_iov, off_t offset) { return -ENOSYS; } static ssize_t qemu_pwritev(int fd, const struct iovec *iov, int nr_iov, off_t offset) { return -ENOSYS; } #endif /* * Check if we need to copy the data in the aiocb into a new * properly aligned buffer. */ static int aiocb_needs_copy(struct qemu_paiocb *aiocb) { if (aiocb->aio_flags & QEMU_AIO_SECTOR_ALIGNED) { int i; for (i = 0; i < aiocb->aio_niov; i++) if ((uintptr_t) aiocb->aio_iov[i].iov_base % 512) return 1; } return 0; } static size_t handle_aiocb_rw_vector(struct qemu_paiocb *aiocb) { size_t offset = 0; ssize_t len; do { if (aiocb->aio_type == QEMU_PAIO_WRITE) len = qemu_pwritev(aiocb->aio_fildes, aiocb->aio_iov, aiocb->aio_niov, aiocb->aio_offset + offset); else len = qemu_preadv(aiocb->aio_fildes, aiocb->aio_iov, aiocb->aio_niov, aiocb->aio_offset + offset); } while (len == -1 && errno == EINTR); if (len == -1) return -errno; return len; } static size_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf) { size_t offset = 0; size_t len; while (offset < aiocb->aio_nbytes) { if (aiocb->aio_type == QEMU_PAIO_WRITE) len = pwrite(aiocb->aio_fildes, (const char *)buf + offset, aiocb->aio_nbytes - offset, aiocb->aio_offset + offset); else len = pread(aiocb->aio_fildes, buf + offset, aiocb->aio_nbytes - offset, aiocb->aio_offset + offset); if (len == -1 && errno == EINTR) continue; else if (len == -1) { offset = -errno; break; } else if (len == 0) break; offset += len; } return offset; } static size_t handle_aiocb_rw(struct qemu_paiocb *aiocb) { size_t nbytes; char *buf; if (!aiocb_needs_copy(aiocb)) { /* * If there is just a single buffer, and it is properly aligned * we can just use plain pread/pwrite without any problems. */ if (aiocb->aio_niov == 1) return handle_aiocb_rw_linear(aiocb, aiocb->aio_iov->iov_base); /* * We have more than one iovec, and all are properly aligned. * * Try preadv/pwritev first and fall back to linearizing the * buffer if it's not supported. */ if (preadv_present) { nbytes = handle_aiocb_rw_vector(aiocb); if (nbytes == aiocb->aio_nbytes) return nbytes; if (nbytes < 0 && nbytes != -ENOSYS) return nbytes; preadv_present = 0; } /* * XXX(hch): short read/write. no easy way to handle the reminder * using these interfaces. For now retry using plain * pread/pwrite? */ } /* * Ok, we have to do it the hard way, copy all segments into * a single aligned buffer. */ buf = qemu_memalign(512, aiocb->aio_nbytes); if (aiocb->aio_type == QEMU_PAIO_WRITE) { char *p = buf; int i; for (i = 0; i < aiocb->aio_niov; ++i) { memcpy(p, aiocb->aio_iov[i].iov_base, aiocb->aio_iov[i].iov_len); p += aiocb->aio_iov[i].iov_len; } } nbytes = handle_aiocb_rw_linear(aiocb, buf); if (aiocb->aio_type != QEMU_PAIO_WRITE) { char *p = buf; size_t count = aiocb->aio_nbytes, copy; int i; for (i = 0; i < aiocb->aio_niov && count; ++i) { copy = count; if (copy > aiocb->aio_iov[i].iov_len) copy = aiocb->aio_iov[i].iov_len; memcpy(aiocb->aio_iov[i].iov_base, p, copy); p += copy; count -= copy; } } qemu_vfree(buf); return nbytes; } static void *aio_thread(void *unused) { pid_t pid; sigset_t set; pid = getpid(); /* block all signals */ if (sigfillset(&set)) die("sigfillset"); if (sigprocmask(SIG_BLOCK, &set, NULL)) die("sigprocmask"); while (1) { struct qemu_paiocb *aiocb; size_t ret = 0; qemu_timeval tv; struct timespec ts; qemu_gettimeofday(&tv); ts.tv_sec = tv.tv_sec + 10; ts.tv_nsec = 0; mutex_lock(&lock); while (TAILQ_EMPTY(&request_list) && !(ret == ETIMEDOUT)) { ret = cond_timedwait(&cond, &lock, &ts); } if (TAILQ_EMPTY(&request_list)) break; aiocb = TAILQ_FIRST(&request_list); TAILQ_REMOVE(&request_list, aiocb, node); aiocb->active = 1; idle_threads--; mutex_unlock(&lock); switch (aiocb->aio_type) { case QEMU_PAIO_READ: case QEMU_PAIO_WRITE: ret = handle_aiocb_rw(aiocb); break; case QEMU_PAIO_IOCTL: ret = handle_aiocb_ioctl(aiocb); break; default: fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type); ret = -EINVAL; break; } mutex_lock(&lock); aiocb->ret = ret; idle_threads++; mutex_unlock(&lock); if (kill(pid, aiocb->ev_signo)) die("kill failed"); } idle_threads--; cur_threads--; mutex_unlock(&lock); return NULL; } static void spawn_thread(void) { cur_threads++; idle_threads++; thread_create(&thread_id, &attr, aio_thread, NULL); } int qemu_paio_init(struct qemu_paioinit *aioinit) { int ret; ret = pthread_attr_init(&attr); if (ret) die2(ret, "pthread_attr_init"); ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); if (ret) die2(ret, "pthread_attr_setdetachstate"); TAILQ_INIT(&request_list); return 0; } static int qemu_paio_submit(struct qemu_paiocb *aiocb, int type) { aiocb->aio_type = type; aiocb->ret = -EINPROGRESS; aiocb->active = 0; mutex_lock(&lock); if (idle_threads == 0 && cur_threads < max_threads) spawn_thread(); TAILQ_INSERT_TAIL(&request_list, aiocb, node); mutex_unlock(&lock); cond_signal(&cond); return 0; } int qemu_paio_read(struct qemu_paiocb *aiocb) { return qemu_paio_submit(aiocb, QEMU_PAIO_READ); } int qemu_paio_write(struct qemu_paiocb *aiocb) { return qemu_paio_submit(aiocb, QEMU_PAIO_WRITE); } int qemu_paio_ioctl(struct qemu_paiocb *aiocb) { return qemu_paio_submit(aiocb, QEMU_PAIO_IOCTL); } ssize_t qemu_paio_return(struct qemu_paiocb *aiocb) { ssize_t ret; mutex_lock(&lock); ret = aiocb->ret; mutex_unlock(&lock); return ret; } int qemu_paio_error(struct qemu_paiocb *aiocb) { ssize_t ret = qemu_paio_return(aiocb); if (ret < 0) ret = -ret; else ret = 0; return ret; } int qemu_paio_cancel(int fd, struct qemu_paiocb *aiocb) { int ret; mutex_lock(&lock); if (!aiocb->active) { TAILQ_REMOVE(&request_list, aiocb, node); aiocb->ret = -ECANCELED; ret = QEMU_PAIO_CANCELED; } else if (aiocb->ret == -EINPROGRESS) ret = QEMU_PAIO_NOTCANCELED; else ret = QEMU_PAIO_ALLDONE; mutex_unlock(&lock); return ret; }