From cb42a1b1461e02efb034582ac5d8f71534723b92 Mon Sep 17 00:00:00 2001 From: David 'Digit' Turner Date: Thu, 23 Dec 2010 02:54:08 +0100 Subject: upstream: integrate block changes This large patch upgrades the block support code to the upstream version available in ba5e7f82169f32ab8163c707d97c799ca09f8924 dated 2010-08-08 Change-Id: I8b24df0c287e72f6620650a4d6a62e1bb315453e --- posix-aio-compat.c | 657 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 657 insertions(+) create mode 100644 posix-aio-compat.c (limited to 'posix-aio-compat.c') diff --git a/posix-aio-compat.c b/posix-aio-compat.c new file mode 100644 index 0000000..a67ffe3 --- /dev/null +++ b/posix-aio-compat.c @@ -0,0 +1,657 @@ +/* + * QEMU posix-aio emulation + * + * Copyright IBM, Corp. 2008 + * + * Authors: + * Anthony Liguori + * + * This work is licensed under the terms of the GNU GPL, version 2. See + * the COPYING file in the top-level directory. + * + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "qemu-queue.h" +#include "osdep.h" +#include "qemu-common.h" +#include "block_int.h" + +#include "block/raw-posix-aio.h" + + +struct qemu_paiocb { + BlockDriverAIOCB common; + int aio_fildes; + union { + struct iovec *aio_iov; + void *aio_ioctl_buf; + }; + int aio_niov; + size_t aio_nbytes; +#define aio_ioctl_cmd aio_nbytes /* for QEMU_AIO_IOCTL */ + int ev_signo; + off_t aio_offset; + + QTAILQ_ENTRY(qemu_paiocb) node; + int aio_type; + ssize_t ret; + int active; + struct qemu_paiocb *next; + + int async_context_id; +}; + +typedef struct PosixAioState { + int rfd, wfd; + struct qemu_paiocb *first_aio; +} PosixAioState; + + +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 QTAILQ_HEAD(, qemu_paiocb) request_list; + +#ifdef CONFIG_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 ssize_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; +} + +static ssize_t handle_aiocb_flush(struct qemu_paiocb *aiocb) +{ + int ret; + + ret = qemu_fdatasync(aiocb->aio_fildes); + if (ret == -1) + return -errno; + return 0; +} + +#ifdef CONFIG_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 + +static ssize_t handle_aiocb_rw_vector(struct qemu_paiocb *aiocb) +{ + size_t offset = 0; + ssize_t len; + + do { + if (aiocb->aio_type & QEMU_AIO_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 ssize_t handle_aiocb_rw_linear(struct qemu_paiocb *aiocb, char *buf) +{ + ssize_t offset = 0; + ssize_t len; + + while (offset < aiocb->aio_nbytes) { + if (aiocb->aio_type & QEMU_AIO_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 ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb) +{ + ssize_t nbytes; + char *buf; + + if (!(aiocb->aio_type & QEMU_AIO_MISALIGNED)) { + /* + * 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_AIO_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_AIO_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; + + pid = getpid(); + + while (1) { + struct qemu_paiocb *aiocb; + ssize_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 (QTAILQ_EMPTY(&request_list) && + !(ret == ETIMEDOUT)) { + ret = cond_timedwait(&cond, &lock, &ts); + } + + if (QTAILQ_EMPTY(&request_list)) + break; + + aiocb = QTAILQ_FIRST(&request_list); + QTAILQ_REMOVE(&request_list, aiocb, node); + aiocb->active = 1; + idle_threads--; + mutex_unlock(&lock); + + switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) { + case QEMU_AIO_READ: + case QEMU_AIO_WRITE: + ret = handle_aiocb_rw(aiocb); + break; + case QEMU_AIO_FLUSH: + ret = handle_aiocb_flush(aiocb); + break; + case QEMU_AIO_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) +{ + sigset_t set, oldset; + + cur_threads++; + idle_threads++; + + /* block all signals */ + if (sigfillset(&set)) die("sigfillset"); + if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask"); + + thread_create(&thread_id, &attr, aio_thread, NULL); + + if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore"); +} + +static void qemu_paio_submit(struct qemu_paiocb *aiocb) +{ + aiocb->ret = -EINPROGRESS; + aiocb->active = 0; + mutex_lock(&lock); + if (idle_threads == 0 && cur_threads < max_threads) + spawn_thread(); + QTAILQ_INSERT_TAIL(&request_list, aiocb, node); + mutex_unlock(&lock); + cond_signal(&cond); +} + +static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb) +{ + ssize_t ret; + + mutex_lock(&lock); + ret = aiocb->ret; + mutex_unlock(&lock); + + return ret; +} + +static 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; +} + +static int posix_aio_process_queue(void *opaque) +{ + PosixAioState *s = opaque; + struct qemu_paiocb *acb, **pacb; + int ret; + int result = 0; + int async_context_id = get_async_context_id(); + + for(;;) { + pacb = &s->first_aio; + for(;;) { + acb = *pacb; + if (!acb) + return result; + + /* we're only interested in requests in the right context */ + if (acb->async_context_id != async_context_id) { + pacb = &acb->next; + continue; + } + + ret = qemu_paio_error(acb); + if (ret == ECANCELED) { + /* remove the request */ + *pacb = acb->next; + qemu_aio_release(acb); + result = 1; + } else if (ret != EINPROGRESS) { + /* end of aio */ + if (ret == 0) { + ret = qemu_paio_return(acb); + if (ret == acb->aio_nbytes) + ret = 0; + else + ret = -EINVAL; + } else { + ret = -ret; + } + /* remove the request */ + *pacb = acb->next; + /* call the callback */ + acb->common.cb(acb->common.opaque, ret); + qemu_aio_release(acb); + result = 1; + break; + } else { + pacb = &acb->next; + } + } + } + + return result; +} + +static void posix_aio_read(void *opaque) +{ + PosixAioState *s = opaque; + ssize_t len; + + /* read all bytes from signal pipe */ + for (;;) { + char bytes[16]; + + len = read(s->rfd, bytes, sizeof(bytes)); + if (len == -1 && errno == EINTR) + continue; /* try again */ + if (len == sizeof(bytes)) + continue; /* more to read */ + break; + } + + posix_aio_process_queue(s); +} + +static int posix_aio_flush(void *opaque) +{ + PosixAioState *s = opaque; + return !!s->first_aio; +} + +static PosixAioState *posix_aio_state; + +static void aio_signal_handler(int signum) +{ + if (posix_aio_state) { + char byte = 0; + ssize_t ret; + + ret = write(posix_aio_state->wfd, &byte, sizeof(byte)); + if (ret < 0 && errno != EAGAIN) + die("write()"); + } + + qemu_service_io(); +} + +static void paio_remove(struct qemu_paiocb *acb) +{ + struct qemu_paiocb **pacb; + + /* remove the callback from the queue */ + pacb = &posix_aio_state->first_aio; + for(;;) { + if (*pacb == NULL) { + fprintf(stderr, "paio_remove: aio request not found!\n"); + break; + } else if (*pacb == acb) { + *pacb = acb->next; + qemu_aio_release(acb); + break; + } + pacb = &(*pacb)->next; + } +} + +static void paio_cancel(BlockDriverAIOCB *blockacb) +{ + struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb; + int active = 0; + + mutex_lock(&lock); + if (!acb->active) { + QTAILQ_REMOVE(&request_list, acb, node); + acb->ret = -ECANCELED; + } else if (acb->ret == -EINPROGRESS) { + active = 1; + } + mutex_unlock(&lock); + + if (active) { + /* fail safe: if the aio could not be canceled, we wait for + it */ + while (qemu_paio_error(acb) == EINPROGRESS) + ; + } + + paio_remove(acb); +} + +static AIOPool raw_aio_pool = { + .aiocb_size = sizeof(struct qemu_paiocb), + .cancel = paio_cancel, +}; + +BlockDriverAIOCB *paio_submit(BlockDriverState *bs, int fd, + int64_t sector_num, QEMUIOVector *qiov, int nb_sectors, + BlockDriverCompletionFunc *cb, void *opaque, int type) +{ + struct qemu_paiocb *acb; + + acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque); + if (!acb) + return NULL; + acb->aio_type = type; + acb->aio_fildes = fd; + acb->ev_signo = SIGUSR2; + acb->async_context_id = get_async_context_id(); + + if (qiov) { + acb->aio_iov = qiov->iov; + acb->aio_niov = qiov->niov; + } + acb->aio_nbytes = nb_sectors * 512; + acb->aio_offset = sector_num * 512; + + acb->next = posix_aio_state->first_aio; + posix_aio_state->first_aio = acb; + + qemu_paio_submit(acb); + return &acb->common; +} + +BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd, + unsigned long int req, void *buf, + BlockDriverCompletionFunc *cb, void *opaque) +{ + struct qemu_paiocb *acb; + + acb = qemu_aio_get(&raw_aio_pool, bs, cb, opaque); + if (!acb) + return NULL; + acb->aio_type = QEMU_AIO_IOCTL; + acb->aio_fildes = fd; + acb->ev_signo = SIGUSR2; + acb->aio_offset = 0; + acb->aio_ioctl_buf = buf; + acb->aio_ioctl_cmd = req; + + acb->next = posix_aio_state->first_aio; + posix_aio_state->first_aio = acb; + + qemu_paio_submit(acb); + return &acb->common; +} + +int paio_init(void) +{ + struct sigaction act; + PosixAioState *s; + int fds[2]; + int ret; + + if (posix_aio_state) + return 0; + + s = qemu_malloc(sizeof(PosixAioState)); + + sigfillset(&act.sa_mask); + act.sa_flags = 0; /* do not restart syscalls to interrupt select() */ + act.sa_handler = aio_signal_handler; + sigaction(SIGUSR2, &act, NULL); + + s->first_aio = NULL; + if (qemu_pipe(fds) == -1) { + fprintf(stderr, "failed to create pipe\n"); + return -1; + } + + s->rfd = fds[0]; + s->wfd = fds[1]; + + fcntl(s->rfd, F_SETFL, O_NONBLOCK); + fcntl(s->wfd, F_SETFL, O_NONBLOCK); + + qemu_aio_set_fd_handler(s->rfd, posix_aio_read, NULL, posix_aio_flush, + posix_aio_process_queue, s); + + 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"); + + QTAILQ_INIT(&request_list); + + posix_aio_state = s; + return 0; +} -- cgit v1.1